Schedule bounty stage changes based on periods

i.e: `fundingPeriod`, `workPeriod`, and `judgingPeriod`

query-node/manifest.yml

@@ -868,4 +868,5 @@ mappings:
     - handler: bootstrapData
       filter:
         height: "[0,0]" # will be executed only at genesis
+    - handler: launchScheduler
   postBlockHooks:

query-node/mappings/src/bounty.ts

@@ -1,6 +1,6 @@
-import { EventContext, StoreContext } from '@joystream/hydra-common'
+import { DatabaseManager, EventContext, StoreContext } from '@joystream/hydra-common'
 import { BountyMetadata } from '@joystream/metadata-protobuf'
-import { AssuranceContractType, BountyActor, BountyId, FundingType } from '@joystream/types/augment'
+import { AssuranceContractType, BountyActor, FundingType } from '@joystream/types/augment'
 import {
   Bounty,
   BountyContractClosed,
@@ -14,6 +14,7 @@ import {
 } from 'query-node/dist/model'
 import { Bounty as BountyEvents } from '../generated/types'
 import { deserializeMetadata, genericEventFields } from './common'
+import { scheduleAtBlock } from './scheduler'
 
 /**
  * Commons helpers
@@ -25,6 +26,82 @@ function bountyActorToMembership(actor: BountyActor): Membership | undefined {
   }
 }
 
+function isBountyFundingLimited(
+  fundingType: BountyFundingPerpetual | BountyFundingLimited
+): fundingType is BountyFundingLimited {
+  return fundingType.isTypeOf === BountyFundingLimited.name
+}
+
+function fundingPeriodEnd(bounty: Bounty): number {
+  return (
+    bounty.maxFundingReachedEvent?.inBlock ??
+    bounty.createdInEvent.inBlock + (bounty.fundingType as BountyFundingLimited).fundingPeriod
+  )
+}
+
+/**
+ * Schedule Periods changes
+ */
+
+export function bountyScheduleFundingEnd(store: DatabaseManager, bounty: Bounty) {
+  const { fundingType } = bounty
+  if (bounty.stage !== BountyStage.Funding || !isBountyFundingLimited(fundingType)) return
+
+  const fundingPeriodEnd = bounty.createdInEvent.inBlock + fundingType.fundingPeriod
+  scheduleAtBlock(fundingPeriodEnd, () => {
+    if (bounty.stage === BountyStage.Funding) {
+      const isFunded = bounty.totalFunding >= fundingType.minFundingAmount
+      endFundingPeriod(store, bounty, isFunded)
+    }
+  })
+}
+
+export function bountyScheduleWorkSubmissionEnd(store: DatabaseManager, bounty: Bounty) {
+  if (bounty.stage !== BountyStage.WorkSubmission) return
+
+  const workingPeriodEnd = fundingPeriodEnd(bounty) + bounty.workPeriod
+  scheduleAtBlock(workingPeriodEnd, () => {
+    if (bounty.stage === BountyStage.WorkSubmission) {
+      endWorkingPeriod(store, bounty)
+    }
+  })
+}
+
+export function bountyScheduleJudgementEnd(store: DatabaseManager, bounty: Bounty) {
+  if (bounty.stage !== BountyStage.Judgment) return
+
+  const judgementPeriodEnd = fundingPeriodEnd(bounty) + bounty.workPeriod + bounty.judgingPeriod
+  scheduleAtBlock(judgementPeriodEnd, () => {
+    if (bounty.stage === BountyStage.Funding) {
+      bounty.updatedAt = new Date()
+      bounty.stage = BountyStage.Failed
+      store.save<Bounty>(bounty)
+    }
+  })
+}
+
+function endFundingPeriod(store: DatabaseManager, bounty: Bounty, isFunded = true) {
+  bounty.updatedAt = new Date()
+  if (isFunded) {
+    bounty.stage = BountyStage.WorkSubmission
+    bountyScheduleWorkSubmissionEnd(store, bounty)
+  } else {
+    bounty.stage = BountyStage[bounty.contributions?.length ? 'Failed' : 'Expired']
+  }
+  return store.save<Bounty>(bounty)
+}
+
+function endWorkingPeriod(store: DatabaseManager, bounty: Bounty) {
+  bounty.updatedAt = new Date()
+  if (bounty.entries?.length) {
+    bounty.stage = BountyStage.Judgment
+    bountyScheduleJudgementEnd(store, bounty)
+  } else {
+    bounty.stage = BountyStage.Failed
+  }
+  return store.save<Bounty>(bounty)
+}
+
 /**
  * Event handlers
  */
@@ -59,6 +136,8 @@ export async function bounty_BountyCreated({ event, store }: EventContext & Stor
 
   await store.save<Bounty>(bounty)
 
+  bountyScheduleFundingEnd(store, bounty)
+
   const createdInEvent = new BountyCreatedEvent({ ...genericEventFields(event), bounty })
   await store.save<BountyCreatedEvent>(createdInEvent)
 

query-node/mappings/src/index.ts

@@ -15,3 +15,4 @@ export * from './proposals'
 export * from './proposalsDiscussion'
 export * from './forum'
 export * from './bootstrap'
+export * from './scheduler'

query-node/mappings/src/scheduler.ts

@@ -0,0 +1,56 @@
+import { In } from 'typeorm'
+import { DatabaseManager, StoreContext } from '@joystream/hydra-common'
+import { eventEmitter, ProcessorEvents } from '@joystream/hydra-processor/lib/start/processor-events'
+import { Bounty, BountyStage } from 'query-node/dist/model'
+import { bountyScheduleWorkSubmissionEnd, bountyScheduleFundingEnd, bountyScheduleJudgementEnd } from './bounty'
+
+let isSchedulerRunning = false
+let toBeScheduled: [number, () => void][] = []
+
+export async function launchScheduler({ store }: StoreContext) {
+  if (!isSchedulerRunning) {
+    runScheduler()
+    await scheduleMissedMappings(store)
+  }
+}
+
+export function scheduleAtBlock(blockNumber: number, job: () => void) {
+  toBeScheduled.push([blockNumber, job])
+}
+
+function runScheduler() {
+  isSchedulerRunning = true
+  const scheduleRecord: { [n: number]: (() => void)[] } = {}
+
+  eventEmitter.on(ProcessorEvents.INDEXER_STATUS_CHANGE, (indexerStatus) => {
+    if (toBeScheduled.length) {
+      toBeScheduled.forEach(([blockNumber, job]) => {
+        if (blockNumber < indexerStatus.chainHeight) {
+          job()
+        } else {
+          scheduleRecord[blockNumber] = [...(scheduleRecord[blockNumber] ?? []), job]
+        }
+      })
+      toBeScheduled = []
+    }
+
+    if (scheduleRecord[indexerStatus.chainHeight]) {
+      scheduleRecord[indexerStatus.chainHeight].forEach((job) => job())
+      delete scheduleRecord[indexerStatus.chainHeight]
+    }
+  })
+}
+
+async function scheduleMissedMappings(store: DatabaseManager) {
+  // Reschedule mappings lost while the processor was off
+
+  // Bounty stage updates
+  const bounties = await store.getMany(Bounty, {
+    where: { stage: In([BountyStage.Funding, BountyStage.WorkSubmission, BountyStage.Judgment]) },
+  })
+  bounties.forEach((bounty) => {
+    bountyScheduleFundingEnd(store, bounty)
+    bountyScheduleWorkSubmissionEnd(store, bounty)
+    bountyScheduleJudgementEnd(store, bounty)
+  })
+}