Stateful transforms

start: async ({ portal, state, logger }) => {
  if (!state.current) return  // first ever run: start empty

  const warmupFrom = Math.max(state.initial, state.current.number - LOOKBACK_BLOCKS)
  if (warmupFrom >= state.current.number) return

  for await (const { blocks } of portal.getStream({
    type: 'evm',
    fromBlock: warmupFrom,
    toBlock: state.current.number,
    fields: { block: { number: true }, log: { data: true } },
    logs: [{ address: [CONTRACT], topic0: [EVENT_TOPIC] }],
  })) {
    for (const block of blocks) {
      for (const log of block.logs) {
        // rebuild in-RAM state from block and log fields
      }
    }
  }
}

fork: async (cursor, { logger }) => {
  recentEntries = recentEntries.filter(e => e.blockNumber <= cursor.number)
}

type CHS = { insert(params: { table: string; values: unknown[]; format: string }): Promise<unknown> }

class WriteQueue {
  private ops: Array<(store: CHS) => Promise<void>> = []
  push(op: (store: CHS) => Promise<void>): void { this.ops.push(op) }
  async flush(store: CHS): Promise<void> { for (const op of this.ops) await op(store) }
}

-- Stores latest balance per pool (AggregatingMergeTree = efficient upsert)
CREATE TABLE current_balances (
    pool_address        String,
    token_a_balance_raw AggregateFunction(argMax, Int256, Tuple(DateTime, UInt16, UInt16)),
    token_b_balance_raw AggregateFunction(argMax, Int256, Tuple(DateTime, UInt16, UInt16))
) ENGINE = AggregatingMergeTree()
ORDER BY pool_address;

-- MV that keeps current_balances up to date
CREATE MATERIALIZED VIEW current_balances_mv TO current_balances AS
SELECT
    pool_address,
    argMaxState(token_a_balance_raw, (timestamp, transaction_index, log_index)) AS token_a_balance_raw,
    argMaxState(token_b_balance_raw, (timestamp, transaction_index, log_index)) AS token_b_balance_raw
FROM balances_history
GROUP BY pool_address;

latest_pool_balances AS (
    SELECT
        pool_address,
        argMaxMerge(token_a_balance_raw) AS balance_token_a_raw,
        argMaxMerge(token_b_balance_raw) AS balance_token_b_raw
    FROM current_balances
    WHERE pool_address IN (SELECT pool_address FROM unique_pools_to_insert)
    GROUP BY pool_address
)

SELECT ...
FROM liquidity_events_raw ml
ASOF JOIN latest_prices wp
    ON wp.pool_address = ml.pool_address
    AND wp.ts_num + wp.transaction_index * 100_000 + wp.log_index
        <= ml.ts_num + ml.transaction_index * 100_000 + ml.log_index
WHERE ml.protocol = 'uniswap_v4'

// Schema
db.exec(`
  CREATE TABLE IF NOT EXISTS balance_deltas (
    address      TEXT    NOT NULL,
    block_number INTEGER NOT NULL,
    delta        TEXT    NOT NULL,
    PRIMARY KEY (address, block_number)
  )
`)

function rollbackTo(blockNumber: number) {
  db.transaction(() => {
    const deltas = db.prepare('SELECT address, delta FROM balance_deltas WHERE block_number > ?').all(blockNumber)
    const net = new Map<string, bigint>()
    for (const { address, delta } of deltas) net.set(address, (net.get(address) ?? 0n) + BigInt(delta))
    for (const [address, delta] of net) {
      const { balance } = db.prepare('SELECT balance FROM balances WHERE address = ?').get(address) as any
      db.prepare('INSERT INTO balances (address, balance) VALUES (?, ?) ON CONFLICT(address) DO UPDATE SET balance = excluded.balance')
        .run(address, (BigInt(balance) - delta).toString())
    }
    db.prepare('DELETE FROM balance_deltas WHERE block_number > ?').run(blockNumber)
    db.prepare('DELETE FROM processed_blocks WHERE block_number > ?').run(blockNumber)
  })()
}

start: async ({ state }) => {
  const sqliteLastBlock  = db.prepare('SELECT MAX(block_number) as m FROM processed_blocks').get().m ?? null
  const pipelineLastBlock = state.current?.number ?? null

  if (sqliteLastBlock !== null && (pipelineLastBlock === null || sqliteLastBlock > pipelineLastBlock)) {
    rollbackTo(pipelineLastBlock ?? -1)  // SQLite crashed ahead of cursor — roll back
  } else if (sqliteLastBlock !== pipelineLastBlock) {
    throw new Error(`State mismatch: SQLite=${sqliteLastBlock}, cursor=${pipelineLastBlock}. Delete the SQLite file to rebuild.`)
  }
}

start: async ({ state }) => {
  if (sqliteLastBlock !== pipelineLastBlock) {
    throw new Error(`Delete ${SQLITE_DB_PATH} to rebuild.`)
  }
}

class WriteQueue {
  private ops: Array<(tx: Transaction) => Promise<void>> = []
  push(op: (tx: Transaction) => Promise<void>): void { this.ops.push(op) }
  async flush(tx: Transaction): Promise<void> { for (const op of this.ops) await op(tx) }
}

function initQueue<T>() {
  return createTransformer<T, Piped<T>>({
    transform: (data) => ({ payload: data, writes: new WriteQueue() }),
  })
}

// Pipeline:
stream
  .pipe(initQueue<DecodedBatch>())
  .pipe(transformerA(db))
  .pipe(transformerB(db))
  .pipeTo(drizzleTarget({
    db,
    tables: [tableA, tableB],
    onData: async ({ tx, data }) => { await data.writes.flush(tx) },
  }))

start: async ({ state }) => {
  const localLastBlock    = /* read your checkpoint */
  const pipelineLastBlock = state.current?.number ?? null

  if (localLastBlock !== null && (pipelineLastBlock === null || localLastBlock > pipelineLastBlock)) {
    rollbackTo(pipelineLastBlock ?? -1)  // crash recovery: undo ahead-of-cursor writes
  }
}

type Piped<T> = { payload: T; writes: WriteQueue }

function initQueue<T>() {
  return createTransformer<T, Piped<T>>({
    transform: (data) => ({ payload: data, writes: new WriteQueue() }),
  })
}