Write your own sink

Implement a 7-method driver, run it through the conformance kit, and any store becomes a first-class sink: for consume loops.

The contract

Three methods, thirteen invariants — each one silent in production when violated, which is why the 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 — `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` — `>=`, 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

createSink(driver, options) from @secondlayer/sdk/sinks/core owns the sequences and guards; you implement the dialect. A complete postgres.js sink:

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 cp = quoteIdent("sl_consumer_checkpoints");
  const driver: SinkDriver<Tx> = {
    transact: (fn) => sql.begin((tx) => fn(tx as Tx)) as Promise<never>,
    async ensureCheckpointStore() {
      await sql.unsafe(`CREATE TABLE IF NOT EXISTS ${cp} (id text PRIMARY KEY, cursor text NOT NULL)`);
    },
    async readCursor() {
      const rows = await sql.unsafe(`SELECT cursor FROM ${cp} WHERE id = $1`, [options.id]);
      return rows[0]?.cursor ?? null;
    },
    async writeCursor(tx, cursor) {
      await tx.unsafe(
        `INSERT INTO ${cp} (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],
      );
    },
    async hasColumn(table, column) {
      const rows = await sql`SELECT EXISTS (SELECT 1 FROM information_schema.columns
        WHERE table_name = ${table} AND column_name = ${column}) AS ok`;
      return rows[0]?.ok === true;
    },
  };
  return createSink(driver, { label: "postgresJsSink", ...options });
}

Identifiers arrive pre-validated ([A-Za-z_][A-Za-z0-9_]*), safe to interpolate after quoteIdent. Values stay bound parameters. acquireLock is the optional eighth 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 { attachSinkConformance } from "@secondlayer/sdk/sinks/testing";

attachSinkConformance(test, {
  makeSink: () => postgresJsSink(sql, { id: "conformance", tables: ["rows"], height: "height" }),
  reset: () => sql`TRUNCATE rows, sl_consumer_checkpoints`,
  insertRow: (tx, height, key) =>
    tx`INSERT INTO rows (key, height) VALUES (${key}, ${height}) ON CONFLICT DO NOTHING`,
  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 becomes one test; the test argument accepts bun:test, vitest, or jest. Every shipped sink passes this exact kit in CI.

insertRow must be replay-safe

Upsert or insert-on-conflict-do-nothing, keyed on key — the same requirement at-least-once delivery puts on real onBatch handlers.

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 then refuses to run without finalizedOnly: true — loudly, at startup, before any data flows. Finalized rows never reorg, and deterministic replay makes cursor a dedup token for exactly-once semantics on the read side.

Where your store fits

Store	Path	Caveats
Postgres via Kysely	`@secondlayer/sdk/sinks/kysely`	ships; advisory lock, typed `ctx.tx`
Postgres or SQLite via drizzle	`@secondlayer/sdk/sinks/drizzle`	ships; schema objects, key→column resolution
bun:sqlite	`@secondlayer/sdk/sinks/bun-sqlite`	ships; zero-dep, no docker, file lock
Prisma	write your own	`$transaction` defaults to a 5s interactive timeout — raise it; writes through the GLOBAL client escape the lent `tx` silently
libsql / Turso remote	write your own	5s write lock; remote statement-per-request can't hold the commit transaction open — local files only
ClickHouse / parquet	write your own	no deletes: `capabilities: { finalizedOnly: true }`, dedup on `cursor`