jsstats/server/joystream/index.ts

import {
  Account,
  Balance,
  Block,
  Category,
  Channel,
  Council,
  Era,
  Event,
  Member,
  Post,
  Proposal,
  Thread,
} from '../db/models'
const models: { [key: string]: any } = {
  channel: Channel,
  proposal: Proposal,
  category: Category,
  thread: Thread,
  post: Post,
  block: Block,
  council: Council,
  member: Member,
  era: Era,
}

import * as get from './lib/getters'
import axios from 'axios'
import moment from 'moment'

import { VoteKind } from '@joystream/types/proposals'
import { EventRecord } from '@polkadot/types/interfaces'
import { AccountInfo } from '@polkadot/types/interfaces/system'
import {
  Api,
  Handles,
  IState,
  MemberType,
  CategoryType,
  ChannelType,
  PostType,
  Seat,
  ThreadType,
  CouncilType,
  ProposalDetail,
  Status,
} from '../types'

import { AccountId, Moment, ActiveEraInfo } from '@polkadot/types/interfaces'
import Option from '@polkadot/types/codec/Option'
import { Vec } from '@polkadot/types'

// queuing
let lastUpdate = 0
const queue: any[] = []
let inProgress = false
const enqueue = (fn: any) => {
  queue.push(fn)
  processNext()
}
const processNext = async () => {
  if (inProgress) return
  inProgress = true
  const task = queue.pop()
  if (task) await task()

  inProgress = false
  //processNext()
  //return queue.length
}

const save = async (model: any, data: any) => {
  const Model = models[model]
  try {
    const exists = await Model.findByPk(data.id)
    if (exists) return exists.update(data)
  } catch (e) {}
  //console.debug(`saving ${data.id}`, `queued tasks: ${queue.length}`)
  try {
    return Model.create(data)
  } catch (e) {
    console.warn(`Failed to save ${Model}`, e.message)
  }
}

const addBlock = async (
  api: Api,
  io: any,
  header: { number: number; author: string },
  status: Status
): Promise<Status> => {
  const id = +header.number
  const last = await Block.findByPk(id - 1)
  const timestamp = moment.utc(await api.query.timestamp.now()).valueOf()
  const blocktime = last ? timestamp - last.timestamp : 6000
  const block = await save('block', { id, timestamp, blocktime })
  io.emit('block', block)

  const author = header.author?.toString()
  const member = await fetchMemberByAccount(api, author)
  if (member && member.id) block.setAuthor(member.id)
  updateBalances(api, id)

  const currentEra = Number(await api.query.staking.currentEra())
  const era = await save('era', { id: currentEra })
  era.addBlock(block.id)

  const handle = member ? member.handle : author
  const queued = `(queued: ${queue.length})`
  console.log(`[Joystream] block ${block.id} ${handle} ${queued}`)

  await processEvents(api, id)
  return updateEra(api, io, status, currentEra)
}

const addBlockRange = async (
  api: Api,
  startBlock: number,
  endBlock: number
) => {
  const previousHash = await api.rpc.chain.getBlockHash(startBlock - 1)
  let previousEra = (await api.query.staking.activeEra.at(
    previousHash
  )) as Option<ActiveEraInfo>

  for (let i = startBlock; i < endBlock; i++) {
    const hash = await api.rpc.chain.getBlockHash(i)
    const blockEra = (await api.query.staking.activeEra.at(
      hash
    )) as Option<ActiveEraInfo>
    if (
      blockEra.unwrap().index.toNumber() ===
      previousEra.unwrap().index.toNumber()
    ) {
      continue
    }

    let totalValidators = (await api.query.staking.snapshotValidators.at(
      hash
    )) as Option<Vec<AccountId>>
    if (totalValidators.isEmpty) {
      continue
    }
    console.log(`found validators: ${totalValidators.unwrap().length}`)

    const totalNrValidators = totalValidators.unwrap().length
    const maxSlots = Number(
      (await api.query.staking.validatorCount.at(hash)).toString()
    )
    const actives = Math.min(maxSlots, totalNrValidators)
    const waiting =
      totalNrValidators > maxSlots ? totalNrValidators - maxSlots : 0

    const timestamp = (await api.query.timestamp.now.at(hash)) as Moment
    const date = new Date(timestamp.toNumber())
    await save('era', {
      id: blockEra.unwrap().index.toNumber(),
      waiting: waiting,
      actives: actives,
      maxSlots: maxSlots,
      timestamp: date,
    })

    previousEra = blockEra
  }
}

const getBlockHash = (api: Api, blockId: number) =>
  api.rpc.chain.getBlockHash(blockId)

