# replicant usage rules

_A framework-agnostic Elixir CDC consumer for Postgres logical replication (`pgoutput`)._

## What replicant is (and is not)

- **Is:** a reliable CDC consumer. Decodes the `pgoutput` logical replication
  stream, assembles committed transactions, validates schema changes, and
  delivers each transaction to a pluggable sink with sink-owned,
  transaction-granularity exactly-once semantics.
- **Is not:** Ash-aware, tenant-aware, or classification-aware. Never put
  multitenancy or sensitive-data logic here — that is `ash_replicant`'s job.
- **Is:** a live streaming client. Owns the replication slot via
  `Postgrex.ReplicationConnection`, acks only after the sink durably commits
  (ack-after-checkpoint), and halts fail-closed on slot invalidation — proven
  by a real-PG16 crash-injection suite (loss = 0, effect-dup = 0).

## Public surface

- **`Replicant`** — the facade module. `start_link/1` starts a supervised
  streaming pipeline (validates opts, enforces the go-forward guard) and
  `stop/1` tears one down; `t:lsn/0` (a `non_neg_integer` 64-bit LSN,
  `(file <<< 32) ||| offset`) and `lsn_to_string/1` (uppercase `"file/offset"`
  hex display, matching Postgres `pg_lsn`).
- **`Replicant.Transaction`** — an assembled, committed transaction: ordered
  changes plus the transaction's single `commit_lsn`. Ordinarily `changes` is a
  `List`; for an oversized **spilled** streamed transaction (opt-in
  `streaming: [spill: [dir: …, max_spill_bytes: …]]`) it is a lazy, single-pass,
  disk-backed `Enumerable` (`Replicant.Spill.Reader`) valid only *during* the
  `handle_transaction/1` / `handle_batch/1` call — iterate it with `Enum`/`Stream`,
  never `length/1` / `Enum.to_list/1` (which force the whole transaction back into
  RAM, defeating spill), and do not retain it past the call. Spill emits value-free
  `[:replicant, :stream, :spilled]` (a tail flushed to disk) and, when disk usage
  would exceed `max_spill_bytes`, `[:replicant, :stream, :spill_exhausted]` before
  the fail-closed halt — attach to alert operators on exhaustion.
- **`Replicant.Change`** — a single row change (`insert`/`update`/`delete`),
  the decoded `record`, and the `unchanged` list of TOASTed columns the source
  UPDATE did not touch (never a value — sinks must leave those columns alone).
- **`Replicant.SchemaChange`** — a detected DDL-shape change (column add/drop,
  type change, replica-identity change). Destructive changes halt fail-closed.
- **`Replicant.Sink`** — the behaviour a consumer implements. In the default **sink-owned
  transaction mode** (`batch_delivery` not set), receives one `Replicant.Transaction` at a time
  and must durably persist it (or raise) before the slot advances past its `commit_lsn`. When
  **sink-owned batch delivery** is enabled (`batch_delivery: [max_transactions: N, max_delay_ms: T]`),
  receives N committed transactions in a single `handle_batch/1` call and must persist all rows
  + checkpoint atomically — the transaction is atomic and the effect-once guarantee (dup=0, loss=0)
  rests on it. Transactions arrive in ascending `commit_lsn` order; the sink must skip any
  `commit_lsn <= checkpoint` and upsert rows by table PK. A non-`{:ok, _}` return (or a
  raise/throw/exit) halts the pipeline fail-closed; the batch is discarded un-acked and
  re-delivered on resume, deduped to zero net effect by the idempotent sink. Optional snapshot
  callbacks (`handle_snapshot/2` + `handle_snapshot_complete/1`) let a sink be bootstrapped
  from a populated source: batches of `%Change{op: :snapshot}` rows (upsert by PK;
  clear the table when `first_for_table?` is true — a hard redo-safety obligation),
  then a durable handoff checkpoint at the snapshot's consistent point. In **lib mode**
  (`:checkpoint_store` configured) the library owns the checkpoint, so a sink implements
  only `handle_transaction/1` — `checkpoint/0` is optional there; its returned LSN is ignored.
  Batch delivery is mutually exclusive with lib mode.
- **`Replicant.Decoder`** — `decode/1` wraps the vendored `pgoutput` byte
  parser; catches and redacts any raise into a value-free `Replicant.Error`.
- **`Replicant.Assembler`** — groups decoded messages into
  `Replicant.Transaction`s by `commit_lsn`.
- **`Replicant.Connection`** — the `Postgrex.ReplicationConnection` that owns
  the replication slot: ack-after-checkpoint keepalive replies, async ack,
  slot-invalidation fail-closed halt, and the bounded in-flight window.
- **`Replicant.AssemblerServer`** — the serial process that applies the sink
  synchronously off the keepalive path.
- **`Replicant.Pipeline`** — the per-slot `:one_for_all` supervisor pairing a
  Connection with an AssemblerServer; started by `Replicant.start_link/1`.
