init.core.template.sql

-- ─────────────────────────────────────────────────────────────────────────────
-- init.core.sql — the read-only half: CDC and snapshots, and nothing that writes.
--
-- Runnable on its own. That is the point: a deployment that only *reads* PostgreSQL
-- should never meet RLS, tombstone guards or tenant scoping, and a newcomer reading this
-- file should not have to skip past them to find out how a change becomes an event.
--
--     envsubst < init.core.template.sql | psql -v ON_ERROR_STOP=1 …
--
-- The write path is `init.write.template.sql`, appended after this one:
--
--     cat init.core.template.sql init.write.template.sql | envsubst | psql …
--
-- ⚠️ It is appended, never merged. The read-only setup is this file *unmodified*, so the
-- two profiles cannot drift: there is no commented-out block here that the other profile
-- uncomments, and no second copy of the publication or the DDL triggers.
--
-- ⚠️ Nothing here may reference ${POSTGRES_WRITER_USER}. That role is created by the
-- write half, so a reference from this file breaks the read-only profile — which is the
-- one invariant that keeps the split honest.
-- ─────────────────────────────────────────────────────────────────────────────
-- Every object below carries a `-- <what it is> — see <doc> §<n>` line above it.
--
-- The reference points at the section that explains *why* the object exists, because this
-- file is where someone is standing when they need that: reading a trigger and wondering
-- whether it can be dropped. The docs describe behaviour a DBA has to preserve;
-- `zebridge_gc_watermark` in particular carries a 🔴 DBA callout in PROTOCOL.md §7.5,
-- because dropping it or unpublishing it removes a client guarantee without failing
-- anything.
--
-- Keep the reference when moving or rewriting an object. A comment that says what the code
-- already says is noise; one that says where the reasoning lives is the only link between
-- the two files.
--
-- The link runs both ways: **PROTOCOL.md §8b** lists every object created here, what it is,
-- and what breaks if it is removed. Read this file and you can find the reasoning; read the
-- docs and you can find the code. Adding an object means adding a row there too — a
-- one-directional link is the one that goes stale.
-- ─────────────────────────────────────────────────────────────────────────────

DO $$
BEGIN
    IF NOT EXISTS (SELECT 1 FROM pg_roles WHERE rolname = '${POSTGRES_READER_USER}') THEN
        CREATE USER ${POSTGRES_READER_USER} WITH PASSWORD '${POSTGRES_READER_PASSWORD}';
        ALTER USER ${POSTGRES_READER_USER} WITH REPLICATION;
    END IF;
    -- The reader's connection BUDGET, DB-enforced (outside the IF: applies to
    -- pre-existing roles on re-apply). Steady state is ~4 backends (producer,
    -- snapshot worker, boot passes) — the replication stream rides a walsender
    -- slot, not a regular backend. 10 leaves boot-burst headroom while making it
    -- impossible for a bridge bug to eat the cluster's default 100.
    ALTER ROLE ${POSTGRES_READER_USER} CONNECTION LIMIT 10;
    IF true THEN
    END IF;
END;
$$;

GRANT CONNECT ON DATABASE ${TARGET_DB} TO ${POSTGRES_READER_USER};

GRANT USAGE ON SCHEMA public TO ${POSTGRES_READER_USER};

-- Grant SELECT on existing tables
GRANT SELECT ON ALL TABLES IN SCHEMA public TO ${POSTGRES_READER_USER};

-- Ensure bridge user gets SELECT permissions on all FUTURE tables created by the DB owner
ALTER DEFAULT PRIVILEGES IN SCHEMA public GRANT SELECT ON TABLES TO ${POSTGRES_READER_USER};

-- Put the tenant column inside the table's REPLICA IDENTITY: a unique index on
-- (tenant, primary key), made the identity. Returns the column list it set (or, with
-- dry_run, would set), NULL when the identity already covers the tenant.
--
-- ⚠️ Every tenant table needs it, read-only or writable. A DELETE carries only the
-- identity columns — by default the primary key alone — so without the tenant it cannot
-- be routed to a tenant subject, and PostgreSQL refuses UPDATE and DELETE once a
-- publication filters on the tenant. The bridge refuses such a table at boot
-- (`tenant_not_in_replica_identity`). Lives in core, not init.write, because a
-- read-only-profile database needs it too; it used to exist only on the write path,
-- and a read-only tenant table came out of zebridge_enable refused (2026-09-25).
-- The caller checks that the tenant column exists and is NOT NULL.
CREATE OR REPLACE FUNCTION public.zebridge_tenant_replica_identity(
    tbl regclass, tenant_col name, dry_run boolean DEFAULT false
) RETURNS text AS $$
DECLARE
    ri_kind  "char";
    covered  boolean;
    pk_cols  name[];
    idx_name name;
    col_list text;
BEGIN
    SELECT relreplident INTO ri_kind FROM pg_class WHERE oid = tbl;
    IF ri_kind = 'f' THEN
        RETURN NULL;                           -- FULL carries every column
    END IF;
    SELECT coalesce(bool_or(a.attname = tenant_col), false) INTO covered
    FROM pg_index i
    JOIN pg_attribute a ON a.attrelid = i.indrelid AND a.attnum = ANY(i.indkey)
    WHERE i.indrelid = tbl
      AND ((ri_kind = 'd' AND i.indisprimary) OR (ri_kind = 'i' AND i.indisreplident));
    IF covered THEN
        RETURN NULL;
    END IF;

    SELECT array_agg(a.attname ORDER BY k.ord) INTO pk_cols
    FROM pg_index i
    JOIN LATERAL unnest(i.indkey) WITH ORDINALITY AS k(attnum, ord) ON true
    JOIN pg_attribute a ON a.attrelid = i.indrelid AND a.attnum = k.attnum
    WHERE i.indrelid = tbl AND i.indisprimary;
    IF pk_cols IS NULL THEN
        RAISE EXCEPTION '% has no primary key, so no replica identity can cover %', tbl, tenant_col;
    END IF;

    idx_name := (SELECT relname FROM pg_class WHERE oid = tbl) || '_zb_ri';
    col_list := quote_ident(tenant_col) || ', ' ||
                (SELECT string_agg(quote_ident(c), ', ') FROM unnest(pk_cols) c);
    IF NOT dry_run THEN
        -- `name` keeps 63 bytes, so the suffix of a 63-character table is cut off and the
        -- "index" is the table itself: CREATE ... IF NOT EXISTS skips it, and the ALTER below
        -- fails with '"<table>" is not an index', which names nothing to do (measured
        -- 2026-10-26). Say what is wrong while there is still nothing to undo.
        IF idx_name = (SELECT relname FROM pg_class WHERE oid = tbl) THEN
            RAISE EXCEPTION 'table name % is % bytes: the replica-identity index name derived from it (<table>_zb_ri) is cut to the table''s own name. Rename the table to at most 62 bytes, or make the tenant part of the replica identity yourself (a unique index on (%, primary key) and ALTER TABLE ... REPLICA IDENTITY USING INDEX)',
                tbl, octet_length(idx_name), tenant_col;
        END IF;
        EXECUTE format('CREATE UNIQUE INDEX IF NOT EXISTS %I ON %s (%s)', idx_name, tbl, col_list);
        EXECUTE format('ALTER TABLE %s REPLICA IDENTITY USING INDEX %I', tbl, idx_name);
        RAISE NOTICE 'replica identity of % moved to (%) so its rows route by tenant', tbl, col_list;
    END IF;
    RETURN col_list;
END;
$$ LANGUAGE plpgsql;

-- Scope the reader's SELECT by tenant, for the snapshot path — see PROTOCOL.md "The
-- Connection Flow" Step 0, NOTES.md §1.12 part 1
--
-- ⚠️ RLS is evaluated against a query, and logical decoding has no query — structurally
-- invisible to CDC (§1.8: a policy returned 1 row to a SELECT while the WAL still carried
-- 2). CDC's tenant correctness is the subject's job and stays that way; this function
-- only ever affects a real SELECT, which today means the snapshot listener's own
-- connection setting `zb.tenant` before it queries.
--
-- Lives in core, not write: it needs nothing from `zebridge_user_tenants` (that table
-- resolves a WRITER's own tenant when omitted from a write — a write-path concern only)
-- and nothing from the writer role. A read-only deployment (`init.core.template.sql`
-- alone, `ZB_PROFILE=readonly`) can scope its snapshots by tenant with no write profile
-- installed at all.
--
-- ⚠️ `zb_reader_all` cannot simply gain a second, more restrictive policy alongside it:
-- PostgreSQL ORs every applicable permissive policy for one role/command together, so a
-- second policy next to an existing `USING (true)` changes nothing — the permissive one
-- always wins. The fix is making the ONE policy conditional instead: wholesale when
-- `zb.tenant` is unset, filtered when it is. The replication connection never calls
-- `set_config('zb.tenant', ...)` — and RLS does not apply to it regardless — so this is a
-- no-op for CDC and preserves exactly what `zb_reader_all USING (true)` already gave every
-- caller that never sets it: the reader sees everything by default, and only narrows on
-- an explicit opt-in.
--
-- ⚠️ `coalesce(..., '') = ''`, NOT `IS NULL` — the same trap `zb_sweeper`'s policy already
-- documents for `zb.principal`, measured again building this one: `current_setting(...,
-- true)` returns an empty string, not SQL NULL, for a GUC that has never been set on this
-- role/session (not only after a `SET LOCAL` resets at COMMIT). `IS NULL` never matched —
-- the "wholesale when unset" branch was dead code, and every reader saw zero rows instead
-- of everything until this was measured directly against the running database.
--
-- ⚠️ The bridge sets `zb.tenant` from the AUTHENTICATED REQUEST SUBJECT, never resolves it
-- through `zebridge_user_tenants` for this purpose — two derivations of one fact can
-- disagree (subject says acme, mapping says globex), and only one of them cannot disagree
-- with itself.
CREATE OR REPLACE FUNCTION public.zebridge_scope_reads_by_tenant(tbl regclass, tenant_col name)
RETURNS void AS $$
DECLARE
    t text := tbl::text;
BEGIN
    IF NOT EXISTS (SELECT 1 FROM pg_attribute
                   WHERE attrelid = tbl AND attname = tenant_col
                     AND attnum > 0 AND NOT attisdropped AND attnotnull) THEN
        RAISE EXCEPTION 'tenant column %.% must exist and be NOT NULL', tbl, tenant_col;
    END IF;

    -- A tenant table's CDC routes by the tenant, reads scoped or not.
    PERFORM public.zebridge_tenant_replica_identity(tbl, tenant_col);

    EXECUTE format('ALTER TABLE %s ENABLE ROW LEVEL SECURITY', tbl);

    EXECUTE format('DROP POLICY IF EXISTS zb_reader_all ON %s', tbl);
    EXECUTE format(
        'CREATE POLICY zb_reader_all ON %s FOR SELECT TO %I'
        -- Same open-tenant carve-out zb_tenant_write already has (init.write.template.sql):
        -- a row mapped to the OPEN tenant is readable by every principal, not just whoever
        -- has zb.tenant set to it. Without this, CDC (subject-routed, RLS-blind) delivers
        -- an open row to every client while a snapshot never grants the SELECT that would
        -- have put it there in the first place — audience says everyone, contents says only
        -- the open tenant itself, and a client's local copy of the row depends on whether it
        -- happened to be connected for a live CDC write, not on what its snapshot returned.
        ' USING (coalesce((SELECT current_setting(''zb.tenant'', true)), '''') = '''' OR %I::text = (SELECT current_setting(''zb.tenant'', true)) OR %I::text = ''${OPEN_TENANT}'')',
        tbl, '${POSTGRES_READER_USER}', tenant_col, tenant_col);

    RAISE NOTICE 'reads on % now filtered by % when zb.tenant is set — CDC is unaffected (RLS does not apply to replication); a client-scoped connection sets zb.tenant to scope it',
                 tbl, tenant_col;
END;
$$ LANGUAGE plpgsql;


-- The widest row actually stored, in the units CDC packs — see PROTOCOL.md §9
--
-- ⚠️ `octet_length`, never `pg_column_size`. The latter reports the *stored, compressed*
-- size: a 40 KB run of one character compresses to almost nothing, so a check built on it
-- reports a table as safe while CDC cannot carry its widest row. Wrong direction for a
-- safety bound.
--
-- ⚠️ Cheap for the types that matter: `octet_length` on text/varchar/bytea reads the length
-- out of the TOAST pointer without fetching the value — measured at 2 shared buffers for a
-- 50 MB table. `jsonb`/arrays need `::text`, which does materialise; they are usually small.
CREATE OR REPLACE FUNCTION public.zebridge_widest_row(tbl regclass)
RETURNS bigint AS $$
DECLARE
    expr   text;
    result bigint;
BEGIN
    SELECT coalesce(string_agg(
        CASE
            WHEN a.atttypid IN ('text'::regtype, 'bytea'::regtype, 'varchar'::regtype)
                THEN format('coalesce(octet_length(%I),0)', a.attname)
            WHEN t.typcategory = 'A'
              OR a.atttypid IN ('json'::regtype, 'jsonb'::regtype, 'xml'::regtype)
                THEN format('coalesce(octet_length(%I::text),0)', a.attname)
            -- Fixed-width columns: a flat allowance rather than a per-type table. The
            -- result is a floor on the row's size, which is what a guard wants.
            ELSE '8'
        END, ' + '), '0')
    INTO expr
    FROM pg_attribute a
    JOIN pg_type t ON t.oid = a.atttypid
    WHERE a.attrelid = tbl AND a.attnum > 0 AND NOT a.attisdropped;

    EXECUTE format('SELECT coalesce(max(%s),0)::bigint FROM %s', expr, tbl) INTO result;
    RETURN result;
END;
$$ LANGUAGE plpgsql STABLE;

-- Column DEFAULTs that alone exceed a byte budget — see PROTOCOL.md §9
--
-- ⚠️ The case `zebridge_widest_row` cannot see: a table with a huge column DEFAULT and no
-- rows yet measures as perfectly safe, and then the *first* insert — a 258-byte mutation
-- that passed every ingress guard — stores a 50 KB row and suspends the table for every
-- client. Measured. The default is the hazard; the row is only where it shows up.
--
-- ⚠️ Two guards make evaluating a default safe:
--
--   * only **variable-width** columns are evaluated. A `nextval()` default lives on an
--     integer column, so it is never reached and no sequence is burned;
--   * the function is **STABLE**, which makes PostgreSQL refuse any write the expression
--     attempts — a side-effecting default raises, and the handler below skips it rather
--     than failing the check.
CREATE OR REPLACE FUNCTION public.zebridge_oversized_defaults(tbl regclass, budget bigint)
RETURNS TABLE(col text, bytes bigint) AS $$
DECLARE
    r record;
    n bigint;
BEGIN
    FOR r IN
        SELECT a.attname AS name, pg_get_expr(ad.adbin, ad.adrelid) AS expr
        FROM pg_attribute a
        JOIN pg_attrdef ad ON ad.adrelid = a.attrelid AND ad.adnum = a.attnum
        JOIN pg_type t ON t.oid = a.atttypid
        WHERE a.attrelid = tbl AND a.attnum > 0 AND NOT a.attisdropped
          AND (a.atttypid IN ('text'::regtype, 'bytea'::regtype, 'varchar'::regtype)
               OR t.typcategory = 'A'
               OR a.atttypid IN ('json'::regtype, 'jsonb'::regtype, 'xml'::regtype))
    LOOP
        BEGIN
            EXECUTE format('SELECT coalesce(octet_length((%s)::text),0)::bigint', r.expr) INTO n;
        EXCEPTION WHEN OTHERS THEN
            -- A default that cannot be evaluated read-only is not a finding. Saying
            -- nothing beats refusing to start over an expression we declined to run.
            CONTINUE;
        END;
        IF n >= budget THEN
            col := r.name;
            bytes := n;
            RETURN NEXT;
        END IF;
    END LOOP;
END;
$$ LANGUAGE plpgsql STABLE;


-- keeps the tracker's own rows out of the DDL feed — see PROTOCOL.md §5. Defined before
-- any table it names is created: on an upgrade the DDL trigger is already installed, and
-- a table created before its name is listed enters the feed as an application table.
CREATE OR REPLACE FUNCTION public.zebridge_is_internal_table(tbl text)
RETURNS boolean AS $$
    -- The namespace, not a list (OPERATIONS "Reserved names"; zebridge_enable refuses the prefix): every
    -- zebridge_ table is the bridge's and is never client-replicated, except the GC floor, which every client
    -- reads. That includes the principal->tenant roster, zebridge_user_tenants: BRIDGE INPUT (its CDC feeds
    -- $KV.tenants), never client-replicated, since replicating it would disclose the roster the per-key
    -- $KV.tenants grant exists to protect (NOTES.md §1.12 part 3). The bridge's own Zig rule
    -- (preflight.isInternalTable) asks a different question and checks the roster like a user table.
    -- A name list here had to be remembered at each new table and drifted; enable_scoping check 11 and
    -- init_sql.zig test this against every table the init SQL creates.
    -- schema_migrations is Ecto's / ActiveRecord's migration bookkeeping.
    SELECT (left(tbl, 9) = 'zebridge_' AND tbl <> 'zebridge_gc_watermark') OR tbl = 'schema_migrations';
$$ LANGUAGE sql IMMUTABLE;

-- THE CATALOGUE comes first: the guard below and every registration in this file
-- read it. Public tables used to live in a separate `zebridge_public_tables`; that
-- was a strict projection of this relation (`tenant_col IS NULL`) and is gone — the
-- deliberate "published to everyone" decision is the `public_reason` column, which
-- the CHECK makes mandatory exactly when a row is public.
-- THE CATALOGUE — one row per replicated table, the disjoint union of public and
-- tenant-scoped as one nullable column (NOTES.md, the static-residue endgame).
-- Written by `zebridge_enable` in the SAME transaction as the guards it installs
-- with these very columns, so nothing can disagree: the bridge and the sweeper read
-- their per-table rules only from here (no env override), and the generation
-- producer reads it per tick.
-- The CHECK absorbs the publication guard's core law into the schema itself: a
-- row is tenant-scoped or publicly justified — never silently neither.
CREATE TABLE IF NOT EXISTS public.zebridge_catalogue (
    tbl           text PRIMARY KEY,
    tenant_col    name,            -- NULL = public; NOT NULL = tenant-scoped
    public_reason text,
    version_col   name NOT NULL DEFAULT 'updated_at',
    tombstone_col name,
    tiebreak_col  name,
    generations   boolean NOT NULL DEFAULT true,
    -- The re-seed lever (NOTES §10df): bumped by zebridge_reseed(), carried in the
    -- schema descriptor and the chain manifest; a client whose stored epoch is lower
    -- forgets its watermark and seeds a fresh full. The producer builds that full.
    seed_epoch    integer NOT NULL DEFAULT 0,
    -- The operator's acceptance of physical deletes on a writable table with no
    -- tombstone (zebridge_enable's allow_physical_deletes), kept so zebridge_check
    -- reads it too: the catalogue is the intent, nothing has to be repeated.
    allow_physical_deletes boolean NOT NULL DEFAULT false,
    CHECK ((tenant_col IS NULL) <> (public_reason IS NULL))
);
ALTER TABLE public.zebridge_catalogue ADD COLUMN IF NOT EXISTS allow_physical_deletes boolean NOT NULL DEFAULT false;
-- A table's incarnations (one lifetime under its name, from a CREATE to the DROP that
-- ends it): `incarnation_relid` the live table's OID, NULL while there is none;
-- `incarnation` how many lifetimes this name has had. Kept by zebridge_observe_incarnation
-- alone. The OID is the evidence of a replacement, the counter is the identity — and the
-- identity must never repeat (chain objects are named with it), so it lives in its OWN
-- table: a catalogue row can be deleted and inserted again, and a counter kept on it
-- started over at 1. Rows here are never deleted.
-- `seed_epoch` the highest seed epoch the name has had: the catalogue's own column is
-- the published copy, and a catalogue row inserted again starts from this floor (the
-- trigger below), never from 0 — a replica compares epochs, so an epoch that came back
-- to its own value told it nothing of the replacement in between.
CREATE TABLE IF NOT EXISTS public.zebridge_incarnations (
    tbl               text PRIMARY KEY,
    incarnation_relid oid,
    incarnation       bigint NOT NULL DEFAULT 0,
    seed_epoch        integer NOT NULL DEFAULT 0
);
ALTER TABLE public.zebridge_incarnations ADD COLUMN IF NOT EXISTS seed_epoch integer NOT NULL DEFAULT 0;
GRANT SELECT ON public.zebridge_incarnations TO ${POSTGRES_READER_USER};
-- The catalogue carried these two columns for a day (dev): their values move here once.
DO $$ BEGIN
    IF EXISTS (SELECT 1 FROM information_schema.columns
                WHERE table_schema = 'public' AND table_name = 'zebridge_catalogue' AND column_name = 'incarnation') THEN
        INSERT INTO public.zebridge_incarnations (tbl, incarnation_relid, incarnation)
        SELECT tbl, incarnation_relid, incarnation FROM public.zebridge_catalogue WHERE incarnation > 0
        ON CONFLICT (tbl) DO NOTHING;
        ALTER TABLE public.zebridge_catalogue DROP COLUMN incarnation, DROP COLUMN incarnation_relid;
    END IF;
END $$;
GRANT SELECT ON public.zebridge_catalogue TO ${POSTGRES_READER_USER};

-- The bridge's runtime verdicts, made queryable (NOTES §10cf). A row_too_large
-- suspension lived only in the bridge's MEMORY plus the client-facing KV descriptor —
-- an operator who missed the log line had no psql answer to "which tables are
-- refused?". This table is that answer:
--
--     SELECT tbl, reason, since FROM zebridge_suspensions;
--
-- Written by the bridge through the SECURITY DEFINER setters below (the
-- zebridge_register_limits pattern — the reader role executes a narrow function
-- rather than holding write privilege on the table). Semantics mirror the in-memory
-- registry exactly: a row per refused table, set on every refusal transition, deleted
-- on every lift — and emptied WHOLESALE at boot, because the registry is memory and
-- a restart forgets (§13/§10bw: the quarantine is lazily re-earned on the next touch,
-- so a surviving row would lie in the opposite direction).
--
-- Its OWN table, deliberately NOT in the publication (NOTES §10cy). It used to be two
-- columns on zebridge_catalogue, and the catalogue rides the publication so the bridge
-- reloads on every catalogue row: each refuse and each lift therefore came back to the
-- bridge as a catalogue move and cost a reload that changed nothing — three reloads
-- per live table birth, one useful. A mirror write is not a declaration; it must not
-- look like one.
CREATE TABLE IF NOT EXISTS public.zebridge_suspensions (
    tbl    name        PRIMARY KEY,
    reason text        NOT NULL,
    since  timestamptz NOT NULL DEFAULT now()
);
GRANT SELECT ON public.zebridge_suspensions TO ${POSTGRES_READER_USER};

CREATE OR REPLACE FUNCTION public.zebridge_set_suspended(p_tbl text, p_reason text)
RETURNS void AS $$
    -- p_reason NULL = lifted. EVERY refused table records a row, catalogued or not
    -- (§10dk): `no_cdc_subject` IS the table with no catalogue row, and a projection
    -- that omitted it answered "which tables are refused?" with one missing.
    DELETE FROM public.zebridge_suspensions WHERE tbl = p_tbl AND p_reason IS NULL;
    INSERT INTO public.zebridge_suspensions (tbl, reason)
    SELECT p_tbl, p_reason
     WHERE p_reason IS NOT NULL
    ON CONFLICT (tbl) DO UPDATE SET reason = EXCLUDED.reason, since = now();
$$ LANGUAGE sql SECURITY DEFINER SET search_path = pg_catalog, public;

CREATE OR REPLACE FUNCTION public.zebridge_clear_suspensions()
RETURNS void AS $$
    DELETE FROM public.zebridge_suspensions;
$$ LANGUAGE sql SECURITY DEFINER SET search_path = pg_catalog, public;

GRANT EXECUTE ON FUNCTION public.zebridge_set_suspended(text, text) TO ${POSTGRES_READER_USER};
GRANT EXECUTE ON FUNCTION public.zebridge_clear_suspensions() TO ${POSTGRES_READER_USER};

-- Where the read side's WAL is, primary or standby (NOTES §10cz). DATABASE_READER_URL
-- may point at a hot standby (PostgreSQL ≥ 16 decodes logically there): on it
-- pg_current_wal_lsn() raises "recovery is in progress", and the honest head is what
-- has been REPLAYED. One function, replicated with the rest of the schema, so the
-- bridge asks the same question on either and carries no mode flag for it.
CREATE OR REPLACE FUNCTION public.zebridge_wal_head() RETURNS pg_lsn AS $$
    SELECT CASE WHEN pg_is_in_recovery() THEN pg_last_wal_replay_lsn()
                ELSE pg_current_wal_lsn() END;
$$ LANGUAGE sql VOLATILE;
GRANT EXECUTE ON FUNCTION public.zebridge_wal_head() TO ${POSTGRES_READER_USER};

-- Force every client to re-seed a table — and every table that REFERENCES it,
-- transitively (NOTES §10df). The one lever for what CDC cannot carry: a volatile
-- column default (now(), gen_random_uuid()), a primary-key re-key, a hand-fixed
-- dataset. A parent re-keyed invalidates the foreign-key values its children hold,
-- and PostgreSQL's rewrite of those children is just as invisible to decoding, so
-- the closure is taken here, once, rather than remembered by hand. Returns what it
-- bumped. SECURITY DEFINER on the zebridge_set_suspended pattern (§10dh): the
-- bridge's reader role pulls it when a suspension that DROPPED events lifts — the
-- rows written meanwhile reached no replica, and only a fresh full brings them —
-- and the DDL trigger pulls it on a re-key whatever role migrates. EXECUTE is
-- revoked from PUBLIC and granted to the reader only; the writer has no use for it.
CREATE OR REPLACE FUNCTION public.zebridge_reseed(p_table regclass)
RETURNS TABLE (tbl text, seed_epoch integer) AS $$
    WITH RECURSIVE closure(oid) AS (
        SELECT p_table::oid
        UNION
        SELECT con.conrelid FROM pg_constraint con JOIN closure ON con.confrelid = closure.oid
         WHERE con.contype = 'f'
    )
    -- Once per table per TRANSACTION, closure members included: a transaction-local
    -- setting marks every table this bumps, and a later call in the same transaction
    -- (a second parent re-keyed, the child's own re-type) skips the marked ones. A
    -- call outside a transaction marks nothing that outlives it. Measured before the
    -- marker: a child of two re-keyed parents was bumped three times in one migration.
    , bumped AS (
        UPDATE public.zebridge_catalogue c
           SET seed_epoch = c.seed_epoch + 1
          FROM closure JOIN pg_class r ON r.oid = closure.oid
         WHERE c.tbl = r.relname
           AND current_setting('zebridge.reseeded_' || md5(c.tbl), true) IS DISTINCT FROM '1'
        RETURNING c.tbl, c.seed_epoch
    )
    SELECT b.tbl, b.seed_epoch
      FROM bumped b, LATERAL set_config('zebridge.reseeded_' || md5(b.tbl), '1', true) AS marked;
$$ LANGUAGE sql SECURITY DEFINER SET search_path = pg_catalog, public;
REVOKE ALL ON FUNCTION public.zebridge_reseed(regclass) FROM PUBLIC;
GRANT EXECUTE ON FUNCTION public.zebridge_reseed(regclass) TO ${POSTGRES_READER_USER};

-- A table's incarnation, observed (the recovery model's §1): ONE operation, called by the
-- DDL triggers inside the DDL's own transaction (prompt), by zebridge_enable, and by the
-- generation producer when a pair finds the OID moved (the backstop for whatever the
-- triggers did not see). "Already handled" is a comparison, never a count: observing the
-- same OID twice changes nothing, so every caller may call it.
--   p_relid NULL   the table is gone: the incarnation ends (incarnation_relid NULL), nothing re-seeds;
--   same OID       nothing;
--   another OID    a new incarnation: the counter moves, and — when a previous one existed
--                  — the seed epoch moves ONCE (the per-transaction marker zebridge_reseed
--                  uses), so every replica of the old table re-seeds. Not once per tenant
--                  pair: the producer used to bump it from each pair's build (0 -> 3 for
--                  three pairs, recreate_recovery.py check 1).
-- A table not catalogued is not observed. The first observation of a table records it
-- without a re-seed (no replica can hold another incarnation of it).
-- Fresh observations only: an OID is applied only while it IS the live table's, read
-- after the row lock (so a recreate that held the lock is committed and visible). A
-- caller that read the OID earlier — the producer reads it, then calls — and lost a race
-- with another recreate supplies a stale OID: it changes nothing. Without this a delayed
-- observation of an earlier OID moved the incarnation again, bumped the epoch again and
-- recorded an OID that was no longer the table's. Likewise NULL (a DROP) applies only
-- while no table of the name exists.
DROP FUNCTION IF EXISTS public.zebridge_observe_incarnation(text, oid);
CREATE FUNCTION public.zebridge_observe_incarnation(p_tbl text, p_relid oid)
RETURNS TABLE (incarnation bigint, status text) AS $$
-- status: 'new' (a new incarnation recorded), 'current' (the OID is already the
-- recorded one), 'ended' (a DROP ended it), 'stale' (the OID — or NULL — does not
-- describe the live table: nothing changed; a caller that read it must read again).
DECLARE
    prev_relid oid;
    prev_inc   bigint;
    now_inc    bigint;
    reseed     boolean;
    live       oid;
