# `Replicant.QueryBuilder`
[🔗](https://github.com/baselabs/replicant/blob/v0.3.1/lib/replicant/query_builder.ex#L1)

Slot and publication SQL/command strings built from **validated** identifiers
(Critical Rule 2). Hardens `walex`'s raw `'#{publication}'` interpolation: every
name passes through `Replicant.Identifier.validate/1` before it reaches the
string; an invalid name returns `{:error, :invalid_identifier}` and builds
nothing. `START_REPLICATION`/`CREATE_REPLICATION_SLOT` are replication commands
(no bind parameters), so validated interpolation is the gate.

# `checkpoint_column_probe`

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

Query probing the `commit_lsn` column's `data_type`. The table name is bound `$1`
(a string value in `information_schema`, not an identifier position), so no
interpolation and no validation are needed here — the caller has already validated
the table for the interpolating builders above.

# `checkpoint_ensure_table`

```elixir
@spec checkpoint_ensure_table(String.t()) ::
  {:ok, String.t()} | {:error, :invalid_identifier}
```

DDL creating the lib-owned checkpoint table if absent. `slot_name` is the PK, one
row per slot; `commit_lsn` is a `bigint` (the `t:Replicant.lsn/0` integer — no
`pg_lsn` text parse at the boundary). The table name is a validated identifier;
`IF NOT EXISTS` is a name check only, so the caller MUST also shape-probe (see
`checkpoint_column_probe/0`).

# `checkpoint_read`

```elixir
@spec checkpoint_read(String.t()) :: {:ok, String.t()} | {:error, :invalid_identifier}
```

Query reading `commit_lsn` for a slot. `slot_name` is bound `$1`; only the validated table is interpolated.

# `checkpoint_upsert`

```elixir
@spec checkpoint_upsert(String.t()) ::
  {:ok, String.t()} | {:error, :invalid_identifier}
```

Upsert of `commit_lsn` for a slot. `slot_name`/`commit_lsn` are bound `$1`/`$2`; only the validated table is interpolated.

# `create_durable_slot`

```elixir
@spec create_durable_slot(String.t(), boolean()) ::
  {:ok, String.t()} | {:error, :invalid_identifier}
```

Replication command to create a durable logical slot with NO exported snapshot. `failover?:
false` (the published default) emits the legacy `NOEXPORT_SNAPSHOT` keyword byte-for-byte;
`failover?: true` emits the PG17 parenthesized `(FAILOVER, SNAPSHOT 'nothing')` grammar (spec
§6). The caller has already gated failover-on-PG16 at connect, so `failover? == true` implies PG17.

# `create_export_slot`

```elixir
@spec create_export_slot(String.t(), boolean()) ::
  {:ok, String.t()} | {:error, :invalid_identifier}
```

Replication command to create a durable logical slot that EXPORTS a consistent snapshot (spec
§4). `failover?: false` emits the legacy `EXPORT_SNAPSHOT` keyword byte-for-byte; `failover?:
true` emits `(FAILOVER, SNAPSHOT 'export')` (spec §6). The result row is `[slot_name,
consistent_point, snapshot_name, output_plugin]` in both forms.

# `is_in_recovery`

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

Query returning `pg_is_in_recovery()` — `true` on a standby (spec §8 R-ISO advisory).

# `keyless_scan`

```elixir
@spec keyless_scan(String.t(), [String.t()]) :: String.t()
```

Whole-table scan for the PK-less fallback (spec §6.4). Every column is projected as
`<col>::text AS <col>` (the SAME cast projection as `keyset_chunk/4`) so a PK-less
table's snapshot rows converge with the stream too (the record values are cast; there
is no keyset bound). `col_quoted` is server-quoted (`table_columns/0`), never raw.

# `keyset_chunk`

```elixir
@spec keyset_chunk(String.t(), [String.t()], [String.t()], non_neg_integer()) ::
  {:ok, String.t()} | {:error, :invalid_identifier}
```

