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

Pipeline entry — the per-mode `Sink` validation and the child-spec wiring for one
supervised pipeline (a `Capstan.CheckpointStore` process, a `Capstan.AssemblerServer`,
and a `Capstan.Connection`).

`Capstan.Supervisor` starts these children in order and threads the pids, because the
`Connection`'s `:receiver` must be the `AssemblerServer`'s PID. The
`Connection` forwards frames with a plain `send/2`, so the receiver is a **PID**: a
send to a terminated `AssemblerServer` is a silent no-op, never a raise that could
restart a fail-closed pipeline into a livelock. The `Connection` also **monitors** that
PID, so the `AssemblerServer`'s fail-closed halt (which stops it without messaging back)
is not silent — the `Connection` stops fail-closed on the `:DOWN` instead of streaming
into the dead pid. Monitoring detects the death without linking, so neither `:temporary`
child restarts into the livelock the plain-`send` avoids.

## Per-mode `Sink` callback required-ness

`Capstan.Sink` declares every callback `@optional_callbacks`; which are REQUIRED
depends on the checkpoint mode, enforced here at start-up via `function_exported?/3` so
a sink missing a required callback is refused **before** a first-call crash:

  * **lib-owned** (`:checkpoint_store` configured): `handle_transaction/1`.
  * **sink-owned** (no `:checkpoint_store`): `checkpoint/0` and `handle_transaction/1`.
  * `handle_schema_change/2` whenever DDL delivery is enabled — in C1 the pipeline
    ALWAYS delivers self-committing DDL (ADR-0003), so it is required in both modes.

Both modes run (C1a): lib-owned wires a store child; sink-owned validates
`checkpoint/0` here and the `AssemblerServer` seeds from it — no store child.

# `coordinator_wiring`

```elixir
@type coordinator_wiring() :: [
  assembler: pid() | atom(),
  snapshot_store: {module(), term()},
  snapshot_state: Capstan.Snapshot.State.t(),
  readers: %{optional({String.t(), String.t()}) => term()},
  processed_set: String.t()
]
```

The bootstrap-provided coordinator wiring (the supervisor supplies these AFTER the
assembler + snapshot-store children are up): the started `AssemblerServer` pid the coordinator
sends `{:capstan_halt, _}` to and observes the watermark from, the started `{impl, handle}`
snapshot store, the initial `%Capstan.Snapshot.State{}`, the per-table opened `ChunkReader`
handles, and the initial processed-watermark string (default `""`).

# `sink_error`

```elixir
@type sink_error() ::
  :invalid_sink
  | :sink_missing_handle_transaction
  | :sink_missing_checkpoint
  | :sink_missing_handle_schema_change
  | :sink_missing_handle_snapshot
  | :sink_missing_handle_batch
```

A value-free sink-validation refusal.

# `assembler_spec`

```elixir
@spec assembler_spec(
  keyword(),
  Capstan.AssemblerServer.checkpoint_store() | nil
) :: Supervisor.child_spec()
```

The `AssemblerServer` child spec, wired to the started checkpoint store `{impl, handle}`.

# `batch_mode?`

```elixir
@spec batch_mode?(keyword()) :: nil | :lib_owned | :sink_owned
```

The configured batch mode: nil (no batching), :lib_owned, or :sink_owned.

# `checkpoint_store`

```elixir
@spec checkpoint_store(keyword()) :: {module(), keyword()}
```

The lib-owned checkpoint store's implementation module and its `start_link/1` options,
drawn from `checkpoint_store: [module: impl, options: keyword()]`.

# `connection_spec`

```elixir
@spec connection_spec(
  keyword(),
  pid(),
  Capstan.Position.t() | nil,
  Capstan.AssemblerServer.checkpoint_store() | nil
) :: Supervisor.child_spec()
```

