# `Capstan.Snapshot.CursorGate`
[🔗](https://github.com/baselabs/capstan/blob/v1.3.1/lib/capstan/snapshot/cursor_gate.ex#L1)

The pure strict-once classification core of the initial snapshot (C2, design §Architecture).

While a table backfills, the C1 stream keeps running. Every streamed `%Change{}` on a
snapshot-active table must be routed so each pre-existing key is delivered EXACTLY ONCE —
once by the stream (for keys already backfilled) or once by a chunk (for keys not yet
backfilled), never both, never neither. This module is the pure decision function that
routes them; it holds no state and does no I/O.

## The gate (`classify/3`)

Given a streamed change, the table's per-table `cursor` (the canonical PK of the last
backfilled row, or `:start` before the first chunk), and the table's `pk_columns` /
`pk_types` / `complete?`, each row-image is classified per its canonical PK `k`:

  * **forward** iff `k ≤ cursor` (already backfilled) OR the table is `complete?` — the
    stream is authoritative for these keys;
  * **suppress** otherwise (`k > cursor`) — a not-yet-backfilled key whose future chunk
    will deliver it. Forwarding it here would double-deliver (stream + chunk).

A `:delete` is the ONE asymmetric case: it gates on `delivered_pk` (the high-water the sink has
RECEIVED) rather than `pk_cursor` (the re-read floor). The two are equal in steady state, so
suppression is unchanged; they diverge only across the emit→cursor-persist crash window, where
`delivered_pk` sits ahead of the rolled-back `pk_cursor`. Forwarding a delete of an
already-delivered key (`k ≤ delivered_pk`) then sweeps the row rather than leaving a permanent
phantom — the re-read chunk, taken as-of a fresh `G` with the row already gone,
would otherwise omit it while the suppressed delete never reached the sink. See the `table_spec`
doc; inserts/updates keep gating on `pk_cursor` so strict-once is preserved.

`k` is derived from the change's record via `Capstan.Snapshot.PrimaryKey.canonical/2`
(the raw column values in PK ordinal order) and compared with `PrimaryKey.compare/2` — the
only ordering the cursor-gate is allowed to use, restricted to order-faithful PK types so
the Elixir comparison provably matches MySQL `ORDER BY` (Ch4).

`classify/3` returns the **list of surviving forward-images** (0, 1, or 2), which the
coordinator folds into the transaction's `changes` and forwards to the real sink.
A suppressed change yields `[]`.

## PK-changing UPDATE split (Ch2, tripwire 17)

A binlog `%Change{op: :update}` that moves a row's PK carries `old_record` (old PK
`k_old`) and `record` (new PK `k_new`) with `k_old ≠ k_new`. Gating such an update on a
single key is wrong when the two straddle the cursor, so it is **split into
`delete(k_old)` + `upsert(k_new)`, each gated on its own key** — the canonical
DELETE(old) + INSERT(new) decomposition of a PK move. Both images carry the FULL
after/before-image (ADR-0002), so each is self-sufficient:

  * `delete(k_old)` — `%Change{op: :delete, old_record: old_record, record: nil}`.
  * `upsert(k_new)` — `%Change{op: :insert, record: record, old_record: nil}`. It is an
    insert because a PK-changing UPDATE makes `k_new` a brand-new key; the sink applies it
    as **upsert-by-PK** (the HARD C2 sink precondition), so it converges even if `k_new`'s
    slot was already backfilled.

Straddle (a) `k_old ≤ cursor < k_new` → forward the delete, suppress the upsert (its
future chunk delivers `k_new`) → no phantom. Straddle (b) `k_new ≤ cursor < k_old` →
suppress the delete (`k_old`'s chunk never ran, nothing was emitted there), forward the
upsert → no gap. A non-PK-mutating update (`k_old = k_new`) is ONE image, unchanged.

## The advance-gate predicate (`advance?/2`)

A buffered chunk read as-of an exact GTID position `G` may only be emitted (and its cursor
advanced) once the stream's processed watermark covers `G`:
`Gtid.subset?(Gtid.parse(g), processed_set)`. This is the ordering that makes suppression
correct — the cursor cannot advance to include a key until every `gtid ≤ G` has been
processed by the stream. All GTID-set math routes through `Capstan.Gtid` (ADR-0001); no
hand-rolled interval arithmetic lives here.

## Rule 1

Pure functions, no I/O: nothing here logs or telemeters a PK value. Canonical PKs and row
values live only in the returned `%Change{}` images (whose `Inspect` already elides the
value maps) and in the caller's in-memory state.

# `cursor`

```elixir
@type cursor() :: Capstan.Snapshot.PrimaryKey.canonical_pk() | :start
```

The per-table backfill cursor: a canonical PK, or `:start` before the first chunk.

# `table_spec`

```elixir
@type table_spec() :: %{
  :pk_columns =&gt; [String.t()],
  :pk_types =&gt; [Capstan.Snapshot.PrimaryKey.pk_type()],
  :complete? =&gt; boolean(),
  optional(:delivered_pk) =&gt; cursor(),
  optional(:weights) =&gt; %{
    optional(String.t()) =&gt; %{optional(binary()) =&gt; binary()}
  },
  optional(atom()) =&gt; term()
}
```

The per-table gate context: the introspected PK shape (`pk_columns` in ordinal order and
their order-faithful `pk_types`), whether the table's backfill is `complete?`, the
optional `delivered_pk` — the high-water the sink has RECEIVED, used as the DELETE threshold
(see the module doc's crash-window section) — and the optional `weights` —
`%{column_name => %{raw => weight_bytes}}` for the STRING pk columns (ADR-0012), resolved
by the coordinator over the shared query connection before gating. When `delivered_pk` is
absent, deletes gate on the `cursor` argument (pre-F1 behaviour). Extra keys are ignored, so
the coordinator may pass a richer per-table state map.

# `advance?`

```elixir
@spec advance?(String.t(), Capstan.Gtid.t() | String.t()) :: boolean()
```

The advance-gate predicate: is the chunk's exact GTID position `g` covered by
`processed_set`? True iff `Gtid.subset?(Gtid.parse(g), processed_set)`.

`g` is always a canonical GTID-set string. `processed_set` may be either an already-parsed
`Capstan.Gtid.t()` (the watermark-observer feed) or a string (parsed here). No interval
math is hand-rolled — the containment decision is `Capstan.Gtid`'s.

# `classify`

```elixir
@spec classify(Capstan.Change.t(), cursor(), table_spec()) :: [Capstan.Change.t()]
```

Classifies one streamed `%Change{}` against the table's `cursor` and `table_spec`,
returning the list of surviving forward-images (`[]`, one, or — for a straddling
PK-changing UPDATE that forwards both halves — two).

An `:insert` is gated on its `record` key, a `:delete` on its `old_record` key, and an
`:update` on its (equal) key unless it moves the PK, in which case it is split into
`delete(k_old)` + `upsert(k_new)` and each half is gated on its own key (see the module
doc). A `:snapshot` change never reaches the gate (the gate classifies streamed images
only) and raises a value-free `ArgumentError` — a loud, fail-closed misuse signal. It is an
EXPLICIT clause rather than a `FunctionClauseError` on purpose (Rule 1): a no-clause-match
error captures the call args — including the `cursor`, a canonical PK / user data — in its
stacktrace frame, whereas the explicit clause leaves the frame at arity 3 (no arg values).

---

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