BlockStream

import { BlockStream } from '@voltaire/block'

const stream = BlockStream({ provider })

// Watch for new blocks with reorg handling
for await (const event of stream.watch()) {
  if (event.type === 'reorg') {
    for (const block of event.removed) {
      rollback(block.hash)
    }
  } else {
    for (const block of event.blocks) {
      process(block)
    }
  }
}

function BlockStream(options: {
  provider: TypedProvider
}): BlockStreamInstance

for await (const { blocks, metadata } of stream.backfill({
  fromBlock: 18000000n,
  toBlock: 19000000n,
  include: 'transactions'
})) {
  for (const block of blocks) {
    indexBlock(block)
  }
}

const controller = new AbortController()

for await (const event of stream.watch({
  signal: controller.signal,
  include: 'receipts'
})) {
  switch (event.type) {
    case 'blocks':
      for (const block of event.blocks) {
        processBlock(block)
      }
      break
    case 'reorg':
      // Undo removed blocks (newest first)
      for (const block of event.removed) {
        rollbackBlock(block.hash)
      }
      // Added blocks will come in subsequent events
      break
  }
}

type BlocksEvent<TInclude> = {
  type: 'blocks'
  blocks: StreamBlock<TInclude>[]
  metadata: { chainHead: bigint }
}

type ReorgEvent<TInclude> = {
  type: 'reorg'
  removed: LightBlock[]           // Blocks removed (newest first)
  added: StreamBlock<TInclude>[]  // New canonical blocks (oldest first)
  commonAncestor: LightBlock      // Last block before divergence
  metadata: { chainHead: bigint }
}

type LightBlock = {
  number: bigint
  hash: BlockHashType
  parentHash: BlockHashType
  timestamp: bigint
}

// Handle reorgs properly
for await (const event of stream.watch()) {
  if (event.type === 'reorg') {
    console.log(`Reorg detected: ${event.removed.length} blocks removed`)
    console.log(`Common ancestor: block ${event.commonAncestor.number}`)

    // Roll back state for removed blocks
    for (const block of event.removed) {
      await db.revertBlock(block.hash)
    }
  }
}

const FINALITY_DEPTH = 64n // ~13 minutes on mainnet

const finalizedBlocks = new Map<bigint, LightBlock>()
const pendingBlocks: LightBlock[] = []

for await (const event of stream.watch()) {
  if (event.type === 'blocks') {
    for (const block of event.blocks) {
      pendingBlocks.push(toLightBlock(block))

      // Check if oldest pending block is now final
      const chainHead = event.metadata.chainHead
      while (pendingBlocks.length > 0) {
        const oldest = pendingBlocks[0]
        if (chainHead - oldest.number >= FINALITY_DEPTH) {
          finalizedBlocks.set(oldest.number, oldest)
          pendingBlocks.shift()
          // Safe to archive/compact finalized data
        } else {
          break
        }
      }
    }
  }
}

import {
  BlockStreamAbortedError,
  UnrecoverableReorgError
} from '@voltaire/block'

try {
  for await (const event of stream.watch({ signal })) {
    // ...
  }
} catch (error) {
  if (error instanceof BlockStreamAbortedError) {
    console.log('Stream aborted')
  } else if (error instanceof UnrecoverableReorgError) {
    console.log('Deep reorg beyond tracked history, resync required')
  } else {
    throw error
  }
}