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

Option validation and the fail-closed server-precondition gate.

Two clearly separated responsibilities:

  * `validate/1` is **pure** — it normalises the user's `start_link` options and
    refuses a mis-shaped or unsafe configuration before any socket is opened.
  * `check_preconditions/1` issues **one** `COM_QUERY` on an already-authenticated
    socket and refuses to start unless the source's binlog is configured for
    lossless row-based CDC.

Two additive C2 helpers share this fail-closed posture: `validate_snapshot/1` (pure —
normalises the `:snapshot` block, or `{:ok, nil}` when absent so the pure-C1 path is
byte-for-byte unchanged) and `read_server_uuid/1` (the source-identity read reused across
BOTH connections, design Q-src / Ch8).

## Server preconditions (ADR-0002, extended by ADR-0013)

`check_preconditions/1` reads six global variables in a single query and refuses
with a DISTINCT reason per violation — degraded row decoding silently guesses
column identity, so the gate fails closed rather than proceed:

  * `binlog_format` must be `ROW` — else `:binlog_format_not_row`
  * `binlog_row_image` must be `FULL` — else `:binlog_row_image_not_full`
  * `binlog_row_metadata` must be `FULL` — else `:binlog_row_metadata_not_full`
  * `binlog_row_value_options` must be empty (`""` = full JSON, not `PARTIAL_JSON`)
    — else `:binlog_row_value_options_not_empty`
  * `gtid_mode` must be `ON` — else `:gtid_mode_not_on`
  * `log_bin` must be enabled (text `"1"`) — else `:binlog_disabled`

`log_bin` (ADR-0013) closes a hole the five-variable gate cannot see on managed
sources: AWS documents that an Aurora cluster group set `binlog_format=OFF` "disables
the `log_bin` session variable … which in turn resets the `binlog_format` session
variable to the default value of `ROW`" — so a binlog-disabled Aurora still answers
`binlog_format=ROW`, and only `log_bin` names the condition before the dump. On a
self-managed server the same check catches `--skip-log-bin`. A boolean flag's text
form over the simple-query protocol is `"1"`/`"0"` (OBSERVED live on MySQL 8.0.46;
`SHOW VARIABLES` renders `ON`/`OFF`, `SELECT @@log_bin` does not), so the enabled
literal is `"1"` — anything else, including a future engine's rendering, refuses
value-free with the variable's name in the reason.