const processEvents = async (api: Api, blockId: number) => {
  const blockHash = await getBlockHash(api, blockId)
  const blockEvents = await api.query.system.events.at(blockHash)
  blockEvents.forEach(({ event }: EventRecord) => {
    let { section, method, data } = event
    Event.create({ blockId, section, method, data: JSON.stringify(data) })
  })
}

const updateEra = async (api: Api, io: any, status: any, era: number) => {
  const now: number = moment().valueOf()
  if (lastUpdate + 60000 > now) return status
  //console.log(`updating status`, lastUpdate)
  lastUpdate = now

  // session.disabledValidators: Vec<u32>
  // check online: imOnline.keys
  // imOnline.authoredBlocks: 2
  // session.currentIndex: 17,081

  const lastReward = Number(await api.query.staking.erasValidatorReward(era))
  console.debug(`last reward`, era, lastReward)
  if (lastReward > 0) {
  } // TODO save lastReward

  const nominatorEntries = await api.query.staking.nominators.entries()
  const nominators = nominatorEntries.map((n: any) => String(n[0].toHuman()))

  const rewardPoints = await api.query.staking.erasRewardPoints(era)
  const validatorEntries = await api.query.session.validators()
  const validators = validatorEntries.map((v: any) => String(v))

  // TODO staking.bondedEras: Vec<(EraIndex,SessionIndex)>
  const stashes = await api.derive.staking.stashes()
  console.debug(`fetching stakes`)

  if (!stashes) return

  for (let validator of stashes) {
    try {
      const prefs = await api.query.staking.erasValidatorPrefs(era, validator)
      const commission = Number(prefs.commission) / 10000000

      const data = await api.query.staking.erasStakers(era, validator)
      let { total, own, others } = data.toJSON()
    } catch (e) {
      console.warn(`Failed to fetch stakes for ${validator} in era ${era}`, e)
    }
  }

  return {
    members: (await api.query.members.nextMemberId()) - 1,
    categories: await get.currentCategoryId(api),
    threads: await get.currentThreadId(api),
    proposals: await get.proposalCount(api),
    channels: await get.currentChannelId(api),
    posts: await get.currentPostId(api),
    proposalPosts: await api.query.proposalsDiscussion.postCount(),
    queued: queue.length,
    era,
  }
}

const validatorStatus = async (api: Api, blockId: number) => {
  const hash = await getBlockHash(api, blockId)
  let totalValidators = await api.query.staking.snapshotValidators.at(hash)
  if (totalValidators.isEmpty) return

  let totalNrValidators = totalValidators.unwrap().length
  const maxSlots = Number(await api.query.staking.validatorCount.at(hash))
  const actives = Math.min(maxSlots, totalNrValidators)
  const waiting =
    totalNrValidators > maxSlots ? totalNrValidators - maxSlots : 0
  let timestamp = await api.query.timestamp.now.at(hash)
  const date = moment(timestamp.toNumber()).valueOf()
  return { blockId, actives, waiting, maxSlots, date }
}

const getAccountAtBlock = (
  api: Api,
  hash: string,
  account: string
): Promise<AccountInfo> => api.query.system.account.at(hash, account)

const fetchAccounts = async (api: Api, blockId: number) => {
  api.query.system.account.entries().then((account: any) => {
    const address = account[0].toHuman()[0]
    Account.create({ address })
  })
}

const updateBalances = async (api: Api, blockId: number) => {
  const blockHash = await getBlockHash(api, blockId)
  const currentEra: number = await api.query.staking.currentEra.at(blockHash)
  const era = await Era.findOrCreate({ where: { id: currentEra } })

  Account.findAll().then(async (account: any) => {
    const { id, address } = account
    if (!address) return
    console.log(`updating balance of`, id, address)

    const { data } = await getAccountAtBlock(api, blockHash, address)
    const { free, reserved, miscFrozen, feeFrozen } = data
    const balance = { available: free, reserved, frozen: miscFrozen }
    Balance.create(balance).then((balance: any) => {
      balance.setAccount(id)
      balance.setEra(era.id)
      console.log(`balance`, era.id, address, balance.available)
    })
  })
}

const fetchTokenomics = async () => {
  console.debug(`Updating tokenomics`)
  const { data } = await axios.get('https://status.joystream.org/status')
  if (!data) return
  // TODO save 'tokenomics', data
}

