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.

With a sink

indexer.tsts

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) => {

^The return value is ignored.

for (const event of events) {

if (event.event_type !== "ft_transfer") continue; // narrows `payload`

await ctx.tx.insertInto("transfers").values({

^Rows and checkpoint commit together.

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.

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

and you double-count on restart. and one block of orphaned rows survives the fork.

Skip onReorg and nothing errors; you keep rows from a chain that no longer exists.

Keyed on the query builder, not the driver

kyselySink(db, …) takes a Kysely instance, not a driver. drizzleSink does the same with your schema objects.

A bare pg.Pool carries no schema type, which would mean:

tables goes unchecked.
ctx.tx is any, exactly where correctness matters.

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) it in onRollback.

It runs in the 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.

sink.tsts

import { sql } from "kysely";

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

kyselySink(db, {

id: "sbtc",

tables: ["transfers"],

height: "block_height",

onRollback: async (tx, { forkPointHeight }) => {

^Same transaction, before the delete.

const doomed = await tx

.selectFrom("transfers")

.where("block_height", ">=", forkPointHeight)

^Inclusive of the fork block.

.selectAll()

.execute();

for (const row of doomed) {

await tx

.updateTable("balances")

.set({ amount: sql`amount - ${row.amount}::numeric` })

.where("holder", "=", row.recipient)

.execute();

await tx

.updateTable("balances")

.set({ amount: sql`amount + ${row.amount}::numeric` })

.where("holder", "=", row.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 two members; the SDK never starts a server for you.

Member	Use
`record`	Feed it from `onProgress`
`handler`	Mount it yourself

Write your own sink

createSink(driver, options) from @secondlayer/sdk/sinks/core owns the transaction sequences and guards; your driver only speaks your database's dialect.

The contract

Three methods, thirteen invariants. Each one is silent in production when violated, so the conformance kit tests them mechanically.

interface ConsumerSink<Tx> {

loadCursor(): Promise<string | null>;

commitBatch(cursor: string, write: (tx: Tx) => Promise<void> | void): Promise<void>;

rollback(forkPointHeight: number, rewindCursor: string): Promise<void>;

}

#	Invariant
1	`loadCursor` runs once, before the first page; it is the init hook
2	Replay is deterministic: same cursor, same rows, so `cursor` is an idempotency key
3	Delivery is at-least-once: a lost commit outcome is re-committed with the same cursor
4	Rows and cursor commit in ONE transaction
5	A throw from `write` aborts the whole transaction
6	The lent `tx` is the real transaction; every handler write goes through it
7	Committing the same cursor twice must not corrupt state
8	Rollback's delete and cursor rewind commit in ONE transaction
9	Delete AT OR ABOVE `forkPointHeight` (`>=`), since the new chain re-supplies the fork block
10	Rollback may be re-applied after a crash; the second pass is a no-op
11	Scope the undo by `forkPointHeight`, never by `rewindCursor`
12	Throw, don't swallow; no internal retry
13	One writer per checkpoint id; a lock must fail loudly, never block

The driver

A complete postgres.js sink:

postgres-sink.tsts

import postgres from "postgres";

import { createSink, quoteIdent, type SinkDriver } from "@secondlayer/sdk/sinks/core";

type Tx = postgres.TransactionSql;

export function postgresJsSink(

sql: postgres.Sql,

options: { id: string; tables: readonly string[]; height: string },

) {

const checkpoints = quoteIdent("sl_consumer_checkpoints");

const driver: SinkDriver<Tx> = {

transact: (work) => sql.begin((tx) => work(tx as Tx)) as Promise<never>,

async ensureCheckpointStore() {

await sql.unsafe(`CREATE TABLE IF NOT EXISTS ${checkpoints} (id text PRIMARY KEY, cursor text NOT NULL)`);

async readCursor() {

const rows = await sql.unsafe(`SELECT cursor FROM ${checkpoints} WHERE id = $1`, [options.id]);

return rows[0]?.cursor ?? null;

async writeCursor(tx, cursor) {

await tx.unsafe(

`INSERT INTO ${checkpoints} (id, cursor) VALUES ($1, $2) ON CONFLICT (id) DO UPDATE SET cursor = excluded.cursor`,

[options.id, cursor],

);

async deleteAtOrAbove(tx, table, height) {

await tx.unsafe(

`DELETE FROM ${quoteIdent(table)} WHERE ${quoteIdent(options.height)} >= $1`,

[height],

);

createSink also takes onRollback: same transaction, before the delete. Fold pattern: Rollback is height-stamped.

Identifiers arrive pre-validated ([A-Za-z_][A-Za-z0-9_]*), safe to interpolate after quoteIdent.
Values stay bound parameters.
acquireLock is the optional seventh method: throw when another writer holds your id.

Prove it

@secondlayer/sdk/sinks/testing probes every invariant against your real store: torn batches, cursor replay, inclusive >=, mismatched rollback pairs, lock contention.

import { test } from "bun:test";

import postgres from "postgres";

import { attachSinkConformance } from "@secondlayer/sdk/sinks/testing";

const sql = postgres(process.env.DATABASE_URL!);

attachSinkConformance(test, {

makeSink: () => postgresJsSink(sql, { id: "conformance", tables: ["rows"], height: "height" }),

reset: async () => {

await sql`TRUNCATE rows, sl_consumer_checkpoints`;

insertRow: async (tx: postgres.TransactionSql, height, key) => {

await tx.unsafe(

"INSERT INTO rows (key, height) VALUES ($1, $2) ON CONFLICT DO NOTHING",

[key, height],

);

readRows: () => sql`SELECT height, key FROM rows`,

readCursor: async () =>

(await sql`SELECT cursor FROM sl_consumer_checkpoints WHERE id = 'conformance'`)[0]?.cursor ?? null,

});

One probe per test; the test argument accepts bun:test, vitest, or jest.
Every shipped sink passes this exact kit in CI.
insertRow must be replay-safe, so upsert keyed on key.

Append-only stores

A store that can't delete (ClickHouse, parquet, an append log) can't implement rollback. Declare it:

const sink: ConsumerSink<MyWriter> = {

capabilities: { finalizedOnly: true },

// rollback is never called; throw if reached

...

};

The loop refuses to run without finalizedOnly: true, at startup, before data flows. Finalized rows never reorg; deterministic replay makes cursor a dedup token for exactly-once reads.

Store caveats

Store	Caveat
Prisma	`$transaction` defaults to a 5s interactive timeout, so raise it; writes through the GLOBAL client escape the lent `tx` silently
libsql / Turso remote	5s write lock; remote statement-per-request can't hold the commit transaction open, so local files only
ClickHouse / parquet	No deletes: `capabilities: { finalizedOnly: true }`, dedup on `cursor`

Edit on GitHub