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

A `COM_QUERY`-only connection for the initial-snapshot path (C2).

`Capstan.Connection` is binlog-dump-only — once it issues `COM_BINLOG_DUMP_GTID` the
socket carries the replication stream and cannot run a `SELECT`. The snapshot needs a
SEPARATE connection to read `information_schema`, capture each chunk's exact GTID under a
brief `LOCK TABLES … READ`, and page the chunk rows. `Capstan.Query` is that connection: it
reuses the C1 handshake / auth / TLS stack (`Capstan.Protocol.Handshake`) and the C1
text-resultset decoder (`Capstan.Protocol.Command`), and exposes the query primitives the
`ChunkReader`, `PrimaryKey` introspection, and the bootstrap `P0` read all build on.

It owns no process. A `%Capstan.Query{}` is a connection HANDLE — an authenticated passive
socket plus the pinned source identity — that a caller (the `ChunkReader`, the bootstrap)
drives synchronously, exactly as `Capstan.Config.check_preconditions/1` drives a socket.

## Source identity (design Ch8)

A query connection that silently (re)connects to a DIFFERENT replica mid-backfill corrupts
the exact-`G` pairing the whole snapshot rests on (the chunk's captured GTID would name a
position on a server the chunk rows did not come from). So on EVERY (re)establish the
`@@server_uuid` is read and compared to the pinned value; a mismatch fails closed
`:snapshot_source_mismatch`. The endpoint + `@@server_uuid` are pinned for the connection's
life: the first `establish/1` sets the pin (or verifies it against an
`:expected_server_uuid` — the STREAM connection's identity, supplied by the bootstrap), and
every `reestablish/1` re-verifies against it. A reconnect that lands on another replica — a
VIP failover, a moved DNS record — is caught even though the endpoint string is unchanged.

## Preconditions (ADR-0002, extended)

`establish/1` reuses `Capstan.Config.check_preconditions/1` on the query socket: the six
binlog variables gate the query connection exactly as they gate the stream, because
`binlog_row_image = FULL` is a HARD reconciliation dependency (a partial after-image would
make a stream-delivered change unable to stand in for a suppressed chunk row). A precondition
VIOLATION is permanent (reconnecting cannot cure a mis-configured server), so it surfaces its
distinct reason without spending the retry budget — the same posture C1 takes.

## Retry budget

A connect/auth failure fails closed `:snapshot_query_connect_failed`. Transient faults are
budgeted through `Capstan.CheckpointStore.retry_decision/2` + `permanent_reason?/1` — the
shared counter the checkpoint-store and connect-read sites use, so the semantics cannot
drift. A permanent connect reason (`:config_invalid`) halts immediately.

## Rule 1

The connection password appears in NO returned term, error, or log emitted here. The struct
carries the `connection` keyword (which holds the password) only to reconnect, and the
derived `Inspect` renders ONLY the structural identity (`endpoint`, `server_uuid`), never
`connection`. Every error is a value-free atom, scrubbed through `Capstan.Error.from/1`; a
raised transport error is discarded (it could reference the SQL or a value) in favour of a
bare reason. No SQL is ever logged.

# `connect_fun`

```elixir
@type connect_fun() :: (keyword() ->
                    {:ok, Capstan.Protocol.Packet.socket(), map()}
                    | {:error, term()})
```

The socket-open + auth function: mirrors `Capstan.Connection`'s `:connect_fun` shape so a
test can inject a scripted transport. Default: `default_connect/1` (gen_tcp + Handshake).

# `endpoint`

```elixir
@type endpoint() :: {charlist(), :inet.port_number()}
```

The pinned `{host, port}` the connection resolves to for its life.

# `establish_opt`

```elixir
@type establish_opt() ::
  {:connection, keyword()}
  | {:connect_fun, connect_fun()}
  | {:max_command_retries, non_neg_integer()}
  | {:expected_server_uuid, String.t()}
```

Options for `establish/1`: `:connection` (the C1 connection keyword), optional `:connect_fun`
(default `default_connect/1`), optional `:max_command_retries` (default
`CheckpointStore.default_max_retries/0`), and optional `:expected_server_uuid` (the stream
connection's identity to verify against at the first connect, design Ch8).

# `t`

```elixir
@type t() :: %Capstan.Query{
  connect_fun: connect_fun(),
  connection: keyword(),
  endpoint: endpoint(),
  max_retries: non_neg_integer(),
  server_uuid: String.t(),
  socket: Capstan.Protocol.Packet.socket()
}
```

# `close`

```elixir
@spec close(t()) :: :ok
```

Closes the pinned socket. Idempotent.

# `default_connect`

```elixir
@spec default_connect(keyword()) ::
  {:ok, Capstan.Protocol.Packet.socket(), map()} | {:error, term()}
```

The default socket-open + auth function (gen_tcp + `Handshake.connect/2`), mirroring the C1
`Capstan.Connection` connect shape. Returns `{:ok, socket, handshake_info}` or
`{:error, reason}`. Public so a live test can drive the real transport.

# `endpoint`

```elixir
@spec endpoint(t()) :: endpoint()
```

The pinned `{host, port}` endpoint.

# `establish`

```elixir
@spec establish([establish_opt()]) :: {:ok, t()} | {:error, atom()}
```

Opens an authenticated `COM_QUERY` connection, checks the ADR-0002 preconditions, reads
`@@server_uuid`, and pins the endpoint + identity for the connection's life.

With `:expected_server_uuid` set (the stream connection's identity), the observed uuid is
verified against it at connect (design Ch8); otherwise the first read SETS the pin. Returns
`{:ok, t()}` or a value-free `{:error, reason}` — `:snapshot_source_mismatch` on an identity
mismatch, a distinct precondition reason on a bad substrate, or `:snapshot_query_connect_failed`
when the connect/auth budget is exhausted.

# `query`

```elixir
@spec query(t(), binary()) ::
  {:ok, [Capstan.Protocol.Command.row()]} | {:error, atom()}
```

Runs `sql` as a `COM_QUERY` on the pinned socket.

Returns `{:ok, rows}` for a text resultset (each row a list of column values, `nil` for a
NULL column), `{:ok, []}` for a statement with no resultset (`SET`, `LOCK TABLES`,
`UNLOCK TABLES`), or a value-free `{:error, reason}` (scrubbed through `Capstan.Error.from/1`;
a raised transport error becomes `:transport`). The SQL is never logged — a chunk `WHERE`
literal can carry a row value (Rule 1).

# `reestablish`

```elixir
@spec reestablish(t()) :: {:ok, t()} | {:error, atom()}
```

Reconnects, RE-verifying `@@server_uuid` against the pinned value (design Ch8).

A reconnect that lands on a different replica halts `:snapshot_source_mismatch`; otherwise a
fresh authenticated handle to the SAME source is returned.

# `server_uuid`

```elixir
@spec server_uuid(t()) :: String.t()
```

The pinned `@@server_uuid` (structural identity — safe to surface, never a row value).

# `verify_source_identity`

```elixir
@spec verify_source_identity(String.t() | nil, String.t()) ::
  :ok | {:error, :snapshot_source_mismatch}
```

The pure source-identity check: `:ok` when there is no pin yet (the first establish) or the
observed uuid matches the pin, else `{:error, :snapshot_source_mismatch}` (design Ch8).

Exposed so the reconnect re-check is provably RED-capable in isolation.

---

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