jsstats/server/joystream/index.ts

import { Op } from 'sequelize'
import {
  Account,
  Balance,
  Block,
  Category,
  Channel,
  Council,
  Consul,
  ConsulStake,
  Era,
  Event,
  Member,
  Post,
  Proposal,
  ProposalVote,
  Thread,
} from '../db/models'

import * as get from './lib/getters'
//import {fetchReports} from './lib/github'
import axios from 'axios'
import moment from 'moment'
import chalk from 'chalk'

import { VoteKind } from '@joystream/types/proposals'
import { Seats } from '@joystream/types/council'
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,
  EventRecord,
} from '@polkadot/types/interfaces'
import Option from '@polkadot/types/codec/Option'
import { Vec } from '@polkadot/types'

// TODO fetch consts from db/chain
const TERMDURATION = 144000
const VOTINGDURATION = 57601
const CYCLE = VOTINGDURATION + TERMDURATION

const DELAY = 0 // ms
let lastUpdate = 0
let queuedAll = false
let processing = false
let queue: any[] = []
let fetching = ''

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

const getEraAtBlock = (api: Api, hash: string) =>
  api.query.staking.activeEra.at(hash)

const getTimestamp = async (api: Api, hash?: string) =>
  moment
    .utc(
      hash
        ? await api.query.timestamp.now.at(hash)
        : await api.query.timestamp.now()
    )
    .valueOf()

const addBlock = async (
  api: Api,
  io: any,
  header: { number: number; author: string },
  status: Status = {
    era: 0,
    round: 0,
    members: 0,
    channels: 0,
    categories: 0,
    threads: 0,
    posts: 0,
    proposals: 0,
    proposalPosts: 0,
  }
): Promise<Status> => {
  const id = +header.number
  const last = await Block.findByPk(id - 1)
  const exists = await Block.findByPk(id)
  if (exists) {
    console.error(`TODO handle fork`, String(header.author))
    return status
  }
  const timestamp = await getTimestamp(api)
  const blocktime = last ? timestamp - last.timestamp : 6000
  const address = header.author?.toString()
  const account = await Account.findOrCreate({ where: { address } })
  const block = await Block.create({ id, timestamp, blocktime })
  block.setValidator(account.id)

  const currentEra = Number(await api.query.staking.currentEra())
  const era = await Era.findOrCreate({ where: { id: currentEra } })
  await block.setEra(currentEra)
  io.emit('block', await Block.findByIdWithIncludes(block.id))

  // logging
  const member = await fetchMemberByAccount(api, address)
  const handle = member ? member.handle : address
  const f = fetching !== '' ? `, fetching ${fetching}` : ''
  const q = queue.length ? ` (${queue.length} queued${f})` : ''
  console.log(`[Joystream] block ${block.id} ${handle}${q}`)

  processEvents(api, id)
  //updateEra(api, io, status, currentEra)
  //updateBalances(api, id)
  return updateStatus(api, status, currentEra)
}

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

  for (let i = startBlock; i < endBlock; i++) {
    const hash = await getBlockHash(api, 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 Era.create({
      id: blockEra.unwrap().index.toNumber(),
      waiting: waiting,
      actives: actives,
      maxSlots: maxSlots,
      timestamp: date,
    })

    previousEra = blockEra
  }
}

const updateStatus = async (api: Api, old: Status, era: number) => {
  const status = {
    era,
    round: Number(await api.query.councilElection.round()),
    members: (await api.query.members.nextMemberId()) - 1,
    channels: await get.currentChannelId(api),

    categories: await get.currentCategoryId(api),
    threads: await get.currentThreadId(api),
    posts: await get.currentPostId(api),

    proposals: await get.proposalCount(api),
    proposalPosts: await api.query.proposalsDiscussion.postCount(),
  }
  if (!queuedAll) fetchAll(api, status)
  else {
    // TODO catch if more than one are added
    status.members > old.members && fetchMember(api, status.members)
    status.posts > old.posts && fetchPost(api, status.posts)
    status.proposals > old.proposals && fetchProposal(api, status.proposals)
    status.channels > old.channels && fetchChannel(api, status.channels)
    status.categories > old.categories && fetchCategory(api, status.categories)
    status.proposalPosts > old.proposalPosts &&
      fetchProposalPosts(api, status.proposalPosts)
  }
  return status
}