const fetchChannel = async (api: Api, id: number) => {
  const exists = await Channel.findByPk(id)
  if (exists) return exists

  console.debug(`Fetching channel ${id}`)
  const data = await api.query.contentWorkingGroup.channelById(id)

  const handle = String(data.handle)
  const title = String(data.title)
  const description = String(data.description)
  const avatar = String(data.avatar)
  const banner = String(data.banner)
  const content = String(data.content)
  const ownerId = Number(data.owner)
  const accountId = String(data.role_account)
  const publicationStatus = data.publication_status === 'Public' ? true : false
  const curation = String(data.curation_status)
  const createdAt = +data.created
  const principal = Number(data.principal_id)

  const channel = {
    id,
    handle,
    title,
    description,
    avatar,
    banner,
    content,
    publicationStatus,
    curation,
    createdAt,
    principal,
  }
  const chan = await save('channel', channel)
  const owner = await fetchMember(api, ownerId)
  chan.setOwner(owner)
  if (id > 1) fetchChannel(api, id - 1)
  return chan
}

const fetchCategory = async (api: Api, id: number) => {
  const exists = await Category.findByPk(id)
  if (exists) return exists

  console.debug(`fetching category ${id}`)
  const data = await api.query.forum.categoryById(id)

  const threadId = +data.thread_id
  const title = String(data.title)
  const description = String(data.description)
  const createdAt = +data.created_at.block
  const deleted = data.deleted
  const archived = data.archived
  const subcategories = Number(data.num_direct_subcategories)
  const moderatedThreads = Number(data.num_direct_moderated_threads)
  const unmoderatedThreads = Number(data.num_direct_unmoderated_threads)
  const position = +data.position_in_parent_category // TODO sometimes NaN
  const moderator: string = String(data.moderator_id) // account

  const cat = {
    id,
    title,
    description,
    createdAt,
    deleted,
    archived,
    subcategories,
    moderatedThreads,
    unmoderatedThreads,
    //position,
  }
  const category = await save('category', cat)
  const mod = await fetchMemberByAccount(api, moderator)
  if (mod) category.setModerator(mod.id)
  if (id > 1) fetchCategory(api, id - 1)
  return category
}

const fetchPost = async (api: Api, id: number) => {
  const exists = await Post.findByPk(id)
  if (exists) return exists

  console.debug(`fetching post ${id}`)
  const data = await api.query.forum.postById(id)

  const threadId = Number(data.thread_id)
  const text = data.current_text
  const moderation = data.moderation
  //const history = data.text_change_history;
  const createdAt = data.created_at.block
  const author: string = String(data.author_id)

  const post = await save('post', { id, text, createdAt })
  const thread = await fetchThread(api, threadId)
  if (thread) post.setThread(thread.id)
  const member = await fetchMemberByAccount(api, author)
  if (member) post.setAuthor(member.id)
  const mod = await fetchMemberByAccount(api, moderation)
  if (id > 1) fetchPost(api, id - 1)
  return post
}

const fetchThread = async (api: Api, id: number) => {
  const exists = await Thread.findByPk(id)
  if (exists) return exists

  console.debug(`fetching thread ${id}`)
  const data = await api.query.forum.threadById(id)

  const title = String(data.title)
  const categoryId = Number(data.category_id)
  const nrInCategory = Number(data.nr_in_category)
  const moderation = data.moderation
  const createdAt = +data.created_at.block
  const account = String(data.author_id)

  const thread = await save('thread', { id, title, nrInCategory, createdAt })
  const category = await fetchCategory(api, categoryId)
  if (category) thread.setCategory(category.id)
  const author = await fetchMemberByAccount(api, account)
  if (author) thread.setAuthor(author.id)
  if (moderation) {
    /* TODO
  Error: Invalid value ModerationAction(3) [Map] {
[1]   'moderated_at' => BlockAndTime(2) [Map] {
[1]     'block' => <BN: 4f4ff>,
[1]     'time' => <BN: 17526e65a40>,
[1]     registry: TypeRegistry {},
[1]     block: [Getter],
[1]     time: [Getter],
[1]     typeDefs: { block: [Function: U32], time: [Function: U64] }
[1]   },
[1]   'moderator_id'
[1]   'rationale' => [String (Text): 'Irrelevant as posted in another thread.'] {
*/
    //const mod = await fetchMemberByAccount(api, moderation)
    //if (mod) thread.setModeration(mod.id)
  }
  if (id > 1) fetchThread(api, id - 1)
  return thread
}

const fetchCouncils = async (api: Api, lastBlock: number) => {
  const round = await api.query.councilElection.round()
  let councils: CouncilType[] = await Council.findAll()
  const cycle = 201600

  for (let round = 0; round < round; round++) {
    const block = 57601 + round * cycle
    if (councils.find((c) => c.round === round) || block > lastBlock) continue
    //enqueue(() => fetchCouncil(api, block))
  }
}

