<!-- source: https://secondlayer.tools/docs/sinks -->

# 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

```ts indexer.ts

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
  }),
  // The return value is ignored.
  onBatch: async (events, _envelope, ctx) => {
    for (const event of events) {
      if (event.event_type !== "ft_transfer") continue; // narrows `payload`
      // Rows and checkpoint commit together.
      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](https://kysely.dev) |
| `drizzleSink` | `@secondlayer/sdk/sinks/drizzle` | Postgres or SQLite via [drizzle](https://orm.drizzle.team) |
| `bunSqliteSink` | `@secondlayer/sdk/sinks/bun-sqlite` | `bun:sqlite` (one file, zero dependencies) |

Any other store: [write your own](#write-your-own-sink) 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:

```ts
// 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.

### Keyed on the query builder, not the driver

`kyselySink(db, …)` takes a [Kysely](https://kysely.dev) 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](/docs/subgraphs) still owns this if you do not want to keep the events.

```ts sink.ts

kyselySink(db, {
  id: "sbtc",
  tables: ["transfers"],
  height: "block_height",
  // Same transaction, before the delete.
  onRollback: async (tx, { forkPointHeight }) => {
    const doomed = await tx
      .selectFrom("transfers")
      // Inclusive of the fork block.
      .where("block_height", ">=", forkPointHeight)
      .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

```ts

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.

```ts
interface ConsumerSink {
  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](https://github.com/porsager/postgres) sink:

```ts postgres-sink.ts

type Tx = postgres.TransactionSql;

  sql: postgres.Sql,
  options: { id: string; tables: readonly string[]; height: string },
) {
  const checkpoints = quoteIdent("sl_consumer_checkpoints");
  const driver: SinkDriver = {
    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],
      );
    },
    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 });
}
```

`createSink` also takes `onRollback`: same transaction, before the delete. Fold pattern: [Rollback is height-stamped](#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.

```ts

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:

```ts
const sink: ConsumerSink = {
  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` |

## Setup

Ensure setup once, skipping any step already done:
- Skill: `bunx skills add ryanwaits/secondlayer --skill secondlayer -y`
- CLI: `bun add -g @secondlayer/cli`
- Instance: `secondlayer init` writes `INSTANCE_TOKEN`. Loopback `/v1` reads need no key.
- Hosted/archive: export `SECONDLAYER_API_KEY` (`sk-sl_*`) or buy credits with `secondlayer credits buy` (`balance`, `refill`), then `secondlayer whoami`