The `Connection` child spec, wired so its `:receiver` is the `AssemblerServer` PID, its
dump resumes from `start_position`, and — given the started checkpoint store
`{impl, handle}` — it re-reads the durable resume position on every establish (so a
reconnect resumes from the current watermark, not the frozen start-up position; design
Q7 / F1). `checkpoint_store` defaults to `nil` so a `Connection` wired without a store
(a unit test) keeps the injected `start_position`.

# `coordinator_spec`

```elixir
@spec coordinator_spec(
  keyword(),
  coordinator_wiring()
) :: Supervisor.child_spec()
```

The `Capstan.Snapshot.Coordinator` child spec (snapshot mode only), mirroring `assembler_spec/2`.

The real (downstream) sink + the retry budget come from the pipeline `opts`; `wiring` carries
the bootstrap-produced pieces (see `t:coordinator_wiring/0`) — the supervisor builds it
once the assembler + snapshot-store children have started. A `:temporary` child: a fail-closed
snapshot halt never restarts.

# `lib_mode?`

```elixir
@spec lib_mode?(keyword()) :: boolean()
```

Is this a lib-owned pipeline? True iff a non-nil `:checkpoint_store` is configured.

# `resolve_start_position`

```elixir
@spec resolve_start_position(
  keyword(),
  Capstan.Position.t() | nil
) ::
  {:ok, Capstan.Position.t() | nil}
  | {:ok, :current, nil}
  | {:error, :config_invalid}
```

Resolve the public `:start_position` (default `:checkpoint`) against the position the
authority (checkpoint store / sink checkpoint) resumed with.

  * `:checkpoint` — the resumed position (a `%Position{}` or `nil` for a fresh start);
  * a `%Capstan.Position{}` — the override itself (C1b): BOTH the dump and the
    assembler watermark seed from it, so the two can never disagree;
  * `:current` — `{:ok, :current, nil}` marker (C1b): the caller resolves it against
    the server's live `@@global.gtid_executed` over an authenticated socket, pre-dump;
  * anything else — `{:error, :config_invalid}`.

# `snapshot_mode?`

```elixir
@spec snapshot_mode?(keyword()) :: boolean()
```

Is this an initial-snapshot pipeline? True iff a non-nil `:snapshot` block is configured
(mirrors `lib_mode?/1`). An absent (or `nil`) `:snapshot` key ⇒ pure C1.

# `snapshot_store_spec`

```elixir
@spec snapshot_store_spec(
  module(),
  keyword()
) :: Supervisor.child_spec()
```

The durable `Capstan.SnapshotStore` child spec (snapshot mode only), mirroring `store_spec/2`.

A `:temporary` child — a snapshot-store fault halts fail-closed, never restarts into a
re-scan-from-zero livelock.

# `store_spec`

```elixir
@spec store_spec(
  module(),
  keyword()
) :: Supervisor.child_spec()
```

The checkpoint-store child spec (a `:temporary` child — a store fault halts, never restarts).

# `validate_sink`

```elixir
@spec validate_sink(keyword()) :: :ok | {:error, sink_error()}
```

Validate the configured `:sink` against its checkpoint mode's required callbacks.

Returns `:ok`, or a distinct value-free refusal naming the missing callback. Every
check is a runtime `function_exported?/3` (the callbacks are `@optional_callbacks`).

# `validate_snapshot_tables`

```elixir
@spec validate_snapshot_tables(
  keyword(),
  Capstan.Config.snapshot_config() | nil
) :: :ok | {:error, :snapshot_table_not_captured}
```

Refuse a snapshot table outside the capture allowlist (`:snapshot_table_not_captured`).

Every snapshot table MUST be captured by the stream: the stream is what delivers the
`gtid > G` changes the cursor-gate suppresses from the chunk, so a snapshot table the stream
does not carry would silently gap. `:all` capture trivially includes every snapshot table.
`snapshot` is the normalised `Capstan.Config.snapshot_config()` (or `nil` in pure C1 — a
no-op `:ok`).

---

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