BEGIN
    -- All identity writers take the catalogue lock first, including epoch recording.
    PERFORM 1 FROM public.zebridge_catalogue c WHERE c.tbl = p_tbl FOR UPDATE;
    IF NOT FOUND THEN
        RETURN;
    END IF;
    INSERT INTO public.zebridge_incarnations (tbl) VALUES (p_tbl) ON CONFLICT (tbl) DO NOTHING;
    SELECT i.incarnation_relid, i.incarnation INTO prev_relid, prev_inc
      FROM public.zebridge_incarnations i WHERE i.tbl = p_tbl FOR UPDATE;
    live := to_regclass(format('%I.%I', 'public', p_tbl))::oid;
    IF p_relid IS DISTINCT FROM live THEN
        RETURN QUERY SELECT prev_inc, 'stale'::text;
        RETURN;
    END IF;
    IF p_relid IS NULL THEN
        IF prev_relid IS NOT NULL THEN
            UPDATE public.zebridge_incarnations i SET incarnation_relid = NULL WHERE i.tbl = p_tbl;
        END IF;
        RETURN QUERY SELECT prev_inc, 'ended'::text;
        RETURN;
    END IF;
    IF prev_relid = p_relid THEN
        RETURN QUERY SELECT prev_inc, 'current'::text;
        RETURN;
    END IF;
    UPDATE public.zebridge_incarnations i
       SET incarnation_relid = p_relid, incarnation = i.incarnation + 1
     WHERE i.tbl = p_tbl
     RETURNING i.incarnation INTO now_inc;
    reseed := prev_inc > 0 AND current_setting('zebridge.reseeded_' || md5(p_tbl), true) IS DISTINCT FROM '1';
    IF reseed THEN
        UPDATE public.zebridge_catalogue c SET seed_epoch = c.seed_epoch + 1 WHERE c.tbl = p_tbl;
        PERFORM set_config('zebridge.reseeded_' || md5(p_tbl), '1', true);
    END IF;
    RETURN QUERY SELECT now_inc, 'new'::text;
END;
$$ LANGUAGE plpgsql SECURITY DEFINER SET search_path = pg_catalog, public;
REVOKE ALL ON FUNCTION public.zebridge_observe_incarnation(text, oid) FROM PUBLIC;
GRANT EXECUTE ON FUNCTION public.zebridge_observe_incarnation(text, oid) TO ${POSTGRES_READER_USER};
-- Every catalogued table that exists has an incarnation. Without this, a table catalogued
-- before the columns existed sat at 0, and its FIRST recreate counted as a first
-- observation: incarnation 1, no re-seed — its replicas kept the old table's rows. Only
-- tables never observed are touched (a row the epoch floor alone created is at 0), so
-- re-applying this file changes nothing.
INSERT INTO public.zebridge_incarnations AS i (tbl, incarnation_relid, incarnation)
SELECT c.tbl, to_regclass(format('%I.%I', 'public', c.tbl))::oid, 1
  FROM public.zebridge_catalogue c
 WHERE to_regclass(format('%I.%I', 'public', c.tbl)) IS NOT NULL
ON CONFLICT (tbl) DO UPDATE SET incarnation_relid = EXCLUDED.incarnation_relid, incarnation = 1
 WHERE i.incarnation = 0;

-- The seed epoch never goes back for a name. Every write of a catalogue row's epoch — an
-- enable inserting it, zebridge_reseed, an observed replacement — passes through this
-- floor: an insert or a lowering is raised to the highest epoch the name has had, and
-- that highest is recorded. Without it a removed and re-inserted catalogue row started
-- at 0, and the replacement's single move brought it back to the epoch its replicas
-- already held (1 -> 0 -> 1): a replica that missed the DROP kept the old table's rows.
CREATE OR REPLACE FUNCTION public.zebridge_catalogue_epoch_floor()
RETURNS trigger AS $$
BEGIN
    -- BEFORE INSERT also runs before an ON CONFLICT row is locked: only read here.
    NEW.seed_epoch := greatest(NEW.seed_epoch, coalesce(
        (SELECT i.seed_epoch FROM public.zebridge_incarnations i WHERE i.tbl = NEW.tbl), 0));
    RETURN NEW;
END;
$$ LANGUAGE plpgsql SECURITY DEFINER SET search_path = pg_catalog, public;
REVOKE ALL ON FUNCTION public.zebridge_catalogue_epoch_floor() FROM PUBLIC;
CREATE OR REPLACE FUNCTION public.zebridge_catalogue_epoch_record()
RETURNS trigger AS $$
DECLARE
    floor_epoch bigint;
BEGIN
    -- AFTER INSERT/UPDATE: the catalogue row is now held before the identity row.
    INSERT INTO public.zebridge_incarnations AS i (tbl, seed_epoch) VALUES (NEW.tbl, NEW.seed_epoch)
    ON CONFLICT (tbl) DO UPDATE SET seed_epoch = greatest(i.seed_epoch, EXCLUDED.seed_epoch)
    RETURNING i.seed_epoch INTO floor_epoch;
    -- An INSERT can wait behind a concurrent DELETE after its BEFORE trigger read
    -- the floor. Recheck after locking the identity; the catalogue row is ours now.
    IF NEW.seed_epoch < floor_epoch THEN
        UPDATE public.zebridge_catalogue SET seed_epoch = floor_epoch
         WHERE tbl = NEW.tbl AND seed_epoch < floor_epoch;
    END IF;
    RETURN NEW;
END;
$$ LANGUAGE plpgsql SECURITY DEFINER SET search_path = pg_catalog, public;
REVOKE ALL ON FUNCTION public.zebridge_catalogue_epoch_record() FROM PUBLIC;
-- An upgrade: the floor starts at the epochs the catalogue holds now.
INSERT INTO public.zebridge_incarnations AS i (tbl, seed_epoch)
SELECT c.tbl, c.seed_epoch FROM public.zebridge_catalogue c
ON CONFLICT (tbl) DO UPDATE SET seed_epoch = greatest(i.seed_epoch, EXCLUDED.seed_epoch);
DROP TRIGGER IF EXISTS zebridge_catalogue_epoch_floor_t ON public.zebridge_catalogue;
CREATE TRIGGER zebridge_catalogue_epoch_floor_t
    BEFORE INSERT OR UPDATE OF tbl, seed_epoch ON public.zebridge_catalogue
    FOR EACH ROW EXECUTE FUNCTION public.zebridge_catalogue_epoch_floor();
DROP TRIGGER IF EXISTS zebridge_catalogue_epoch_record_t ON public.zebridge_catalogue;
CREATE TRIGGER zebridge_catalogue_epoch_record_t
    AFTER INSERT OR UPDATE OF tbl, seed_epoch ON public.zebridge_catalogue
    FOR EACH ROW EXECUTE FUNCTION public.zebridge_catalogue_epoch_record();
-- The writer's grant lives in init.write.template.sql, NOT here: roles are
-- cluster-wide, so this line passed silently on any cluster where the write half
-- had ever run — and failed with `role does not exist` only on a fresh cluster
-- running the read-only profile, which is precisely the profile it breaks. See
-- this file's header: nothing here may reference the writer role.

-- Refuses `ALTER PUBLICATION ... ADD TABLE` for a table that is neither tenant-scoped nor
-- recorded as public. The bridge cannot check this for itself — it is a pass-through, and
-- an unscoped table looks exactly like a scoped one from the WAL — so the check belongs
-- here, where the mistake is actually made.
--
-- ⚠️ Only the table the command touched is checked. Validating the whole publication would
-- make a pre-existing unscoped table block every unrelated change, and a guard that blocks
-- unrelated work is a guard people drop.
--
-- `zebridge_scope_publication_to_one_tenant()` passes this because it creates the policies and the
-- row filter *before* adding to the publication. That ordering is load-bearing, not
-- incidental.
-- refuses an unscoped ALTER PUBLICATION — see NOTES.md §1.8
CREATE OR REPLACE FUNCTION public.zebridge_publication_guard()
 RETURNS event_trigger
 LANGUAGE plpgsql
AS $$
DECLARE
    r       record;
    relid   oid;
    scoped  boolean;
    rls     boolean;