const fetchAll = async (api: Api, status: Status) => {
  // trying to avoid SequelizeUniqueConstraintError

  for (let id = status.members; id > 0; id--) {
    queue.push(() => fetchMember(api, id))
  }
  for (let id = status.round; id > 0; id--) {
    queue.push(() => fetchCouncil(api, id))
  }

  for (let id = status.proposals; id > 0; id--) {
    queue.push(() => fetchProposal(api, id))
  }
  queue.push(() => fetchProposalPosts(api, status.proposalPosts))

  for (let id = status.channels; id > 0; id--) {
    queue.push(() => fetchChannel(api, id))
  }
  for (let id = status.categories; id > 0; id--) {
    queue.push(() => fetchCategory(api, id))
  }
  for (let id = status.threads; id > 0; id--) {
    queue.push(() => fetchThread(api, id))
  }
  for (let id = status.posts; id > 0; id--) {
    queue.push(() => fetchPost(api, id))
  }

  queuedAll = true
  processNext()
}

const processNext = async () => {
  if (processing) return
  processing = true
  const task = queue.shift()
  if (!task) return
  const result = await task()
  processing = false
  setTimeout(() => processNext(), DELAY)
}

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) })
  })
  // TODO catch votes, posts, proposals?
}

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()
  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)
    }
  }
}

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)

// TODO when to cal
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) => {
  if (id <= 0) return
  const exists = await Channel.findByPk(id)
  if (exists) return exists

  fetching = `channel ${id}`
  const data = await api.query.contentWorkingGroup.channelById(id)
  const { handle, title, description, avatar, banner, content, created } = data
  //const accountId = String(data.role_account)
  const channel = {
    id,
    handle: String(handle),
    title: String(title),
    description: String(description),
    avatar: String(avatar),
    banner: String(banner),
    content: String(content),
    publicationStatus: data.publication_status === 'Public' ? true : false,
    curation: String(data.curation_status),
    createdAt: +created,
    principal: Number(data.principal_id),
  }
  const chan = await Channel.create(channel)
  const owner = await fetchMember(api, data.owner)
  chan.setOwner(owner)
  return chan
}

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

  fetching = `category ${id}`
  const data = await api.query.forum.categoryById(id)
  const { title, description, deleted, archived } = data
  const threadId = +data.thread_id // TODO needed?
  const moderator: string = String(data.moderator_id) // account

  const cat = {
    id,
    title,
    description,
    createdAt: +data.created_at.block,
    deleted,
    archived,
    subcategories: Number(data.num_direct_subcategories),
    moderatedThreads: Number(data.num_direct_moderated_threads),
    unmoderatedThreads: Number(data.num_direct_unmoderated_threads),
    //position:+data.position_in_parent_category // TODO sometimes NaN,
  }
  const category = await Category.create(cat)
  const mod = await fetchMemberByAccount(api, moderator)
  if (mod) category.setModerator(mod.id)
  return category
}

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

  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 Post.create({ 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)
    member.addPost(post.id)
  }
  if (moderation) {
    const mod = await fetchMemberByAccount(api, moderation)
    post.setModerator(mod)
  }
  return post
}

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

  fetching = `thread ${id}`
  const data = await api.query.forum.threadById(id)
  const { title, moderation, nr_in_category } = data
  const account = String(data.author_id)
  const t = {
    id,
    title,
    nrInCategory: +nr_in_category,
    createdAt: +data.created_at.block,
  }
  const thread = await Thread.create(t)
  const category = await fetchCategory(api, +data.category_id)
  if (category) thread.setCategory(category.id)
  const author = await fetchMemberByAccount(api, account)
  if (author) thread.setCreator(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.'] {
*/
    //console.log(`thread mod`, moderation
    //const mod = await fetchMemberByAccount(api, moderation)
    //if (mod) thread.setModeration(mod.id)
  }
  return thread
}

