defmodule Replicant do @moduledoc """ Framework-agnostic Elixir CDC consumer for Postgres logical replication (`pgoutput`), delivering committed row changes to a pluggable **sink** with **zero data loss**: the replication slot advances only after the sink has durably persisted the transaction. `replicant` is **tenant-blind and classification-blind** — the reliable CDC consumer sibling to `arcadic` and `ash_age`. Multitenancy, classification, and Ash resources live one layer up, in a future `ash_replicant` sink adapter. ## LSN representation A Postgres LSN is exposed as a single `non_neg_integer` — the 64-bit value `(xlog_file <<< 32) ||| xlog_offset` — so that ordinary integer comparison is correct WAL ordering, and the same value feeds the wire-level standby status update. Use `lsn_to_string/1` for display (`"0/16E3778"`); LSNs are WAL positions, not row data, so they are permitted in telemetry metadata. ## Starting a pipeline Configure each pipeline explicitly at `start_link/1` (no global mutable state): Replicant.start_link( connection: [hostname: "standby.internal", port: 5432, username: "u", password: "p", database: "orders", ssl: true], # point at a STANDBY (R-ISO) slot_name: "replicant_orders", publication: "orders_pub", sink: MyApp.OrdersSink, # implements Replicant.Sink go_forward_only: false # must be true to start a :state_mirror sink from empty ) The `Replicant.Connection` owns the slot and advances it only after the sink has durably persisted a transaction (ack-after-checkpoint); the `Replicant.AssemblerServer` applies the sink synchronously off the keepalive path. """ @typedoc """ A Postgres LSN as a single 64-bit integer `(file <<< 32) ||| offset`. Integer comparison (`<=`, `>`) is correct WAL ordering, so the exactly-once watermark is `txn.commit_lsn <= checkpoint`. """ @type lsn :: non_neg_integer() @doc """ The hex display form of an LSN, e.g. `lsn_to_string(0x16E3778) == "0/16E3778"`. Postgres displays `pg_lsn` as uppercase `file/offset` hex with no padding. """ @spec lsn_to_string(lsn()) :: String.t() def lsn_to_string(lsn) when is_integer(lsn) and lsn >= 0 do file = Bitwise.bsr(lsn, 32) offset = Bitwise.band(lsn, 0xFFFFFFFF) "#{Integer.to_string(file, 16)}/#{Integer.to_string(offset, 16)}" end @doc """ Parse a Postgres `pg_lsn` display string (`"0/16E3778"`) into the uint64 LSN — the inverse of `lsn_to_string/1`. The two hex halves are `file`/`offset`; `String.to_integer/2` accepts either case (Postgres renders uppercase). """ @spec lsn_from_string(String.t()) :: lsn() def lsn_from_string(string) when is_binary(string) do [file, offset] = String.split(string, "/", parts: 2) Bitwise.bsl(String.to_integer(file, 16), 32) + String.to_integer(offset, 16) end @doc """ Start a CDC pipeline. Validates `opts`, enforces the go-forward-only start guard, and supervises a `Replicant.Connection` + `Replicant.AssemblerServer` under the pipeline `DynamicSupervisor`. Options: `:connection` (a Postgrex connection keyword list — point at a STANDBY per R-ISO), `:slot_name`, `:publication` (both allowlist-validated identifiers), `:sink` (a module implementing `Replicant.Sink`), and `:go_forward_only` (default `false`; must be `true` to start a `:state_mirror` sink from an empty checkpoint — see `Replicant.Config`). Returns `{:ok, pipeline_pid}` or `{:error, reason}` where `reason` is `:invalid_identifier` / `:invalid_sink` / `:config_invalid` / `:go_forward_required`. """ @spec start_link(keyword()) :: {:ok, pid()} | {:error, term()} def start_link(opts) when is_list(opts) do with {:ok, config} <- Replicant.Config.validate(opts), :ok <- Replicant.Config.guard(config) do Replicant.Supervisor.start_pipeline(config) end end @doc "Stop a running pipeline by slot name (idempotent)." @spec stop(String.t()) :: :ok def stop(slot_name) when is_binary(slot_name) do Replicant.Supervisor.stop_pipeline(slot_name) end end