BEGIN
    FOR r IN SELECT * FROM pg_event_trigger_ddl_commands()
             WHERE command_tag = 'ALTER PUBLICATION' AND object_type = 'publication relation'
    LOOP
        -- Only the table this command touched: a pre-existing unscoped table must not
        -- block an unrelated change, or the guard becomes something people disable.
        SELECT pr.prrelid, pr.prqual IS NOT NULL, c.relrowsecurity
          INTO relid, scoped, rls
        FROM pg_publication_rel pr JOIN pg_class c ON c.oid = pr.prrelid
        WHERE pr.oid = r.objid;

        CONTINUE WHEN relid IS NULL;
        CONTINUE WHEN relid::regclass::text LIKE '%zebridge_ddl_events';
        -- Same reasoning as zebridge_ddl_events: this table's rows are never published as
        -- ordinary CDC (the bridge special-cases its relation name and diverts every row
        -- into $KV.tenants.<principal> instead — NOTES.md §1.12 part 3). It carries the
        -- full principal→tenant roster, which the whole point of that KV bucket's
        -- principal-first key order is to avoid ever exposing wholesale; publishing it as
        -- `cdc.zebridge_user_tenants.*` would hand every client with a broad CDC grant
        -- exactly that roster.
        CONTINUE WHEN relid::regclass::text LIKE '%zebridge_user_tenants';
        -- A control table like the two above: the bridge diverts its rows into a
        -- catalogue reload and never publishes them (NOTES §10bj).
        CONTINUE WHEN relid::regclass::text LIKE '%zebridge_catalogue';
        CONTINUE WHEN EXISTS (SELECT 1 FROM public.zebridge_catalogue cat
                              WHERE cat.tbl = (SELECT relname FROM pg_class WHERE oid = relid)
                                AND cat.tenant_col IS NULL);
        -- Either model counts as scoped:
        --   A  a publication row filter  → PostgreSQL bounds CDC itself
        --   B  RLS enabled               → RLS bounds writes and snapshots, and the
        --                                  bridge bounds CDC via the catalogue's tenant_col
        -- Requiring both would refuse every model-B table; requiring neither would let an
        -- unscoped table through. So: at least one deliberate act of scoping.
        CONTINUE WHEN scoped OR rls;

        RAISE EXCEPTION
            'refusing to publish % unscoped: row filter=%, RLS=%',
            relid::regclass, scoped, rls
        USING HINT =
            'Use SELECT zebridge_scope_publication_to_one_tenant(''' || relid::regclass ||
            ''', ''<tenant_col>'', ''<tenant>'', ''<publication>''), or record the decision '
            'with SELECT zebridge_enable(''' || relid::regclass ||
            ''', public_reason => ''why everyone may read this'', dry_run => false).';
    END LOOP;
END $$;

DROP EVENT TRIGGER IF EXISTS zebridge_publication_guard_t;
CREATE EVENT TRIGGER zebridge_publication_guard_t ON ddl_command_end
    WHEN TAG IN ('ALTER PUBLICATION') EXECUTE FUNCTION public.zebridge_publication_guard();
-- Answers "is anything published without being scoped?" — the invariant a pass-through
-- bridge cannot check for itself. Run it after any publication change.
-- is anything published without being scoped? — see NOTES.md §1.8
CREATE OR REPLACE FUNCTION public.zebridge_audit_publications()
RETURNS TABLE (publication name, tbl text, row_filter text, rls boolean, verdict text) AS $$
    SELECT p.pubname,
           pr.prrelid::regclass::text,
           coalesce(pg_get_expr(pr.prqual, pr.prrelid), '(none)'),
           c.relrowsecurity,
           CASE
             WHEN pr.prrelid::regclass::text LIKE '%zebridge_ddl_events' THEN 'internal — expected'
             WHEN pr.prrelid::regclass::text LIKE '%zebridge_catalogue' THEN 'internal — expected'
             WHEN pr.prqual IS NULL AND NOT c.relrowsecurity THEN
               'PASS-THROUGH — every row reaches every subscriber of this publication'
             WHEN pr.prqual IS NULL AND c.relrowsecurity THEN
               'CDC UNSCOPED — RLS bounds the snapshot, pgoutput ignores it'
             WHEN pr.prqual IS NOT NULL AND NOT c.relrowsecurity THEN
               'reads scoped, WRITES UNSCOPED — no policy bounds what a principal may write'
             ELSE 'scoped'
           END
    FROM pg_publication_rel pr
    JOIN pg_publication p ON p.oid = pr.prpubid
    JOIN pg_class c ON c.oid = pr.prrelid
    ORDER BY p.pubname, 2;
$$ LANGUAGE sql;

-- ── making a publication is an OPERATION, not a line in a template ──────────
--
-- Nothing in these files creates a publication any more. This one used to:
-- init.core rendered `CREATE PUBLICATION` with BRIDGE_CDC_PUBLICATION substituted in from the
-- environment, so the name a bridge streams was spelled in two places — the
-- template's env and the bridge's own `--pub` — with nothing to make them agree.
--
-- And that block was not the only reader of the variable. THREE internal tables
-- get attached to the publication, and every consuming bridge needs all three:
--
--   zebridge_ddl_events    the DDL transport. Without it a bridge never learns a
--                          schema changed.
--   zebridge_gc_watermark  the offline-window watermark every client reads.
--   zebridge_user_tenants  the principal->tenant roster, diverted to $KV.tenants.
--                          Without it PROTOCOL Step 0 fails for every client of
--                          that bridge. (Write profile only.)
--
-- So the obvious recipe for a second bridge — `CREATE PUBLICATION pub_x` by hand,
-- then enables — produced a QUIET CRIPPLE: user tables replicate, DDL
-- propagation, the GC watermark and tenant resolution silently do not, and every
-- check stays green because the publication is not empty.
--
-- One fix for both: the publication is created HERE, by name, as a deliberate
-- act, and this is the only supported way to make one — so the three cannot be
-- forgotten. Idempotent, and worth re-running after init.write.template.sql to
-- pick up zebridge_user_tenants.
CREATE OR REPLACE FUNCTION public.zebridge_create_publication(p_name name)
RETURNS TABLE (step text, status text, detail text) AS $$
DECLARE
    t        text;
    -- Existence-checked one by one rather than assumed: zebridge_user_tenants
    -- lives in init.write, which a readonly-profile database never applies.
    -- zebridge_catalogue rides along so a running bridge sees zebridge_enable's row
    -- the way it sees DDL — through the WAL — and reloads its routing without a
    -- restart (NOTES §10bj). Its rows are never published as CDC.
    internal text[] := ARRAY['zebridge_ddl_events', 'zebridge_gc_watermark', 'zebridge_user_tenants', 'zebridge_catalogue'];
BEGIN
    IF p_name IS NULL OR p_name = '' THEN
        RAISE EXCEPTION 'zebridge_create_publication: no name given'
            USING HINT = 'Name it: SELECT * FROM zebridge_create_publication(''my_pub''). '
                         'The same name the bridge is given as --pub / BRIDGE_CDC_PUBLICATION.';
    END IF;

    IF EXISTS (SELECT 1 FROM pg_publication WHERE pubname = p_name) THEN
        RETURN QUERY SELECT 'publication', 'already', format('%I exists', p_name);
    ELSE
        EXECUTE format('CREATE PUBLICATION %I', p_name);
        RETURN QUERY SELECT 'publication', 'created',
            format('CREATE PUBLICATION %I — carries only the internal tables until the '
                   'first zebridge_enable', p_name);
    END IF;

    FOREACH t IN ARRAY internal LOOP
        IF to_regclass('public.' || quote_ident(t)) IS NULL THEN
            RETURN QUERY SELECT 'internal', 'ABSENT',
                format('%I does not exist in this database. If this is a write-enabled '
                       'deployment, apply init.write.template.sql and re-run this function — '
                       'a bridge on %I loses what that table carries.', t, p_name);
        ELSIF EXISTS (SELECT 1 FROM pg_publication_tables
                      WHERE pubname = p_name AND schemaname = 'public' AND tablename = t) THEN
            RETURN QUERY SELECT 'internal', 'already', format('%I already carries %I', p_name, t);
        ELSE
            EXECUTE format('ALTER PUBLICATION %I ADD TABLE public.%I', p_name, t);
            RETURN QUERY SELECT 'internal', 'added',
                format('ALTER PUBLICATION %I ADD TABLE %I', p_name, t);
        END IF;
    END LOOP;
END $$ LANGUAGE plpgsql;

-- ---------------------------------------------------------
-- ZeBridge DDL Event Tracking
-- Guarantees strict ordering of schema changes with CDC data
-- ---------------------------------------------------------

-- DDL transport: the INSERT is what reaches the bridge — see PROTOCOL.md §5
CREATE TABLE IF NOT EXISTS public.zebridge_ddl_events (
    id SERIAL PRIMARY KEY,
    schema_name TEXT NOT NULL,
    table_name TEXT NOT NULL,
    command_tag TEXT NOT NULL,
    -- Schema captured INSIDE the DDL transaction, so it is the schema as of this
    -- LSN. The bridge must not re-query the catalog at processing time: that reads
    -- "now" rather than "then", which reorders bursts and corrupts WAL replay after
    -- a restart. Shape: {"columns":[{"name":..,"type":..}], "pk":".."|null}
    schema_def JSONB,
    emitted_at TIMESTAMPTZ DEFAULT NOW()
);
-- A DROP TABLE row: the incarnation that DROP ended (NULL: not catalogued). The bridge's
-- cleanup removes only chains of that incarnation or older, never a later one's.
ALTER TABLE public.zebridge_ddl_events ADD COLUMN IF NOT EXISTS incarnation bigint;

-- Backfill for databases initialised before schema_def existed
ALTER TABLE public.zebridge_ddl_events ADD COLUMN IF NOT EXISTS schema_def JSONB;

-- Tables whose DDL is infrastructure, not application data. Publishing their schemas
-- would tell every client to build a local replica of bookkeeping it has no use for.
-- Defined once and used by BOTH triggers: keeping two copies of this list in sync by
-- hand is exactly how one of them ends up stale.
--
-- Add your migration tool's table here if it differs: flyway_schema_history,
-- alembic_version, __diesel_schema_migrations, goose_db_version, ...
-- ---------------------------------------------------------
-- GC watermark — the oldest tombstone still standing
-- ---------------------------------------------------------
--
-- `GC_THRESHOLD_MS` is a promise to clients: the maximum offline window this deployment
-- supports. A client offline for longer can resurrect a deleted row, because the tombstone
-- that would have overruled its queued edit has been reaped. Until now a client had no way
-- to know where that line sits — it could only assume the sweeper was running on schedule.
--
-- ⚠️ Carried as a **replicated table**, not published to KV by the sweeper. The sweeper is
-- a PostgreSQL client and nothing else: giving it a NATS identity would widen what a
-- compromised sweeper can reach, in order to publish one number the existing CDC pipeline
-- already carries for free. It writes a row; every client that consumes CDC gets it.
--
-- A singleton (`id = 1`, CHECKed), because there is one watermark per deployment and a
-- table that can grow to two rows will eventually have two disagreeing answers.
--
-- ⚠️ Deliberately NOT in `zebridge_is_internal_table`: internal tables are hidden from
-- clients, and this one exists precisely to reach them.
-- the GC watermark clients read before flushing an outbox — see PROTOCOL.md §7.5
CREATE TABLE IF NOT EXISTS public.zebridge_gc_watermark (
    id            smallint PRIMARY KEY DEFAULT 1 CHECK (id = 1),
    -- Nothing soft-deleted before this is guaranteed to still exist. A client whose oldest
    -- queued write predates it must re-snapshot rather than flush.
    watermark     timestamptz NOT NULL,
    -- What the sweeper was configured with, so a client can tell a *moving* watermark from
    -- a sweeper that has stopped: if `swept_at` is old, the watermark is stale whatever it
    -- says.
    threshold_ms  bigint NOT NULL,
    swept_at      timestamptz NOT NULL DEFAULT now(),
    -- Rows removed in the last pass. Zero is the normal steady state; a large number after
    -- a long quiet period usually means the sweeper was down.
    reaped        bigint NOT NULL DEFAULT 0,
    -- The version column, so this table satisfies the same contract as any other CDC
    -- table. It is never written from the edge — see the grant below.
    updated_at    timestamptz NOT NULL DEFAULT now()
);

-- The sweeper writes it; everyone else reads it through CDC. ⚠️ SELECT for the reader is
-- what puts it in snapshots; UPDATE/INSERT for the writer is what lets the sweeper stamp
-- it. No DELETE: a singleton is never removed, and the row must survive a sweeper bug.
GRANT SELECT ON public.zebridge_gc_watermark TO ${POSTGRES_READER_USER};

INSERT INTO public.zebridge_gc_watermark (id, watermark, threshold_ms)
VALUES (1, now(), 0)
ON CONFLICT (id) DO NOTHING;

-- ⚠️ The floor only moves forward, and the DATABASE says so (NOTES "One floor, one sweeper").
-- The watermark is a fact about the database's history: once a tombstone is reaped, no writer
-- can un-reap it, so no value below the floor can ever be true again. A second sweeper with a
-- shorter or longer retention, a bridge, a restored script, a forged mutation (§10aq dragged it
-- back six years with one) or a re-applied init file must not be able to lower it. The sweeper's
-- own SQL uses GREATEST, but that is the sweeper behaving; this is what holds when something else
-- does not. A raise, not a silent clamp: whoever tried should know.
--
-- Break glass, for an operator who has restored the database from a backup taken BEFORE the
-- reaping, or a test that stages a floor: ALTER TABLE public.zebridge_gc_watermark DISABLE TRIGGER
-- zebridge_gc_watermark_monotone; (and the same for _no_delete), the UPDATE, then ENABLE TRIGGER.
CREATE OR REPLACE FUNCTION public.zebridge_gc_watermark_guard() RETURNS trigger
LANGUAGE plpgsql AS $$
BEGIN
    IF TG_OP = 'UPDATE' THEN
        IF NEW.watermark < OLD.watermark THEN
            RAISE EXCEPTION 'zebridge_gc_watermark: the floor only moves forward (% -> % refused)', OLD.watermark, NEW.watermark
                USING ERRCODE = 'check_violation',
                      HINT = 'Tombstones older than the floor are gone; no value below it can be true again. A longer retention takes effect as time catches up.';
        END IF;
        RETURN NEW;
    END IF;
    RAISE EXCEPTION 'zebridge_gc_watermark: the floor is never removed (% refused)', TG_OP
        USING ERRCODE = 'check_violation';
END $$;

DROP TRIGGER IF EXISTS zebridge_gc_watermark_monotone ON public.zebridge_gc_watermark;
CREATE TRIGGER zebridge_gc_watermark_monotone BEFORE UPDATE ON public.zebridge_gc_watermark
    FOR EACH ROW EXECUTE FUNCTION public.zebridge_gc_watermark_guard();
DROP TRIGGER IF EXISTS zebridge_gc_watermark_no_delete ON public.zebridge_gc_watermark;
CREATE TRIGGER zebridge_gc_watermark_no_delete BEFORE DELETE OR TRUNCATE ON public.zebridge_gc_watermark
    FOR EACH STATEMENT EXECUTE FUNCTION public.zebridge_gc_watermark_guard();

-- ⚠️ Published, or the whole design is inert: the row reaches clients through CDC and
-- nothing else, so a table nobody publishes is a watermark nobody can read.
--
-- Published **here**, hardcoded, rather than through `zebridge_scope_publication_to_one_tenant()`.
-- That is the rule for bridge-owned tables and it is worth stating, because there are now
-- two publication paths and the difference is not obvious:
--
--   this file          the bridge's own tables (zebridge_ddl_events, zebridge_gc_watermark).
--                      They exist before any migration runs, they have no tenant, and their
--                      shape is fixed by the bridge rather than chosen by a DBA. Publishing
--                      them anywhere else would mean the bridge could boot without them.
--   a migration        application tables. Tenant-scoped ones go through
--                      `zebridge_scope_publication_to_one_tenant()`; genuinely public ones are recorded
--                      in `zebridge_catalogue` (public_reason) and added directly.
--
-- Both paths still pass through `zebridge_publication_guard`, which is why the registration
-- below is not optional even here: the guard refuses an unscoped ADD TABLE regardless of who
-- issues it, and this table genuinely is public — one row, no tenant, identical for every
-- consumer.
DO $$
BEGIN
    INSERT INTO public.zebridge_catalogue (tbl, public_reason)
    VALUES ('zebridge_gc_watermark',
            'GC watermark: one row, no tenant, every client needs it')
    ON CONFLICT (tbl) DO NOTHING;

    -- The catalogue row is written here and the publication attachment is NOT:
    -- `zebridge_create_publication` attaches this table to whichever publication
    -- is being made. The row must exist first either way — it is what carries this
    -- table past zebridge_publication_guard (tenant_col IS NULL == public), and
    -- the guard fires on the ALTER, not here.
END $$;

-- ---------------------------------------------------------
-- Generation bookkeeping — the producer's own memory (NOTES.md §1.13, delta generations)
-- ---------------------------------------------------------
--
-- One row per built generation of a (tenant, table) pair: which cutoff it used and at
-- which LSN it was taken. The generation producer reads its LAST row to compute the
-- next delta and appends its new one — from PostgreSQL, never from NATS, because "the
-- bridge never reads its own output back" is a founding invariant. Append-only with
-- pruning past the chain depth; the rows double as the audit trail ("what did G17
-- cover and when").
--
-- ⚠️ `cutoff_lsn` is stamped by the producer itself: `SELECT pg_current_wal_lsn()`
-- taken BEFORE the REPEATABLE READ snapshot that reads the content, and this row
-- committed in that same transaction — overlap-never-gap; see NOTES.md §1.13 for why
-- the reverse order loses rows.
--
-- ⚠️ NOT published and listed in `zebridge_is_internal_table`: clients must never build
-- a local replica of producer bookkeeping, and its DDL must not reach `$KV.schemas`.
CREATE TABLE IF NOT EXISTS public.zebridge_generations (
    tenant         text NOT NULL,
    tbl            text NOT NULL,
    gen            bigint NOT NULL,
    -- The version watermark this generation's content query cut at. The next delta
    -- selects `version > cutoff_version - clamp_tolerance` (the margin rule).
    cutoff_version timestamptz NOT NULL,
    -- Where CDC resumes after applying this generation's chain: clients skip events
    -- with lsn <= cutoff_lsn (every CDC event carries lsn).
    cutoff_lsn     pg_lsn NOT NULL,
    -- The previous generation's cutoff — the delta's lower bound (minus the margin).
    -- Stored, not derived: after pruning, the oldest kept delta's lower bound refers
    -- to a generation whose row is gone. NULL on a chain's first generation.
    prev_cutoff    timestamptz,
    -- Whether this generation also shipped a full object (`<tbl>-g<N>-full`): the
    -- chain's jump-in point, refreshed before it can age out of the kept window.
    has_full       boolean NOT NULL DEFAULT false,
    built_at       timestamptz NOT NULL DEFAULT now(),
    -- The catalogue's seed_epoch this generation was built under (§10df): a later
    -- bump forces the next build to be a FULL, whatever the counts say.
    seed_epoch     integer NOT NULL DEFAULT 0,
    -- The table's column shape (`name:type,…` in attnum order) at this build (§10dg):
    -- a later DROP/RENAME/re-type/ADD forces a FULL, so a chain object never names a
    -- column the replica no longer has.
    col_shape      text,
    -- The table's OID at this build (§10dl): a table dropped and re-created under the
    -- same name within one tick is a DIFFERENT table with the same chain name — the
    -- producer sweeps the old chain, bumps the epoch, and starts it at g1.
    relid          oid,
    PRIMARY KEY (tenant, tbl, gen)
);
-- The incarnation this generation was built under: chain objects are named with it
-- (`<table>-i<incarnation>-g<gen>-<kind>`), so a recreated table's g1 never shares a
-- name with its predecessor's, and whoever deletes an old chain cannot hit the new one.
ALTER TABLE public.zebridge_generations ADD COLUMN IF NOT EXISTS incarnation bigint NOT NULL DEFAULT 0;
-- 2026-09-24: the per-era compression dictionary is gone (NOTES §10iy). Measured on the
-- 668 MB test_types full: 53% smaller at 4 KB, 3% from 64 KB up — for a training pass,
-- five columns here, an object per era, and a decoder a phone did not have. Every
-- object is plain zstd, level 3. The eras cut with a dictionary are retired here in one
-- go, so the next cut is a plain full (numbered above every row, retired included);
-- their objects leave with the retirement grace, the dictionary objects among them.
DO $$ BEGIN
    IF EXISTS (SELECT 1 FROM information_schema.columns
                WHERE table_schema = 'public' AND table_name = 'zebridge_generations' AND column_name = 'dict_object') THEN
        UPDATE public.zebridge_generations SET retired_at = now()
         WHERE retired_at IS NULL AND (tenant, tbl) IN (
               SELECT tenant, tbl FROM public.zebridge_generations
                WHERE dict_object IS NOT NULL OR full_dict_object IS NOT NULL OR ckpt_dict_object IS NOT NULL);
    END IF;
END $$;
ALTER TABLE public.zebridge_generations
    DROP COLUMN IF EXISTS dict,
    DROP COLUMN IF EXISTS dict_object,
    DROP COLUMN IF EXISTS dict_ratio,
    DROP COLUMN IF EXISTS full_dict_object,
    DROP COLUMN IF EXISTS ckpt_dict_object;
-- §10gl: the lower bound of the NEXT delta. `now()` is a transaction's start, so a row
-- committed after this cut can carry a version older than the cutoff; it came from a
-- transaction already open when the cut was read. The floor is the start of the oldest
-- of those (or the cut itself), and the next delta reads `version >= floor`.
ALTER TABLE public.zebridge_generations ADD COLUMN IF NOT EXISTS open_xact_floor timestamptz;
-- §10gt: the incremental chain's middle level. A checkpoint carries every row whose
-- version moved inside its window — tombstoned rows included, which is what lets a
-- returning client learn about a delete without reloading the table — and attaches to an
-- existing generation, like a background full. `ckpt_lower` is the window's lower bound:
-- the previous checkpoint's cutoff, or the base's.
ALTER TABLE public.zebridge_generations ADD COLUMN IF NOT EXISTS has_checkpoint boolean NOT NULL DEFAULT false;
ALTER TABLE public.zebridge_generations ADD COLUMN IF NOT EXISTS ckpt_lower timestamptz;
-- §10ja: whether each object this row carries was cut in primary-key order. The manifest
-- says `"sorted": true` for those, and a client applies them without staging or sorting.
-- False for everything cut before 2026-09-25 — those objects stay readable as they were.
ALTER TABLE public.zebridge_generations ADD COLUMN IF NOT EXISTS delta_sorted boolean NOT NULL DEFAULT false;
ALTER TABLE public.zebridge_generations ADD COLUMN IF NOT EXISTS full_sorted boolean NOT NULL DEFAULT false;
ALTER TABLE public.zebridge_generations ADD COLUMN IF NOT EXISTS ckpt_sorted boolean NOT NULL DEFAULT false;
-- §10gw: the compressed size of the full or checkpoint attached to this row. The base is
-- rebuilt when the checkpoints since it outweigh it — a client applying them would
-- otherwise pay more than a reload — and that comparison needs both sides on the record.
ALTER TABLE public.zebridge_generations ADD COLUMN IF NOT EXISTS obj_bytes bigint;
-- §10gq: when this generation left the manifest. Its objects stay in the store for
-- GENERATION_RETIRE_GRACE_SECONDS after it, because a client may be READING them: a seed
-- of a big full takes minutes while a busy pair replaces its full every few seconds. One
-- retired window per pair is kept — the newest — so the cost is one full and its deltas.
ALTER TABLE public.zebridge_generations ADD COLUMN IF NOT EXISTS retired_at timestamptz;
-- §10gl: the table's relfilenode at the cut. TRUNCATE changes it (transactionally, so the
-- snapshot sees it exactly); on a guarded table it replaces count(*) as the sign that rows
-- went without a DELETE.
ALTER TABLE public.zebridge_generations ADD COLUMN IF NOT EXISTS filenode oid;
-- Databases created before the delta milestone: same columns, idempotently.
ALTER TABLE public.zebridge_generations ADD COLUMN IF NOT EXISTS prev_cutoff timestamptz;
ALTER TABLE public.zebridge_generations ADD COLUMN IF NOT EXISTS has_full boolean NOT NULL DEFAULT false;
-- The tenant-scoped row count at this generation's cutoff (NOTES §10bb). The
-- producer's skip test is version-based and cannot see a hard delete; comparing
-- count(*) against this catches it and forces a full — the only artifact that can
-- tell a client a row is gone. NULL on rows from before the column: built once.
ALTER TABLE public.zebridge_generations ADD COLUMN IF NOT EXISTS row_count bigint;
-- pg_stat_user_tables.n_tup_del for the table at this cutoff: cumulative deletes,
-- which offsetting inserts cannot hide from count(*). Moved since the last
-- generation → a full is forced (NOTES §10bb).
ALTER TABLE public.zebridge_generations ADD COLUMN IF NOT EXISTS del_count bigint;

-- ⚠️ The ONE deliberate write grant the read role holds, and why it is not a breach of
-- "the read role is physically unable to write": the generation's content query MUST
-- run as the reader (SELECT-everywhere + `zb.tenant` RLS scoping belong to it), and
-- the bookkeeping row MUST commit in that same transaction — so the same connection
-- writes it. The grant is INSERT + DELETE (pruning) on THIS table only; no UPDATE of the
-- history, so a cut is append-only by privilege, not by convention.
GRANT SELECT, INSERT, DELETE ON public.zebridge_generations TO ${POSTGRES_READER_USER};
-- §10gf: the one column-level UPDATE — the columns that say a generation also carries a
-- full or a checkpoint, and its retirement. The background lane attaches to an existing
-- generation's row; the cut itself (cutoff_version, cutoff_lsn, prev_cutoff, the counts,
-- the epoch, the shape) stays unwritable once inserted.
GRANT UPDATE (has_full, retired_at, has_checkpoint, ckpt_lower, obj_bytes, full_sorted, ckpt_sorted) ON public.zebridge_generations TO ${POSTGRES_READER_USER};

-- §10gl: the start of the oldest transaction open in this database, for the generation
-- producer's delta floor. pg_stat_activity hides other roles' sessions from the reader
-- (measured: it saw none), and pg_read_all_stats would also show it every session's query
-- text; this returns one timestamp and nothing else. The caller's own sessions are left
-- out: the reader role never writes a published table, and its snapshot transactions (a
-- background full holds one for tens of seconds) would pull the floor down for nothing.
-- The wal sender and autovacuum write no rows a delta reads.
CREATE OR REPLACE FUNCTION public.zebridge_oldest_open_xact()
RETURNS timestamptz
LANGUAGE sql STABLE SECURITY DEFINER
SET search_path = pg_catalog
AS $$
    SELECT min(xact_start) FROM pg_stat_activity
    WHERE datname = current_database()
      AND xact_start IS NOT NULL
      AND pid <> pg_backend_pid()
      AND usename IS DISTINCT FROM session_user
      AND backend_type NOT IN ('walsender', 'autovacuum worker', 'autovacuum launcher')
$$;
REVOKE ALL ON FUNCTION public.zebridge_oldest_open_xact() FROM PUBLIC;
GRANT EXECUTE ON FUNCTION public.zebridge_oldest_open_xact() TO ${POSTGRES_READER_USER};


-- Retention. The bridge reads these events from the WAL, never from the table, so
-- rows are disposable the moment they are committed — pruning cannot cause a missed
-- or mis-replayed schema event even if the bridge is down and later replays old WAL.
-- The table is kept only as a human-readable audit trail, so a short window is
-- plenty and unbounded growth is pure cost.
--
-- Pruning runs inside the DDL triggers rather than on a schedule: DDL is rare, so
-- the cost is negligible and there is no external cron to deploy or forget.
-- Deliberately a constant, not a setting. PostgreSQL has no table-level TTL, so a
-- function is the only way to express this at all — but the *value* governs nothing
-- beyond disk.
--
-- What bounds it: **nothing functional.** The bridge never reads this table, it reads the
-- WAL, so a row is disposable the moment it commits. The pruning DELETE even runs in the
-- same transaction as the INSERT that carried the schema, so no consumer can be waiting on
-- a row this removes. The only reader is a human answering "what changed on Tuesday"
-- during an incident.
--
-- 7 days was arbitrary. 2 days is arbitrary too, and smaller: a schema change nobody
-- noticed within two days is one the audit trail was not going to explain anyway, and the
-- table grows with every DDL forever otherwise. Raise it freely — nothing breaks either
-- way, which is exactly why it is not a tuneable.
-- retention for the DDL audit trail — see PROTOCOL.md §5
CREATE OR REPLACE FUNCTION public.zebridge_prune_ddl_events()
RETURNS void AS $$
BEGIN
    DELETE FROM public.zebridge_ddl_events
    WHERE emitted_at < NOW() - INTERVAL '2 days';
END;
$$ LANGUAGE plpgsql;

-- Full replica identity ensures ZeBridge gets all columns in the WAL
ALTER TABLE public.zebridge_ddl_events REPLICA IDENTITY FULL;
GRANT SELECT ON TABLE public.zebridge_ddl_events TO ${POSTGRES_READER_USER};

-- on ddl_command_end: writes the schema-change row — see PROTOCOL.md §5
-- The DDL-ready literal of a CONSTANT column default, or NULL when the default is an
-- expression (now(), nextval(), gen_random_uuid(), …). Numbers and booleans verbatim;
-- a quoted literal keeps its quotes and loses PostgreSQL's ::cast suffix. Consumed by
-- the DDL trigger below and by the bridge's boot descriptor (NOTES §10df).
CREATE OR REPLACE FUNCTION public.zebridge_constant_default(expr text) RETURNS text AS $$
    SELECT CASE
        WHEN expr IS NULL THEN NULL
        WHEN expr ~ '^-?[0-9]+(\.[0-9]+)?$' THEN expr
        WHEN expr IN ('true', 'false') THEN expr
        WHEN expr ~ '^''([^'']|'''')*''(::[a-zA-Z_ ]+(\([0-9, ]*\))?(\[\])?)?$' THEN regexp_replace(expr, '::.*$', '')
        ELSE NULL
    END;
$$ LANGUAGE sql IMMUTABLE;
GRANT EXECUTE ON FUNCTION public.zebridge_constant_default(text) TO ${POSTGRES_READER_USER};

