Sinks

A sink is the destination your consume loop writes through. It owns the checkpoint, the transaction boundary, and the reorg rollback, so your handler only inserts rows.

The problem it removes

Hand-rolled, a correct consumer is about ninety lines, most of them 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. Skip onReorg and nothing errors; you keep rows from a chain that no longer exists.

With a sink

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

const sl = new SecondLayer({
  apiKey: process.env.INSTANCE_TOKEN,
});

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. Write only through ctx.tx, the sink's transaction, and rows plus checkpoint commit together. onBatch's return value is ignored: the sink owns the cursor.

A sink without reorg handling would be a no-op

A sink makes rollback unconditional: omitting onReorg no longer skips reorgs. onReorg still runs after the sink transaction commits. Fold current-state (balances) in the sink's onRollback, not there.

Three sinks ship

Sink	Import	Store
`kyselySink`	`@secondlayer/sdk/sinks/kysely`	Postgres via Kysely
`drizzleSink`	`@secondlayer/sdk/sinks/drizzle`	Postgres or SQLite via drizzle
`bunSqliteSink`	`@secondlayer/sdk/sinks/bun-sqlite`	`bun:sqlite` (one file, zero dependencies)

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

Keyed on the query builder, not the driver

kyselySink(db, …) takes a Kysely instance because a bare pg.Pool carries no schema type: tables would go unchecked and ctx.tx would be any exactly where correctness matters. drizzleSink does the same with your schema objects.

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

Rollback is height-stamped

v1 deletes height >= fork_point_height from every declared table. That is correct for one-row-per-event tables, which is why height is required.

A running balance is a fold of those rows, not a row to delete. Keep it off tables and invert (or recompute) in onRollback, same transaction, while the doomed facts are still visible. Remaining facts are height < fork. Throwing aborts the rewind. A subgraph still owns this if you do not want to keep the events.

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

kyselySink(db, {
  id: "sbtc",
  tables: ["transfers"],
  height: "block_height",
  onRollback: async (tx, { forkPointHeight }) => {
    const doomed = await tx
      .selectFrom("transfers")
      .where("block_height", ">=", forkPointHeight)
      .selectAll()
      .execute();
    for (const e of doomed) {
      await tx
        .updateTable("balances")
        .set({ amount: sql`amount - ${e.amount}::numeric` })
        .where("holder", "=", e.recipient)
        .execute();
      await tx
        .updateTable("balances")
        .set({ amount: sql`amount + ${e.amount}::numeric` })
        .where("holder", "=", e.sender)
        .execute();
    }
  },
});

Resuming

The sink loads its committed cursor on start, so a restart continues with no arguments. An explicit fromCursor overrides the position for replaying a range; the sink still initializes.

Health and shutdown

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.