(with-open [conn (admin/connect {:host "localhost" :database "app"
                                 :user "repl" :password "..."})]
  (admin/ensure-slot! conn "my_slot")
  (admin/ensure-publication! conn "my_pub" {:tables ["public.users"]}))

Manage the server-side objects a logical-replication consumer
needs: replication slots and publications, plus the introspection
(WAL level, current LSN, slot state) that operations around a
consumer want.

Everything here runs as ordinary SQL on a regular connection -- no
replication protocol -- so it works with any JDBC connection the
caller already has.  Connect with a map or a JDBC URL:

```clojure
(with-open [conn (admin/connect {:host "localhost" :database "app"
                                 :user "repl" :password "..."})]
  (admin/ensure-slot! conn "my_slot")
  (admin/ensure-publication! conn "my_pub" {:tables ["public.users"]}))
```

A user creating slots or publications needs enough privilege
(superuser, or ``pg_create_subscription``/ownership as the server
version allows) and the server must run ``wal_level = logical``.

(def c (consumer/start!
        {:db           {:host "localhost" :database "app"
                        :user "repl" :password "..."}
         :slot-name    "my_slot"
         :publication-name "my_pub"
         :tables       ["public.users"]
         :on-change    (fn [change] ...)       ; per change
         ;; and/or
         :on-transaction (fn [tx] ...)}))      ; per commit
(consumer/stop! c)

The replication consumer loop: open a stream, assemble decoded
messages into transactions, deliver them to handlers, and ack the
flushed LSN only after delivery succeeds.

Delivery is ordered and at-least-once.  The slot's confirmed
position advances only at transaction boundaries after the handler
returns, so a handler failure or a crash redelivers from the last
acked transaction; handlers must tolerate (or dedupe) duplicates.

```clojure
(def c (consumer/start!
        {:db           {:host "localhost" :database "app"
                        :user "repl" :password "..."}
         :slot-name    "my_slot"
         :publication-name "my_pub"
         :tables       ["public.users"]
         :on-change    (fn [change] ...)       ; per change
         ;; and/or
         :on-transaction (fn [tx] ...)}))      ; per commit
(consumer/stop! c)
```

Connections (exactly one source is required):

:db        spec map; postlog opens and owns both the replication
           and admin connections, closing them on stop.
:conn-fn   (fn [] Connection); called for every (re)connect and
           must return a fresh, dedicated, un-pooled replication
           connection (``replication=database``) which postlog
           never closes.  A bare connection instance is not
           accepted on purpose: a replication connection that dies
           mid-stream cannot be revived -- resuming the slot means
           a *fresh* connection, which only a factory can produce.
:admin-conn  optional Connection for slot/publication bootstrap;
           never closed by postlog.  Without :db or :admin-conn,
           the slot and publication must already exist.

:stream-fn  test escape hatch (fn [] LogicalStream); bypasses
            connection handling entirely.

:binary?    stream values in the binary wire format (default
            true); false selects text.  decode/decode-value
            decodes both, so the choice matters mainly for custom
            value-decoders and for unknown types, which pass
            through as raw bytes (binary) or raw text (text).

Handlers run on the consumer's single thread, one at a time --
that is what keeps ordering and provides backpressure.  Handler
exceptions retry with exponential backoff (see :retry) and fail
the consumer when exhausted; a failed consumer acks nothing past
the failure.  Connection loss reconnects with the same backoff,
and the slot resumes from its confirmed position.

Transactions with no delivered changes (nothing published, or only
filtered operations) still ack -- the position must advance -- but
do not invoke :on-transaction.

Decode the column values pgoutput sends into Clojure data, keyed by
type OID.

pgoutput renders columns in one of two formats per stream: every
column as its text representation (``42``, ``t``, ``2500.00``), or
with the protocol-v2 ``binary`` option, each type's binary wire form
(int4 as four big-endian bytes, timestamps as microseconds since
2000-01-01, and so on — the layouts Postgres' typsend functions
emit, verified identical on servers 14-17).  Relation messages name
each column's type only by OID, so both formats decode through an
OID-keyed lookup: [[decode-value]] dispatches on the value being a
string (text) or bytes (binary).

Values of unknown or custom types keep their raw text or bytes, and
other values (NULL, unchanged-TOAST markers) pass through
untouched.

Postgres log sequence numbers.

An LSN identifies a byte offset into the write-ahead log.  Postgres
prints it as two hex halves joined by a slash, e.g. ``0/16B374D``;
the first half is the high 32 bits, the second the low 32.

This library represents an LSN as a plain 64-bit long, so ordinary
comparisons and arithmetic work on it: a later LSN is a bigger
number, and subtracting two LSNs yields the bytes of WAL between
them.  Real-world LSNs stay far below ``Long/MAX_VALUE``, so the
unsigned interpretation Postgres uses never matters in practice.

Decode the binary stream produced by Postgres' ``pgoutput`` logical
replication output plugin.

Each message starts with a one-byte type tag followed by
big-endian fixed-width integers and NUL-terminated (cstring) text.
[[decode]] turns one message's bytes into a map carrying an ``:op``
keyword plus the fields of that message type; LSNs are longs (see
[[ebifriedrice.postlog.lsn]]) and commit timestamps are
``java.time.Instant``s.

Tuple values arrive in one of four encodings, exposed as:
``nil`` (SQL NULL), a UTF-8 string (text format), a byte array
(binary format, proto v2+ only), or [[unchanged-toast]] when a TOASTed
value was not modified by the change and the server skipped sending
it.

This decoder covers protocol version 1 message layouts, which is
what postlog streams with.  The in-progress-transaction stream
messages (proto v2+) are rejected with a clear error rather than
misparsed.

Track the relations a stream announces and turn positional tuples
into row maps.

pgoutput identifies tables by OID and sends column values
positionally, with a Relation message (see
[[ebifriedrice.postlog.pgoutput]]) arriving before the first change
that references a table — and again whenever its definition
changes.  A consumer keeps the latest relation per OID in a map and
resolves changes against it.

Key-only before-images (updates that changed a replica-identity
column, and deletes on tables without REPLICA IDENTITY FULL)
arrive full-width with NULL in every non-identity column; rows
below pick out the identity columns.  Full tuples carry every
column.

Rows are maps from column name (string) to value, because column
names come from the database and stay data here.

Open replication connections and stream pgoutput messages.

pgjdbc exposes the walsender protocol on a connection opened with
``replication=database``: unwrap to ``org.postgresql.PGConnection``,
build a logical stream against a slot, and receive the raw pgoutput
payload of each WAL record.  The driver owns the protocol mechanics
this library is happy not to reimplement: XLogData framing,
keepalive replies, and periodic standby status updates reporting
our flushed LSN.

[[start!]] returns anything implementing [[LogicalStream]] -- by
default a thin adapter over pgjdbc's stream, but the consumer loop
only talks to the protocol, so tests drive it with fakes.