@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.

@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 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).

@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.

@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.

@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.

@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.

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

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

@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.

@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.

@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.

@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).

@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.

@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.

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

Query returning the pubnames 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.

@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.

@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.

@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.

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

Query returning the active flag for the replication slot.

@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.

@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 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.

@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].

@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).