Keyset chunk SELECT (spec §6.6). `col_quoted` (all columns) and `pk_quoted` (the PK
subset) come from `table_columns/0`/`pk_columns/0` (server-quoted — never validated
client-side, never raw). `bound_arity` is the number of PK columns in the resume bound:
`0` emits the first-chunk form (no WHERE); `n > 0` emits a ROW() comparison whose bounds
are BIND PARAMETERS `$2..$n+1` (`$1` is the LIMIT) — a literal bound would be
simultaneously an injection surface and a Rule-1 leak into logged SQL text. An empty PK
list is a caller error.

EVERY column is projected as `<col>::text AS <col>` so each delivered value is the
type's text output — byte-identical to pgoutput text, which the stream casts through the
SAME `Casting.Types.cast_record/2` path (spec §2 convergence). The WHERE/ORDER BY reference
the REAL typed PK columns — TABLE-QUALIFIED (`<qualified>.<pk>`, NOT bare `<pk>`) so they
bind to the typed table columns, NEVER the same-named `::text` OUTPUT ALIAS: a bare
`ORDER BY <pk>` resolves to the projected alias (SQL lets ORDER BY see output names),
which would sort keyset pages LEXICOGRAPHICALLY (`1,10,100,…`) and silently skip rows.

# `pk_columns`

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

Query returning, per publication table, its ordered PRIMARY KEY columns — BOTH the
raw `attname` (to read `%Change{}.record` string keys) and the server-quoted form
via `quote_ident` (the ONLY form ever interpolated into keyset SQL — spec §6.6,
Critical Rule 2; the `format('%I')` precedent). Publication name is bound `$1`.
A publication table with NO primary key returns no row here (PK-less fallback,
spec §6.4). Row shape: `[schemaname, tablename, qualified, pk_raw, pk_quoted]`.

Per-column TYPE oids come from `table_columns/0` (the full column set already covers
the PK columns), so no per-PK type array is duplicated here.

# `progress_ensure_table`

```elixir
@spec progress_ensure_table(String.t()) ::
  {:ok, String.t()} | {:error, :invalid_identifier}
```

DDL creating the lib-owned snapshot-progress table if absent (token is an opaque bytea; spec §6.2).

# `progress_read`

```elixir
@spec progress_read(String.t()) :: {:ok, String.t()} | {:error, :invalid_identifier}
```

Query reading the progress token for a slot. `slot_name` bound `$1`; only the validated table is interpolated.

# `progress_upsert`

```elixir
@spec progress_upsert(String.t()) :: {:ok, String.t()} | {:error, :invalid_identifier}
```

Upsert of the progress token for a slot. `slot_name`/`token` bound `$1`/`$2`.

# `publication_exists`

```elixir
@spec publication_exists([String.t()]) ::
  {:ok, String.t()} | {:error, :invalid_identifier}
```