-- The KEY SHAPE of a schema_def (§10dg): the pk columns in pk order, each with its
-- type — `id:bigint`, `tenant_id:uuid,seq:integer`. NULL for a table without a pk.
-- Compared by the DDL trigger between the previous and the new definition: a
-- changed key shape (a pk column re-typed, a column added to or taken from the pk)
-- is a RE-KEY — every row's identity moves — and is treated as a re-seed.
-- ⚠️ `pk` is the JSON scalar `null` for a keyless table, and jsonb_array_elements_text
-- raises on a scalar — inside an event trigger that raise would fail the DDL itself
-- (measured: a CREATE TABLE refused with "cannot extract elements from a scalar").
CREATE OR REPLACE FUNCTION public.zebridge_key_shape(def jsonb) RETURNS text AS $$
    SELECT CASE WHEN jsonb_typeof(def -> 'pk') = 'array' AND jsonb_typeof(def -> 'columns') = 'array' THEN (
        SELECT string_agg(p.name || ':' || COALESCE(c.type, '?'), ',' ORDER BY p.ord)
        FROM jsonb_array_elements_text(def -> 'pk') WITH ORDINALITY AS p(name, ord)
        LEFT JOIN jsonb_to_recordset(def -> 'columns') AS c(name text, type text) ON c.name = p.name)
    END;
$$ LANGUAGE sql IMMUTABLE;

CREATE OR REPLACE FUNCTION public.zebridge_ddl_trigger_fn()
RETURNS event_trigger AS $$
DECLARE
    obj         record;
    tbl         text;
    tables      text[] := '{}';
    cmd_tag     text   := 'UNKNOWN';
    catalogued  boolean;
    schema_json jsonb;
    last_def    jsonb;
    pre_def     jsonb;
    reseed_why  text;
    renames     jsonb;
BEGIN
    -- Pass 1: collect the distinct tables this command touched.
    -- object_identity is 'public.users' for a table but 'public.users.email' for a
    -- column, so split_part(..., 2) normalises both to the table name. ALTER TABLE
    -- ADD COLUMN emits a 'table' row AND a 'table column' row; deduping here stops
    -- the same schema being published twice.
    FOR obj IN SELECT * FROM pg_event_trigger_ddl_commands()
    LOOP
        IF obj.object_type IN ('table', 'table column') AND obj.schema_name = 'public' THEN
            tbl := split_part(obj.object_identity, '.', 2);
            cmd_tag := obj.command_tag;

            -- Skip infrastructure tables (our tracker, migration bookkeeping): their
            -- schemas are meaningless to clients and would provoke pointless migrations.
            IF NOT public.zebridge_is_internal_table(tbl) AND NOT (tbl = ANY(tables)) THEN
                tables := array_append(tables, tbl);
            END IF;
        END IF;
    END LOOP;

    -- §10ko: the width guard is a generated function that names the table's columns, so a
    -- DROP or RENAME COLUMN left it referring to a column that no longer exists, and every
    -- INSERT and UPDATE on the table failed ("record "new" has no field …") until someone
    -- re-ran zebridge_enable. Re-install it for every catalogued table this DDL touched:
    -- the install reads the current columns, and drops the trigger when none is needed.
    FOREACH tbl IN ARRAY tables
    LOOP
        IF to_regclass('public.' || quote_ident(tbl)) IS NOT NULL THEN
            EXECUTE format('SELECT EXISTS (SELECT 1 FROM public.zebridge_catalogue WHERE tbl = %L)', tbl)
                INTO catalogued;
            IF catalogued THEN
                PERFORM public.zebridge_install_width_guard(('public.' || quote_ident(tbl))::regclass);
                -- The incarnation, in the DDL's own transaction: a CREATE under a name
                -- that had a table is a new one (the epoch moves once, the catalogue row
                -- the bridge replicates says so at once). Any other DDL observes the same
                -- OID and changes nothing.
                BEGIN
                    PERFORM public.zebridge_observe_incarnation(tbl, to_regclass('public.' || quote_ident(tbl))::oid);
                EXCEPTION WHEN insufficient_privilege THEN
                    RAISE WARNING 'zebridge: % — this role may not observe its incarnation; the generation producer will (a re-created table re-seeds at its next build)', tbl;
                END;
            END IF;
        END IF;
    END LOOP;

    -- Pass 2: snapshot each table's schema while we are still inside the DDL
    -- transaction, so catalog state and LSN are bound together atomically.
    FOREACH tbl IN ARRAY tables
    LOOP
        SELECT jsonb_build_object(
                 -- `oid` and `typtype` are for the BRIDGE, not for clients: its CDC
                 -- decoder switches on numeric OIDs and cannot know a per-database one
                 -- (extension and user-defined types), so it needs `typtype` to tell an
                 -- enum (safe to pass through as text) from anything else (must refuse).
                 -- Produced here because the event trigger already runs inside Postgres
                 -- with the catalog open — the alternative was a blocking pg_type query
                 -- on the replication hot path. `type` is `format_type` with modifiers —
                 -- the form PROTOCOL.md §3 promises and the boot path publishes.
                 -- information_schema's data_type said `USER-DEFINED` for every
                 -- extension type and `ARRAY` for every array (§10ex): a PostGIS column
                 -- mapped to TEXT at CREATE TABLE and to BLOB at the next boot, and every
                 -- replica rebuilt the table between the two.
                 'columns', COALESCE((
                     SELECT jsonb_agg(jsonb_build_object(
                              'name', c.column_name,
                              -- §10fl: with the declared dimensions (`int[][]` is `integer[]` to
                              -- format_type; attndims keeps the count) — a DuckDB list is typed by depth
                              'type', format_type(a.atttypid, a.atttypmod) || repeat('[]', greatest(a.attndims - 1, 0)),
                              -- Required-ness, so a client can build a valid INSERT.
                              -- Without these the descriptor said which columns exist and
                              -- not which ones a write must carry: an omitted NOT NULL
                              -- column with no DEFAULT is refused by PostgreSQL (23502),
                              -- and the only way a client could learn that was the
                              -- rejection. `required` is the single fact it needs —
                              -- `nullable`/`has_default` are kept because "why" matters
                              -- when a DBA is reading the descriptor.
                              'nullable', (c.is_nullable = 'YES'),
                              'has_default', (c.column_default IS NOT NULL),
                              'required', (c.is_nullable = 'NO' AND c.column_default IS NULL),
                              -- A CONSTANT default, as a DDL-ready literal (NOTES §10df):
                              -- a client adds the column WITH it and its engine fills the
                              -- existing rows, so an ADD COLUMN ... DEFAULT x converges
                              -- without a re-seed. Volatile defaults (now(), nextval) are
                              -- NULL here — those need zebridge_reseed.
                              'default', public.zebridge_constant_default(c.column_default),
                              'oid', a.atttypid::int,
                              'typtype', t.typtype,
                              -- `attnum` is what survives a RENAME COLUMN: Postgres
                              -- renames by updating pg_attribute.attname alone, never
                              -- attnum (which is also never reused — a dropped column
                              -- keeps its attnum with attisdropped). Same-attnum,
                              -- different-name between two events IS the rename, with
                              -- no need to parse ALTER TABLE's command tag. For the
                              -- BRIDGE like oid/typtype; never forwarded to clients.
                              'attnum', a.attnum
                            ) ORDER BY c.ordinal_position)
                     FROM information_schema.columns c
                     JOIN pg_attribute a
                       ON a.attrelid = to_regclass(format('%I.%I', 'public', tbl))
                      AND a.attname = c.column_name
                      AND NOT a.attisdropped
                     JOIN pg_type t ON t.oid = a.atttypid
                     WHERE c.table_schema = 'public'
                       AND c.table_name   = tbl
                       -- §10ff: every column, column list or not. The trigger does
                       -- not know which publication a bridge follows, so it does not
                       -- guess: the bridge narrows the list to its own publication's
                       -- when it publishes the descriptor (publishedColumns).
                 ), '[]'::jsonb),
                 -- Every primary key column, in key order — NOT just single-column
                 -- keys. This used to filter on array_length(indkey,1) = 1, which made
                 -- a composite key report NULL, indistinguishable from having no key at
                 -- all. Those two cases need opposite treatment: a composite key
                 -- identifies rows exactly (CDC is fine, only snapshot pagination is
                 -- unimplemented), whereas no key at all makes DELETE ambiguous.
                 -- NULL here now means genuinely no primary key.
                 'pk', (
                     SELECT jsonb_agg(a.attname ORDER BY k.ord)
                     FROM pg_index i
                     JOIN LATERAL unnest(i.indkey) WITH ORDINALITY AS k(attnum, ord)
                       ON true
                     JOIN pg_attribute a
                       ON a.attrelid = i.indrelid AND a.attnum = k.attnum
                     -- to_regclass, not ::regclass: returns NULL instead of raising
                     -- if the table is gone (e.g. dropped later in the same tx)
                     WHERE i.indrelid = to_regclass(format('%I.%I', 'public', tbl))
                       AND i.indisprimary
                 ),
                 -- Secondary indexes, so a replica can be QUERIED rather than only
                 -- scanned. "Query your replica directly — joins, aggregates, offline"
                 -- is the product's central promise, and without these a 100k-row table
                 -- answers every predicate with a sequential scan.
                 --
                 -- Dialect-NEUTRAL and at the root, like 'pk': CREATE [UNIQUE] INDEX
                 -- name ON tbl (cols) is the same statement in PostgreSQL/PGlite and in
                 -- SQLite, so one list serves both consumer shapes.
                 --
                 -- Only what actually translates:
                 --   NOT indisprimary — the PK is created inline with the table
                 --   indpred IS NULL  — partial indexes carry a PG WHERE expression
                 --   indexprs IS NULL — expression indexes carry PG expressions
                 --   amname = 'btree' — gin/gist/brin have no SQLite equivalent
                 --   not <table>_zb_version — zebridge_enable's index for the producer's
                 --     delta query (§10gc); a replica never reads by version, and the
                 --     index would slow its seed and every applied update (§10gd)
                 -- Anything excluded is a performance loss on the replica, never a
                 -- correctness one: an index is never the reason a row is right.
                 'indexes', COALESCE((
                     SELECT jsonb_agg(jsonb_build_object(
                              'name', ci.relname,
                              'unique', i.indisunique,
                              'columns', (
                                  SELECT jsonb_agg(a.attname ORDER BY k.ord)
                                  FROM unnest(i.indkey) WITH ORDINALITY AS k(attnum, ord)
                                  JOIN pg_attribute a
                                    ON a.attrelid = i.indrelid AND a.attnum = k.attnum
                              )
                            ) ORDER BY ci.relname)
                     FROM pg_index i
                     JOIN pg_class ci ON ci.oid = i.indexrelid
                     JOIN pg_am am ON am.oid = ci.relam
                     WHERE i.indrelid = to_regclass(format('%I.%I', 'public', tbl))
                       AND NOT i.indisprimary
                       AND i.indpred IS NULL
                       AND i.indexprs IS NULL
                       AND am.amname = 'btree'
                       AND ci.relname NOT LIKE '%\_zb\_version'
                 ), '[]'::jsonb),
                 -- Foreign keys, so a replica can enforce referential integrity
                 -- rather than merely resemble it (PROTOCOL §4; NOTES §10d).
                 --
                 -- ⚠️ Filtered to what SQLite will ACCEPT, which is stricter than what
                 -- PostgreSQL allows. SQLite requires the parent key to be the parent's
                 -- PRIMARY KEY or to carry a UNIQUE INDEX, and refuses anything else at
                 -- DML time with `foreign key mismatch` — a failure no amount of waiting
                 -- resolves. Measured, both directions. So a constraint travels only if
                 -- its parent columns are the parent's PK, or are covered by a unique
                 -- index we ACTUALLY PORT (btree, non-partial, non-expression): a parent
                 -- key unique only under a partial or expression index is unique to
                 -- PostgreSQL and unsatisfiable on the replica.
                 --
                 -- Composite keys included: confrelid/conkey/confkey are arrays, and the
                 -- child and parent column lists must stay in matching order.
                 'foreign_keys', COALESCE((
                     SELECT jsonb_agg(jsonb_build_object(
                              'name', con.conname,
                              'columns', child_cols.cols,
                              'references', pc.relname,
                              'parent_columns', parent_cols.cols
                            ) ORDER BY con.conname)
                     FROM pg_constraint con
                     JOIN pg_class pc ON pc.oid = con.confrelid
                     JOIN LATERAL (
                         SELECT jsonb_agg(a.attname ORDER BY k.ord) AS cols
                         FROM unnest(con.conkey) WITH ORDINALITY AS k(attnum, ord)
                         JOIN pg_attribute a ON a.attrelid = con.conrelid AND a.attnum = k.attnum
                     ) child_cols ON true
                     JOIN LATERAL (
                         SELECT jsonb_agg(a.attname ORDER BY k.ord) AS cols,
                                array_agg(k.attnum ORDER BY k.ord) AS attnums
                         FROM unnest(con.confkey) WITH ORDINALITY AS k(attnum, ord)
                         JOIN pg_attribute a ON a.attrelid = con.confrelid AND a.attnum = k.attnum
                     ) parent_cols ON true
                     WHERE con.conrelid = to_regclass(format('%I.%I', 'public', tbl))
                       AND con.contype = 'f'
                       -- the parent table must itself be replicated, or the child
                       -- references something the replica will never have
                       -- ⚠️ catalogue.tbl is TEXT (the bare relname), not regclass —
                       -- comparing it to an oid raises `operator does not exist:
                       -- text = oid`, which inside this trigger would fail the DDL
                       -- itself. Same comparison the generation producer uses.
                       AND EXISTS (SELECT 1 FROM public.zebridge_catalogue cat
                                    WHERE cat.tbl = pc.relname)
                       -- parent key is the PK, or has a unique index we port
                       AND EXISTS (
                           SELECT 1 FROM pg_index i
                           JOIN pg_class ci ON ci.oid = i.indexrelid
                           JOIN pg_am am ON am.oid = ci.relam
                           WHERE i.indrelid = con.confrelid
                             AND i.indisunique
                             AND i.indpred IS NULL AND i.indexprs IS NULL
                             AND am.amname = 'btree'
                             AND i.indkey::int2[] @> parent_cols.attnums::int2[]
                             AND parent_cols.attnums::int2[] @> i.indkey::int2[]
                       )
                 ), '[]'::jsonb),
                 -- Replica identity governs what UPDATE/DELETE actually carry, so both
                 -- the bridge and its clients need it: without a PK, DEFAULT means
                 -- PostgreSQL rejects writes outright, and only FULL yields old.* values
                 -- (and therefore working transition rules). Neither is discoverable
                 -- from the column list alone.
                 'replica_identity', (
                     SELECT CASE c.relreplident
                              WHEN 'd' THEN 'default'
                              WHEN 'f' THEN 'full'
                              WHEN 'n' THEN 'nothing'
                              WHEN 'i' THEN 'index'
                            END
                     FROM pg_class c
                     WHERE c.oid = to_regclass(format('%I.%I', 'public', tbl))
                 )
               )
        INTO schema_json;

        -- No columns means the table did not survive the transaction; nothing useful
        -- to publish, and an empty schema would tell clients to drop every column.
        IF schema_json -> 'columns' <> '[]'::jsonb THEN
            -- One migration commonly fires several ddl_command_end events for the same
            -- table (CREATE TABLE, then ALTER TABLE ... REPLICA IDENTITY, then a GRANT).
            -- Emitting an identical schema each time makes clients re-run a migration
            -- that changes nothing. Compare against the newest row for this table and
            -- skip if nothing actually changed — replica_identity is part of schema_def,
            -- so a genuine identity change still produces an event.
            SELECT e.schema_def INTO last_def
            FROM public.zebridge_ddl_events e
            WHERE e.table_name = tbl
            ORDER BY e.id DESC
            LIMIT 1;

            -- RENAME COLUMN detection (§1.2): same attnum, different name between this
            -- schema and the previous event's. Without this hint a rename reads as
            -- drop+add and the client nulls every existing value of the renamed column.
            -- Emitted as {"new_name": "old_name"}. An older last_def without attnum
            -- joins nothing and yields NULL — renames are simply not detected across
            -- the trigger upgrade, which is the pre-existing behaviour.
            renames := NULL;
            IF last_def IS NOT NULL THEN
                SELECT jsonb_object_agg(n.name, o.name) INTO renames
                FROM jsonb_to_recordset(schema_json -> 'columns') AS n(name text, attnum int)
                JOIN jsonb_to_recordset(last_def -> 'columns') AS o(name text, attnum int)
                  ON n.attnum = o.attnum AND n.name <> o.name;
            END IF;
            IF renames IS NOT NULL THEN
                schema_json := schema_json || jsonb_build_object('renamed', renames);
            END IF;

            -- Compare with 'renamed' stripped from BOTH sides: it describes the
            -- transition, not the shape. Comparing it too would emit one spurious
            -- schema event on the first unrelated DDL after a rename.
            IF (last_def - 'renamed') IS DISTINCT FROM (schema_json - 'renamed') THEN
                INSERT INTO public.zebridge_ddl_events
                       (schema_name, table_name, command_tag, schema_def)
                VALUES ('public', tbl, cmd_tag, schema_json);
            END IF;

            -- §10dg: two shapes of DDL that rewrite VALUES a replica holds, without a
            -- decoded event to carry them — so a re-seed is the only way back:
            --   * a changed KEY SHAPE — a pk column re-typed (bigserial → uuid), a
            --     column added to or taken from the pk: every row's identity moves;
            --   * a RE-TYPED column — same name, different type (ALTER COLUMN TYPE …
            --     USING rewrites the table with no row events at all).
            -- Compared against the definition from BEFORE this transaction, not the
            -- previous step's: a real re-key is many statements (add the new column,
            -- populate, drop the old, rename, add the pk) whose per-step diffs read as
            -- add/remove/rename and never as a re-type. Bumped ONCE per table per
            -- transaction (a transaction-local setting remembers), through the FK
            -- closure of zebridge_reseed: the producer's next build is a full, and each
            -- replica rebuilds/re-types locally and seeds afresh. The DDL must not fail
            -- on a role that may not touch the catalogue: the bump degrades to a
            -- warning, and `SELECT zebridge_reseed('<table>')` by hand completes it.
            SELECT e.schema_def INTO pre_def
            FROM public.zebridge_ddl_events e
            WHERE e.table_name = tbl AND e.emitted_at < now()
            ORDER BY e.id DESC
            LIMIT 1;
            IF pre_def IS NOT NULL
               AND current_setting('zebridge.reseeded_' || md5(tbl), true) IS DISTINCT FROM '1' THEN
                reseed_why := NULL;
                IF public.zebridge_key_shape(pre_def) IS DISTINCT FROM public.zebridge_key_shape(schema_json) THEN
                    reseed_why := format('key shape %s -> %s', public.zebridge_key_shape(pre_def), public.zebridge_key_shape(schema_json));
                ELSE
                    SELECT string_agg(n.name || ' ' || o.type || ' -> ' || n.type, ', ') INTO reseed_why
                    FROM jsonb_to_recordset(schema_json -> 'columns') AS n(name text, type text)
                    JOIN jsonb_to_recordset(pre_def -> 'columns') AS o(name text, type text) ON o.name = n.name
                    WHERE o.type IS DISTINCT FROM n.type;
                    IF reseed_why IS NOT NULL THEN reseed_why := 're-typed: ' || reseed_why; END IF;
                END IF;
                IF reseed_why IS NOT NULL THEN
                    BEGIN
                        PERFORM public.zebridge_reseed(to_regclass(format('%I.%I', 'public', tbl)));
                        RAISE NOTICE 'zebridge: % (%): seed epoch bumped, re-seed forced', tbl, reseed_why;
                    EXCEPTION WHEN insufficient_privilege THEN
                        RAISE WARNING 'zebridge: % (%) but this role may not bump its seed epoch — run SELECT public.zebridge_reseed(''%'') as the owner', tbl, reseed_why, tbl;
                    END;
                END IF;
            END IF;
        END IF;
    END LOOP;

    PERFORM public.zebridge_prune_ddl_events();
END;
$$ LANGUAGE plpgsql;

-- ⚠️ This trigger is the ONLY way a schema change reaches a client — PROTOCOL.md §0: the bridge never
-- polls the catalog for shape, so a table whose DDL does not land here keeps serving its
-- old schema forever. Tested end to end by scripts/scenarios/invalidate.py §1-2.
DROP EVENT TRIGGER IF EXISTS zebridge_ddl_trigger;
CREATE EVENT TRIGGER zebridge_ddl_trigger ON ddl_command_end
EXECUTE FUNCTION public.zebridge_ddl_trigger_fn();

-- DROP TABLE never reaches ddl_command_end — by the time it fires the object is
-- gone and pg_event_trigger_ddl_commands() does not report it. Drops therefore
-- need their own trigger on sql_drop, or clients never learn a table disappeared
-- and keep a local replica of something that no longer exists upstream.
-- on sql_drop: drops never reach ddl_command_end — see PROTOCOL.md §5
CREATE OR REPLACE FUNCTION public.zebridge_drop_trigger_fn()
RETURNS event_trigger AS $$
DECLARE
    obj record;
    tbl text;
    dropped_inc bigint;
BEGIN
    FOR obj IN SELECT * FROM pg_event_trigger_dropped_objects()
    LOOP
        IF obj.object_type = 'table' AND obj.schema_name = 'public' THEN
            tbl := split_part(obj.object_identity, '.', 2);

            IF NOT public.zebridge_is_internal_table(tbl) THEN
                -- schema_def is NULL rather than '{"columns":[]}': the table is gone,
                -- so there is no schema to state. The bridge treats a DROP TABLE tag
                -- as "forget this table" and never reads schema_def for it.
                -- Through EXECUTE: `tbl` is this function's variable AND the catalogue's
                -- column, and plpgsql refuses the ambiguity at run time.
                EXECUTE format('SELECT incarnation FROM public.zebridge_incarnations WHERE tbl = %L', tbl)
                    INTO dropped_inc;
                INSERT INTO public.zebridge_ddl_events
                       (schema_name, table_name, command_tag, schema_def, incarnation)
                VALUES ('public', tbl, 'DROP TABLE', NULL, dropped_inc);
                -- The incarnation ends here, in the DROP's transaction: a later CREATE
                -- under the name is a new one even if PostgreSQL reuses the OID.
                BEGIN
                    PERFORM public.zebridge_observe_incarnation(tbl, NULL);
                EXCEPTION WHEN insufficient_privilege THEN
                    RAISE WARNING 'zebridge: % dropped — this role may not end its incarnation; the generation producer will see the new OID', tbl;
                END;

                -- Reap the table's generated width guard. `zebridge_install_width_guard`
                -- creates ONE FUNCTION PER TABLE, and DROP TABLE ... CASCADE removes the
                -- trigger (which belongs to the table) but NOT the function (which does
                -- not) — so every dropped table used to leave a dead
                -- zebridge_width_guard_<tbl>() behind, permanently. Measured: two were
                -- found in a running database, and one still carried the pre-2026-08-26
                -- `WHERE id = 1` budget lookup against a column that no longer exists,
                -- so calling it raised. Dead code that is also a landmine.
                --
                -- Here rather than in the bridge because it is DDL about the database's
                -- own objects, it runs in the same transaction as the DROP, and it needs
                -- no connection the bridge would have to open. `IF EXISTS` because a
                -- table with no unbounded columns never had a guard installed.
                EXECUTE format('DROP FUNCTION IF EXISTS public.%I()',
                               'zebridge_width_guard_' || tbl);
            END IF;
        END IF;
    END LOOP;

    PERFORM public.zebridge_prune_ddl_events();
