@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()) ::
  {:ok, String.t()} | {:error, :invalid_identifier}

Replication command to create a durable logical slot (NOEXPORT_SNAPSHOT).

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

Replication command to create a durable logical slot that EXPORTS a consistent snapshot (spec §4). The result row is [slot_name, consistent_point, snapshot_name, output_plugin] — the caller reads consistent_point (a pg_lsn string) and snapshot_name from it.

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

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

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

Query returning 1 if the publication exists.

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

Query returning each publication table's schemaname, tablename, and PG-quoted fully-qualified name (format('%I.%I', …), spec §9). The quoted qualified column is interpolation-safe for any valid identifier (mixed-case/quoted tables the streaming path also supports); the raw schemaname/tablename fill the %Change{} fields. The publication name is validated (already an allowlisted identifier) then interpolated.

@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()) ::
  {:ok, String.t()} | {:error, :invalid_identifier}

Query returning wal_status and conflicting for the replication slot — the PG16 invalidation signals (spec §8). wal_status = 'lost' means WAL the slot needs was removed (max_slot_wal_keep_size exceeded); conflicting = true means a standby recovery conflict invalidated the slot. Both are unrecoverable data gaps → fail-closed halt. (PG16 has no invalidation_reason column — that is PG17+; wal_status/conflicting are the PG16-correct signals.)

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

opts[:start_lsn] is a Replicant.lsn/0 (non_neg_integer, default 0). A non-integer or negative value raises (caller contract, not attacker input) — pass the uint64 WAL position from checkpoint/0.

opts[:streaming], when truthy, selects proto_version '2', streaming 'on' (in-progress transaction streaming, spec §5). Absent or falsy (the default) emits the byte-for-byte v1 command (proto_version '1', no streaming clause).