Query returning the `pubname`s that exist for any name in the validated list (spec §5.3 /
decision #18). The caller halts fail-closed if the returned set ≠ the requested set —
`START_REPLICATION` with a missing publication silently streams the existing subset, so it
CANNOT be the fail-closed gate (probe-verified).

Unlike `publication_tables/1` / `pk_columns/0` / `table_columns/0` (which run via
`Postgrex.query!/3` and bind `$1`), this query runs in the `Postgrex.ReplicationConnection`
connect chain, whose `{:query, sql, state}` dispatch uses the SIMPLE query protocol — which
cannot bind `$1`. So the validated names are interpolated into a single-quoted `IN (...)` list,
the exact precedent `slot_invalidation_status/2` sets with `slot_name`. Injection-safe per
Critical Rule 2: every name passed `Identifier.validate/1` (`[a-z_][a-z0-9_]{0,62}`) BEFORE
reaching here — the allowlist excludes `'`, `)`, `;`, and every other breakout character, so
the interpolated literal cannot escape the `IN (...)` position.

# `publication_tables`

```elixir
@spec publication_tables([String.t()]) ::
  {:ok, String.t()} | {:error, :invalid_identifier}
```

Query returning each DISTINCT publication table's `schemaname`, `tablename`, and PG-quoted
fully-qualified name across the publication LIST (spec §5.3 / decision #19). The `DISTINCT`
collapses the pubname dimension BEFORE it fans the join, so a table in two publications yields
ONE row (not two). The publication names are bound `$1` (an array), so no interpolation.

# `recovery_and_version`

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

Connect-time probe: `pg_is_in_recovery()` (standby detection, spec §8) plus the numeric
server version (`server_version_num`, e.g. `170010`) in one round trip on the replication
connection. The version gates the PG17+ invalidation columns (spec §5) and the `FAILOVER`
slot grammar (spec §6). No identifier to validate — a constant query.

# `set_transaction_snapshot`

```elixir
@spec set_transaction_snapshot(term()) ::
  {:ok, String.t()} | {:error, :invalid_snapshot_name}
```

Command adopting an exported snapshot by name (spec §9). The name is validated as a
snapshot-name LITERAL — not an identifier — before interpolation into the quoted
string; a name with a quote/whitespace/other injection returns
`{:error, :invalid_snapshot_name}` and builds nothing.

# `slot_exists`

```elixir
@spec slot_exists(String.t()) :: {:ok, String.t()} | {:error, :invalid_identifier}
```

Query returning the `active` flag for the replication slot.

# `slot_invalidation_status`

```elixir
@spec slot_invalidation_status(String.t(), non_neg_integer()) ::
  {:ok, String.t()} | {:error, :invalid_identifier}
```

Query returning the slot's invalidation signals (spec §5/§8). On **PG < 17** (`version <
170000`): `wal_status` + `conflicting` (the PG16 columns; `invalidation_reason` errors there).
On **PG ≥ 17**: also `invalidation_reason` (Postgres's authoritative invalidation field) and
`synced` (true on a standby holding a slot synced from the primary). `wal_status = 'lost'` =
WAL removed; `conflicting = true` = standby recovery conflict; any non-null `invalidation_reason`
= invalidated. All are unrecoverable → fail-closed halt.

# `start_replication`

```elixir
@spec start_replication(String.t(), [String.t()], keyword()) ::
  {:ok, String.t()} | {:error, :invalid_identifier}
```

Replication command that starts streaming WAL from `start_lsn` for the publication set.

`publication` is a non-empty list of validated identifiers (Critical Rule 2): a single-element
list reproduces the published-0.1.0 command byte-for-byte; a multi-element list is comma-joined
into `publication_names 'p1,p2'` (each name is an allowlisted bare identifier, so the bare
comma-list is injection-safe — probe-verified). `opts[:start_lsn]` is a `t:Replicant.lsn/0`
(`non_neg_integer`, default `0`).

`opts[:streaming]`, when truthy, selects `proto_version '2', streaming 'on'` (spec §5).
`opts[:messages]`, when truthy, adds `messages 'true'` (spec §6, A2). Absent or falsy (the
default) emits the byte-for-byte v1 command.

# `table_columns`

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

Query returning, per publication table, ALL its non-dropped user columns
(`attnum > 0 AND NOT attisdropped`, the `SELECT *` column set) ordered by `attnum` —
the raw `attname` (the `%Change{}.record` string key and the result column name), the
server-quoted form via `quote_ident` (the ONLY form interpolated into the keyset/keyless
`::text` projection — Critical Rule 2, the `format('%I')` precedent), and each column's
`atttypid`. The reader maps the OID to a pgoutput type name (`OidDatabase.name_for_type_id/1`)
and casts the `<col>::text` value through the SAME `Casting.Types.cast_record/2` path the
stream uses, so snapshot and stream deliver byte-identical `%Change{}.record` values for
EVERY column (spec §2 convergence — the F1 fix generalized from PK-only to all columns).
Publication name is bound `$1`. Row shape:
`[schemaname, tablename, qualified, col_raw, col_quoted, col_type_oids]`.

# `watermark_lsn`

```elixir
@spec watermark_lsn(boolean()) :: String.t()
```

Read-only watermark position (spec §2/§4): `pg_current_wal_lsn()` on a primary,
`pg_last_wal_replay_lsn()` on a standby (the chunk reader connects to the same host
as the replication connection, so its snapshot visibility is bounded by replay).

---

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