- **`Replicant.Config`** — validates pipeline options + the start-mode guard
  (`go_forward_only` / `snapshot` / resume; `go_forward_only` + `snapshot` both
  set is refused as `:conflicting_start_mode`).
- **`Replicant.Snapshotter`** — reads a consistent snapshot of the publication's
  tables at the `EXPORT_SNAPSHOT` LSN (a `REPEATABLE READ` cursor on a separate
  connection) and pushes `%Change{op: :snapshot}` batches to the sink, behind a
  value-free error boundary; the `Connection` hands off to streaming at that LSN.
- **`Replicant.CheckpointStore`** — in **lib mode** (a `:checkpoint_store` option
  is present), a supervised GenServer owning one `replicant_checkpoints` row per slot:
  the library writes the checkpoint (`commit_lsn bigint`) to this durable Postgres table
  **after** the sink persists, so a non-transactional sink (files, S3, Kafka, external
  APIs) needs no atomic data+checkpoint unit. Value-free boundary; lazy table create +
  shape-probe. A store fault is bounded by two `:checkpoint_store` knobs — `max_retries`
  (default 5) and `retry_backoff_ms` (default 1000) — shared by both fault sites: a
  transient connect-read fault paces N fresh reconnects, a transient mid-stream write fault
  retries N (blocking the applier, so dup-bounded-to-one holds), then both **halt
  fail-closed** on exhaustion (`Supervisor.halt`, loss = 0). A permanent fault (schema
  mismatch / `:config_invalid`) halts immediately; `max_retries: 0` opts out of retry
  (halt-now). Each retry emits `[:replicant, :checkpoint_store, :retrying]`.
- **`Replicant.QueryBuilder`** — builds the identifier-validated SQL used to
  create/manage slots and publications.
- **`Replicant.Identifier`** — allowlist validation for slot, publication, and
  other identifiers that reach SQL.
- **`Replicant.Telemetry`** — value-free `:telemetry` spans (LSNs, table
  names, counts, durations, error classes — never row values).
- **`Replicant.Error`** — the typed, value-free error struct raised/returned
  at decode and validation boundaries.

## Non-negotiable rules

- **No row value in an error, log, or telemetry event.** Assume every value is
  PII or a secret. Column names are strings, never atoms (`String.to_atom` on
  a wide or attacker-influenced schema exhausts the atom table).
- **Validate identifiers.** Slot and publication names go through
  `Replicant.Identifier.validate/1` before reaching SQL. A failure carries the
  invalid-shape fact only, never the offending string.
- **Exactly-once is at-least-once + a transaction-watermark-idempotent sink.**
  The watermark is the commit LSN at transaction granularity — skip any
  transaction whose `commit_lsn <= checkpoint`; upsert rows by table PK.
  There is no naked exactly-once without two-phase commit or an idempotent
  sink; never claim one.
- **Lib mode is at-least-once, never effect-once.** With a `:checkpoint_store`, the
  library writes the checkpoint after the sink persists (checkpoint-after-persist), so a
  crash between persist and checkpoint re-delivers exactly one transaction on resume:
  **duplicate bounded to one transaction, never loss.** A non-transactional sink cannot
  dedup — do not claim effect-once for it.
- **Batching is opt-in and lib-mode only.** `checkpoint_store: [batch: [max_transactions: N, max_delay_ms: T]]`. It batches the checkpoint write + ack, NOT sink delivery (`handle_transaction/1` is still per-transaction). A crash or graceful stop mid-batch re-delivers up to one batch — size `max_transactions` for your dup tolerance. Do not set `:batch` at the top level (it belongs under `:checkpoint_store`; a misplaced top-level `:batch` is rejected at start).
- **Sink-owned atomic batch delivery (`handle_batch/1`) preserves effect-once.** The `batch_delivery: [max_transactions: N, max_delay_ms: T]` config (top-level, sink-owned only; mutually exclusive with `:checkpoint_store`) routes delivery through `handle_batch/1` instead of `handle_transaction/1`. The HARD OBLIGATION is that the data + checkpoint write is ATOMIC — the effect-once guarantee (dup=0 across mid-batch teardown) rests on it. Transactions arrive in ascending `commit_lsn` order; the sink must skip any `commit_lsn <= checkpoint` and upsert rows by table PK, exactly as `handle_transaction/1` does. A non-`{:ok, _}` return (or a raise/throw/exit) halts the pipeline fail-closed; the batch is discarded un-acked and re-delivered on resume, deduped to zero net effect by the idempotent sink.
- **Unchanged TOAST is a sentinel, not a value.** It surfaces only as
  `Replicant.Change`'s `unchanged` list of column names, never in `record`.
  Sinks must leave those columns untouched on upsert.
- **Stay tenant-blind.** No multitenancy, scope, or classification logic
  belongs here — that boundary is the whole reason `replicant` and
  `ash_replicant` are separate libraries.

See `AGENTS.md` for the full working rules.