const fetchCouncil = async (api: Api, round: number) => {
  if (round <= 0) return console.log(chalk.red(`[fetchCouncil] round:${round}`))

  const exists = await Council.findByPk(round)
  if (exists) return exists

  fetching = `council ${round}`
  const start = 57601 + (round - 1) * CYCLE
  const end = start + TERMDURATION
  let council = { round, start, end, startDate: 0, endDate: 0 }
  let seats: Seats
  try {
    const startHash = await getBlockHash(api, start)
    council.startDate = await getTimestamp(api, startHash)
    seats = await api.query.council.activeCouncil.at(startHash)
  } catch (e) {
    return console.log(`council term ${round} lies in the future ${start}`)
  }

  try {
    const endHash = await getBlockHash(api, end)
    council.endDate = await getTimestamp(api, endHash)
  } catch (e) {
    console.warn(`end of council term ${round} lies in the future ${end}`)
  }

  try {
    Council.create(council).then(({ round }: any) =>
      seats.map(({ member, stake, backers }) =>
        fetchMemberByAccount(api, member.toHuman()).then((m: any) =>
          Consul.create({
            stake: Number(stake),
            councilRound: round,
            memberId: m.id,
          }).then((consul: any) =>
            backers.map(async ({ member, stake }) =>
              fetchMemberByAccount(api, member.toHuman()).then(({ id }: any) =>
                ConsulStake.create({
                  stake: Number(stake),
                  consulId: consul.id,
                  memberId: id,
                })
              )
            )
          )
        )
      )
    )
  } catch (e) {
    console.error(`Failed to save council ${round}`, e)
  }
}

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

  fetching = `proposal ${id}`
  const proposal = await get.proposalDetail(api, id)
  fetchProposalVotes(api, proposal)
  return Proposal.create(proposal)
}

const fetchProposalPosts = async (api: Api, max: number) => {
  console.log(`posts`, max)
  let postId = 1
  for (let threadId = 1; postId <= max; threadId++) {
    fetching = `proposal posts ${threadId} ${postId}`
    const post = await api.query.proposalsDiscussion.postThreadIdByPostId(
      threadId,
      postId
    )
    if (post.text.length) {
      console.log(postId, threadId, post.text.toHuman())
      postId++
    }
  }
}

const findCouncilAtBlock = (api: Api, block: number) =>
  Council.findOne({
    where: {
      start: { [Op.lte]: block },
      end: { [Op.gte]: block - VOTINGDURATION },
    },
  })

const fetchProposalVotes = async (api: Api, proposal: ProposalDetail) => {
  if (!proposal) return console.error(`[fetchProposalVotes] empty proposal`)
  fetching = `votes proposal ${proposal.id}`
  const { createdAt } = proposal
  if (!createdAt) return console.error(`empty start block`, proposal)
  try {
    const start = await findCouncilAtBlock(api, createdAt)
    if (start) start.addProposal(proposal.id)
    else return console.error(`no council found for proposal ${proposal.id}`)
    // some proposals make it into a second term
    const end = await findCouncilAtBlock(api, proposal.finalizedAt)
    const councils = [start.round, end && end.round]
    const consuls = await Consul.findAll({
      where: { councilRound: { [Op.or]: councils } },
    })
    consuls.map(({ id, memberId }: any) =>
      fetchProposalVoteByConsul(api, proposal.id, id, memberId)
    )
  } catch (e) {
    console.log(`failed to fetch votes of proposal ${proposal.id}`, e)
  }
}

const fetchProposalVoteByConsul = async (
  api: Api,
  proposalId: number,
  consulId: number,
  memberId: number
): Promise<any> => {
  fetching = `vote by ${consulId} for proposal ${proposalId}`
  const exists = await ProposalVote.findOne({
    where: { proposalId, memberId, consulId },
  })
  if (exists) return exists

  const query = api.query.proposalsEngine
  const args = [proposalId, memberId]

  const hasVoted = await query.voteExistsByProposalByVoter.size(...args)
  if (!hasVoted.toNumber()) return

  const vote = (await query.voteExistsByProposalByVoter(...args)).toHuman()
  return ProposalVote.create({ vote: vote, proposalId, consulId, memberId })
}

// accounts
const fetchMemberByAccount = async (
  api: Api,
  account: string
): Promise<MemberType | undefined> => {
  if (!account) {
    console.error(`fetchMemberByAccount called without account`)
    return 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 | undefined> => {
  if (id <= 0) return
  const exists = await Member.findByPk(+id)
  if (exists) return exists

  fetching = `member ${id}`
  const membership = await get.membership(api, id)
  const about = String(membership.about)
  const account = String(membership.root_account)
  const handle = String(membership.handle)
  const createdAt = +membership.registered_at_block
  return Member.create({ id, about, account, createdAt, handle })
}

module.exports = { addBlock, addBlockRange }