END;
$$ LANGUAGE plpgsql;

DROP EVENT TRIGGER IF EXISTS zebridge_drop_trigger;
CREATE EVENT TRIGGER zebridge_drop_trigger ON sql_drop
EXECUTE FUNCTION public.zebridge_drop_trigger_fn();

-- No publication attachment here: `zebridge_create_publication` adds this table
-- to every publication it makes. Attaching it to one name rendered from the
-- environment is what made a second, hand-made publication a quiet cripple —
-- it replicated user tables and never carried a schema change.

-- Refuses CREATE/ALTER TABLE that introduces a `timestamp without time zone` column.
-- The version protocol depends on it: version columns travel in §7.2's wire format
-- (UTC, trailing Z), are compared and clamped as absolute instants, and a naive
-- timestamp makes two writers in different zones disagree about which write is newer —
-- silently, per row. The rule lived in prose ("use timestamptz") until a migration
-- forgot it; this is the mechanical form, same pattern as zebridge_publication_guard:
-- refuse at the source, inside the DDL transaction, so the migration fails whole.
--
-- ⚠️ `zebridge_is_internal_table` names are exempt — Ecto's own `schema_migrations`
-- uses naive timestamps by design, and blocking it would stop `mix ecto.migrate`'s
-- very first statement on a fresh database. Scoped to ordinary tables in `public`;
-- a deliberate exception is a DBA act:
--   ALTER EVENT TRIGGER zebridge_timestamp_guard_t DISABLE;  -- migrate; then ENABLE
-- refuses naive-timestamp columns — see SECURITY.md §1.3
CREATE OR REPLACE FUNCTION public.zebridge_timestamp_guard()
RETURNS event_trigger AS $$
DECLARE
    r       record;
    bad_col text;
    rel     text;
BEGIN
    FOR r IN SELECT * FROM pg_event_trigger_ddl_commands()
             WHERE command_tag IN ('CREATE TABLE', 'CREATE TABLE AS', 'ALTER TABLE')
               AND object_type IN ('table', 'table column')
    LOOP
        SELECT c.relname INTO rel
        FROM pg_class c
        JOIN pg_namespace n ON n.oid = c.relnamespace
        WHERE c.oid = r.objid AND n.nspname = 'public' AND c.relkind IN ('r', 'p');
        CONTINUE WHEN rel IS NULL;
        CONTINUE WHEN public.zebridge_is_internal_table(rel);

        SELECT a.attname INTO bad_col
        FROM pg_attribute a
        JOIN pg_type t ON t.oid = a.atttypid
        WHERE a.attrelid = r.objid
          AND t.typname = 'timestamp'
          AND a.attnum > 0 AND NOT a.attisdropped
        LIMIT 1;

        IF FOUND THEN
            RAISE EXCEPTION
                'migration rejected: column "%" on table "%" is "timestamp without time zone" — use timestamptz (Ecto: timestamps(type: :timestamptz)). Version comparison and the §7.2 wire format need absolute instants.',
                bad_col, rel;
        END IF;
        -- §10dn: the same contradiction zebridge_enable refuses, caught when the
        -- constraint arrives AFTER the enable — the catalogue names the tombstone.
        IF EXISTS (SELECT 1 FROM public.zebridge_catalogue cat WHERE cat.tbl = rel AND cat.tombstone_col IS NOT NULL) THEN
            SELECT con.conname INTO bad_col
            FROM pg_constraint con
            WHERE con.conrelid = r.objid AND con.contype = 'f' AND con.confdeltype = 'c'
            LIMIT 1;
            IF FOUND THEN
                RAISE EXCEPTION
                    'migration rejected: foreign key "%" on "%" is ON DELETE CASCADE, but the table keeps tombstones (zebridge_catalogue). A cascade removes rows the bridge never forwards for a tombstone table. Declare it ON DELETE NO ACTION.',
                    bad_col, rel;
            END IF;
        END IF;
        rel := NULL;
    END LOOP;
END;
$$ LANGUAGE plpgsql;

DROP EVENT TRIGGER IF EXISTS zebridge_timestamp_guard_t;
CREATE EVENT TRIGGER zebridge_timestamp_guard_t ON ddl_command_end
    WHEN TAG IN ('CREATE TABLE', 'CREATE TABLE AS', 'ALTER TABLE')
    EXECUTE FUNCTION public.zebridge_timestamp_guard();

-- ─────────────────────────────────────────────────────────────────────────────
-- The T2 entry point — one call instead of six remembered ones (NOTES.md §4.5)
-- ─────────────────────────────────────────────────────────────────────────────
--
-- Activation is spread over four places and two times: the templates *define* functions
-- at bootstrap, the DBA *invokes* them per table after the application's migrations, and
-- two more settings live outside the database entirely. Nothing here can set those last
-- two — so this composes what it can and **prints the rest**, which is the difference
-- between a workflow you can follow and one you have to remember.
--
-- ⚠️ `dry_run` defaults to TRUE. This grants, guards and enables RLS in one statement;
-- showing the plan first is the cheap half of that trade. Pass `dry_run => false` to apply.
--
--     SELECT * FROM zebridge_enable('public.orders');                       -- plan only
--     SELECT * FROM zebridge_enable('public.orders',
--                                   tenant_col => 'tenant_id',
--                                   writable   => true,
--                                   version_col => 'updated_at',
--                                   tombstone_col => 'deleted_at',
--                                   dry_run    => false);
--
-- ⚠️ Lives in the core half but dispatches the write-path calls through EXECUTE, so a
-- read-only install can still use it for the one step it needs (publishing a table) and
-- fails with a *named* reason rather than "function does not exist" if asked for writes.
-- ─────────────────────────────────────────────────────────────────────────────
-- The change-feed row-width budget, and the guard that enforces it AT THE ROW.
--
-- Two doors write rows: the edge (mutations through the bridge) and psql/backend
-- apps. The bridge's ingress check measures only the mutation PAYLOAD, so a small
-- edit to an already-wide row — or any direct backend write — could commit a row
-- the change feed cannot carry, and the first CDC touch would then suspend the
-- table for every reader ("accepted, then everybody freezes" — NOTES.md §1.13,
-- case C). A BEFORE trigger closes BOTH doors at the only place they meet: the
-- row, inside the writer's own transaction. The edge write becomes a verdict
-- (`rejected`, SQLSTATE 23514 — class 23 is already classified permanent by the
-- mutation listener); the psql write becomes an ordinary ERROR. Rules that only
-- live in prose are the ones that bite.
--
-- ⚠️ The budget is PER BRIDGE INSTANCE, not per database. `BASE_BUF` is a
-- per-process memory setting, so two bridges on the same database can legitimately
-- carry different row ceilings; a single global limit was simply wrong. It was also
-- a coupling maintained BY HAND — SECURITY.md said so, and it was forgotten the
-- first time BASE_BUF changed (measured: buffer 4 KB, table still 16384, so
-- PostgreSQL would accept rows the feed could not carry and suspend the table).
--
-- So the bridge WRITES this table at boot from its own resolved BASE_BUF. Nobody
-- maintains it; there is nothing to forget.
--
--   slot          — the instance. It is the identity PostgreSQL already keeps
--                   (`pg_replication_slots`), so a retired bridge's row is detectable
--                   and GC'd on the next boot of any bridge. Keyed by publication
--                   instead, a decommissioned 4 KB instance would pin the budget low
--                   forever with nothing able to tell.
--   publication   — what it carries. `pg_publication_tables` expands it to tables.
--   max_row_bytes — that instance's `2^BASE_BUF`.
--
-- ⚠️ This table is the INPUT, not what the guard reads. Each table's guard carries its
-- budget as a LITERAL, baked at boot by `zebridge_register_limits` from
--
--     zebridge_limits (slot -> budget)  x  pg_publication_tables (publication -> tables)
--
-- taking MIN per table: a row must fit the NARROWEST instance carrying it, or it is
-- accepted here and suspends the feed over there. Measured, and the reason the join is
-- not in the trigger: a per-row publication join cost 22 us against 2.2 for a plain
-- lookup, and a plain lookup still cost +4.81 us/row against a baked literal, which is
-- free. The cold path resolves; the hot path reads nothing at all.
CREATE TABLE IF NOT EXISTS public.zebridge_limits (
    slot          text PRIMARY KEY,
    publication   name    NOT NULL,
    max_row_bytes integer NOT NULL,
    updated_at    timestamptz NOT NULL DEFAULT now(),
    -- The instance's slot lives on a STANDBY (§10cz): this primary never lists it, so
    -- the GC below must not read its absence as "retired". Such a row is never reaped;
    -- it is replaced when the same slot registers again. A stale one keeps the budget
    -- at the MIN — the safe direction.
    on_standby    boolean NOT NULL DEFAULT false
);
GRANT SELECT ON public.zebridge_limits TO ${POSTGRES_READER_USER};

-- The bridge registers its OWN budget here at boot. One call, over the connection
-- that always exists (the reader's), so this works identically in the read-only and
-- read/write profiles — a read-only deployment still has a width guard, and it is
-- still the truth about that instance's buffer.
--
-- ⚠️ SECURITY DEFINER, and that is the whole point: ${POSTGRES_READER_USER} keeps
-- SELECT + REPLICATION and NO write privilege on any table. "The reader is
-- physically unable to write" stays literally true and testable in
-- `role_table_grants` — the reader's only power here is to call this one function,
-- which writes rows describing itself and nothing else. Granting the reader
-- INSERT/UPDATE on the table instead would have bought the same behaviour by
-- weakening the one invariant the whole read path leans on.
--
-- Three steps, in this order:
--   1. GC instances PostgreSQL no longer knows — a retired bridge must stop
--      constraining everyone else, and its slot vanishing is the signal. Any
--      bridge's boot cleans up after every dead one.
--   2. drop this slot's rows for tables it no longer carries.
--   3. upsert one row per table currently in its publication.
--
-- ⚠️ Call it AFTER the replication slot exists, or step 1 deletes what step 3
-- writes on the next boot.
CREATE OR REPLACE FUNCTION public.zebridge_register_limits(
    p_slot          text,
    p_publication   name,
    p_max_row_bytes integer,
    p_on_standby    boolean DEFAULT false
) RETURNS integer AS $$
DECLARE
    r       record;
    n_guard integer := 0;
BEGIN
    IF p_max_row_bytes IS NULL OR p_max_row_bytes < 1024 THEN
        RAISE EXCEPTION 'refusing a row-width budget of %: below any sane floor', p_max_row_bytes
            USING ERRCODE = 'check_violation';
    END IF;

    -- 1. GC instances PostgreSQL no longer knows. A retired bridge must stop
    --    constraining everyone else, and its slot vanishing is the signal. An
    --    INACTIVE slot is NOT gone — that bridge is down, its WAL is retained, and
    --    it will replay; its budget still binds.
    --    A slot on a STANDBY is invisible from here and is not gone (on_standby).
    DELETE FROM public.zebridge_limits l
     WHERE NOT l.on_standby
       AND NOT EXISTS (SELECT 1 FROM pg_replication_slots s WHERE s.slot_name = l.slot);

    -- 2. this instance's own row.
    INSERT INTO public.zebridge_limits (slot, publication, max_row_bytes, updated_at, on_standby)
    VALUES (p_slot, p_publication, p_max_row_bytes, now(), p_on_standby)
    ON CONFLICT (slot) DO UPDATE
       SET publication   = EXCLUDED.publication,
           max_row_bytes = EXCLUDED.max_row_bytes,
           updated_at    = now(),
           on_standby    = EXCLUDED.on_standby;

    -- 3. bake the literal into every guard this instance carries.
    --
    -- MIN over the instances that carry the table: a row must fit the NARROWEST of
    -- them. Taking MAX would admit rows the smallest instance cannot encode, and it
    -- would suspend the table for every reader on ITS feed while the wider instance
    -- carried on — a partial, per-feed outage that reads as a mystery.
    FOR r IN
        SELECT format('%I.%I', pt.schemaname, pt.tablename)::regclass AS tbl,
               MIN(l.max_row_bytes) AS budget
          FROM public.zebridge_limits l
          JOIN pg_publication_tables pt ON pt.pubname = l.publication
         WHERE format('%I.%I', pt.schemaname, pt.tablename)::regclass IN (
                   SELECT format('%I.%I', p2.schemaname, p2.tablename)::regclass
                     FROM pg_publication_tables p2 WHERE p2.pubname = p_publication)
         GROUP BY 1
    LOOP
        IF public.zebridge_rebudget_width_guard(r.tbl, r.budget) THEN
            n_guard := n_guard + 1;
        END IF;
    END LOOP;

    RETURN n_guard;
END;
$$ LANGUAGE plpgsql SECURITY DEFINER SET search_path = pg_catalog, public;

-- Rewrite ONE guard's budget, body only. Returns true if a guard was rewritten.
--
-- ⚠️ **Never touches the TRIGGER**, and that is the whole point of it existing
-- separately from `zebridge_install_width_guard`. Measured against a table with one
-- open writer: `CREATE OR REPLACE FUNCTION` completed in 203 ms, while `DROP TRIGGER`
-- blocked past a 6 s timeout waiting for ACCESS EXCLUSIVE. A pending exclusive lock
-- also queues AHEAD of new writers, so doing that at every bridge boot would freeze a
-- busy table for the length of someone else's transaction. Replacing the body needs no
-- table lock at all: the trigger keeps pointing at the same function, and PostgreSQL
-- invalidates the cached plan, so even a pool connection open for months uses the new
-- literal on its very next write (measured both ways).
--
-- The width EXPRESSION is regenerated too, since the column set may have changed; only
-- the trigger creation is skipped, and only when the trigger is already there.
CREATE OR REPLACE FUNCTION public.zebridge_rebudget_width_guard(tbl regclass, p_budget integer)
RETURNS boolean AS $$
DECLARE
    short       text := (SELECT relname FROM pg_class WHERE oid = tbl);
    expr        text := '';
    n_cols      int  := 0;
    n_unbounded int  := 0;
    col         record;
    fn_body     text;
BEGIN
    -- A table with no guard installed has none to rebudget. Installing one here would
    -- need the trigger DDL this function exists to avoid.
    IF NOT EXISTS (SELECT 1 FROM pg_trigger g
                    WHERE g.tgrelid = tbl AND g.tgname = 'zebridge_width_guard'
                      AND NOT g.tgisinternal) THEN
        RETURN false;
    END IF;

    FOR col IN
        SELECT a.attname, t.typname, a.atttypmod, t.typcategory
        FROM pg_attribute a JOIN pg_type t ON t.oid = a.atttypid
        WHERE a.attrelid = tbl AND a.attnum > 0 AND NOT a.attisdropped
    LOOP
        n_cols := n_cols + 1;
        -- Bytes ride the feed as MessagePack bin (§10ex): the bytes plus a header of at
        -- most five. The hex form (2 per byte + 2) this measured until 2026-09-12 counted
        -- every bytea twice and refused a 300 KB tile at 600 KB. PostGIS geometry and
        -- geography ride as EWKB, which pg_column_size approximates from above without
        -- calling a PostGIS function from a guard that must work without the extension.
        IF col.typname = 'bytea' THEN
            n_unbounded := n_unbounded + 1;
            expr := expr || format(' + coalesce(octet_length(NEW.%I) + 5, 0)', col.attname);
        -- §10fg: pgvector and bit(n) ride as bytes too, a header lighter than the
        -- stored form pg_column_size reports — again an approximation from above.
        ELSIF col.typname IN ('geometry', 'geography', 'vector', 'halfvec', 'sparsevec', 'bit') THEN
            n_unbounded := n_unbounded + 1;
            expr := expr || format(' + coalesce(pg_column_size(NEW.%I), 0)', col.attname);
        ELSIF col.typname IN ('text', 'json', 'jsonb', 'xml')
              OR (col.typname = 'varchar' AND col.atttypmod = -1)
              OR col.typcategory = 'A' THEN
            n_unbounded := n_unbounded + 1;
            expr := expr || format(' + coalesce(octet_length(NEW.%I::text), 0)', col.attname);
        END IF;
    END LOOP;

    IF n_unbounded = 0 THEN
        RETURN false;
    END IF;

    fn_body := format(
        'DECLARE '
        '  budget integer := %s; '
        '  width  integer := %s + %s; '
        'BEGIN '
        '  IF width >= budget THEN '
        '    RAISE EXCEPTION ''row width %% bytes is at or over the %%-byte change-feed budget: the feed could not carry this row (zebridge_limits, narrowest instance carrying this table; store a reference, not the blob)'', width, budget '
        '      USING ERRCODE = ''check_violation''; '
        '  END IF; '
        '  RETURN NEW; '
        'END',
        p_budget, substr(expr, 4), 32 * n_cols + 256);

    EXECUTE 'CREATE OR REPLACE FUNCTION public.' || quote_ident('zebridge_width_guard_' || short)
         || '() RETURNS trigger AS ' || quote_literal(fn_body)
         || ' LANGUAGE plpgsql SECURITY DEFINER SET search_path = pg_catalog, public';
    -- A guard created here for the first time would keep PUBLIC's EXECUTE (CREATE OR
    -- REPLACE keeps an existing function's grants, so a re-bake changes nothing).
    EXECUTE format('REVOKE EXECUTE ON FUNCTION public.%I() FROM PUBLIC', 'zebridge_width_guard_' || short);
    IF EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'anon') THEN
        EXECUTE format('REVOKE EXECUTE ON FUNCTION public.%I() FROM anon', 'zebridge_width_guard_' || short);
    END IF;
    IF EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'authenticated') THEN
        EXECUTE format('REVOKE EXECUTE ON FUNCTION public.%I() FROM authenticated', 'zebridge_width_guard_' || short);
    END IF;
    RETURN true;
END;
$$ LANGUAGE plpgsql SECURITY DEFINER SET search_path = pg_catalog, public;

GRANT EXECUTE ON FUNCTION public.zebridge_register_limits(text, name, integer, boolean)
    TO ${POSTGRES_READER_USER};



-- The generation producer's per-tenant chain set: tenants found in the DATA,
-- UNIONed with every tenant the system KNOWS (zebridge_user_tenants).
--
-- Data alone was the original rule (dyntenant-correct: a new tenant's first row
-- creates its chain on the next tick) — but data alone made "no chain" AMBIGUOUS.
-- Measured (NOTES §10h): test_types holds rows for acme/dynten only, so no
-- kilo chain was ever built, and a kilo client could not tell "producer has not
-- built yet" from "genuinely empty for me" — the retired snapshot path used to
-- answer that with a 0-row RLS dump. Including known tenants means an EMPTY
-- MANIFEST (a 0-row full) is built for them, which is the explicit "nothing to
-- seed": the client seeds zero rows, anchors its watermark, and follows CDC.
--
-- A tenant that exists ONLY as data (dyntenant, not yet registered) still gets
-- its chain from the data half of the union — neither source is dropped.
--
-- ⚠️ `zebridge_user_tenants` is created by init.write, and THIS file must stay
-- runnable (and its functions callable) in the read-only profile — hence the
-- to_regclass guard rather than a bare reference: plpgsql only resolves tables
-- at call time, so an unguarded reference would pass the apply and then fail
-- every producer tick on a read-only deployment.
--
-- SECURITY DEFINER because the reader's own RLS would scope the DISTINCT to one
-- tenant — PG functions hold the rules.
CREATE OR REPLACE FUNCTION public.zebridge_tenants_of(tbl regclass, tenant_col name)
RETURNS SETOF text AS $$
BEGIN
    IF to_regclass('public.zebridge_user_tenants') IS NOT NULL THEN
        RETURN QUERY EXECUTE format(
            'SELECT DISTINCT %I::text FROM %s WHERE %I IS NOT NULL '
            'UNION SELECT DISTINCT tenant_id::text FROM public.zebridge_user_tenants',
            tenant_col, tbl, tenant_col);
    ELSE
        RETURN QUERY EXECUTE format(
            'SELECT DISTINCT %I::text FROM %s WHERE %I IS NOT NULL', tenant_col, tbl, tenant_col);
    END IF;
END;
$$ LANGUAGE plpgsql STABLE SECURITY DEFINER SET search_path = pg_catalog, public;

-- Install (or refresh) the per-table width guard. Only unbounded columns are
-- measured — text, unbounded varchar, bytea (×2: the feed renders it as hex),
-- json/jsonb/xml, and arrays of any type — because a table with none of them is
-- statically inside every budget and gets no trigger at all (hot-path cost only
-- where overflow is possible). The generated trigger body is STATIC per table:
-- explicit column references, no runtime catalog reads, one SELECT on the one-row
-- limits table.
CREATE OR REPLACE FUNCTION public.zebridge_install_width_guard(tbl regclass)
RETURNS text AS $$
DECLARE
    short       text := (SELECT relname FROM pg_class WHERE oid = tbl);
    expr        text := '';
    n_cols      int  := 0;
    n_unbounded int  := 0;
    col         record;
    fn_body     text;
    budget_lit  integer;
