#CDC-out — declarative reverse-ETL

@voltro/plugin-cdc-out streams your table changes OUTWARD to an external sink. Declare { table → sink, map } and the plugin mirrors every change through a durable outbox in your app's own database — no separate Debezium / Fivetran pipeline. The data-team wedge: app data in the warehouse, declared in code, composed onto the change stream you already have.

#Wiring

// app.config.ts
import { cdcOutPlugin, webhookSink } from '@voltro/plugin-cdc-out'

export default {
  type: 'api' as const,
  name: 'api',
  plugins: [
    cdcOutPlugin({
      sinks: [
        {
          table: 'orders',
          sink:  webhookSink('https://warehouse.example.com/ingest'),
          // Map a row → the outbound record body (default: the row unchanged).
          map:   (row) => ({ id: row.id, total: row.total, status: row.status }),
          // Optional: only mirror some changes.
          filter: (event) => event.op !== 'delete',
          batchSize: 500,          // max records per delivered batch (default 100)
        },
      ],
      maxAttempts: 5,              // per-record delivery attempts before dead-letter
      backoffBaseMs: 200,          // first retry delay; doubles per attempt, jittered
      deliveryTimeoutMs: 10_000,   // per-attempt timeout — aborts the sink call
      leaseTtlMs: 15_000,          // leader lease; heartbeat renews at ttl/3
      // Fleet-scope handoff (see "Delivery guarantees"):
      dedupWindowMs: 60_000,       // how long an enqueue claim is kept — must exceed leaseTtlMs
      handoffBufferMs: 60_000,     // how far back each replica buffers for a takeover to drain
      handoffBufferSize: 10_000,   // hard ceiling on buffered changes per replica
    }),
  ],
}

One sink config per table per instance — a duplicate table is a loud boot error. For multiple sinks on one table, wire a second instance with a name (cdcOutPlugin({ name: 'analytics', sinks: […] })); it suffixes the plugin name (@voltro/plugin-cdc-out#analytics) and the inspect mount (/_voltro/inspect/plugins/cdc-out--analytics/…).

#Sinks

• memorySink() — in-process, records every delivered batch. Dev + tests.
• webhookSink(url, { headers? }) — POSTs each batch as { records: [...] } JSON, honoring the engine's per-attempt abort signal. Its host is declared as a network:outbound:<host> permission automatically.
• kafkaSink({ topic }) — from @voltro/plugin-queue, producing through the same provider your consumers use: message key = the row id (one row's changes stay ordered in one partition), value = the change record, and the x-voltro-delivery-key header carries the dedupe handle below.
• Warehouse (Snowflake / BigQuery / …) — implement the CdcSink interface ({ name, deliver(batch, ctx), outboundHost? }). deliver may return a Promise or an Effect — both compose without wrapping. The engine is connector-agnostic; the sink is the only thing that changes.

#Delivery guarantees — exactly what holds

Every mirrored change becomes a row in _voltro_cdcout_outbox (contributed via extendSchema, migrated by voltro dev). The row's TypeID id is the record's deliveryKey — unique across replicas, stable across restarts and retries.

• Enqueue is de-duplicated per observed change, fleet-wide, across leadership handovers — with one hole: a replica that dies between winning a change's claim and inserting its outbox row loses that change, because the claim survives and nothing rescans orphan claims. (Those are two statements with no transaction around them, which is why this does not say "exactly-once".)

  On changeScope: 'local' stores each replica enqueues only its OWN commits (injected cross-replica events are skipped), so a change is observed by exactly one process and needs nothing further.

  On 'fleet' stores (postgres changeStrategy: 'cdc' LISTEN/NOTIFY, mysql binlog) EVERY replica sees the full stream, so every replica buffers it in memory (handoffBufferMs, capped at handoffBufferSize entries, oldest dropped first). The holder of the leader lease (_voltro_cdcout_leases, TTL-heartbeat) enqueues as it goes; a replica that WINS the lease drains the window its predecessor never got to. Enqueue is idempotent per change identity: each change is keyed by a changeKey every replica computes alike — a digest of (pipe, op, row id, new image, old image) plus an occurrence counter that keeps two byte-identical changes to one row apart — and claimed in _voltro_cdcout_claims under unique(pipe, changeKey). So the rows the dying leader already wrote collapse instead of duplicating, and the ones it never reached are written by its successor.

  A replica that boots into a fleet that is already running adopts the fleet's occurrence counters from the claims already in the database before it keys anything — otherwise its first sighting of an already-claimed change would key occurrence 0, collide, and be dropped as a duplicate it is not. GET /_voltro/inspect/plugins/cdc-out/sinks reports handoff.seeded and handoff.awaitingSeed so you can see that happen rather than assume it.

  What still bounds it, stated plainly: a change no surviving replica observed is gone (the transport delivered it only to the dead process); a handoff that takes longer than handoffBufferMs loses whatever aged out of the buffer, and that count is reported as handoff.dropped on GET /_voltro/inspect/plugins/cdc-out/sinks rather than dropped silently. dedupWindowMs (default max(60_000, 4 × leaseTtlMs)) is how long a claim is kept; it must exceed leaseTtlMs — a claim that expires mid-handoff is a duplicate window, and the plugin refuses to boot with one.

• At-least-once FROM ENQUEUE. The tap is post-commit — a crash in the narrow window between commit and the outbox insert loses that one event; the plugin does not claim better. From enqueue on, delivery survives restarts, retries with exponential backoff + jitter, and re-sends the SAME deliveryKeys — a sink that upserts on them dedupes safely.

• Ordered per pipe. One in-flight batch per pipe, consumed strictly in commit (id) order; the queue is head-blocking, so a batch waiting out its retry backoff is never overtaken.

• Dead-letter, never silent drop. A record that exhausts maxAttempts moves to status dead with its last error — queryable at GET /_voltro/inspect/plugins/cdc-out/dead-letter — and unblocks the pipe.

• Bounded storage. Delivered/dead rows are purged by the framework retention sweep after retentionHours (default 72, env CDCOUT_RETENTION_HOURS); pending rows are never purged. Enqueue claims are short-lived by design — purged after dedupWindowMs, which only has to outlive a leadership handoff.

#Multi-tenancy

The change tap fires for every tenant's rows, and a sink is app-global, developer-authored config: whatever map returns (default: the raw row, including tenantId) is mirrored to ONE external endpoint. Use filter / map to scope or strip tenant data, and only ever point sinks at operator-controlled URLs — never at a tenant-supplied one.

#Backfill

Seed a pipe with existing rows once — they flow through the SAME durable outbox as live changes (ordered, retried, dead-lettered, stable per-row deliveryKeys). Call from a *.startup.tsx or a CLI with the app's store:

import { enqueueBackfill } from '@voltro/plugin-cdc-out'
import { sinkConfig } from './cdc'

const n = await enqueueBackfill(store, sinkConfig, rows) // every current row, enqueued as an insert