@callback checkpoint() :: {:ok, Replicant.lsn() | nil} | {:error, term()}

Last durably-persisted commit LSN, for resume AND as the dedup watermark.

@callback handle_batch([Replicant.Transaction.t()]) ::
  {:ok, Replicant.lsn()} | {:error, term()}

Optional. Deliver a BATCH of committed transactions as ONE atomic unit. Enabled by the top-level batch_delivery config (sink-owned mode only). transactions arrives in ascending commit_lsn order; persist ALL of their rows AND checkpoint := the last (highest) commit_lsn in ONE atomic database transaction (or, at minimum, checkpoint-after-all-persist), returning that LSN. Idempotency is unchanged: skip any commit_lsn <= checkpoint; upsert rows by PK.

HARD OBLIGATION: the data + checkpoint write is atomic. The effect-once guarantee (dup = 0 across a mid-batch teardown) rests on it, exactly as handle_snapshot/2's redo-safety rests on first_for_table?. 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).

@callback handle_message(Replicant.Decoder.Messages.Message.t(), context) ::
  :ok | {:ok, term()} | {:error, term()}
when context: %{lsn: Replicant.lsn()}

Optional. Deliver a NON-TRANSACTIONAL logical-decoding message (pg_logical_emit_message with transactional => false). Such a message arrives STANDALONE (no Begin/Commit bracket) and has no PK or commit_lsn dedup key, so the guarantee is at-least-once, duplicates possible on a reconnect between the {:ok} return and the checkpoint advancing — mirroring the lib-mode handle_transaction/1 downgrade. A transactional message (transactional => true) does NOT route here: it rides %Transaction.messages and inherits the txn path's effect-once.

On :ok / {:ok, _} the library advances the checkpoint to the message's LSN; a non-:ok return (or a raise/throw/exit) halts the pipeline fail-closed. context carries %{lsn: lsn}. Enabled by the top-level messages: true config; a sink missing this callback is rejected at start (:messages_unsupported).

@callback handle_schema_change(Replicant.SchemaChange.t(), map()) ::
  :ok | {:error, term()}

Optional. When a sink does not implement this, the Assembler (Task 13) applies the default: an :additive change auto-applies; a :destructive change halts the pipeline fail-closed. Implement it to accept (:ok) or decline ({:error, _}) a Replicant.SchemaChange.

@callback handle_snapshot([Replicant.Change.t()], context) :: :ok | {:error, term()}
when context: %{
       :snapshot_lsn => Replicant.lsn(),
       :table => String.t(),
       :first_for_table? => boolean(),
       optional(:progress) => binary() | nil,
       optional(:backfill_complete?) => boolean()
     }

Optional. Persist a batch of snapshot (backfill) rows for context.table — each a %Replicant.Change{op: :snapshot} — upserting by table PK.

This callback serves BOTH snapshot modes with mode-conditional obligations (spec §6.1 — normative):

In BOTH modes, the library guarantees at least one call per publication table (even a zero-row table), so first_for_table? always fires and the redo-safety reset always happens. A :state_mirror sink MUST distinguish snapshot-origin rows from stream-origin rows for that reset: first_for_table? clears only the SNAPSHOT-loaded rows and must PRESERVE rows the live stream (handle_transaction/1) applied — under frontier ordering a stream update can land BEFORE the first chunk closes, and a blanket clear would lose it. context.snapshot_lsn is the snapshot's consistent point.

@callback handle_snapshot_complete(snapshot_lsn :: Replicant.lsn()) ::
  {:ok, Replicant.lsn()} | {:error, term()}

Optional. The snapshot handoff commit, called once after all batches. Durably persist checkpoint := snapshot_lsn and return it. Until this succeeds the checkpoint stays nil, so a crash before it re-runs the whole snapshot. A non-{:ok, _} return (or a raise/throw/exit) halts the pipeline fail-closed.

@callback handle_transaction(Replicant.Transaction.t()) ::
  {:ok, Replicant.lsn()} | {:error, term()}

Persist the whole transaction AND its checkpoint atomically; return the commit LSN. Idempotency: skip if txn.commit_lsn <= checkpoint(); upsert rows by PK.

@callback sink_kind() :: :state_mirror | :append_log

Optional (default :state_mirror). :append_log sinks receive appends.

@callback snapshot_progress() :: {:ok, binary() | nil} | {:error, term()}

Optional (REQUIRED for sink-owned incremental snapshot — Config gates on it). Return the most recent context.progress token persisted atomically with a chunk (nil = no backfill in progress / never started). Read ONCE at pipeline start for resume. A read fault halts fail-closed — the resume decision is not dedup-recoverable (spec §6.1/§8).