# `Capstan.Assembler`
[🔗](https://github.com/baselabs/capstan/blob/v1.3.1/lib/capstan/assembler.ex#L1)

The three-terminator fold — a **pure** assembler over a `%Capstan.Binlog.Event{}`
stream (ADR-0003; the owning GenServer is `Capstan.AssemblerServer`).

The fold turns raw binlog events into the committed units a `Sink` consumes:
`%Capstan.Transaction{}` for DML, `%Capstan.SchemaChange{}` for self-committing
DDL. It threads a running `Capstan.Position` (the processed GTID set) from a
caller-supplied start, and a `Capstan.Binlog.TableRegistry` built as `TABLE_MAP`
events arrive.

## Why it folds `%Event{}`, not decoded terms

`Capstan.Binlog.Decoder.decode/1` returns only the per-type body term and **drops**
the event header. But a `Transaction`'s `commit_ts` is the terminator event's
header `timestamp` and its diagnostic `pos` is the header `log_pos`. So the fold
decodes each `%Event{}` via `Decoder.decode/1` for the body term **and** reads the
header off the same struct for `timestamp`/`log_pos`.

## The single biggest failure mode this fold is built AROUND (risk-first)

A **phantom / rolled-back transaction delivered effect-once** — the silent one.
Two shapes of it, both defended structurally rather than only by convention:

  * An `XA_PREPARE` (`Decoder` `{:halt, :unsupported_transaction_shape}`): its row
    images may later be `XA ROLLBACK`-ed, so the rows accumulated so far in that
    transaction must be **discarded**, not delivered (ADR-0003). The halt is keyed
    on the `Decoder`'s type-byte signal, never re-derived here.
  * Any mid-transaction failure — an unmapped `table_id`, a `Rows.decode/2` error,
    or a `Decoder` `{:error, _}` — aborts **fail-closed**. A partial transaction
    carrying only the rows decoded before the failure is never emitted.

Both are guaranteed by the fold's shape: rows accumulate in the in-flight
transaction buffer and are released **only** when a terminator is reached. A
`{:halt, _}`/`{:error, _}` short-circuits with that buffer still un-released, so it
is structurally impossible to leak. The happy path (row → `%Change{}`) is secondary.

Stream **desyncs** fail closed the same way rather than being silently absorbed: a
GTID arriving while a transaction is still open (`:gtid_within_open_transaction`),
or a QUERY / row / terminator with no open GTID scope (`:begin_without_gtid`,
`:query_without_gtid`, `:rows_without_transaction`, `:terminator_without_transaction`)
— each aborts rather than dropping or overwriting an open buffer, since a silently
absorbed desync can mask exactly the loss this fold exists to prevent.

## The three terminators (ADR-0003) — anchored on the proven `fixture_capture`
`classify`/`apply_event` precedent so a terminator is never misclassified (itself a
silent-loss risk: a `COMMIT` read as DDL would drop the transaction's rows):

  * `XID` — a transactional (InnoDB) commit.
  * `QUERY("COMMIT")` — a non-transactional (MyISAM) commit; the `COMMIT` is the
    terminator, not a change.
  * a self-committing DDL `QUERY` — a `QUERY` that is **not** `BEGIN`/`COMMIT` and
    arrives with **no open `BEGIN`**. Yields a `%SchemaChange{}` and advances the
    position. `BEGIN` opens a DML block; `COMMIT` closes it.

## Fully-filtered transactions keep the watermark moving (ADR-0003)

The checkpoint is a **processed-GTID watermark**, so a committed transaction whose
changes are all filtered (or which has none) still advances the position and is
delivered as a `%Transaction{}` with **empty** `changes`. `Capstan.AssemblerServer`
checkpoints it
without calling the sink. A long run of filtered transactions therefore never
stalls the position.

`Capstan.Config` carries no table filter in C1, so the filter is a **parameter**:
`tables: :all` (default — nothing filtered) or an allowlist of `{schema, table}`
pairs, applied **before** row decode.

## Rule 1

No `%Change{}` value, `%SchemaChange{}` DDL text, or error tuple ever carries a row
value or statement literal. DDL is reduced to `schema`/`table`/`kind` here; the raw
SQL is inspected only to classify and is then dropped.

## Interface

  * `new/2` + `step/2` + `position/1` — the incremental primitive the owning
    GenServer (`Capstan.AssemblerServer`) drives to checkpoint per transaction.
  * `run/3` — folds a whole event sequence, returning every output plus the final
    watermark; the batch entry the tests drive.

# `filter`

```elixir
@type filter() :: :all | MapSet.t({String.t(), String.t()})
```

The table filter: everything, or an allowlist of `{schema, table}` pairs.

# `output`

```elixir
@type output() :: Capstan.Transaction.t() | Capstan.SchemaChange.t()
```

An emitted committed unit.

# `run_result`

```elixir
@type run_result() ::
  {:ok, [output()], Capstan.Position.t()}
  | {:halt, xa_halt(), [output()], Capstan.Position.t()}
  | {:error, term(), [output()], Capstan.Position.t()}
```

The result of `run/3`. On a `{:halt, _}`/`{:error, _}` short-circuit the
already-COMMITTED outputs (and the watermark up to the failure) are still returned —
a transaction that reached its terminator before the halt is genuinely committed and
must not be lost, only the in-flight one is discarded.

# `step_result`

```elixir
@type step_result() ::
  {:cont, [output()], t()} | {:halt, xa_halt()} | {:error, term()}
```

The result of `step/2`.

# `t`

```elixir
@type t() :: %Capstan.Assembler{
  file: String.t() | nil,
  filter: filter(),
  gtid_set: Capstan.Gtid.t(),
  max_prepared: pos_integer(),
  pos: non_neg_integer() | nil,
  prepared: %{optional(Capstan.Xa.Id.t()) => pooled()},
  registry: Capstan.Binlog.TableRegistry.t(),
  startup_xids: MapSet.t(),
  txn: inflight() | nil,
  xa: xa_policy()
}
```

# `xa_halt`

```elixir
@type xa_halt() ::
  :unsupported_transaction_shape
  | :xa_prepared_pool_exhausted
  | :xa_prepared_gtid_mismatch
  | :xa_commit_without_prepare
  | :xa_rollback_without_prepare
```

The XA-related fail-closed halts (`step/2` / `run/3`).

# `xa_policy`

```elixir
@type xa_policy() :: :refuse | :track
```

# `new`

```elixir
@spec new(
  Capstan.Position.t(),
  keyword()
) :: t()
```

Builds a fresh assembler state resuming from `start_position`.

`opts` accepts `tables: :all` (default) or `tables: [{schema, table}, ...]` — the
allowlist of tables whose rows are delivered (everything else is filtered before row
decode) — plus the XA policy (ADR-0006): `xa: :refuse` (default — a TWO-PHASE
`XA_PREPARE` halts `:unsupported_transaction_shape`, the C1 posture; a `one_phase:
true` prepare is an ordinary atomic commit and is DELIVERED in both modes — the one
deliberate deviation from C1's unconditional type-38 halt, per ADR-0006 §2) or
`xa: :track`; `max_prepared_transactions:` (positive integer, default
10000) bounds the in-memory prepared pool (halt
`:xa_prepared_pool_exhausted` — never evict, eviction is the silent-loss class); and
`startup_xids:` (a list of `Capstan.Xa.Id` digests from the connect-time `XA RECOVER`)
pre-seeds the dangling-prepare discriminator so a pre-start prepare's resolution is a
correct row-less watermark advance.

# `position`

```elixir
@spec position(t()) :: Capstan.Position.t()
```

The running processed-GTID watermark as a `Capstan.Position`.

`Capstan.AssemblerServer` reads this after a `step/2` to checkpoint a transaction — including a
self-committing DDL or a fully-filtered transaction, whose position advances even
though the emitted output carries none (or an empty `changes`).

# `run`

```elixir
@spec run([Capstan.Binlog.Event.t()], Capstan.Position.t(), keyword()) :: run_result()
```

Folds an entire event sequence from `start_position`.

Returns `{:ok, outputs, final_position}` when every event is consumed. On the first
`{:halt, _}`/`{:error, _}` a `step/2` produces, returns `{:halt, reason, committed,
watermark}` / `{:error, reason, committed, watermark}` — the transactions that
already terminated before the failure are genuinely committed and returned; only the
in-flight (e.g. XA-prepared) transaction is discarded. `opts` is as for `new/2`.

# `step`

```elixir
@spec step(t(), Capstan.Binlog.Event.t()) :: step_result()
```

Folds one `%Capstan.Binlog.Event{}` into the state.

Returns `{:cont, outputs, state}` (`outputs` is `[]` for every event except a
terminator), or short-circuits fail-closed with `{:halt,
:unsupported_transaction_shape}` or `{:error, reason}` — in either case the
in-flight transaction is discarded and nothing is emitted (see the module doc).

---

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