BEGIN
    FOR col IN
        SELECT a.attname, t.typname, a.atttypmod, t.typcategory
        FROM pg_attribute a JOIN pg_type t ON t.oid = a.atttypid
        WHERE a.attrelid = tbl AND a.attnum > 0 AND NOT a.attisdropped
    LOOP
        n_cols := n_cols + 1;
        -- Bytes ride the feed as MessagePack bin (§10ex): the bytes plus a header of at
        -- most five. The hex form (2 per byte + 2) this measured until 2026-09-12 counted
        -- every bytea twice and refused a 300 KB tile at 600 KB. PostGIS geometry and
        -- geography ride as EWKB, which pg_column_size approximates from above without
        -- calling a PostGIS function from a guard that must work without the extension.
        IF col.typname = 'bytea' THEN
            n_unbounded := n_unbounded + 1;
            expr := expr || format(' + coalesce(octet_length(NEW.%I) + 5, 0)', col.attname);
        -- §10fg: pgvector and bit(n) ride as bytes too, a header lighter than the
        -- stored form pg_column_size reports — again an approximation from above.
        ELSIF col.typname IN ('geometry', 'geography', 'vector', 'halfvec', 'sparsevec', 'bit') THEN
            n_unbounded := n_unbounded + 1;
            expr := expr || format(' + coalesce(pg_column_size(NEW.%I), 0)', col.attname);
        ELSIF col.typname IN ('text', 'json', 'jsonb', 'xml')
              OR (col.typname = 'varchar' AND col.atttypmod = -1)
              OR col.typcategory = 'A' THEN
            n_unbounded := n_unbounded + 1;
            expr := expr || format(' + coalesce(octet_length(NEW.%I::text), 0)', col.attname);
        END IF;
    END LOOP;

    -- §10kt: a table that needs no guard must not KEEP one. Dropping its last unbounded
    -- column (a text added, then dropped) left the previous guard naming the dropped
    -- column, and every INSERT and UPDATE failed (measured on Supabase: "record "new"
    -- has no field …"). §10ko's re-install on DDL came through here and changed nothing.
    IF n_unbounded = 0 THEN
        -- Only when there is one: a bare DROP … IF EXISTS prints a notice for every
        -- table that never had a guard, which is most of them.
        IF EXISTS (SELECT 1 FROM pg_trigger WHERE tgrelid = tbl AND tgname = 'zebridge_width_guard') THEN
            EXECUTE format('DROP TRIGGER zebridge_width_guard ON %s', tbl);
        END IF;
        IF to_regprocedure('public.' || quote_ident('zebridge_width_guard_' || short) || '()') IS NOT NULL THEN
            EXECUTE 'DROP FUNCTION public.' || quote_ident('zebridge_width_guard_' || short) || '()';
        END IF;
        RETURN format('%s: no unbounded columns — statically inside every budget, no guard (any earlier one removed)', short);
    END IF;

    -- ⚠️ The generated body is assembled as a STRING and attached via quote_literal,
    -- never via nested dollar-quote tags (dollar-f-dollar and the like): the
    -- templates pass through envsubst at init, which eats any `$word` as an (empty)
    -- shell variable and leaves an unterminated quote that swallows every statement
    -- after it. Measured on the first fresh boot: this function AND zebridge_enable
    -- simply did not exist. Only the plain double-dollar delimiter survives envsubst
    -- — and it must never be SPELLED inside a body either, comments included: the
    -- lexer ends the body at the first occurrence, wherever it sits. (This very
    -- comment broke two boots by naming it literally.)
    --
    -- SECURITY DEFINER: the guard reads zebridge_limits as whoever writes the row
    -- (bridge_writer, a backend app), and none of them should need a grant on bridge
    -- internals just to be told no. search_path pinned, as every SECURITY DEFINER must.
    -- The budget is baked as a LITERAL, resolved HERE, once. MIN over the instances
    -- carrying this table: a row must fit the NARROWEST of them, or it is accepted
    -- here and suspends the feed over there. COALESCE covers "no bridge has
    -- registered yet" — nothing replicates the table, so there is no feed to
    -- overflow. `zebridge_register_limits` re-bakes it at every bridge boot.
    --
    -- Measured, and the reason it is a literal rather than a lookup: per row, a
    -- publication join cost 22 us, a plain indexed lookup 3.0 us, and a literal is
    -- free (within noise of no guard at all). This trigger fires on every insert and
    -- update of the table.
    SELECT COALESCE(MIN(l.max_row_bytes), 16384) INTO budget_lit
      FROM public.zebridge_limits l
      JOIN pg_publication_tables pt ON pt.pubname = l.publication
     WHERE format('%I.%I', pt.schemaname, pt.tablename)::regclass = tbl;

    fn_body := format(
        'DECLARE '
        '  budget integer := %s; '
        '  width  integer := %s + %s; '
        'BEGIN '
        '  IF width >= budget THEN '
        '    RAISE EXCEPTION ''row width %% bytes is at or over the %%-byte change-feed budget: the feed could not carry this row (zebridge_limits, narrowest instance carrying this table; store a reference, not the blob)'', width, budget '
        '      USING ERRCODE = ''check_violation''; '
        '  END IF; '
        '  RETURN NEW; '
        'END',
        budget_lit, substr(expr, 4), 32 * n_cols + 256);

    EXECUTE 'CREATE OR REPLACE FUNCTION public.' || quote_ident('zebridge_width_guard_' || short)
         || '() RETURNS trigger AS ' || quote_literal(fn_body)
         || ' LANGUAGE plpgsql SECURITY DEFINER SET search_path = pg_catalog, public';
    -- PostgreSQL grants EXECUTE on a new function to PUBLIC, and Supabase's API roles get
    -- every new object in `public`. The sweep at the end of this file takes that back, but
    -- only for the functions that exist when it runs: a guard installed later, by
    -- zebridge_enable on a new table, kept it (bridge --diagnose, 2026-10-06). A trigger
    -- fires without EXECUTE, so nobody needs it.
    EXECUTE format('REVOKE EXECUTE ON FUNCTION public.%I() FROM PUBLIC', 'zebridge_width_guard_' || short);
    IF EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'anon') THEN
        EXECUTE format('REVOKE EXECUTE ON FUNCTION public.%I() FROM anon', 'zebridge_width_guard_' || short);
    END IF;
    IF EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'authenticated') THEN
        EXECUTE format('REVOKE EXECUTE ON FUNCTION public.%I() FROM authenticated', 'zebridge_width_guard_' || short);
    END IF;

    IF EXISTS (SELECT 1 FROM pg_trigger WHERE tgrelid = tbl AND tgname = 'zebridge_width_guard') THEN
        EXECUTE format('DROP TRIGGER zebridge_width_guard ON %s', tbl);
    END IF;
    EXECUTE format('CREATE TRIGGER zebridge_width_guard BEFORE INSERT OR UPDATE ON %s FOR EACH ROW EXECUTE FUNCTION public.%I()',
                   tbl, 'zebridge_width_guard_' || short);
    RETURN format('%s: width guard installed over %s unbounded column(s)', short, n_unbounded);
END;
$$ LANGUAGE plpgsql;

-- Adding a parameter changes the function's identity: without this DROP the old
-- signature would survive CREATE OR REPLACE and every named-argument call would
-- be ambiguous between the two.
DROP FUNCTION IF EXISTS public.zebridge_enable(regclass, name, boolean, name[], name, name, boolean, text, name, boolean);
DROP FUNCTION IF EXISTS public.zebridge_enable(regclass, name, boolean, name[], name, name, name, boolean, text, name, boolean);
DROP FUNCTION IF EXISTS public.zebridge_enable(regclass, name, boolean, name[], name, name, name, boolean, boolean, text, name, boolean);
DROP FUNCTION IF EXISTS public.zebridge_enable(regclass, name, boolean, name[], name, name, name, boolean, boolean, text, name, boolean, boolean);
DROP FUNCTION IF EXISTS public.zebridge_enable(regclass, name, boolean, name[], name, name, name, boolean, boolean, boolean, text, name, boolean, boolean);
CREATE OR REPLACE FUNCTION public.zebridge_enable(
    tbl           regclass,
    tenant_col    name    DEFAULT NULL,
    writable      boolean DEFAULT false,
    columns       name[]  DEFAULT NULL,
    version_col   name    DEFAULT NULL,
    tombstone_col name    DEFAULT NULL,
    tiebreak_col  name    DEFAULT NULL,  -- where the winning writer's id is stored; makes equal versions resolvable
    -- jsonb columns of registers {v, t, w} (COOPERATIVE_EDITING.md): PostgreSQL merges each
    -- accepted write into them register by register, so a late write never rolls back a
    -- field someone else edited meanwhile. Needs writable => true.
    register_cols name[]  DEFAULT NULL,
    -- ⚠️ The conscious opt-OUT of the tombstone requirement (see the gate below).
    -- Physical deletes are INEXPRESSIBLE in a generation delta (deltas are
    -- upsert-only), and generations are the only seed path — so a hard-deleted row
    -- RESURRECTS on any fresh or re-seeding client until the next full (measured,
    -- NOTES §10i/§10j). Setting this true says the operator accepts that.
    allow_physical_deletes boolean DEFAULT false,
    generations   boolean DEFAULT true,
    -- The index on the version column that keeps a delta from scanning the table (§10gc).
    -- On by default; false is a decision for a table of hot rows updated again and again,
    -- where the index costs up to a fifth more WAL and every HOT update (§10gd).
    version_index boolean DEFAULT true,
    public_reason text    DEFAULT NULL,
    -- ⚠️ No default. It used to render BRIDGE_CDC_PUBLICATION here, which made
    -- this name a SECOND source of truth against the bridge's own `--pub` —
    -- two spellings that had to agree, with nothing checking that they did.
    -- Naming it is one word, and it decides which tables get replicated.
    publication   name    DEFAULT NULL,
    -- Opt-in, same shape as allow_physical_deletes: a typo must not manufacture
    -- a publication (that would be a silent, empty, perfectly healthy-looking
    -- second feed), so an unknown name is refused unless this says otherwise.
    create_publication boolean DEFAULT false,
    dry_run       boolean DEFAULT true
) RETURNS TABLE (step text, status text, detail text) AS $$
DECLARE
    bad_fks  text;
    vcol_eff   name;
    vcol_num   smallint;
    vidx       text;
    short      text := (SELECT relname FROM pg_class WHERE oid = tbl);
    have_write boolean := to_regprocedure('public.zebridge_grant_edge_writes(regclass)') IS NOT NULL;
    published  boolean;
    scoped     boolean;
    col_list   text;
    excluded   name[];
    cur_list   name[];
    verb       text := CASE WHEN dry_run THEN 'would' ELSE 'done' END;
    r           record;
    ri_cols     text;
