# `AshReplicant.Sink.Impl`
[🔗](https://github.com/baselabs/ash_replicant/blob/v0.3.0/lib/ash_replicant/sink/impl.ex#L1)

The config-parameterized sink implementation the `use AshReplicant.Sink`
macro delegates to. `handle_transaction/2` is the effect-once core: one
`Repo.transaction` wrapping {dedup-check → single-pass apply → checkpoint
upsert}, value-free and fail-closed.

Mirrored writes run inside the host transaction and pass
`return_notifications?: true`, so Ash bundles any notification into the return
value (never firing a notifier) and the sink discards it — Ash notifiers/pubsub
do NOT fire for mirrored changes. (Ash cannot emit notifications from within a
host-managed transaction anyway; `notify?: false` is not honored by
single-record create/destroy in Ash 3.x, only `return_notifications?`.)

# `checkpoint`

```elixir
@spec checkpoint(map()) :: {:ok, Replicant.lsn() | nil} | {:error, term()}
```

Last durably-persisted commit LSN for the slot (`nil` = never), the dedup watermark.

# `handle_schema_change`

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

Accept or decline a schema change. An `:additive` change auto-applies; a
`:destructive` change on a resource whose `on_schema_change` is
`:halt_destructive` (default) halts fail-closed. The context map is not
value-inspected. Unmapped tables use the behaviour default.

# `handle_snapshot`

```elixir
@spec handle_snapshot(map(), [Replicant.Change.t()], map()) :: :ok | {:error, term()}
```

Persist a snapshot batch for `ctx.table`, upserting by PK. On
`first_for_table?`, clear the resource's mirror rows in-txn first (redo-safety).
Plain SCD1 non-tenant, non-sensitive resources use a bulk upsert; the
load-bearing fail-closed guard is the `case result.status` check — anything
other than `:success` (including the default-options `:partial_success`) rolls
the snapshot transaction back, so a failing row is never silently dropped.
`stop_on_error?: true` is a defensible early-stop on top of that, not the loss
guard. Sensitive, tenant-scoped, OR SCD2 resources apply per-record — SCD2
stamps the batch's snapshot LSN onto each change so each version opens at
`valid_from_lsn = snapshot_lsn`. Does not advance the checkpoint.

This sink implements replicant's v1 snapshot only (no `snapshot_progress/0`
callback). The whole-resource `first_for_table?` clear is correct under v1
because the snapshot runs as a separate phase before the stream starts
(EXPORT_SNAPSHOT -> COPY -> START_REPLICATION at the consistent_point), so
there are no concurrent `handle_transaction` rows to wipe when the clear
runs. If/when this sink adopts replicant's incremental snapshot mode
(`snapshot: [mode: :incremental]`, which requires implementing
`snapshot_progress/0` and interleaves snapshot chunks with the live stream),
this clear must change to preserve stream-applied rows (clear only
snapshot-origin rows) — otherwise a stream update that lands before the
first chunk closes is lost (replicant incremental "Bug C", proven by the
replicant marquee 2026-07-10).

# `handle_snapshot_complete`

```elixir
@spec handle_snapshot_complete(map(), Replicant.lsn()) ::
  {:ok, Replicant.lsn()} | {:error, term()}
```

Durably set `checkpoint := snapshot_lsn` and return it (the snapshot handoff commit).

# `handle_transaction`

```elixir
@spec handle_transaction(map(), Replicant.Transaction.t()) ::
  {:ok, Replicant.lsn()} | {:error, term()}
```

Persist the transaction's changes AND the checkpoint atomically; skip if
`commit_lsn <= checkpoint`. Returns `{:ok, commit_lsn}` or a value-free
`{:error, %AshReplicant.Error{}}` (the pipeline halts fail-closed and
re-delivers on resume).

`Apply.apply_change/3` RAISES on failure, so a failing change propagates out of
`Repo.transaction` (Ecto rolls back, then re-raises) and lands on the outer
`rescue` — NOT the `{:error, _}` branch of the result match. Both halt paths
route through `halt/2`, so `:halted` telemetry fires on the real raise path too.

---

*Consult [api-reference.md](api-reference.md) for complete listing*