Sinks

A sink is the destination your consume loop writes through. It owns the three things everyone gets subtly wrong — the checkpoint, the transaction boundary, and the reorg rollback — so your handler only inserts rows.

Written by hand, a correct consumer is about ninety lines, and most of them are not about your data:

// load the cursor, in your own table
// begin a transaction
//   insert rows
//   write the cursor — in the SAME transaction, or a crash between them
//   either replays the batch or skips it
// commit
// on reorg: delete rows at _block_height >= fork_point_height (INCLUSIVE),
//   rewind the cursor, both atomically

Miss the shared transaction and you double-count on restart. Use > instead of >= and one block of orphaned rows survives the fork forever. Forget onReorg entirely and the loop silently skips reorg handling — there is no error, just rows from a chain that no longer exists.

import { Streams } from "@secondlayer/sdk";
import { kyselySink } from "@secondlayer/sdk/sinks/kysely";

await sl.streams.events.consume({
  types: ["ft_transfer"],
  sink: kyselySink(db, {
    id: "sbtc-flows",          // checkpoint identity
    tables: ["transfers"],     // rolled back on reorg
    height: "height",          // the column carrying block height
  }),
  onBatch: async (events, _envelope, ctx) => {
    for (const event of events) {
      await ctx.tx.insertInto("transfers").values({
        cursor: event.cursor,
        sender: event.payload.sender,
        recipient: event.payload.recipient,
        amount: event.payload.amount,
        height: event.block_height,
      }).execute();
    }
  },
});

That is the whole indexer. ctx.tx is the sink's transaction — write only through it, and rows plus checkpoint commit together. onBatch's return value is ignored: the sink owns the cursor.

Any other store: write your own on the same base and prove it with the conformance kit.

kyselySink(db, …) takes a Kysely instance because a bare pg.Pool carries no schema type. Without it, tables could not be checked against your database and ctx.tx could not be typed — you would be back to any at the exact place correctness matters. drizzleSink plays the same trick with your schema objects.

kysely and drizzle-orm are optional peer dependencies; the root entry stays dependency-free.

v1 rolls back by deleting rows at height >= fork_point_height. That is correct for append-only projections, and it is why height is required. A declared table that lacks the column is a type error when your Kysely schema knows about it, and a loud throw on the consumer's first cursor load otherwise — either way you find out before the first batch, rather than discovering at reorg time that rollback deletes nothing.

Aggregates — running balances, counters — are not append-only, so a delete cannot reverse them. Use a Subgraph for those: its runtime journals pre-images and restores them on rewind.

The sink loads its own committed cursor on start, so a restart continues where it left off with no arguments. An explicit fromCursor overrides that position — useful for replaying a range — and the sink still initializes.

import { consumerHealth, shutdownSignal } from "@secondlayer/sdk";

const health = consumerHealth({ staleAfterMs: 60_000 });
Bun.serve({ port: 8080, fetch: health.handler });

await sl.streams.events.consume({
  signal: shutdownSignal(),   // SIGTERM/SIGINT → clean stop at a batch boundary
  onProgress: health.record,
  sink,
  onBatch: …,
});

consumerHealth returns a record to feed from onProgress and a handler you mount yourself — the SDK never starts a server for you, which would pin your runtime.