BEGIN
    -- ── the zebridge_ prefix is the bridge's own ──────────────────────────────
    --
    -- Every table named zebridge_* is the bridge's (the catalogue, the invites, the GC floor,
    -- the key roster...), and the bridge refuses a client write to any of them by that prefix
    -- (mutation_listener.isForbiddenTable). A user table with such a name would be silently
    -- unreplicated and unwritable, so it is refused HERE, loudly, in the same shape as the
    -- `lf-` prefix for principals (zebridge_invite). A RAISE, for the same reason as the
    -- publication check below: the call is malformed, whatever dry_run says.
    IF lower(left(short, 9)) = 'zebridge_'
       OR (SELECT n.nspname FROM pg_class c JOIN pg_namespace n ON n.oid = c.relnamespace WHERE c.oid = tbl) = 'zebridge_private' THEN
        RAISE EXCEPTION 'zebridge_enable(%): the zebridge_ prefix is reserved for the bridge''s own tables', tbl
            USING HINT = 'Rename the table. Names starting with zebridge_ (and the zebridge_private schema) belong '
                         'to the bridge: it never replicates them and never accepts a client write to them.';
    END IF;

    -- ── the publication must be NAMED, and may only be created on request ─────
    --
    -- A RAISE, not one of this function's ERROR rows: an enable that does not name
    -- its publication is a malformed CALL, not a table failing a rule — and the
    -- rows are easy to not read (`PERFORM * FROM zebridge_enable(...)` in a
    -- migration discards every one of them). Same reasoning as the bridge's own
    -- refusal to invent a publication name: this argument decides WHICH TABLES get
    -- replicated, so there is nothing sensible to guess.
    IF publication IS NULL THEN
        RAISE EXCEPTION 'zebridge_enable(%): no publication named', tbl
            USING HINT = 'Pass publication => ''my_pub'' — the same name the bridge is given as '
                         '--pub / BRIDGE_CDC_PUBLICATION. There is no default. Add '
                         'create_publication => true if it does not exist yet.';
    END IF;

    IF NOT EXISTS (SELECT 1 FROM pg_publication WHERE pubname = publication) THEN
        IF NOT create_publication THEN
            RAISE EXCEPTION 'zebridge_enable(%): publication % does not exist', tbl, publication
                USING HINT = 'Refused by default so a typo cannot manufacture a silent, empty, '
                             'healthy-looking second feed. If you meant to create it: '
                             'create_publication => true, or SELECT * FROM '
                             'zebridge_create_publication(''<name>'') first.';
        END IF;
        -- Reported through the same rows as everything else, so the operator sees
        -- the three internal tables land in the enable's own output.
        IF dry_run THEN
            RETURN QUERY SELECT 'publication', 'would',
                format('CREATE PUBLICATION %I, with zebridge_ddl_events / _gc_watermark / '
                       '_user_tenants attached', publication);
        ELSE
            RETURN QUERY SELECT * FROM public.zebridge_create_publication(publication);
        END IF;
    END IF;

    IF writable AND NOT have_write THEN
        RETURN QUERY SELECT 'preflight', 'ERROR',
            'writable => true, but the write half is not installed — this database was '
            'initialised with ZB_PROFILE=readonly (init.core.template.sql alone).';
        RETURN;
    END IF;

    -- ── the tombstone gate ────────────────────────────────────────────────────
    -- A writable table without a tombstone column deletes PHYSICALLY, and a
    -- physical delete cannot travel in a generation delta (deltas are upsert-only).
    -- With snapshot-on-demand retired, chains are the ONLY seed path — so a
    -- hard-deleted row silently RESURRECTS on every fresh or re-seeding client
    -- until the next full rebuild. That stopped being a "documented weaker
    -- guarantee" and became a divergence generator the day snapshots retired
    -- (measured: NOTES §10i, trigger-B). Refused mechanically, same pattern as the
    -- timestamp and publication guards; `allow_physical_deletes => true` is the
    -- recorded, deliberate opt-out.
    IF writable AND tombstone_col IS NULL AND NOT allow_physical_deletes THEN
        RETURN QUERY SELECT 'preflight', 'ERROR',
            format('%s is writable but has no tombstone_col: DELETEs would be physical, '
                   'and a physical delete cannot travel in a generation delta — hard-deleted '
                   'rows RESURRECT on fresh clients until the next full (NOTES §10i). '
                   'Add a tombstone column (e.g. deleted_at timestamptz) and pass '
                   'tombstone_col => %L, or state the acceptance explicitly with '
                   'allow_physical_deletes => true.', tbl, 'deleted_at');
        RETURN;
    END IF;
    IF writable AND tombstone_col IS NULL AND allow_physical_deletes THEN
        RETURN QUERY SELECT 'tombstone', 'WARNING',
            format('%s: physical deletes ACCEPTED by the operator — hard-deleted rows '
                   'resurrect on fresh seeds until the next full rebuild.', tbl);
    END IF;

    IF tenant_col IS NOT NULL AND NOT EXISTS (
        SELECT 1 FROM pg_attribute WHERE attrelid = tbl AND attname = tenant_col
          AND attnum > 0 AND NOT attisdropped
    ) THEN
        RETURN QUERY SELECT 'preflight', 'ERROR', format('%s has no column %I', tbl, tenant_col);
        RETURN;
    END IF;
    -- §10lz: register columns are merged on WRITE; a read-only table has no writes to merge.
    IF register_cols IS NOT NULL AND cardinality(register_cols) > 0 AND NOT writable THEN
        RETURN QUERY SELECT 'preflight', 'ERROR',
            format('register_cols %s needs writable => true: the merge runs on the writes clients send', register_cols::text);
        RETURN;
    END IF;
    -- A nullable tenant column cannot route a row (NULL has no stream) and the scoping
    -- functions refuse it; said here, before anything is applied, rather than as an
    -- exception halfway through the activation.
    IF tenant_col IS NOT NULL AND EXISTS (
        SELECT 1 FROM pg_attribute WHERE attrelid = tbl AND attname = tenant_col
          AND attnum > 0 AND NOT attisdropped AND NOT attnotnull
    ) THEN
        RETURN QUERY SELECT 'preflight', 'ERROR',
            format('%s.%I is nullable: a tenant column must be NOT NULL (a row with no tenant has no stream to ride)', tbl, tenant_col);
        RETURN;
    END IF;

    -- ⚠️ **Scoping must exist BEFORE the table is published**, and that ordering is not a
    -- preference — `zebridge_publication_guard` is an event trigger that refuses
    -- `ALTER PUBLICATION ... ADD TABLE` for a table with neither a row filter nor RLS.
    -- Publishing first and scoping second leaves a window in which the change feed carries
    -- every tenant's rows, so the guard closes it at the source. Predicted here rather than
    -- discovered halfway through: an ERROR row costs nothing, a half-applied activation does.
    -- §10dn: a foreign key ON DELETE CASCADE into a table that keeps tombstones is a
    -- contradiction — the cascade removes rows physically, the bridge never forwards a
    -- physical delete on a tombstone table, and every replica keeps them. Refused here,
    -- and by the DDL guard for a constraint added later.
    IF tombstone_col IS NOT NULL THEN
        SELECT string_agg(con.conname || ' (from ' || con.confrelid::regclass::text || ')', ', ') INTO bad_fks
        FROM pg_constraint con
        WHERE con.conrelid = tbl AND con.contype = 'f' AND con.confdeltype = 'c';
        IF bad_fks IS NOT NULL THEN
            RETURN QUERY SELECT 'preflight', 'ERROR',
                format('%s keeps tombstones, but a foreign key on it cascades physical deletes into it: %s. '
                       'A cascade removes rows the bridge never forwards for a tombstone table, and every replica keeps them. '
                       'Declare the constraint ON DELETE NO ACTION, or enable the table without a tombstone.', tbl, bad_fks);
            RETURN;
        END IF;
    END IF;

    scoped := (SELECT relrowsecurity FROM pg_class WHERE oid = tbl)
              OR EXISTS (SELECT 1 FROM pg_publication_rel WHERE prrelid = tbl AND prqual IS NOT NULL)
              -- ⚠️ Aliased and qualified: the parameter is also named `tbl`, and this
              -- table has a column of that name, so the unqualified form is ambiguous.
              OR EXISTS (SELECT 1 FROM public.zebridge_catalogue cat
                         WHERE cat.tbl = short AND cat.tenant_col IS NULL)
              OR public_reason IS NOT NULL
              -- A tenant column scopes the table either way: writable, through
              -- scope_writes_by_tenant; read-only, through scope_reads_by_tenant (the
              -- ELSIF branch below). Both enable RLS before the publication step.
              -- §10fx: this used to read `(writable AND tenant_col IS NOT NULL)`, which
              -- refused every read-only tenant table and left that branch unreachable.
              OR tenant_col IS NOT NULL;
    IF NOT scoped THEN
        RETURN QUERY SELECT 'preflight', 'ERROR',
            format('%s would be published unscoped, which zebridge_publication_guard refuses. '
                   'Pick one: (a) tenant_col => ''<column>'', which enables RLS (writable or read-only); '
                   '(b) public_reason => ''why everyone may read this'', recorded in '
                   'zebridge_catalogue; (c) call zebridge_scope_publication_to_one_tenant() '
                   'first for a one-bridge-per-tenant deployment.', tbl);
        RETURN;
    END IF;

    -- A tenant table routes its CDC by the tenant, so the tenant must be in the replica
    -- identity — the scoping functions below put it there. Planned now, while nothing is
    -- applied: the row below reports it, and a table with no primary key (no identity to
    -- extend) is an ERROR row here rather than an exception halfway through.
    IF tenant_col IS NOT NULL THEN
        IF NOT EXISTS (SELECT 1 FROM pg_index WHERE indrelid = tbl AND indisprimary) THEN
            RETURN QUERY SELECT 'preflight', 'ERROR',
                format('%s has no primary key: no replica identity can carry %I, and the bridge '
                       'refuses a keyless table anyway', tbl, tenant_col);
            RETURN;
        END IF;
        ri_cols := public.zebridge_tenant_replica_identity(tbl, tenant_col, dry_run => true);
    END IF;

    -- ── writes, guards and RLS come FIRST, because the guard demands it ───────
    IF writable THEN
        IF NOT dry_run THEN EXECUTE format('SELECT public.zebridge_grant_edge_writes(%L::regclass)', tbl); END IF;
        RETURN QUERY SELECT 'grants', verb, format('zebridge_grant_edge_writes(%L)', tbl::text);

        IF version_col IS NULL THEN
            RETURN QUERY SELECT 'guards', 'skipped',
                'no version_col given — writes are granted but unguarded: a client that omits '
                'the version is not corrected, and a direct DELETE is not tombstoned';
        ELSE
            IF NOT dry_run THEN
                -- tenant_col passed through: when the table is tenant-routed, the same call
                -- installs the tenant guard (absent -> the open tenant, OPEN_TENANT; malformed -> rejected).
                EXECUTE format('SELECT public.zebridge_install_write_guards(%L::regclass, %L, %L, %L)',
                               tbl, version_col, tombstone_col, tenant_col);
            END IF;
            RETURN QUERY SELECT 'guards', verb,
                format('zebridge_install_write_guards(%L, %L, %L, %L)', tbl::text, version_col, tombstone_col, tenant_col);
        END IF;

        IF register_cols IS NOT NULL AND cardinality(register_cols) > 0 THEN
            IF NOT dry_run THEN
                EXECUTE format('SELECT public.zebridge_install_register_merge(%L::regclass, %L::name[])', tbl, register_cols);
            END IF;
            RETURN QUERY SELECT 'registers', verb,
                format('zebridge_install_register_merge(%L, %L) — each accepted write merged register by register',
                       tbl::text, register_cols::text);
        END IF;

        IF tenant_col IS NOT NULL THEN
            IF NOT dry_run THEN
                EXECUTE format('SELECT public.zebridge_scope_writes_by_tenant(%L::regclass, %L)', tbl, tenant_col);
            END IF;
            RETURN QUERY SELECT 'rls', verb,
                format('zebridge_scope_writes_by_tenant(%L, %L) — writes AND snapshot reads scoped; '
                       'CDC stays scoped by the subject, not RLS (RLS cannot see it)',
                       tbl::text, tenant_col);
        END IF;
    ELSIF tenant_col IS NOT NULL THEN
        -- writable => false: nothing to scope on the WRITE side, but the READ side still
        -- can be — zebridge_scope_reads_by_tenant needs nothing this branch doesn't
        -- already have (init.core.template.sql, no write profile required).
        IF NOT dry_run THEN
            EXECUTE format('SELECT public.zebridge_scope_reads_by_tenant(%L::regclass, %L)', tbl, tenant_col);
        END IF;
        RETURN QUERY SELECT 'rls', verb,
            format('zebridge_scope_reads_by_tenant(%L, %L) — snapshot reads scoped; '
                   'CDC stays scoped by the subject, not RLS (RLS cannot see it)',
                   tbl::text, tenant_col);
    END IF;

    -- §10lz: register_cols is declared on every call, like the other options: a call
    -- without it turns the merge off, rather than leaving an earlier call's trigger behind.
    IF (register_cols IS NULL OR cardinality(register_cols) = 0) AND EXISTS (
        SELECT 1 FROM pg_trigger WHERE tgrelid = tbl AND tgname = 'zebridge_merge_registers_t'
    ) THEN
        IF NOT dry_run THEN
            EXECUTE format('DROP TRIGGER zebridge_merge_registers_t ON %s', tbl);
        END IF;
        RETURN QUERY SELECT 'registers', verb,
            format('register merge removed from %s: register_cols not given', tbl::text);
    END IF;

    -- ── the sweeper's reach on a read-only table that keeps tombstones ─────────
    -- A loader's soft deletes (examples/08-map/load_fuel.py) are reaped like a client's;
    -- the sweeper runs as the writer role, which a read-only table never granted. A
    -- writable table already has the grant (zebridge_grant_edge_writes, above).
    IF NOT writable AND tombstone_col IS NOT NULL THEN
        IF to_regprocedure('public.zebridge_grant_sweeper_reaps(regclass)') IS NULL THEN
            RETURN QUERY SELECT 'sweeper', 'skipped',
                'the write half (init.write) is not installed: no writer role, no sweeper to grant';
        ELSE
            IF NOT dry_run THEN EXECUTE format('SELECT public.zebridge_grant_sweeper_reaps(%L::regclass)', tbl); END IF;
            RETURN QUERY SELECT 'sweeper', verb,
                format('zebridge_grant_sweeper_reaps(%L): SELECT and DELETE for the tombstone sweeper', tbl::text);
        END IF;
    END IF;

    -- Done inside the scoping call above, writable or read-only.
    IF tenant_col IS NOT NULL THEN
        RETURN QUERY SELECT 'replica identity',
            CASE WHEN ri_cols IS NULL THEN 'ok' ELSE verb END,
            CASE WHEN ri_cols IS NULL
                 THEN format('already carries %I: deletes route by tenant', tenant_col)
                 ELSE format('unique index %s_zb_ri on (%s), made the replica identity: deletes route by tenant',
                             short, ri_cols) END;
    END IF;

    -- ── key order: advice, never a refusal (§10ja) ────────────────────────────
    -- Clients mint the keys of a writable table. A random uuid scatters every replica's
    -- key index on live inserts; a time-ordered one appends. Seeds do not care — the
    -- producer writes every chain object in key order — so a read-only table, or a
    -- bigint key (already ordered), gets nothing here.
    IF writable THEN
        FOR r IN
            SELECT a.attname, pg_get_expr(d.adbin, d.adrelid) AS def
            FROM pg_index i
            JOIN pg_attribute a ON a.attrelid = i.indrelid AND a.attnum = ANY(i.indkey)
            LEFT JOIN pg_attrdef d ON d.adrelid = a.attrelid AND d.adnum = a.attnum
            WHERE i.indrelid = tbl AND i.indisprimary AND a.atttypid = 'uuid'::regtype
              AND coalesce(pg_get_expr(d.adbin, d.adrelid), '') NOT ILIKE '%uuidv7(%'
        LOOP
            RETURN QUERY SELECT 'key order', 'advice',
                format('%s.%I is a uuid defaulting to %s. Clients mint keys for a writable table: '
                       'mint uuid v7, and ALTER TABLE %s ALTER COLUMN %I SET DEFAULT uuidv7(). '
                       'Time-ordered keys append to every replica''s index instead of scattering it: '
                       'live apply measured 4-11x faster (NOTES §10ja). Existing keys can stay.',
                       tbl, r.attname, coalesce(r.def, 'nothing'), tbl, r.attname);
        END LOOP;
    END IF;

    -- ── the width guard: both doors, one trigger ──────────────────────────────
    IF NOT dry_run THEN
        EXECUTE format('SELECT public.zebridge_install_width_guard(%L::regclass)', tbl);
    END IF;
    RETURN QUERY SELECT 'width guard', verb,
        format('zebridge_install_width_guard(%L) — a row the change feed cannot carry is '
               'refused at write time: edge writes get a rejected verdict (SQLSTATE 23514), '
               'psql gets an ordinary ERROR. No-op on tables without unbounded columns.', tbl::text);

    -- ── THE CATALOGUE ROW: the declaration itself, atomically with the guards ──
    --
    -- ⚠️ `seed_epoch` is absent from the column list AND from the DO UPDATE SET, and
    -- that absence is the invariant, not an oversight (§10df, MIGRATIONS.md). A
    -- re-enable declares the table's RULES; it must not re-seed every replica. The
    -- INSERT lets the DEFAULT 0 stand; the UPDATE leaves whatever zebridge_reseed(),
    -- the DDL trigger or a suspension lift put there. Setting it here would re-seed
    -- the world on every re-run, and could LOWER it — after which a replica holding a
    -- higher epoch never sees a descriptor above its own again, and the re-seed rule
    -- is dead for that table. The value is a monotone counter: only the comparison
    -- with the replica's stored epoch means anything.
    IF NOT dry_run THEN
        EXECUTE format(
            'INSERT INTO public.zebridge_catalogue (tbl, tenant_col, public_reason, version_col, tombstone_col, tiebreak_col, generations, allow_physical_deletes) '
            'VALUES (%L, %L, %L, COALESCE(%L, ''updated_at''), %L, %L, %L, %L) '
            'ON CONFLICT (tbl) DO UPDATE SET tenant_col = EXCLUDED.tenant_col, '
            'public_reason = EXCLUDED.public_reason, version_col = EXCLUDED.version_col, '
            'tombstone_col = EXCLUDED.tombstone_col, tiebreak_col = EXCLUDED.tiebreak_col, '
            'generations = EXCLUDED.generations, allow_physical_deletes = EXCLUDED.allow_physical_deletes',
            short, tenant_col, public_reason, version_col, tombstone_col, tiebreak_col, generations, allow_physical_deletes);
        -- The incarnation enabled: recorded on a first enable (no re-seed), unchanged on
        -- a re-run, and a new one if the table was re-created while nothing observed it.
        PERFORM public.zebridge_observe_incarnation(short, tbl::oid);
    END IF;
    RETURN QUERY SELECT 'catalogue', verb,
        format('zebridge_catalogue[%s]: tenant_col=%s version_col=%s tombstone=%s tiebreak=%s generations=%s allow_physical_deletes=%s '
               '— the bridge reads this at boot (env rules become overrides) and the '
               'generation producer per tick', short,
               COALESCE(tenant_col::text, 'NULL(public)'), COALESCE(version_col::text, 'updated_at'),
               COALESCE(tombstone_col::text, '-'), COALESCE(tiebreak_col::text, '-'), generations, allow_physical_deletes);

    -- ── the version index (§10gc): a delta must not read the whole table ──────
    -- The generation producer selects a delta's rows `WHERE version > previous cutoff`.
    -- Without an index that leads with the version column, that is a scan of the whole
    -- table on every cut, however few rows moved: 587 ms on a 3M-row table where the
    -- index answers in 0.011 ms (measured 2026-09-15). A cut that grows with the table
    -- is what lets a pruning CDC stream slide past the chain. The index has costs, said
    -- in the detail: its build blocks writes on the table, and an update that changes an
    -- indexed column is no longer HOT, so every update also writes every index.
    IF generations AND NOT version_index THEN
        RETURN QUERY SELECT 'version index', 'skipped',
            'version_index => false: every delta and every "did anything change" check scans the whole table, '
            'once per tenant per tick (330 ms on 2M rows, §10gd). enable never drops an index: one built earlier stays until DROP INDEX.';
    ELSIF generations THEN
        vcol_eff := COALESCE(version_col, 'updated_at');
        SELECT attnum INTO vcol_num FROM pg_attribute
         WHERE attrelid = tbl AND attname = vcol_eff AND attnum > 0 AND NOT attisdropped;
        IF vcol_num IS NULL THEN
            RETURN QUERY SELECT 'version index', 'skipped',
                format('%s has no column %I — the producer cuts fulls only for it', tbl, vcol_eff);
        ELSE
            SELECT i.indexrelid::regclass::text INTO vidx FROM pg_index i
             WHERE i.indrelid = tbl AND i.indkey[0] = vcol_num
             ORDER BY i.indnatts LIMIT 1;
            IF vidx IS NOT NULL THEN
                RETURN QUERY SELECT 'version index', 'already',
                    format('%s leads with %I — deltas read only the rows since the last cut.%s', vidx, vcol_eff,
                           CASE WHEN vidx LIKE '%\_zb\_version' THEN ''
                                ELSE ' Its name does not end in _zb_version, so clients receive it and build it too.' END);
            ELSE
                IF NOT dry_run THEN
                    EXECUTE format('CREATE INDEX %I ON %s (%I)', left(short, 48) || '_zb_version', tbl, vcol_eff);
                END IF;
                RETURN QUERY SELECT 'version index', verb,
                    format('CREATE INDEX %I ON %s (%I) — a delta then reads the rows since the last cut instead '
                           'of the whole table. Costs: the build blocks writes on the table (on a large live table, '
                           'run CREATE INDEX CONCURRENTLY %I ON %s (%I) first and this step finds it), and updates that '
                           'change %I are no longer HOT, so each update also writes every index. Clients do not '
                           'receive this index: they never read by version.',
                           left(short, 48) || '_zb_version', tbl, vcol_eff, left(short, 48) || '_zb_version', tbl, vcol_eff, vcol_eff);
            END IF;
        END IF;
    ELSE
        RETURN QUERY SELECT 'version index', 'skipped', 'generations => false: no chain is cut for this table';
    END IF;

    -- ── the column list (§10ff): index-like types never travel ────────────────
    -- A tsvector, tsquery, xml or range column is an index representation of other
    -- columns, not data a replica can use — FTS5 builds its own from the text. They
    -- are left out of the publication's column list, with whatever the caller left
    -- out through `columns`; the descriptor, the chain and CDC then agree on what a
    -- table is. A column added later to a table WITH a list is not published until
    -- zebridge_enable runs again (it refreshes the list below).
    SELECT array_agg(a.attname ORDER BY a.attnum) INTO excluded
      FROM pg_attribute a JOIN pg_type t ON t.oid = a.atttypid
     WHERE a.attrelid = tbl AND a.attnum > 0 AND NOT a.attisdropped
       AND (t.typname IN ('tsvector', 'tsquery', 'xml') OR t.typtype IN ('r', 'm'))
       AND (columns IS NULL OR a.attname = ANY(columns));
    IF excluded IS NOT NULL THEN
        SELECT array_agg(a.attname ORDER BY a.attnum) INTO columns
          FROM pg_attribute a
         WHERE a.attrelid = tbl AND a.attnum > 0 AND NOT a.attisdropped
           AND (columns IS NULL OR a.attname = ANY(columns))
           AND NOT (a.attname = ANY(excluded));
        RETURN QUERY SELECT 'columns', verb,
            format('%s column(s) left out of the publication — index-like types never travel: %s',
                   array_length(excluded, 1), array_to_string(excluded, ', '));
    END IF;
    col_list := CASE WHEN columns IS NULL THEN ''
                ELSE ' (' || array_to_string(
                     ARRAY(SELECT quote_ident(c) FROM unnest(columns) c), ', ') || ')' END;

    -- ── publication LAST ──────────────────────────────────────────────────────
    SELECT EXISTS (SELECT 1 FROM pg_publication_tables
                   WHERE pubname = publication AND tablename = short) INTO published;
    IF published THEN
        SELECT attnames INTO cur_list FROM pg_publication_tables
         WHERE pubname = publication AND tablename = short;
        IF columns IS NOT NULL AND (cur_list IS NULL OR cur_list <> columns) THEN
            IF NOT dry_run THEN
                EXECUTE format('ALTER PUBLICATION %I DROP TABLE %s', publication, tbl);
                EXECUTE format('ALTER PUBLICATION %I ADD TABLE %s%s', publication, tbl, col_list);
            END IF;
            RETURN QUERY SELECT 'publication', verb,
                format('%I: column list refreshed — ALTER PUBLICATION %I ADD TABLE %s%s', publication, publication, tbl, col_list);
        ELSE
            RETURN QUERY SELECT 'publication', 'already', format('%I already carries %s', publication, tbl);
        END IF;
    ELSE
        IF NOT dry_run THEN
            EXECUTE format('ALTER PUBLICATION %I ADD TABLE %s%s', publication, tbl, col_list);
        END IF;
        RETURN QUERY SELECT 'publication', verb,
            format('ALTER PUBLICATION %I ADD TABLE %s%s', publication, tbl, col_list);
    END IF;

    -- ── the same table in a SECOND publication: warn, never refuse ────────────
    -- Legitimate (two NATS deployments, a staged migration), and dangerous in one
    -- specific way: the width guard bakes MIN(max_row_bytes) across every instance
    -- whose publication carries this table, so the NARROWEST bridge sets the ceiling
    -- for everyone. Rows already stored above that ceiling stay valid in Postgres and
    -- on the wider feed — but the narrow bridge cannot encode them, so IT quarantines
    -- the table (preflight at boot, or on the next touch) while the other carries on.
    -- A per-feed outage that reads as a mystery unless someone says this out loud.
    FOR r IN
        SELECT p.pubname
          FROM pg_publication_tables p
         WHERE format('%I.%I', p.schemaname, p.tablename)::regclass = tbl
           AND p.pubname <> publication
    LOOP
        RETURN QUERY SELECT 'publication'::text, 'WARNING'::text,
            format('%s is also in publication %I. The width guard takes the MINIMUM '
                   'budget across every instance carrying it, so the narrowest bridge '
                   'sets the ceiling for ALL writers — and any bridge whose buffer is '
                   'smaller than an existing row quarantines this table on its own feed. '
                   'Keep their BASE_BUF equal, or keep the publications disjoint.',
                   tbl::text, r.pubname);
    END LOOP;

    -- ── re-bake the width guard NOW THAT THE TABLE IS PUBLISHED ───────────────
    -- The guard is installed BEFORE the publication step (its own required
    -- ordering), but its baked budget derives from a join through
    -- pg_publication_tables — which at install time contains NOTHING for this
    -- table, so a newborn's guard silently baked the 16384 default and ignored
    -- every registered instance budget until the next bridge restart. Measured:
    -- livebirth's 4096-budget probe stored a 4096-byte row the guard should have
    -- refused (case C, the exact failure the guard exists for). Re-derive with
    -- the publication row in place; rebudget is body-only, so no table lock.
    IF NOT dry_run THEN
        DECLARE
            eff_budget integer;
        BEGIN
            -- Schema-qualified regclass comparison, IDENTICAL to
            -- zebridge_install_width_guard's own derivation: `pt.tablename = short`
            -- matches by bare name across every schema in the publication, so two
            -- same-named tables in different schemas would share a budget.
            SELECT COALESCE((SELECT MIN(l.max_row_bytes)
                             FROM public.zebridge_limits l
                             JOIN pg_publication_tables pt ON pt.pubname = l.publication
                             WHERE format('%I.%I', pt.schemaname, pt.tablename)::regclass = tbl), 16384)
              INTO eff_budget;
            IF public.zebridge_rebudget_width_guard(tbl, eff_budget) THEN
                RETURN QUERY SELECT 'width guard', 'done',
                    format('budget re-baked at %s bytes now the publication carries %s', eff_budget, tbl);
            END IF;
        END;
    END IF;

    IF columns IS NOT NULL AND tenant_col IS NOT NULL THEN
        RETURN QUERY SELECT 'publication', 'NOTE',
            format('column list + tenant scoping: put %I inside the replica identity, or '
                   'UPDATE/DELETE are refused at the source (NOTES.md §1.8)', tenant_col);
    END IF;

    -- ── T3 and T4: the bridge and NATS, outside the database ──────────────────
    RETURN QUERY SELECT 'T3 bridge', 'LIVE',
        CASE WHEN tenant_col IS NULL
             THEN 'nothing to do — the catalogue row this wrote reaches a running bridge '
                  'through the WAL; it reconciles CDC_PUBLIC''s subject filter, lifts the '
                  'table''s refusal and publishes its schema on the spot (NOTES §10bj). '
                  'A bridge started later reads the same row at boot.'
             ELSE 'nothing to do — the catalogue row this wrote reaches a running bridge '
                  'through the WAL; it reloads its routing, lifts the table''s refusal and '
                  'publishes its schema on the spot (NOTES §10bj). The generation chain '
                  'needs nothing either — the producer reads the catalogue per tick.' END;
    -- §10kk: the grants are the client role's template (bridge --init-nats), written per
    -- stream, not per table: a new table changes nothing in NATS.
    RETURN QUERY SELECT 'T4 nats', 'LIVE',
        CASE WHEN tenant_col IS NULL
             THEN 'nothing to do — the table rides CDC_PUBLIC, which every client may read '
                  'already; the bridge adds its subject to the stream (T3)'
             ELSE 'nothing to do — the table rides each tenant''s CDC_<tenant> stream, which '
                  'the client role grants per tenant tag already; no grant, no NATS reload' END;

    IF dry_run THEN
        RETURN QUERY SELECT 'summary', 'DRY RUN', 'nothing was applied — re-run with dry_run => false';
    END IF;
END;
$$ LANGUAGE plpgsql;

-- ═══════════════════════════════════════════════════════════════════════════
-- zebridge_check: does this table conform to what the bridge needs — and to what
-- YOU intend for it? Run it after a migration, from psql, with no bridge running:
--
--     SELECT * FROM zebridge_check('orders', 'writable');
--     SELECT * FROM zebridge_check('orders', 'writable', 'updated_at', 'deleted_at', 'last_writer', 'tenant_id');
--     SELECT * FROM zebridge_check_all();   -- every catalogued table, against its row
--
-- It asks pg_catalog and our own tables the same questions the bridge's preflight
-- asks at boot and at every DDL event (preflight.zig, mutation_listener.zig), from
-- the same source of truth, so a green run here is a green boot there. `intent` is
-- what you MEAN the table to be; the checks compare it with what the grants and the
-- catalogue actually say — the mismatch is the finding a doctor exists to make.
-- Statuses: ok | NOTE | WARNING | ERROR. An ERROR is something the bridge refuses.
-- ═══════════════════════════════════════════════════════════════════════════
-- The column names are the part you must DECLARE for the check to mean anything
-- before a migration: which column is the version, the tombstone, the tiebreak, the
-- tenant. Name them and the check compares your naming with the schema AND with the
-- catalogue (a disagreement is a finding); leave them NULL and the catalogue's row is
-- taken at its word (`updated_at` when there is no row yet).
CREATE OR REPLACE FUNCTION public.zebridge_check(
    tbl           regclass,
    intent        text DEFAULT NULL,   -- 'read_only' | 'writable' | NULL = whatever the grants say
    version_col   name DEFAULT NULL,
    tombstone_col name DEFAULT NULL,
    tiebreak_col  name DEFAULT NULL,
    tenant_col    name DEFAULT NULL,
    -- The acceptance of physical deletes for a table not yet catalogued. A catalogued
    -- table's own acceptance (zebridge_enable's allow_physical_deletes, kept in the
    -- catalogue) counts without being repeated here. Either way, a missing tombstone
    -- is then a WARNING — the resurrection risk stated — not an ERROR.
    allow_physical_deletes boolean DEFAULT false
) RETURNS TABLE (check_name text, status text, detail text) AS $$
DECLARE
    tname    text := (SELECT relname FROM pg_class WHERE oid = tbl);
    owner    text := (SELECT pg_get_userbyid(relowner) FROM pg_class WHERE oid = tbl);
    cat      public.zebridge_catalogue%ROWTYPE;
    have_cat boolean;
    pkcols   name[];
    ident    "char";
    idx_cols name[];
    pubs     text[];
    writers  text[];
    want_writable boolean;
    seq_key  boolean;
    vname    name;
    vcol     record;
    tcol     record;
    budget   integer;
    widest   bigint;
    errors   integer := 0;
    warns    integer := 0;
