Fork handling

// WORKAROUND — see explanation above the tab
function pipeToIterator<T>(
  source: { pipeTo(t: ReturnType<typeof createTarget<T>>): Promise<void> },
  initialCursor: BlockCursor | undefined,
  onFork: (previousBlocks: BlockCursor[]) => Promise<BlockCursor | null>
): AsyncIterableIterator<{ data: T; ctx: any }> {

  type Slot =
    | { k: 'batch'; v: { data: T; ctx: any } }
    | { k: 'end' }
    | { k: 'error'; err: unknown }

  const queue: Slot[] = []
  let consumerWake: (() => void) | null = null
  let producerAck:  (() => void) | null = null
  const wake = () => { consumerWake?.(); consumerWake = null }

  ;(source.pipeTo as any)(createTarget({
    write: async ({ read }: any) => {
      for await (const batch of read(initialCursor)) {
        queue.push({ k: 'batch', v: batch })
        wake()
        await new Promise<void>(r => { producerAck = r })
      }
      queue.push({ k: 'end' })
      wake()
    },
    fork: onFork,
  })).catch((err: unknown) => { queue.push({ k: 'error', err }); wake() })

  return {
    async next(): Promise<IteratorResult<{ data: T; ctx: any }>> {
      if (!queue.length) await new Promise<void>(r => { consumerWake = r })
      const slot = queue.shift()!
      if (slot.k === 'end')   return { done: true,  value: undefined as any }
      if (slot.k === 'error') throw slot.err
      producerAck?.(); producerAck = null
      return { done: false, value: slot.v }
    },
    [Symbol.asyncIterator]() { return this },
  }
}

import { BlockCursor, createTarget } from '@subsquid/pipes'
import { solanaPortalStream, solanaInstructionDecoder } from '@subsquid/pipes/solana'
import * as orcaWhirlpool from './abi/orca_whirlpool/index.js'

// WORKAROUND — pipeToIterator defined above (see implementation expandable)

async function main() {
  let recentUnfinalizedSlots: BlockCursor[] = []
  let finalizedHighWatermark: BlockCursor | undefined

  const stream = pipeToIterator(
    solanaPortalStream({
      id: 'forks-async',
      portal: 'https://portal.sqd.dev/datasets/solana-mainnet',
      outputs: solanaInstructionDecoder({
        programId: orcaWhirlpool.programId,
        instructions: { swap: orcaWhirlpool.instructions.swap },
        range: { from: 'latest' }
      }),
    }),
    recentUnfinalizedSlots.at(-1),
    async (newConsensusSlots) => {
      const rollbackIndex = findRollbackIndex(recentUnfinalizedSlots, newConsensusSlots)
      if (rollbackIndex >= 0) {
        recentUnfinalizedSlots.length = rollbackIndex + 1
        return recentUnfinalizedSlots[rollbackIndex]
      }
      if (finalizedHighWatermark &&
          newConsensusSlots.every(b => b.number < finalizedHighWatermark!.number)) {
        recentUnfinalizedSlots = recentUnfinalizedSlots.filter(b => b.number <= finalizedHighWatermark!.number)
        return finalizedHighWatermark
      }
      recentUnfinalizedSlots.length = 0
      return null
    }
  )

  for await (const {data, ctx} of stream) {
    console.log(`Got ${data.swap.length} swaps`)
    ctx.stream.state.rollbackChain.forEach((bc: BlockCursor) => { recentUnfinalizedSlots.push(bc) })
    if (ctx.stream.head.finalized) {
      if (!finalizedHighWatermark || ctx.stream.head.finalized.number > finalizedHighWatermark.number) {
        finalizedHighWatermark = ctx.stream.head.finalized
      }
      recentUnfinalizedSlots = recentUnfinalizedSlots.filter(b => b.number >= finalizedHighWatermark!.number)
    }
    recentUnfinalizedSlots = recentUnfinalizedSlots.slice(recentUnfinalizedSlots.length - 1000)
  }
}

main().then(() => { console.log('\ndone') })


function findRollbackIndex(chainA: BlockCursor[], chainB: BlockCursor[]): number {
  let aIndex = 0, bIndex = 0, lastCommonIndex = -1
  while (aIndex < chainA.length && bIndex < chainB.length) {
    const a = chainA[aIndex], b = chainB[bIndex]
    if (a.number < b.number) { aIndex++; continue }
    if (a.number > b.number) { bIndex++; continue }
    if (a.hash !== b.hash) return lastCommonIndex
    lastCommonIndex = aIndex; aIndex++; bIndex++
  }
  return lastCommonIndex
}