const fetchCouncil = async (api: Api, block: number) => {
  console.debug(`Fetching council at block ${block}`)
  const blockHash = await api.rpc.chain.getBlockHash(block)
  if (!blockHash)
    return console.error(`Error: empty blockHash fetchCouncil ${block}`)
  const council = await api.query.council.activeCouncil.at(blockHash)
  return save('council', council)
}

const fetchProposal = async (api: Api, id: number) => {
  const exists = await Proposal.findByPk(id)
  if (exists) return exists

  //if (exists && exists.stage === 'Finalized')
  //if (exists.votesByAccount && exists.votesByAccount.length) return
  //else return //TODO fetchVotesPerProposal(api, exists)

  console.debug(`Fetching proposal ${id}`)
  const proposal = await get.proposalDetail(api, id)
  if (id > 1) fetchProposal(api, id - 1)
  return save('proposal', proposal)
  //TODO fetchVotesPerProposal(api, proposal)
}

const fetchVotesPerProposal = async (api: Api, proposal: ProposalDetail) => {
  if (proposal.votesByAccount && proposal.votesByAccount.length) return

  const proposals = await Proposal.findAll()
  const councils = await Council.findAll()

  console.debug(`Fetching proposal votes (${proposal.id})`)
  let members: MemberType[] = []
  councils.map((seats: Seat[]) =>
    seats.forEach(async (seat: Seat) => {
      if (members.find((member) => member.account === seat.member)) return
      const member = await Member.findOne({ where: { account: seat.member } })
      member && members.push(member)
    })
  )

  const { id } = proposal
  const votesByAccount = await Promise.all(
    members.map(async (member) => {
      const vote = await fetchVoteByProposalByVoter(api, id, member.id)
      return { vote, handle: member.handle }
    })
  )
  Proposal.findByPk(id).then((p: any) => p.update({ votesByAccount }))
}

const fetchVoteByProposalByVoter = async (
  api: Api,
  proposalId: number,
  voterId: number
): Promise<string> => {
  console.debug(`Fetching vote by ${voterId} for proposal ${proposalId}`)
  const vote: VoteKind = await api.query.proposalsEngine.voteExistsByProposalByVoter(
    proposalId,
    voterId
  )
  const hasVoted: number = (
    await api.query.proposalsEngine.voteExistsByProposalByVoter.size(
      proposalId,
      voterId
    )
  ).toNumber()

  return hasVoted ? String(vote) : ''
}

// accounts
const fetchMemberByAccount = async (
  api: Api,
  account: string
): Promise<MemberType | undefined> => {
  const exists = await Member.findOne({ where: { account } })
  if (exists) return exists

  const id: number = Number(await get.memberIdByAccount(api, account))
  return id ? fetchMember(api, id) : undefined
}

const fetchMember = async (api: Api, id: number): Promise<MemberType> => {
  try {
    const exists = await Member.findByPk(id)
    if (exists) return exists
  } catch (e) {
    console.debug(`Fetching member ${id}`)
  }
  const membership = await get.membership(api, id)
  const handle = String(membership.handle)
  const account = String(membership.root_account)
  const about = String(membership.about)
  const createdAt = +membership.registered_at_block
  if (id > 1) fetchMember(api, id - 1)
  return save('member', { id, handle, createdAt, about })
}

const fetchReports = () => {
  const domain = `https://raw.githubusercontent.com/Joystream/community-repo/master/council-reports`
  const apiBase = `https://api.github.com/repos/joystream/community-repo/contents/council-reports`

  const urls: { [key: string]: string } = {
    alexandria: `${apiBase}/alexandria-testnet`,
    archive: `${apiBase}/archived-reports`,
    template: `${domain}/templates/council_report_template_v1.md`,
  }

  ;['alexandria', 'archive'].map((folder) => fetchGithubDir(urls[folder]))

  fetchGithubFile(urls.template)
}

const fetchGithubFile = async (url: string): Promise<string> => {
  const { data } = await axios.get(url)
  return data
}

const fetchGithubDir = async (url: string) => {
  const { data } = await axios.get(url)
  data.forEach(
    async (o: {
      name: string
      type: string
      url: string
      download_url: string
    }) => {
      const match = o.name.match(/^(.+)\.md$/)
      const name = match ? match[1] : o.name
      if (o.type === 'file') {
        const file = await fetchGithubFile(o.download_url)
        // TODO save file
      } else fetchGithubDir(o.url)
    }
  )
}

module.exports = { addBlock, addBlockRange }