BEGIN
    IF intent IS NOT NULL AND intent NOT IN ('read_only', 'writable') THEN
        RAISE EXCEPTION 'zebridge_check: intent must be ''read_only'' or ''writable'' (got %)', intent;
    END IF;

    -- ── 1. the catalogue row: what the bridge routes by ─────────────────────
    SELECT * INTO cat FROM public.zebridge_catalogue c WHERE c.tbl = tname;
    have_cat := FOUND;
    IF NOT have_cat THEN
        errors := errors + 1;
        RETURN QUERY SELECT 'catalogue', 'ERROR',
            format('%s has no zebridge_catalogue row: the bridge cannot route it (no_cdc_subject). '
                   'Declare it with zebridge_enable(...)', tname);
        -- No row: the declared names are the only ones there are.
        cat.tbl := tname;
        cat.version_col := coalesce(version_col, 'updated_at');
        cat.tombstone_col := tombstone_col;
        cat.tiebreak_col := tiebreak_col;
        cat.tenant_col := tenant_col;
        cat.generations := false;
    ELSE
        -- Declared names against the catalogue's: a disagreement means the migration
        -- and the operator's intent have drifted apart, which is exactly what a
        -- check after a migration exists to catch. The declared name wins for the
        -- checks below, so the rest of the report is about what YOU meant.
        IF version_col IS NOT NULL AND version_col IS DISTINCT FROM cat.version_col THEN
            errors := errors + 1;
            RETURN QUERY SELECT 'naming', 'ERROR', format('you declared version_col %I, the catalogue says %I', version_col, cat.version_col);
            cat.version_col := version_col;
        END IF;
        IF tombstone_col IS NOT NULL AND tombstone_col IS DISTINCT FROM cat.tombstone_col THEN
            errors := errors + 1;
            RETURN QUERY SELECT 'naming', 'ERROR', format('you declared tombstone_col %I, the catalogue says %s', tombstone_col, coalesce(cat.tombstone_col::text, 'none'));
            cat.tombstone_col := tombstone_col;
        END IF;
        IF tiebreak_col IS NOT NULL AND tiebreak_col IS DISTINCT FROM cat.tiebreak_col THEN
            errors := errors + 1;
            RETURN QUERY SELECT 'naming', 'ERROR', format('you declared tiebreak_col %I, the catalogue says %s', tiebreak_col, coalesce(cat.tiebreak_col::text, 'none'));
            cat.tiebreak_col := tiebreak_col;
        END IF;
        IF tenant_col IS NOT NULL AND tenant_col IS DISTINCT FROM cat.tenant_col THEN
            errors := errors + 1;
            RETURN QUERY SELECT 'naming', 'ERROR', format('you declared tenant_col %I, the catalogue says %s', tenant_col, coalesce(cat.tenant_col::text, 'none (public)'));
            cat.tenant_col := tenant_col;
        END IF;
        RETURN QUERY SELECT 'catalogue', 'ok',
            format('%s: %s; version_col=%s tombstone_col=%s tiebreak_col=%s generations=%s', tname,
                   CASE WHEN cat.tenant_col IS NULL THEN 'public (' || cat.public_reason || ')'
                        ELSE 'tenant-scoped on ' || cat.tenant_col END,
                   cat.version_col, coalesce(cat.tombstone_col::text, '∅'),
                   coalesce(cat.tiebreak_col::text, '∅'), cat.generations);
    END IF;

    -- ── 2. writers: who holds INSERT AND UPDATE (the bridge's writability verdict) ─
    SELECT array_agg(rolname ORDER BY rolname) INTO writers
    FROM pg_roles r
    WHERE NOT r.rolsuper AND r.rolcanlogin AND r.rolname <> owner
      AND has_table_privilege(r.rolname, tbl, 'INSERT')
      AND has_table_privilege(r.rolname, tbl, 'UPDATE');
    want_writable := CASE intent WHEN 'writable' THEN true
                                 WHEN 'read_only' THEN false
                                 ELSE writers IS NOT NULL END;
    IF tname = 'zebridge_gc_watermark' THEN
        -- Bridge-owned: the sweeper writes it through the writer role, every client
        -- reads it, nobody edge-writes it (the ingress refuses it by name).
        RETURN QUERY SELECT 'writers', 'NOTE', 'bridge-owned: written by the sweeper, never edge-writable; the writer grant is by design';
        want_writable := false;
    ELSIF intent = 'writable' AND writers IS NULL THEN
        errors := errors + 1;
        RETURN QUERY SELECT 'writers', 'ERROR',
            'declared writable, but no login role holds INSERT+UPDATE on it: every edge write '
            'will be refused. zebridge_enable(..., writable => true) grants the writer role';
    ELSIF intent = 'read_only' AND writers IS NOT NULL THEN
        errors := errors + 1;
        RETURN QUERY SELECT 'writers', 'ERROR',
            format('declared read_only, but %s hold INSERT+UPDATE: the bridge would publish it as '
                   'writable and accept edge writes. REVOKE, or declare it writable', writers);
    ELSE
        RETURN QUERY SELECT 'writers', 'ok',
            CASE WHEN writers IS NULL THEN 'read-only: no login role holds INSERT+UPDATE'
                 ELSE format('writable by %s', writers) END;
    END IF;

    -- ── 3. key and replica identity ─────────────────────────────────────────
    SELECT array_agg(a.attname ORDER BY k.n) INTO pkcols
    FROM pg_index i
    JOIN LATERAL unnest(i.indkey) WITH ORDINALITY AS k(attnum, n) ON true
    JOIN pg_attribute a ON a.attrelid = i.indrelid AND a.attnum = k.attnum
    WHERE i.indrelid = tbl AND i.indisprimary;
    ident := (SELECT relreplident FROM pg_class WHERE oid = tbl);
    IF pkcols IS NULL THEN
        errors := errors + 1;
        RETURN QUERY SELECT 'primary key', 'ERROR',
            'no primary key: rows cannot be identified, so the bridge refuses the table '
            '(no_primary_key) and publishes a suspension. Add one with a migration';
    ELSE
        RETURN QUERY SELECT 'primary key', 'ok', format('(%s)', array_to_string(pkcols, ', '));
        seq_key := EXISTS (
            SELECT 1 FROM pg_attribute a
            LEFT JOIN pg_attrdef d ON d.adrelid = a.attrelid AND d.adnum = a.attnum
            WHERE a.attrelid = tbl AND a.attname = ANY (pkcols)
              AND (a.attidentity <> '' OR pg_get_expr(d.adbin, d.adrelid) LIKE 'nextval%'));
        IF seq_key AND want_writable THEN
            errors := errors + 1;
            RETURN QUERY SELECT 'key allocation', 'ERROR',
                'the primary key is database-allocated (serial/identity): a client-minted key does '
                'not advance the sequence and would later collide with the application''s own inserts, '
                'so edge writes are refused (DbAllocatedKey). Use a uuid key, or drop the DEFAULT';
        ELSIF seq_key THEN
            RETURN QUERY SELECT 'key allocation', 'NOTE',
                'sequence-backed primary key: fine read-only; would refuse edge writes if made writable';
        END IF;
    END IF;
    IF ident = 'n' THEN
        errors := errors + 1;
        RETURN QUERY SELECT 'replica identity', 'ERROR',
            'REPLICA IDENTITY NOTHING: UPDATE and DELETE carry no key on the wire. '
            'ALTER TABLE ... REPLICA IDENTITY DEFAULT (or USING INDEX)';
    ELSIF ident = 'd' AND pkcols IS NULL THEN
        errors := errors + 1;
        RETURN QUERY SELECT 'replica identity', 'ERROR',
            'REPLICA IDENTITY DEFAULT with no primary key is no identity at all';
    ELSE
        IF ident = 'i' THEN
            SELECT array_agg(a.attname ORDER BY k.n) INTO idx_cols
            FROM pg_index i
            JOIN LATERAL unnest(i.indkey) WITH ORDINALITY AS k(attnum, n) ON true
            JOIN pg_attribute a ON a.attrelid = i.indrelid AND a.attnum = k.attnum
            WHERE i.indrelid = tbl AND i.indisreplident;
        END IF;
        RETURN QUERY SELECT 'replica identity', 'ok',
            CASE ident WHEN 'd' THEN 'DEFAULT (the primary key)'
                       WHEN 'f' THEN 'FULL'
                       WHEN 'i' THEN format('USING INDEX (%s)', array_to_string(idx_cols, ', ')) END;
        -- A tenant-scoped table must carry its tenant in the identity, or UPDATE/DELETE
        -- arrive without it and are refused at the source (NOTES §1.8).
        IF cat.tenant_col IS NOT NULL THEN
            IF ident = 'f' OR (ident = 'd' AND cat.tenant_col = ANY (pkcols))
               OR (ident = 'i' AND cat.tenant_col = ANY (idx_cols)) THEN
                RETURN QUERY SELECT 'tenant in identity', 'ok', format('%s is part of the replica identity', cat.tenant_col);
            ELSE
                errors := errors + 1;
                RETURN QUERY SELECT 'tenant in identity', 'ERROR',
                    format('%s is not in the replica identity: UPDATE/DELETE would arrive without a tenant '
                           'and be refused. CREATE UNIQUE INDEX ... (%s, %s) and REPLICA IDENTITY USING INDEX',
                           cat.tenant_col, cat.tenant_col, array_to_string(pkcols, ', '));
            END IF;
        END IF;
    END IF;

    IF cat.tenant_col IS NOT NULL AND NOT EXISTS (
        SELECT 1 FROM pg_attribute WHERE attrelid = tbl AND attname = cat.tenant_col AND NOT attisdropped) THEN
        errors := errors + 1;
        RETURN QUERY SELECT 'tenant column', 'ERROR', format('%s has no column %I', tname, cat.tenant_col);
    END IF;

    -- ── 4. publication ──────────────────────────────────────────────────────
    SELECT array_agg(pubname ORDER BY pubname) INTO pubs
    FROM pg_publication_tables WHERE schemaname = 'public' AND tablename = tname;
    IF pubs IS NULL THEN
        errors := errors + 1;
        RETURN QUERY SELECT 'publication', 'ERROR',
            'in no publication: no WAL row ever reaches a bridge. zebridge_enable(..., publication => ''<name>'')';
    ELSE
        RETURN QUERY SELECT 'publication', 'ok', format('published by %s', pubs);
    END IF;

    -- ── 5. the version column (last-write-wins) ─────────────────────────────
    vname := coalesce(cat.version_col, 'updated_at');
    SELECT a.attname, t.typname, a.atttypmod, a.attnotnull INTO vcol
    FROM pg_attribute a JOIN pg_type t ON t.oid = a.atttypid
    WHERE a.attrelid = tbl AND a.attname = vname AND NOT a.attisdropped;
    IF NOT FOUND THEN
        errors := errors + 1;
        RETURN QUERY SELECT 'version column', 'ERROR',
            format('%s has no column %I: the table is outbound-only, and a writable one refuses every write', tname, vname);
    ELSIF vcol.typname NOT IN ('timestamp', 'timestamptz', 'int4', 'int8') THEN
        errors := errors + 1;
        RETURN QUERY SELECT 'version column', 'ERROR',
            format('%I is %s: a version must be a timestamp or an integer to be compared', vname, vcol.typname);
    ELSE
        IF vname IN ('inserted_at', 'created_at') THEN
            errors := errors + 1;
            RETURN QUERY SELECT 'version column', 'ERROR',
                format('%I is a creation stamp: it never changes, so no update can ever win', vname);
        ELSE
            RETURN QUERY SELECT 'version column', 'ok', format('%I %s%s', vname, vcol.typname,
                CASE WHEN vcol.attnotnull THEN ' NOT NULL' ELSE '' END);
        END IF;
        IF vcol.typname = 'timestamp' THEN
            warns := warns + 1;
            RETURN QUERY SELECT 'version column', 'WARNING',
                format('%I is a naive timestamp: two clients in two zones disagree on order. Prefer timestamptz', vname);
        END IF;
        IF vcol.typname IN ('timestamp', 'timestamptz') AND vcol.atttypmod BETWEEN 0 AND 5 THEN
            warns := warns + 1;
            RETURN QUERY SELECT 'version column', 'WARNING',
                format('%I keeps %s fractional digits: versions issued within the same instant collide. Use precision 6', vname, vcol.atttypmod);
        END IF;
        IF NOT vcol.attnotnull THEN
            warns := warns + 1;
            RETURN QUERY SELECT 'version column', 'WARNING',
                format('%I is nullable: a NULL version never wins a comparison. Add NOT NULL DEFAULT now()', vname);
        END IF;
    END IF;

    -- ── 6. the tombstone (soft delete, PROTOCOL §7.5) ───────────────────────
    IF cat.tombstone_col IS NOT NULL THEN
        SELECT a.attname, t.typname, a.attnotnull INTO tcol
        FROM pg_attribute a JOIN pg_type t ON t.oid = a.atttypid
        WHERE a.attrelid = tbl AND a.attname = cat.tombstone_col AND NOT a.attisdropped;
        IF NOT FOUND THEN
            errors := errors + 1;
            RETURN QUERY SELECT 'tombstone', 'ERROR', format('tombstone_col %I does not exist on %s', cat.tombstone_col, tname);
        ELSIF tcol.attnotnull THEN
            errors := errors + 1;
            RETURN QUERY SELECT 'tombstone', 'ERROR', format('%I is NOT NULL: a tombstone must be NULL for a live row', cat.tombstone_col);
        ELSE
            RETURN QUERY SELECT 'tombstone', 'ok', format('%I %s', cat.tombstone_col, tcol.typname);
        END IF;
    ELSIF want_writable AND (allow_physical_deletes OR coalesce(cat.allow_physical_deletes, false)) THEN
        warns := warns + 1;
        RETURN QUERY SELECT 'tombstone', 'WARNING',
            'physical deletes ACCEPTED: a hard-deleted row resurrects on a fresh seed until the next full '
            '(the producer forces one when rows were deleted, §10bb) — stated, not refused';
    ELSIF want_writable THEN
        errors := errors + 1;
        RETURN QUERY SELECT 'tombstone', 'ERROR',
            'Accepts writes from devices but has no "deleted" column: a row deleted in PostgreSQL '
            'stays on devices that re-sync from scratch. Fix: add a deleted_at timestamptz column, then '
            'zebridge_enable(''<table>'', tombstone_col => ''deleted_at''). Or allow it: '
            'zebridge_enable(''<table>'', allow_physical_deletes => true)';
    ELSE
        RETURN QUERY SELECT 'tombstone', 'NOTE', 'none: read-only, physical deletes forwarded as deletes';
    END IF;

    -- ── 7. the tiebreak column ──────────────────────────────────────────────
    IF cat.tiebreak_col IS NOT NULL THEN
        IF EXISTS (SELECT 1 FROM pg_attribute WHERE attrelid = tbl AND attname = cat.tiebreak_col AND NOT attisdropped) THEN
            RETURN QUERY SELECT 'tiebreak', 'ok', format('%I (equal versions resolved by writer id)', cat.tiebreak_col);
        ELSE
            errors := errors + 1;
            RETURN QUERY SELECT 'tiebreak', 'ERROR', format('tiebreak_col %I does not exist on %s', cat.tiebreak_col, tname);
        END IF;
    ELSIF want_writable THEN
        RETURN QUERY SELECT 'tiebreak', 'NOTE', 'none: two writes with the same version are refused rather than resolved';
    END IF;

    -- ── 8. the guards zebridge_enable installs ──────────────────────────────
    -- The width guard is a trigger only where a row CAN outgrow the budget; a table whose
    -- widest possible row already fits is statically safe and zebridge_install_width_guard
    -- leaves it alone on purpose (no hot-path cost) — so "no trigger" is a finding only on
    -- an unbounded table. The width check (9) computes the same number.
    SELECT min(max_row_bytes) INTO budget FROM public.zebridge_limits;
    widest := public.zebridge_widest_row(tbl);
    IF EXISTS (SELECT 1 FROM pg_trigger WHERE tgrelid = tbl AND tgname = 'zebridge_width_guard') THEN
        RETURN QUERY SELECT 'width guard', 'ok', 'zebridge_width_guard trigger present';
    ELSIF budget IS NOT NULL AND widest <= budget THEN
        RETURN QUERY SELECT 'width guard', 'NOTE', format('bounded (widest possible row %s ≤ %s): statically safe, no trigger needed', widest, budget);
    ELSE
        warns := warns + 1;
        RETURN QUERY SELECT 'width guard', 'WARNING',
            'no zebridge_width_guard trigger on an unbounded table: an oversized row is caught by the bridge '
            '(table suspended) instead of by PostgreSQL (statement refused). zebridge_enable installs it';
    END IF;
    IF want_writable AND cat.tombstone_col IS NOT NULL
       AND NOT EXISTS (SELECT 1 FROM pg_trigger WHERE tgrelid = tbl AND tgname = 'zebridge_soft_delete_t') THEN
        warns := warns + 1;
        RETURN QUERY SELECT 'write guards', 'WARNING',
            'no zebridge_soft_delete_t trigger: a DELETE from psql is physical here even though the table has a '
            'tombstone. zebridge_install_write_guards(...) turns it into a soft delete';
    END IF;

    -- ── 9. width against the narrowest registered bridge ────────────────────
    IF budget IS NULL THEN
        RETURN QUERY SELECT 'row width', 'NOTE',
            format('widest possible row %s bytes; no bridge has registered its budget yet (zebridge_limits is empty)', widest);
    ELSIF widest > budget THEN
        errors := errors + 1;
        RETURN QUERY SELECT 'row width', 'ERROR',
            format('widest possible row %s bytes exceeds the narrowest bridge budget %s (BASE_BUF): such a row suspends the table', widest, budget);
    ELSE
        RETURN QUERY SELECT 'row width', 'ok', format('widest possible row %s bytes within the %s budget', widest, budget);
    END IF;

    -- ── 10. the chain ───────────────────────────────────────────────────────
    IF have_cat AND cat.generations THEN
        IF EXISTS (SELECT 1 FROM public.zebridge_generations g WHERE g.tbl = tname) THEN
            RETURN QUERY SELECT 'generations', 'ok', 'a chain exists: fresh clients seed from it';
        ELSE
            RETURN QUERY SELECT 'generations', 'NOTE', 'no chain yet: the producer builds one at its next tick, once a bridge runs';
        END IF;
    END IF;

    -- ── 11. the API roles of a PostgREST deployment (Supabase) ──────────────
    -- `anon` and `authenticated` reach `public` over HTTP, with the project's public key.
    -- A grant to either without RLS is the table, readable or writable by anyone who has
    -- that key, outside ZeBridge: no tenant, no version rule, no verdict.
    IF EXISTS (SELECT 1 FROM pg_roles WHERE rolname IN ('anon', 'authenticated'))
       AND NOT (SELECT relrowsecurity FROM pg_class WHERE oid = tbl) THEN
        DECLARE open_to text;
        BEGIN
            SELECT string_agg(r, ', ') INTO open_to FROM unnest(ARRAY['anon', 'authenticated']) r
             WHERE EXISTS (SELECT 1 FROM pg_roles WHERE rolname = r)
               AND has_table_privilege(r, tbl, 'SELECT, INSERT, UPDATE, DELETE');
            IF open_to IS NOT NULL THEN
                warns := warns + 1;
                RETURN QUERY SELECT 'api roles', 'WARNING',
                    format('%s can use %s over the REST API, and RLS is off: revoke it (REVOKE ALL ON %s FROM %s), or enable RLS', open_to, tname, tname, open_to);
            END IF;
        END;
    END IF;

    RETURN QUERY SELECT 'summary', CASE WHEN errors > 0 THEN 'ERROR' WHEN warns > 0 THEN 'WARNING' ELSE 'ok' END,
        format('%s: %s error(s), %s warning(s)%s', tname, errors, warns,
               CASE WHEN intent IS NOT NULL THEN ' against intent ' || intent ELSE '' END);
END $$ LANGUAGE plpgsql STABLE;

-- Every catalogued table, checked against its catalogue row: the intent the DBA already
-- wrote with zebridge_enable, in the migration. Nothing to repeat, no second file.
-- A catalogue row whose table is gone is an ERROR. In psql:
--   SELECT * FROM zebridge_check_all() WHERE status = 'ERROR';   -- empty: a clean bill
DROP FUNCTION IF EXISTS public.zebridge_check_all(jsonb, boolean);
CREATE OR REPLACE FUNCTION public.zebridge_check_all()
RETURNS TABLE (tbl text, check_name text, status text, detail text) AS $$
BEGIN
    RETURN QUERY
        SELECT c.tbl, k.check_name, k.status, k.detail
        FROM (SELECT cat.tbl, to_regclass('public.' || quote_ident(cat.tbl)) AS rc
                FROM public.zebridge_catalogue cat
               WHERE NOT public.zebridge_is_internal_table(cat.tbl)) c
        CROSS JOIN LATERAL public.zebridge_check(c.rc) k
        WHERE c.rc IS NOT NULL
        ORDER BY c.tbl;
    RETURN QUERY
        -- A NOTE, not an error: the row outlives its table on purpose (MIGRATIONS.md), so
        -- a table re-created under the name syncs again with the same rules.
        SELECT cat.tbl, 'dropped'::text, 'NOTE'::text,
               'Dropped, its sync rules kept: a table re-created under the same name syncs again with them. '
               'To forget one: DELETE FROM zebridge_catalogue WHERE tbl = ''<table>'''::text
        FROM public.zebridge_catalogue cat
        WHERE NOT public.zebridge_is_internal_table(cat.tbl)
          AND to_regclass('public.' || quote_ident(cat.tbl)) IS NULL
        ORDER BY cat.tbl;
    -- ZeBridge's own tables and functions, open to a PostgREST deployment's API roles
    -- (Supabase grants them every new object in `public`): anyone with the project's
    -- public key could write an invite, a tenant mapping, or call a definer function.
    -- The init SQL revokes them; a later grant, or an older install, shows here.
    RETURN QUERY
        SELECT c.relname::text, 'api roles'::text, 'ERROR'::text,
               format('%s is open to %s over the REST API: re-run the init SQL, or REVOKE ALL ON public.%I FROM anon, authenticated', c.relname, r.rolname, c.relname)
        FROM pg_class c JOIN pg_namespace n ON n.oid = c.relnamespace
        CROSS JOIN pg_roles r
        WHERE n.nspname = 'public' AND c.relkind IN ('r', 'p', 'v') AND c.relname LIKE 'zebridge\_%'
          AND r.rolname IN ('anon', 'authenticated')
          AND has_table_privilege(r.oid, c.oid, 'SELECT, INSERT, UPDATE, DELETE')
        ORDER BY 1;
    RETURN QUERY
        SELECT p.proname::text, 'definer function'::text, 'ERROR'::text,
               format('%s runs as its owner and anyone may call it (EXECUTE for PUBLIC%s): re-run the init SQL', p.oid::regprocedure,
                      CASE WHEN EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'anon') THEN ', so Supabase''s anon role too' ELSE '' END)
        FROM pg_proc p JOIN pg_namespace n ON n.oid = p.pronamespace
        WHERE n.nspname = 'public' AND p.proname LIKE 'zebridge\_%' AND p.prosecdef
          AND has_function_privilege('public', p.oid, 'EXECUTE')
        ORDER BY 1;
END $$ LANGUAGE plpgsql STABLE;


-- ─── Who may call ZeBridge's functions, and what Supabase's API roles may touch ───────
-- Last, so it covers every function this file created. Three rules:
--   * every ZeBridge function runs with a fixed search_path, pg_catalog first: a function
--     named like a built-in, created in `public` by another role, is never called in its
--     place — which matters most for SECURITY DEFINER functions, run as their owner;
--   * PostgreSQL grants EXECUTE on a new function to PUBLIC, every role: taken back here,
--     and given to the bridge's reader. The writer gets the same in init.write;
--   * a PostgREST deployment (Supabase) exposes `public` to its API roles `anon` and
--     `authenticated`, and grants them every new object: they lose ZeBridge's tables and
--     functions. The application's own tables are the application's to decide; the
--     catalogue check says when one is open without RLS.
DO $$
DECLARE
    f regprocedure;
    api text[] := ARRAY(SELECT rolname::text FROM pg_roles WHERE rolname IN ('anon', 'authenticated'));
    t record;
BEGIN
    FOR f IN SELECT p.oid::regprocedure FROM pg_proc p JOIN pg_namespace n ON n.oid = p.pronamespace
              WHERE n.nspname = 'public' AND p.proname LIKE 'zebridge\_%'
    LOOP
        IF NOT EXISTS (SELECT 1 FROM pg_proc WHERE oid = f AND proconfig::text LIKE '%search_path=%') THEN
            EXECUTE format('ALTER FUNCTION %s SET search_path = pg_catalog, public', f);
        END IF;
        EXECUTE format('REVOKE EXECUTE ON FUNCTION %s FROM PUBLIC', f);
        EXECUTE format('GRANT EXECUTE ON FUNCTION %s TO %I', f, '${POSTGRES_READER_USER}');
        IF cardinality(api) > 0 THEN
            EXECUTE format('REVOKE EXECUTE ON FUNCTION %s FROM %s', f, array_to_string(ARRAY(SELECT quote_ident(a) FROM unnest(api) a), ', '));
        END IF;
    END LOOP;
    IF cardinality(api) > 0 THEN
        FOR t IN SELECT c.oid::regclass AS rel FROM pg_class c JOIN pg_namespace n ON n.oid = c.relnamespace
                  WHERE n.nspname = 'public' AND c.relkind IN ('r', 'p', 'v', 'S') AND c.relname LIKE 'zebridge\_%'
        LOOP
            EXECUTE format('REVOKE ALL ON %s FROM %s', t.rel, array_to_string(ARRAY(SELECT quote_ident(a) FROM unnest(api) a), ', '));
        END LOOP;
    END IF;
    -- The reader policies made before this rule re-evaluated current_setting() per row:
    -- rewritten in place, once (the wrapped form contains `SELECT current_setting`).
    FOR t IN SELECT pol.polname, pol.polrelid::regclass AS rel, pg_get_expr(pol.polqual, pol.polrelid) AS q
               FROM pg_policy pol WHERE pol.polname = 'zb_reader_all'
                AND pg_get_expr(pol.polqual, pol.polrelid) LIKE '%current_setting(%'
                AND pg_get_expr(pol.polqual, pol.polrelid) NOT LIKE '%SELECT current_setting(%'
    LOOP
        EXECUTE format('ALTER POLICY %I ON %s USING (%s)', t.polname, t.rel,
            regexp_replace(t.q, 'current_setting\(''([a-z_.]+)''::text, true\)', '(SELECT current_setting(''\1''::text, true))', 'g'));
    END LOOP;
END $$;