feat(hydra-processor): string-based ranges in manifest (Joystream#347)

* refactor(hydra-processor): refactor event queue for processing

affects: @dzlzv/hydra-processor, sample

* refactor(hydra-processor): block-based processing of events

affects: @dzlzv/hydra-processor, sample, sample-mappings

events are processed in blocks, so that the transaction granularity always corresponds to the
block-level

* feat(hydra-processor): support block ranges for mappings

affects: @dzlzv/hydra-processor

* test(sample): add hooks to sample

affects: sample, sample-mappings

* test(hydra-processor): add validation and tests for handlers in manifest

* test(hydra-e2e-tests): add e2e-tests for ranges and block hooks

affects: hydra-e2e-tests

* feat(hydra-processor): support semiopen ranges

affects: hydra-e2e-tests, @dzlzv/hydra-processor

support mapping ranges defined as strings, e.g. (3,] or (3, 34]

packages/hydra-e2e-tests/fixtures/manifest.yml

@@ -23,9 +23,14 @@ mappings:
     - event: balances.Transfer 
       handler: balancesTransfer(DatabaseManager, Balances.TransferEvent)
   extrinsicHandlers:
-    # infer defaults
     - extrinsic: timestamp.set 
-      handler: timestampCall(DatabaseManager, Timestamp.SetCall)
+      # if no argument types were provided, a single 
+      # context argument is passed
+      handler: timestampCall
   preBlockHooks:
+    - handler: preHook
+      range: '[1, 2]'
   postBlockHooks:
+    - handler: postHook
+      range: '[2, 4)'
 

packages/hydra-e2e-tests/fixtures/mappings/index.ts

@@ -0,0 +1,4 @@
+// Export here all the event handler functions
+// so that the indexer picks them up
+// export { balancesTransfer as balances_Transfer } from './transfer'
+export { balancesTransfer, timestampCall, preHook, postHook } from './mappings'

packages/hydra-e2e-tests/fixtures/mappings/mappings.ts

@@ -0,0 +1,71 @@
+import { DatabaseManager } from '@dzlzv/hydra-db-utils'
+import BN from 'bn.js'
+import { SubstrateEvent } from '@dzlzv/hydra-common'
+
+import { Transfer } from '../generated/graphql-server/src/modules/transfer/transfer.model'
+import { BlockTimestamp } from '../generated/graphql-server/src/modules/block-timestamp/block-timestamp.model'
+import {
+  BlockHook,
+  HookType,
+} from '../generated/graphql-server/src/modules/block-hook/block-hook.model'
+
+// run 'NODE_URL=<RPC_ENDPOINT> EVENTS=<comma separated list of events> yarn codegen:mappings-types'
+// to genenerate typescript classes for events, such as Balances.TransferEvent
+import { Balances, Timestamp } from './generated/types'
+
+// positional arguments
+export async function balancesTransfer(
+  store: DatabaseManager,
+  event: Balances.TransferEvent
+) {
+  const transfer = new Transfer()
+  transfer.from = Buffer.from(event.data.accountIds[0].toHex())
+  transfer.to = Buffer.from(event.data.accountIds[1].toHex())
+  transfer.value = event.data.balance.toBn()
+  transfer.block = event.ctx.blockNumber
+  transfer.comment = `Transferred ${transfer.value} from ${transfer.from} to ${transfer.to}`
+  transfer.insertedAt = new Date()
+  await store.save<Transfer>(transfer)
+}
+
+// context argument
+export async function timestampCall({
+  store,
+  event,
+}: {
+  store: DatabaseManager
+  event: SubstrateEvent
+}) {
+  const call = new Timestamp.SetCall(event)
+  const block = new BlockTimestamp()
+  block.timestamp = call.args.now.toBn()
+  block.blockNumber = new BN(call.ctx.blockNumber)
+
+  await store.save<BlockTimestamp>(block)
+}
+
+export async function preHook({
+  block: { blockNumber },
+  store,
+}: {
+  block: { blockNumber: BN }
+  store: DatabaseManager
+}) {
+  const hook = new BlockHook()
+  hook.blockNumber = blockNumber
+  hook.type = HookType.PRE
+  await store.save<BlockHook>(hook)
+}
+
+export async function postHook({
+  block: { blockNumber },
+  store,
+}: {
+  block: { blockNumber: BN }
+  store: DatabaseManager
+}) {
+  const hook = new BlockHook()
+  hook.blockNumber = blockNumber
+  hook.type = HookType.POST
+  await store.save<BlockHook>(hook)
+}

packages/hydra-e2e-tests/fixtures/schema.graphql

@@ -14,6 +14,17 @@ type BlockTimestamp @entity {
   timestamp: BigInt!
 }
 
+" Tracks block hooks "
+type BlockHook @entity {
+  blockNumber: BigInt!
+  type: HookType!
+}
+
+enum HookType {
+  PRE
+  POST
+}
+
 ############## Test purpose only @hydra-e2e-tests #################
 # To make sure graphql api is generated as expected
 type Extrinsic @entity {

packages/hydra-e2e-tests/test/e2e/api/graphql-queries.ts

@@ -85,3 +85,12 @@ export const PROCESSOR_SUBSCRIPTION = gql`
     }
   }
 `
+
+export const HOOKS = gql`
+  query {
+    blockHooks {
+      blockNumber
+      type
+    }
+  }
+`

packages/hydra-e2e-tests/test/e2e/api/processor-api.ts

@@ -30,8 +30,9 @@ export interface ProcessorStatus {
 
 export let processorStatus: ProcessorStatus | undefined
 
-const getGQLClient = () => Container.get<GraphQLClient>('ProcessorClient')
-const getSubClient = () =>
+export const getGQLClient = () =>
+  Container.get<GraphQLClient>('ProcessorClient')
+export const getSubClient = () =>
   Container.get<SubscriptionClient>('SubscriptionClient')
 
 export async function findTransfersByComment(text: string): Promise<string[]> {

packages/hydra-e2e-tests/test/e2e/hooks.test.ts

@@ -0,0 +1,40 @@
+import pWaitFor from 'p-wait-for'
+import { expect } from 'chai'
+import { HOOKS } from './api/graphql-queries'
+import { getGQLClient, getProcessorStatus } from './api/processor-api'
+
+const hooksTo = 4
+
+describe('end-to-end hook tests', () => {
+  let hooks: { type: string; blockNumber: number }[]
+
+  before(async () => {
+    // wait until the indexer indexes the block and the processor picks it up
+    await pWaitFor(
+      async () => {
+        return (await getProcessorStatus()).lastCompleteBlock > hooksTo
+      },
+      { interval: 50 }
+    )
+    hooks = (
+      await getGQLClient().request<{
+        blockHooks: { type: string; blockNumber: number }[]
+      }>(HOOKS)
+    ).blockHooks
+    console.log(`Executed hooks: ${JSON.stringify(hooks, null, 2)}`)
+  })
+
+  it('finds pre hooks', async () => {
+    const preHooks = hooks
+      .filter((h) => h.type === 'PRE')
+      .map((h) => h.blockNumber)
+    expect(preHooks).to.have.members(['1', '2'])
+  })
+
+  it('finds post hooks', async () => {
+    const postHooks = hooks
+      .filter((h) => h.type === 'POST')
+      .map((h) => h.blockNumber)
+    expect(postHooks).to.have.members(['2', '3'])
+  })
+})

packages/hydra-processor/package.json

@@ -38,15 +38,17 @@
     "@oclif/command": "^1.8.0",
     "@oclif/config": "^1",
     "@oclif/errors": "^1.3.3",
-    "bn.js": "^5.1.3",
+    "bn.js": "^5.2.0",
     "chalk": "^4.1.0",
     "delay": "^5.0.0",
     "dotenv": "^8.2.0",
     "envalid": "^6.0.2",
     "express": "^4.17.1",
     "graphql": "^15.4.0",
     "graphql-request": "^3.3.0",
+    "p-throttle": "^4.1.1",
     "p-wait-for": "^3.2.0",
+    "p-whilst": "^2.1.0",
     "prom-client": "^12.0.0",
     "semver": "^7.3.4",
     "typedi": "^0.8.0",
@@ -55,12 +57,18 @@
   },
   "devDependencies": {
     "@types/bn.js": "^4.11.6",
+    "@types/chai-spies": "^1.0.3",
     "@types/express": "^4.17.8",
     "@types/figlet": "^1.2.1",
     "@types/graphql": "^14.5.0",
     "@types/node": "^12",
+    "chai-spies": "^1.0.0",
     "eslint": "^7.12.1",
+    "p-immediate": "^3.2.0",
+    "ts-auto-mock": "^3.1.2",
+    "ts-mock-imports": "^1.3.3",
     "ts-node": "^9.0.0",
+    "ts-sinon": "^2.0.1",
     "tslib": "^2.0.3",
     "typescript": "^3.8"
   }

packages/hydra-processor/src/executor/IMappingExecutor.ts

@@ -0,0 +1,8 @@
+import { BlockContext } from '../queue'
+
+export interface IMappingExecutor {
+  executeBlock(
+    blockCtx: BlockContext,
+    onSuccess: (ctx: BlockContext) => Promise<void>
+  ): Promise<void>
+}

packages/hydra-processor/src/executor/IMappingsLookup.ts

@@ -0,0 +1,25 @@
+import { DatabaseManager } from '@dzlzv/hydra-db-utils'
+import { BlockContext, MappingContext } from '../queue'
+import { MappingHandler } from '../start/manifest'
+
+export interface StoreContext {
+  store: DatabaseManager
+}
+
+export type BlockHookContext = StoreContext & BlockContext
+
+export type EventContext = StoreContext & MappingContext
+
+export type ExecContext = BlockHookContext | EventContext
+
+export interface BlockMappings {
+  pre: MappingHandler[]
+  post: MappingHandler[]
+  mappings: MappingHandler[]
+}
+
+export interface IMappingsLookup {
+  lookupHandlers(ctx: BlockContext): BlockMappings
+
+  call(handler: MappingHandler, ctx: ExecContext): Promise<void>
+}