`binlog_transaction_compression` is deliberately NOT gated: compression is
source-unilateral (a consumer cannot opt out; MySQL 8.0.20+) and capstan
CONSUMES compressed transactions — the pure-Elixir zstd decoder inflates each
`TRANSACTION_PAYLOAD` event (ADR-0011's consume arm; `Capstan.Zstd`). A
malformed or non-ZSTD payload still halts fail-closed at decode time.

MySQL simple-query results are **all text strings**, so every value is compared as
text against the expected literal and never coerced to a typed term — an empty
`binlog_row_value_options` arrives as `""`, not `nil` or `0`.

## TLS verification posture (ADR-0002)

`ssl` defaults **true**. Peer verification is an explicit operator choice, never a
silent default: with TLS on, `ssl_opts` must carry EITHER a `cacertfile`/`cacerts`
OR an explicit `verify:`. Given neither, `validate/1` fails closed with
`:tls_verification_unspecified` rather than let OTP's `verify_peer` default select
a posture nobody chose. This mirrors the guard `Capstan.Protocol.Handshake` applies
at connect time, moved to config time so a bad TLS configuration is refused before
any socket is opened.

**Authenticated TLS against MySQL's auto-generated certificate.** A `cacertfile`
alone drives `verify: :verify_peer`, but MySQL's auto-generated server certificate
is self-signed with CN `…Auto_Generated_Server_Certificate` and no SAN, so
`verify_peer`'s hostname check fails against an address such as `127.0.0.1`. An
operator taking the authenticated route must ALSO pass
`server_name_indication: :disable` in `ssl_opts` — the chain is verified, the
hostname is not (VERIFY_CA semantics). `validate/1` does NOT inject this: that would
silently weaken every `cacertfile` user, so the operator supplies it deliberately.

# `batch_config`

```elixir
@type batch_config() :: %{
  max_transactions: pos_integer(),
  flush_ms: pos_integer(),
  mode: :lib_owned | :sink_owned
}
```

The normalised C3 batching configuration, or `nil` when absent (no batching).

# `precondition_error`

```elixir
@type precondition_error() ::
  :binlog_format_not_row
  | :binlog_row_image_not_full
  | :binlog_row_metadata_not_full
  | :binlog_row_value_options_not_empty
  | :gtid_mode_not_on
  | :binlog_disabled
  | :precondition_query_failed
```

A value-free precondition-gate refusal (ADR-0002, extended by ADR-0013).

# `snapshot_config`

```elixir
@type snapshot_config() :: %{
  tables: [{String.t(), String.t()}] | :all,
  store: {module(), keyword()},
  chunk_size: pos_integer()
}
```

The normalised initial-snapshot configuration (C2), or `nil` when `:snapshot` is absent
(pure C1). `tables` is the snapshot set (a concrete `{schema, table}` list, or `:all` when
it defaults to an `:all` capture — resolved to a concrete list at bootstrap). `store` is the
durable snapshot store `{module, start_link_options}`. `chunk_size` bounds the brief lock's
hold + buffered memory.

# `t`

```elixir
@type t() :: %{
  connection: keyword(),
  server_id: pos_integer(),
  max_command_retries: non_neg_integer(),
  reconnect_backoff: pos_integer(),
  heartbeat_period_ms: pos_integer(),
  stream_timeout_ms: pos_integer(),
  xa: :refuse | :track,
  max_prepared_transactions: pos_integer(),
  batch: batch_config() | nil
}
```

The normalised configuration `validate/1` returns on success.

# `validation_error`

```elixir
@type validation_error() ::
  :config_invalid
  | :server_id_required
  | :tls_verification_unspecified
  | :invalid_liveness_config
  | :unknown_option
```

A value-free option-validation refusal.

# `check_preconditions`

```elixir
@spec check_preconditions(Capstan.Protocol.Packet.socket()) ::
  :ok
  | {:error,
     precondition_error()
     | {:query_error, non_neg_integer()}
     | {:transport, term()}}
```

Reads the six server preconditions over `socket` and returns `:ok` iff all pass.

Issues ONE `COM_QUERY` on the already-authenticated socket and compares each value
as text (ADR-0002; `log_bin` per ADR-0013). A wrong variable refuses with its
distinct reason; a server or transport error is surfaced fail-closed, never
swallowed into a spurious `:ok`.

# `read_server_uuid`

```elixir
@spec read_server_uuid(Capstan.Protocol.Packet.socket()) ::
  {:ok, String.t()} | {:error, :server_uuid_read_failed}
```

Reads `@@server_uuid` over an already-authenticated `socket` — the source-identity primitive
(design Q-src / Ch8).

The bootstrap reads the STREAM connection's identity through this helper and pins the
`Capstan.Query` connection against it (`:expected_server_uuid`), so a query connection that
silently reconnects to a DIFFERENT replica mid-backfill is caught (`:snapshot_source_mismatch`)
— `@@server_uuid` is compared across BOTH connections. Returns `{:ok, uuid}` (a value-free
structural identity string) or the value-free `{:error, :server_uuid_read_failed}`; a
transport/query fault is scrubbed to the bare reason, never the raw term (Rule 1).

# `validate`

```elixir
@spec validate(keyword()) :: {:ok, t()} | {:error, validation_error()}
```

Validates raw `start_link` options into a normalised config map, or returns a
value-free error.

Refuses: `:server_id_required` (missing or non-positive `server_id`),
`:tls_verification_unspecified` (`ssl: true` with no CA source and no explicit
`verify:`), `:invalid_liveness_config` (`stream_timeout_ms <= heartbeat_period_ms`,
compared after defaults are applied — a window at or below the heartbeat interval would
false-drop a healthy idle stream), `:unknown_option` (a key outside the documented
surface — top level, `connection:`, `snapshot:`, or a store block — never silently
defaulted), and `:config_invalid` (any other missing or mis-shaped option, including a
liveness value beyond the schedulable timer ceiling).
Defaults applied: `ssl` true, `ssl_opts` `[]`,
`auth_plugins` `[:caching_sha2_password]`, `password` `""`, `database` `nil`,
`max_command_retries` `5`, `reconnect_backoff` `1_000`, `heartbeat_period_ms` `15_000`,
`stream_timeout_ms` `60_000`.

# `validate_snapshot`

```elixir
@spec validate_snapshot(keyword()) ::
  {:ok, snapshot_config() | nil} | {:error, :config_invalid | :unknown_option}
```

Normalises the initial-snapshot `:snapshot` block, or `{:ok, nil}` when it is absent (pure C1).

An absent `:snapshot` key returns `{:ok, nil}` and touches nothing — the pure-C1 path is
byte-for-byte unchanged. A present block normalises to a `snapshot_config()`:

  * `tables` — the snapshot set. Defaults to the capture allowlist (`:tables`, itself `:all`
    by default). When given it MUST be a non-empty list of `{schema, table}` binary tuples;
    the `⊆ captured` check is `Capstan.Pipeline.validate_snapshot_tables/2`.
  * `store` — REQUIRED, shaped like `checkpoint_store`: `[module: impl, options: keyword()]`.
    A missing or mis-shaped store fails closed `:config_invalid` (a half-configured backfill
    would silently lose its resumability durability).
  * `chunk_size` — a POSITIVE integer, default `4096`.

Any mis-shaped field fails closed with the value-free `:config_invalid` (the generic
mis-shaped-option refusal, as elsewhere in this module); an unrecognized key in the
block or its store refuses `:unknown_option`.

---

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