init.write.template.sql

-- ─────────────────────────────────────────────────────────────────────────────
-- init.write.sql — the write half: ingress, guards, and optional tenant scoping.
--
-- ⚠️ **Appended to `init.core.template.sql`, never run alone.** It needs the reader role,
-- the publication and the DDL triggers that file creates:
--
--     cat init.core.template.sql init.write.template.sql | envsubst | psql -v ON_ERROR_STOP=1 …
--
-- Everything below is **defined, not applied**. These are functions a DBA calls per
-- table; loading this file grants nobody anything beyond the writer role's CONNECT and
-- USAGE. Making a table writable is a deliberate act:
--
--     SELECT zebridge_grant_edge_writes('public.orders');
--     SELECT zebridge_install_write_guards('public.orders', 'updated_at', 'deleted_at');
--
-- Multi-tenant is not a build flag, it is two more calls, both live without a restart:
--
--     INSERT INTO zebridge_user_tenants (principal, tenant_id) VALUES ('alice', 'acme');
--     SELECT zebridge_enable('public.orders', tenant_col => 'tenant_id', dry_run => false);
--     -- the catalogue row it writes is what routes the *subject*
--
-- ⚠️ `zebridge_scope_publication_to_one_tenant()` is a different shape entirely — it pins
-- a publication to ONE tenant value, for one-bridge-per-tenant deployments. It is not the
-- multi-tenant path above.
-- ─────────────────────────────────────────────────────────────────────────────

-- ---------------------------------------------------------
-- The WRITE role — separate from the read role on purpose
-- ---------------------------------------------------------
--
-- ${POSTGRES_READER_USER} stays exactly what it is: SELECT + REPLICATION, physically
-- unable to write. The ingress path (NATS -> bridge -> PostgreSQL) uses a *second*
-- connection under this role, so adding the write feature never widens the privileges
-- of the read path.
--
-- It is created with **no table privileges at all**. That is the backstop that makes
-- "the bridge is a dumb dispatcher" safe rather than merely simple: even a bug in
-- subject parsing cannot reach a table the DBA has not explicitly opened, because the
-- role has no path to one. Grants are per table, via
-- zebridge_grant_edge_writes() below — never ALL TABLES, and never by default.
--
-- Two attributes are spelled out even though they are the defaults, because each fails
-- **open** if it is ever wrong:
--   NOBYPASSRLS  - with BYPASSRLS every row-level policy becomes decorative
--   NOREPLICATION- the write role must not be able to read the WAL
--
-- A third rule cannot be expressed here and belongs in review: this role must **not own
-- any table**. Table owners are exempt from row-level security unless the table is set
-- to FORCE ROW LEVEL SECURITY, so an owning writer silently bypasses every policy.
DO $$
BEGIN
    IF NOT EXISTS (SELECT 1 FROM pg_roles WHERE rolname = '${POSTGRES_WRITER_USER}') THEN
        CREATE USER ${POSTGRES_WRITER_USER} WITH PASSWORD '${POSTGRES_WRITER_PASSWORD}'
            NOSUPERUSER NOCREATEDB NOCREATEROLE NOREPLICATION NOBYPASSRLS;
    END IF;
    -- The writer's connection BUDGET, DB-enforced (outside the IF: applies on
    -- re-apply to existing roles). Inventory: one per mutation listener lane
    -- (ZB_INGRESS_LANES, at most 8) + sweeper 1 + enrollment permits ≤4
    -- (http_server caps them, connect_timeout=3 bounds a hang) + margin. 20 total,
    -- with the reader's 10 still under a third of the cluster's default 100: the
    -- bridge cannot starve it even through a bug — the budget is a rule in PG, not
    -- prose. The bridge checks the arithmetic at every boot (preflight's 🔌 line).
    ALTER ROLE ${POSTGRES_WRITER_USER} CONNECTION LIMIT 20;
END;
$$;

GRANT CONNECT ON DATABASE ${TARGET_DB} TO ${POSTGRES_WRITER_USER};
GRANT USAGE ON SCHEMA public TO ${POSTGRES_WRITER_USER};

-- Read-only catalogue access (moved here from init.core: that file must never name
-- this role, or the read-only profile cannot apply on a fresh cluster).
GRANT SELECT ON public.zebridge_catalogue TO ${POSTGRES_WRITER_USER};

-- Bookkeeping the bridge records THROUGH THE READER by design (its width budget, the
-- refusal mirror, zebridge_generations) so a read-only deployment can record it at
-- all. When DATABASE_READER_URL is a hot standby the server refuses every write
-- there, and the bridge sends the same bookkeeping over DATABASE_WRITER_URL instead
-- (NOTES §10cz) — so the writer needs exactly the reader's bookkeeping privileges.
GRANT SELECT, INSERT, DELETE ON public.zebridge_generations TO ${POSTGRES_WRITER_USER};
GRANT UPDATE (has_full, retired_at, has_checkpoint, ckpt_lower, obj_bytes, full_sorted, ckpt_sorted) ON public.zebridge_generations TO ${POSTGRES_WRITER_USER};  -- §10gf/§10gq, the reader's column grant
GRANT SELECT ON public.zebridge_limits TO ${POSTGRES_WRITER_USER};  -- the registration reads MIN and its previous row around the call
GRANT EXECUTE ON FUNCTION public.zebridge_register_limits(text, name, integer, boolean) TO ${POSTGRES_WRITER_USER};
GRANT EXECUTE ON FUNCTION public.zebridge_set_suspended(text, text) TO ${POSTGRES_WRITER_USER};
GRANT EXECUTE ON FUNCTION public.zebridge_clear_suspensions() TO ${POSTGRES_WRITER_USER};
GRANT EXECUTE ON FUNCTION public.zebridge_observe_incarnation(text, oid) TO ${POSTGRES_WRITER_USER};


-- Deliberately NOT granted: SELECT/INSERT/UPDATE on any table, and no ALTER DEFAULT
-- PRIVILEGES. A new table is not edge-writable until someone says so.


-- Open one table to edge writes. Idempotent; call it once per table.
--
-- SELECT is required alongside INSERT/UPDATE because the last-write-wins guard
-- (`WHERE stored.version < EXCLUDED.version`) reads the existing row.
--
-- zebridge_ddl_events is refused by name: it is the bridge's own DDL tracker, and a
-- client able to insert into it could forge a schema, a suspension or a tombstone for
-- every other client. That is the one table where "can write my own rows" becomes
-- "can control every replica".
-- opens one table to edge writes — see PROTOCOL.md §7.4
CREATE OR REPLACE FUNCTION public.zebridge_grant_edge_writes(tbl regclass)
RETURNS void AS $$
DECLARE
    tbl_name text := tbl::text;
    rel_name text := (SELECT relname FROM pg_class WHERE oid = tbl);
BEGIN
    -- ⚠️ Every bridge-owned table, not just the DDL transport. Each is a
    -- different escalation, and the second one was proven exploitable on 2026-08-28
    -- (NOTES §10aq) before the ingress path learned to refuse it by name:
    --
    --   zebridge_ddl_events    a forged row publishes a fabricated schema, suspension
    --                          or tombstone to EVERY client
    --   zebridge_gc_watermark  the offline window for every client — one principal
    --                          dragged it back six years with a single mutation
    --   zebridge_user_tenants  the principal->tenant roster: a forged row is
    --                          cross-tenant privilege escalation
    --   zebridge_invites       the enrollment table. bridge_writer holds UPDATE so
    --                          redemption can stamp used_at — and UPDATE covers every
    --                          column, so a known code could be un-spent, un-expired,
    --                          or re-pointed at another principal, tenant and role
    --
    -- This does not revoke what the BRIDGE itself needs (the sweeper writes the
    -- watermark as bridge_writer, and must keep that grant). It refuses to hand those
    -- tables to the EDGE, which is a different question and the one that was wrong.
    -- The whole zebridge_ namespace (zebridge_enable refuses a user table with that prefix), so a
    -- table a later release adds is covered without anyone remembering a list: it used to be
    -- five names, in both spellings of the regclass text.
    IF lower(left(rel_name, 9)) = 'zebridge_' THEN
        RAISE EXCEPTION 'refusing to grant edge writes on %: this is a bridge-owned '
                        'table — a forged row here reaches every client', tbl_name;
    END IF;
    -- Mutation receipts, by schema: the parent, every monthly partition (present or
    -- future) and the policy row. A forged receipt silences a write that never ran.
    IF (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 'refusing to grant edge writes on %: zebridge_private holds the bridge''s '
                        'mutation receipts', tbl_name;
    END IF;

    -- DELETE as well as SELECT/INSERT/UPDATE, for two paths that both need it:
    --
    --   * a table with **no tombstone column** deletes physically — that is the documented
    --     weaker guarantee, and it is an actual DELETE;
    --   * the **tombstone sweeper** (src/gc.zig) reaps soft-deleted rows past
    --     GC_THRESHOLD_MS, and runs as this role precisely so it is not the admin.
    --
    -- Marginal risk is small: this role can already UPDATE any row it may write, and RLS
    -- bounds which rows those are. Withholding DELETE did not prevent data loss, it only
    -- made the sweeper fail with `permission denied` — silently, in a sidecar nobody reads.
    EXECUTE format('GRANT SELECT, INSERT, UPDATE, DELETE ON %s TO ${POSTGRES_WRITER_USER}', tbl);
    RAISE NOTICE 'edge writes enabled on % for ${POSTGRES_WRITER_USER}', tbl;
END;
$$ LANGUAGE plpgsql;

-- lets the tombstone sweeper reap a table that keeps tombstones but takes no edge writes — see PROTOCOL.md §7.5
--
-- A table can keep tombstones without being writable from the edge: a loader that
-- soft-deletes and revives rows (examples/08-map/load_fuel.py), so a returning client
-- learns what went. Those tombstones must be reaped too, and the sweeper runs as this
-- role: without SELECT and DELETE it fails with `permission denied` on every pass, and
-- the tombstones pile up for ever. zebridge_enable calls this for a read-only table with
-- a tombstone_col; a writable one already has both (zebridge_grant_edge_writes).
--
-- Under row-level security (a tenant-scoped read-only table: the reader's SELECT policy
-- only), the grant alone reaps nothing — the writer sees no row. So the sweeper's session,
-- and only it (zb.principal = 'zb_sweeper', as the tombstone guard knows it), gets its
-- own SELECT and DELETE policies. No INSERT, no UPDATE: the table stays read-only.
CREATE OR REPLACE FUNCTION public.zebridge_grant_sweeper_reaps(tbl regclass)
RETURNS void AS $$
BEGIN
    EXECUTE format('GRANT SELECT, DELETE ON %s TO ${POSTGRES_WRITER_USER}', tbl);
    IF (SELECT relrowsecurity FROM pg_class WHERE oid = tbl) THEN
        EXECUTE format('DROP POLICY IF EXISTS zb_sweeper_read ON %s', tbl);
        EXECUTE format('CREATE POLICY zb_sweeper_read ON %s FOR SELECT TO ${POSTGRES_WRITER_USER} '
                       'USING (coalesce(current_setting(''zb.principal'', true), '''') = ''zb_sweeper'')', tbl);
        EXECUTE format('DROP POLICY IF EXISTS zb_sweeper_reap ON %s', tbl);
        EXECUTE format('CREATE POLICY zb_sweeper_reap ON %s FOR DELETE TO ${POSTGRES_WRITER_USER} '
                       'USING (coalesce(current_setting(''zb.principal'', true), '''') = ''zb_sweeper'')', tbl);
    END IF;
    RAISE NOTICE 'the sweeper may reap % (SELECT, DELETE for ${POSTGRES_WRITER_USER})', tbl;
END;
$$ LANGUAGE plpgsql;


-- ---------------------------------------------------------
-- Write guards — making the version column true for EVERY writer
-- ---------------------------------------------------------
--
-- Last-write-wins compares `stored.version < incoming.version`, and the bridge stamps the
-- version on every write it applies. The bridge is not the only writer. A cron job, a psql
-- session, another service, or an ORM path that forgets can run
--
--     UPDATE users SET name = 'x' WHERE id = 1;      -- updated_at untouched
--
-- and the row changes while the version does not. The stored version then no longer
-- describes the row, so a client's OLDER edit beats a NEWER server write — silently, and
-- "correctly" by the rule. See PROTOCOL.md §7.3.
--
-- A trigger fixes it without a migration: it changes no shape and breaks no query, which
-- is what makes it usable on a database you do not control.

-- Stamp the version column when the statement did not — see PROTOCOL.md §7.3
CREATE OR REPLACE FUNCTION public.zebridge_bump_version()
RETURNS trigger AS $$
DECLARE
    col text := TG_ARGV[0];
BEGIN
    -- Only when this statement left the version alone. A writer that DID set it — the
    -- bridge, which sends the client's value and clamps it (§7.3), or a well-behaved ORM —
    -- keeps what it wrote. Overwriting unconditionally would make the version
    -- server-assigned and quietly discard the value the client was told to send back.
    IF (to_jsonb(OLD) -> col) IS NOT DISTINCT FROM (to_jsonb(NEW) -> col) THEN
        -- jsonb round trip because plpgsql cannot assign to a column named at runtime.
        -- One conversion per updated row, on tables the operator opted in.
        NEW := jsonb_populate_record(NEW, jsonb_build_object(col, now()));
    END IF;
    RETURN NEW;
END;
$$ LANGUAGE plpgsql;

-- Turn a physical DELETE into a tombstone — see PROTOCOL.md §7.5
CREATE OR REPLACE FUNCTION public.zebridge_soft_delete()
RETURNS trigger AS $$
DECLARE
    tombstone text := TG_ARGV[0];
    version   text := TG_ARGV[1];
BEGIN
    -- The sweeper MUST get through, or tombstones accumulate forever and the reaping this
    -- whole design depends on can never happen. It identifies itself the same way every
    -- other writer does — `zb.principal`, the setting the RLS policies read — so no new
    -- mechanism is needed here.
    IF coalesce(current_setting('zb.principal', true), '') = 'zb_sweeper' THEN
        RETURN OLD;
    END IF;

    -- Located by `ctid`, not by primary key: the key columns are not known here without a
    -- catalog lookup per row, and the tuple's physical address is stable for the statement
    -- that is deleting it.
    --
    -- The version is stamped too, not just the tombstone. A soft delete that did not move
    -- the version would be refused by the next edge write's LWW guard — the row would read
    -- as deleted while still accepting edits.
    EXECUTE format(
        'UPDATE %I.%I SET %I = now(), %I = now() WHERE ctid = $1',
        TG_TABLE_SCHEMA, TG_TABLE_NAME, tombstone, version
    ) USING OLD.ctid;

    -- Suppress the DELETE. The UPDATE above is what reaches CDC, so clients see the
    -- tombstone rather than a physical removal they cannot overrule.
    RETURN NULL;
END;
$$ LANGUAGE plpgsql;

-- Attach both to one table — see PROTOCOL.md §7.3
--
-- The column names must match this table's zebridge_catalogue row. zebridge_enable writes
-- both in one transaction; called by hand, nothing cross-checks them: name a
-- different column here and the bridge and the database disagree about what "version"
-- means, with no error. `zebridge_audit_write_guards()` reports what is actually attached
-- so the two can be compared.
-- Tenant guard: every row leaves with a legal tenant token, or the write is refused.
--
-- Attached per table by `zebridge_install_write_guards`, which passes the column name as a
-- trigger argument. It fills in a missing tenant and refuses an unusable one — because the
-- bridge copies the column straight into `cdc.<tenant>.<table>.<op>` without interpreting it.
--
--   1. **absence** — omitted, NULL, or empty → **the writer's own tenant**, looked up from
--      `zebridge_user_tenants` by `zb.principal`. This is the correction that matters: a
--      tenanted client that forgets the column must NOT have its (sensitive) row silently
--      published to everyone. Omission means "mine", never "open" — and only while
--      "mine" is one tenant: a member of several (§10fn) must name it, the omission
--      is refused rather than guessed.
--
--      ⚠️ **Fail-CLOSED on no derivable tenant.** If `zb.principal` is unset or unmapped,
--      there is no tenant to stamp and the write is REJECTED — a row on a tenant-routed
--      table with no tenant cannot be routed, so producing one is worse than refusing. An
--      admin seeding such a table must state the tenant, exactly as a client does.
--
--      The OPEN tenant (`${OPEN_TENANT}`) is no longer a coalesce target — it is simply the
--      tenant a principal *mapped to it* carries, so its rows land in `CDC_PUBLIC` and are
--      read by everyone. "Open" is a mapping, not a default: to write open data you are a
--      member of the open tenant, audited in the one table like every other writer.
--
--   2. **malformed** — a real value with a NATS subject metacharacter (`.`, space, `*`, `>`)
--      → RAISES. A dot splits the token and the row routes to a subject no consumer
--      receives. Not corrected: a caller that sent `evil.acme` meant something, and
--      rerouting it silently is worse than refusing.
--
--   3. **valid** — passed through untouched; `zb_tenant_write` (RLS `WITH CHECK`) is what
--      confirms the writer may actually write that tenant, so a supplied value is validated,
--      not trusted.
--
-- ⚠️ `set_config('zb.principal', …, true)` MUST share the write's transaction (the bridge's
-- pipeline does). In autocommit each statement is its own transaction and the setting is
-- gone by the INSERT — the lookup then finds nothing and this fails closed.
--
-- ⚠️ Runs BEFORE the NOT NULL check (Postgres fires BEFORE-row triggers first), so the
-- column stays NOT NULL with no DEFAULT and this still satisfies it.
--
-- ⚠️ The jsonb round trip is how a dynamically-named column is written — plain plpgsql
-- cannot assign `NEW.<variable>`. Same trick as `zebridge_bump_version()`; no extension, and
-- a generated per-table function would need nested dollar-quoting, which `envsubst`
-- (the bridge-init pipe) destroys.
CREATE OR REPLACE FUNCTION public.zebridge_guard_tenant()
RETURNS trigger AS $$
DECLARE
    col  text := TG_ARGV[0];
    val  text;
    mine text;
    n    bigint;
    who  text := current_setting('zb.principal', true);
BEGIN
    EXECUTE format('SELECT ($1).%I::text', col) USING NEW INTO val;

    IF val IS NULL OR val = '' THEN
        -- Derive from identity — exact only when the principal belongs to ONE tenant.
        -- A member of several (§10fn) must say which: guessing one would route the row
        -- to a tenant the writer did not name, so the omission is refused, not filled.
        SELECT min(tenant_id), count(*) INTO mine, n FROM public.zebridge_user_tenants
            WHERE principal = who;
        IF n > 1 THEN
            RAISE EXCEPTION 'no tenant for a write to %: principal % belongs to % tenants, '
                'so % must be supplied explicitly.',
                TG_TABLE_NAME, quote_literal(who), n, col
                USING ERRCODE = 'check_violation';
        END IF;
        IF mine IS NULL THEN
            RAISE EXCEPTION 'no tenant for a write to %: principal % is unset or not in '
                'zebridge_user_tenants, and a tenant-routed row cannot be left unrouted. '
                'Set zb.principal in the transaction, or supply % explicitly.',
                TG_TABLE_NAME, coalesce(quote_literal(who), 'NULL'), col
                USING ERRCODE = 'check_violation';
        END IF;
        NEW := jsonb_populate_record(NEW, jsonb_build_object(col, mine));
        RETURN NEW;
    END IF;

    -- A tenant is a NATS subject token, measured in BYTES (octet_length, not length): no `.`, `*`,
    -- `>`, space or control character, and at most ${MAX_TENANT_BYTES} bytes. The bridge drops a row
    -- that gets past this (counted: bridge_dropped_events_total{reason="tenant_unusable"}), so this
    -- trigger is what keeps the replica from silently missing it. Existing rows are not rechecked by
    -- a trigger; `bridge --diagnose` lists them.
    IF val ~ '[\x01-\x20\x7f.*>]' THEN
        RAISE EXCEPTION 'tenant %=% on % contains a NATS subject metacharacter '
            '(. * > space or a control character); it would route to a subject no consumer receives',
            col, quote_literal(val), TG_TABLE_NAME USING ERRCODE = 'check_violation';
    END IF;
    IF octet_length(val) > ${MAX_TENANT_BYTES} THEN
        RAISE EXCEPTION 'tenant %=% on % is % bytes; a tenant is at most ${MAX_TENANT_BYTES} bytes '
            '(measured in bytes, not characters)',
            col, quote_literal(left(val, 24) || '…'), TG_TABLE_NAME, octet_length(val)
            USING ERRCODE = 'check_violation';
    END IF;

    RETURN NEW;
END $$ LANGUAGE plpgsql;


-- A parent's tombstone while a live child still references it (§10dn). Every foreign
-- key pointing at this table is looked up in pg_constraint; a referencing row counts
-- as live unless the child's own tombstone column (from the catalogue) is set. Raised
-- as foreign_key_violation (23503): the bridge turns the failed UPDATE into a
-- `rejected` verdict naming the child table, exactly as a physical delete would have
-- been refused by NO ACTION. Fresh tombstones only: an already-tombstoned row's
-- update (a version bump, a reap-time touch) passes.
CREATE OR REPLACE FUNCTION public.zebridge_tombstone_children_guard()
RETURNS trigger AS $$
DECLARE
    tomb       text := TG_ARGV[0];
    fk         record;
    cond       text;
    i          int;
    child_tomb text;
    n          bigint;
BEGIN
    IF (to_jsonb(NEW) ->> tomb) IS NULL OR (to_jsonb(OLD) ->> tomb) IS NOT NULL THEN
        RETURN NEW;
    END IF;
    FOR fk IN SELECT con.conname, con.conrelid, con.conkey, con.confkey
              FROM pg_constraint con
              WHERE con.confrelid = TG_RELID AND con.contype = 'f'
    LOOP
        cond := '';
        FOR i IN 1..array_length(fk.conkey, 1) LOOP
            cond := cond || CASE WHEN i > 1 THEN ' AND ' ELSE '' END
                 || format('%I = %L',
                           (SELECT attname FROM pg_attribute WHERE attrelid = fk.conrelid AND attnum = fk.conkey[i]),
                           to_jsonb(OLD) ->> (SELECT attname FROM pg_attribute WHERE attrelid = TG_RELID AND attnum = fk.confkey[i]));
        END LOOP;
        SELECT c.tombstone_col INTO child_tomb
        FROM public.zebridge_catalogue c
        WHERE c.tbl = (SELECT relname FROM pg_class WHERE oid = fk.conrelid);
        IF child_tomb IS NOT NULL THEN
            cond := cond || format(' AND %I IS NULL', child_tomb);
        END IF;
        EXECUTE format('SELECT count(*) FROM %s WHERE %s', fk.conrelid::regclass, cond) INTO n;
        IF n > 0 THEN
            RAISE EXCEPTION USING
                ERRCODE = 'foreign_key_violation',
                MESSAGE = format('cannot tombstone %s: %s live row(s) in %s still reference it (constraint %s) — delete the children first',
                                 TG_TABLE_NAME, n, fk.conrelid::regclass, fk.conname);
        END IF;
    END LOOP;
    RETURN NEW;
END;
$$ LANGUAGE plpgsql;

CREATE OR REPLACE FUNCTION public.zebridge_install_write_guards(
    tbl regclass,
    version_col text,
    tombstone_col text DEFAULT NULL,
    tenant_col text DEFAULT NULL
)
RETURNS void AS $$
DECLARE
    t text := tbl::text;
BEGIN
    IF NOT EXISTS (SELECT 1 FROM pg_attribute WHERE attrelid = tbl
                   AND attname = version_col AND attnum > 0 AND NOT attisdropped) THEN
        RAISE EXCEPTION '% has no column %', t, version_col;
    END IF;

    EXECUTE format('DROP TRIGGER IF EXISTS zebridge_bump_version_t ON %s', t);
    EXECUTE format(
        'CREATE TRIGGER zebridge_bump_version_t BEFORE UPDATE ON %s '
        'FOR EACH ROW EXECUTE FUNCTION public.zebridge_bump_version(%L)', t, version_col
    );
    RAISE NOTICE 'version guard on %: % is stamped even when a writer forgets', t, version_col;

    EXECUTE format('DROP TRIGGER IF EXISTS zebridge_soft_delete_t ON %s', t);
    IF tombstone_col IS NOT NULL THEN
        IF NOT EXISTS (SELECT 1 FROM pg_attribute WHERE attrelid = tbl
                       AND attname = tombstone_col AND attnum > 0 AND NOT attisdropped) THEN
            RAISE EXCEPTION '% has no column %', t, tombstone_col;
        END IF;
        EXECUTE format(
            'CREATE TRIGGER zebridge_soft_delete_t BEFORE DELETE ON %s '
            'FOR EACH ROW EXECUTE FUNCTION public.zebridge_soft_delete(%L, %L)',
            t, tombstone_col, version_col
        );
        RAISE NOTICE 'delete guard on %: DELETE writes % instead of removing the row', t, tombstone_col;
        -- §10dn: a tombstone is a DELETE on every replica, and a replica cannot delete a
        -- parent whose children are still there (its foreign keys have no cascade). So
        -- the tombstone is refused HERE, where the truth lives, while a live child still
        -- references the row: the client gets `rejected`, and deletes the children first.
        EXECUTE format('DROP TRIGGER IF EXISTS zebridge_tombstone_children_t ON %s', t);
        EXECUTE format(
            'CREATE TRIGGER zebridge_tombstone_children_t BEFORE UPDATE OF %I ON %s '
            'FOR EACH ROW EXECUTE FUNCTION public.zebridge_tombstone_children_guard(%L)',
            tombstone_col, t, tombstone_col
        );
    ELSE
        -- No tombstone column means deletes are physical for this table — the documented
        -- weaker guarantee, not an oversight. Saying so beats silence.
        RAISE NOTICE 'no tombstone for %: deletes stay physical, so an offline client can resurrect a row', t;
    END IF;

    -- The tenant guard, when the table is tenant-routed. One shared function,
    -- `zebridge_guard_tenant()`, with the column name passed as a trigger argument.
    EXECUTE format('DROP TRIGGER IF EXISTS zebridge_guard_tenant_t ON %s', t);
    IF tenant_col IS NOT NULL THEN
        IF NOT EXISTS (SELECT 1 FROM pg_attribute WHERE attrelid = tbl
                       AND attname = tenant_col AND attnum > 0 AND NOT attisdropped) THEN
            RAISE EXCEPTION '% has no column %', t, tenant_col;
        END IF;
        EXECUTE format(
            'CREATE TRIGGER zebridge_guard_tenant_t BEFORE INSERT OR UPDATE ON %s '
            'FOR EACH ROW EXECUTE FUNCTION public.zebridge_guard_tenant(%L)', t, tenant_col
        );
        RAISE NOTICE 'tenant guard on %: % filled from the writer''s tenant when omitted (fail-closed if unmapped), malformed or longer than ${MAX_TENANT_BYTES} bytes refused', t, tenant_col;
    END IF;
END;
$$ LANGUAGE plpgsql;

-- ── Registers (COOPERATIVE_EDITING.md) ───────────────────────────────────────
-- A register column is one jsonb object of named registers, each {"v": value, "t": stamp,
-- "w": writer}. The row's last-writer-wins decides whether a write is ACCEPTED; for a
-- register column it would then replace the whole document with what its writer last saw
-- — a late write from an editor who was offline rolled back registers written meanwhile
-- by others, accepted, with no verdict saying so (measured in example 15). So on every
-- accepted UPDATE the column is merged instead, register by register: the newest edit of
-- each field survives, whoever wrote the row last — Cassandra's per-cell timestamps,
-- Figma's per-property last-writer-wins, on a plain jsonb column.
--
-- zebridge_merge_registers is mergeRegisters (libzb core.zig, which zb-client-ts runs as WebAssembly), pinned
-- by the same fixtures: per key the later `t` wins, an equal `t` breaks on `w`, a key on
-- one side only is kept, a value without `t` counts as the oldest. Stamps and writers are
-- compared byte by byte (COLLATE "C"), as JavaScript and Zig compare strings — a locale
-- collation could order two stamps differently. Registers only gain: removing one is
-- writing it again with a newer `t` and an empty `v`.
CREATE OR REPLACE FUNCTION public.zebridge_merge_registers(a jsonb, b jsonb) RETURNS jsonb AS $$
    SELECT COALESCE(jsonb_object_agg(k, CASE
               WHEN rb IS NULL THEN ra
               WHEN ra IS NULL THEN rb
               WHEN COALESCE(rb ->> 't', '') COLLATE "C" > COALESCE(ra ->> 't', '') COLLATE "C" THEN rb
               WHEN COALESCE(rb ->> 't', '') = COALESCE(ra ->> 't', '')
                AND COALESCE(rb ->> 'w', '') COLLATE "C" > COALESCE(ra ->> 'w', '') COLLATE "C" THEN rb
               ELSE ra END), '{}'::jsonb)
    FROM (SELECT ea.key AS k, ea.value AS ra, NULL::jsonb AS rb
            FROM jsonb_each(CASE WHEN jsonb_typeof(a) = 'object' THEN a ELSE '{}'::jsonb END) ea
           WHERE NOT (CASE WHEN jsonb_typeof(b) = 'object' THEN b ELSE '{}'::jsonb END) ? ea.key
          UNION ALL
          SELECT eb.key, (CASE WHEN jsonb_typeof(a) = 'object' THEN a ELSE '{}'::jsonb END) -> eb.key, eb.value
            FROM jsonb_each(CASE WHEN jsonb_typeof(b) = 'object' THEN b ELSE '{}'::jsonb END) eb) m
$$ LANGUAGE sql IMMUTABLE PARALLEL SAFE;

-- The trigger: TG_ARGV names the register columns. BEFORE INSERT, and BEFORE UPDATE OF
-- those columns, so a write that does not set one (part A's plain columns) costs nothing;
-- a write the version guard refuses (stale) never reaches it — the row's rule still
-- decides WHETHER, this decides WHAT.
--
-- A register column holds a JSON object of registers, or NULL. Anything else (an array, a
-- string, a number) is refused: merged, it would count as an empty document, so the write
-- would change nothing and still be reported applied. SQLSTATE 22023 is permanent to the
-- bridge, so the client gets `rejected` at once, with this message as the detail.
CREATE OR REPLACE FUNCTION public.zebridge_merge_register_cols() RETURNS trigger AS $$
DECLARE
    col   text;
    val   jsonb;
    patch jsonb := '{}'::jsonb;
BEGIN
    FOREACH col IN ARRAY TG_ARGV LOOP
        val := to_jsonb(NEW) -> col;
        IF jsonb_typeof(val) NOT IN ('object', 'null') THEN
            RAISE EXCEPTION '%.%: a register column holds a JSON object of registers (got %)', TG_TABLE_NAME, col, jsonb_typeof(val)
                USING ERRCODE = '22023';
        END IF;
        IF TG_OP = 'UPDATE' THEN
            patch := patch || jsonb_build_object(col, public.zebridge_merge_registers(to_jsonb(OLD) -> col, val));
        END IF;
    END LOOP;
    IF TG_OP = 'UPDATE' THEN
        NEW := jsonb_populate_record(NEW, patch);
    END IF;
    RETURN NEW;
END;
$$ LANGUAGE plpgsql;

-- Installs the merge on `cols` (each a jsonb column); idempotent, replaces the column list.
CREATE OR REPLACE FUNCTION public.zebridge_install_register_merge(tbl regclass, cols name[]) RETURNS void AS $$
DECLARE
    c   name;
    typ text;
BEGIN
    IF cols IS NULL OR cardinality(cols) = 0 THEN
        EXECUTE format('DROP TRIGGER IF EXISTS zebridge_merge_registers_t ON %s', tbl);
        RETURN;
    END IF;
    FOREACH c IN ARRAY cols LOOP
        SELECT format_type(atttypid, atttypmod) INTO typ
          FROM pg_attribute WHERE attrelid = tbl AND attname = c AND NOT attisdropped;
        IF typ IS NULL THEN
            RAISE EXCEPTION '%: no column %', tbl, c;
        ELSIF typ <> 'jsonb' THEN
            RAISE EXCEPTION '%.%: a register column is jsonb, not %', tbl, c, typ;
        END IF;
    END LOOP;
    EXECUTE format('DROP TRIGGER IF EXISTS zebridge_merge_registers_t ON %s', tbl);
    EXECUTE format('CREATE TRIGGER zebridge_merge_registers_t BEFORE INSERT OR UPDATE OF %s ON %s '
                   'FOR EACH ROW EXECUTE FUNCTION public.zebridge_merge_register_cols(%s)',
                   (SELECT string_agg(quote_ident(x), ', ') FROM unnest(cols) x), tbl,
                   (SELECT string_agg(quote_literal(x), ', ') FROM unnest(cols) x));
END;
$$ LANGUAGE plpgsql;

-- Take the guards off again — the counterpart, and needed more often than it looks
--
-- ⚠️ Once the delete guard is on, **nobody but the sweeper can physically delete from this
-- table**. An admin running `DELETE FROM t WHERE …` in psql gets a tombstone, not a
-- removal. That is the whole point, and it is also a genuine surprise the first time it
-- happens during a cleanup. Two ways out, in order of preference:
--
--   SELECT set_config('zb.principal', 'zb_sweeper', false);   -- for this session only
--   SELECT zebridge_remove_write_guards('public.t');          -- permanently
CREATE OR REPLACE FUNCTION public.zebridge_remove_write_guards(tbl regclass)
RETURNS void AS $$
DECLARE
    t text := tbl::text;
BEGIN
    EXECUTE format('DROP TRIGGER IF EXISTS zebridge_bump_version_t ON %s', t);
    EXECUTE format('DROP TRIGGER IF EXISTS zebridge_soft_delete_t ON %s', t);
    RAISE NOTICE 'write guards removed from %: versions are only as true as their writers again', t;
END;
$$ LANGUAGE plpgsql;
-- Scope WRITES by tenant, without pinning the publication — see PROTOCOL.md §7.4
--
-- ⚠️ The companion to `zebridge_scope_publication_to_one_tenant`, and the difference is the whole
-- point. That one adds a publication row filter (`WHERE tenant = 'acme'`), which pins the
-- publication — and therefore the bridge — to **one** tenant. This one does not touch the
-- publication at all, so a single bridge can carry many tenants and route them by subject
-- (the catalogue's `tenant_col` → `cdc.<tenant>.<table>.<op>`).
--
-- ⚠️ **RLS does not bound reads.** It bounds what the writer role may write and what a
-- snapshot may select. What stops one tenant *reading* another's rows is the CDC subject
-- plus the NATS permission on it. Enabling this and granting clients `cdc.>` gives you
-- airtight writes and no read isolation whatsoever.
CREATE OR REPLACE FUNCTION public.zebridge_scope_writes_by_tenant(tbl regclass, tenant_col name)
RETURNS void AS $$
BEGIN
    -- NOT NULL because a replica-identity index requires it, and because a nullable tenant
    -- is a row no policy matches and no subject can carry.
    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;

    -- ⚠️ The tenant column must be inside the REPLICA IDENTITY, or the table is refused
    -- with `tenant_not_in_replica_identity` and nothing replicates (init.core).
    PERFORM public.zebridge_tenant_replica_identity(tbl, tenant_col);

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

    -- The writer may touch only rows whose tenant is mapped to the principal it declared.
    -- `USING` bounds what it can see and change; `WITH CHECK` bounds what it can leave
    -- behind — both are needed, or a writer can move a row into another tenant.
    --
    -- ⚠️ **Open bar on the OPEN tenant (`${OPEN_TENANT}`): every principal may write it**, in
    -- both clauses, even one mapped to no tenant but its own. This is the write half of the
    -- open-data model — the open tenant is a shared pool anyone may post to and anyone may
    -- edit (it is also read by everyone via CDC_PUBLIC). Integrity on *private* tenants is
    -- untouched: alice still cannot write or delete a `globex` row. The carve-out only widens
    -- reach to the tenant that is public by definition, so it grants nothing that reading the
    -- open feed did not already expose.
    --
    -- ⚠️ It is in USING as well as WITH CHECK on purpose. Edge writes are upserts
    -- (INSERT … ON CONFLICT DO UPDATE); updating an existing open row needs USING to see it,
    -- so a WITH-CHECK-only carve-out would let a restricted user CREATE an open row but fail
    -- to UPDATE one — a half-open state worse than either choice.
    EXECUTE format('DROP POLICY IF EXISTS zb_tenant_write ON %s', tbl);
    EXECUTE format(
        'CREATE POLICY zb_tenant_write ON %s FOR ALL TO %I'
        ' USING (%I IN (SELECT tenant_id FROM public.zebridge_user_tenants'
        '   WHERE principal = (SELECT current_setting(''zb.principal'', true))) OR %I = ''${OPEN_TENANT}'')'
        ' WITH CHECK (%I IN (SELECT tenant_id FROM public.zebridge_user_tenants'
        '   WHERE principal = (SELECT current_setting(''zb.principal'', true))) OR %I = ''${OPEN_TENANT}'')',
        tbl, '${POSTGRES_WRITER_USER}', tenant_col, tenant_col, tenant_col, tenant_col);

    -- Reads: CDC stays subject-scoped regardless (RLS cannot see it — see
    -- zebridge_scope_reads_by_tenant's own comment, init.core.template.sql). Snapshots
    -- now ARE RLS-scoped, via the same function a read-only deployment would call
    -- directly — defined once in core so it needs nothing from this file.
    PERFORM public.zebridge_scope_reads_by_tenant(tbl, tenant_col);

    RAISE NOTICE 'writes and scoped reads on % are now scoped by %; CDC stays scoped by the subject, not RLS — zebridge_enable records %.% in zebridge_catalogue, and a running bridge reloads the catalogue row live',
                 tbl, tenant_col, tbl, tenant_col;
END;
$$ LANGUAGE plpgsql;

-- Answers "which tables are guarded, and on which columns?" — compare against the catalogue
CREATE OR REPLACE FUNCTION public.zebridge_audit_write_guards()
RETURNS TABLE (tbl text, version_guard boolean, delete_guard boolean, detail text) AS $$
    SELECT c.relname::text,
           EXISTS (SELECT 1 FROM pg_trigger g WHERE g.tgrelid = c.oid
                   AND g.tgname = 'zebridge_bump_version_t'),
           EXISTS (SELECT 1 FROM pg_trigger g WHERE g.tgrelid = c.oid
                   AND g.tgname = 'zebridge_soft_delete_t'),
           coalesce((SELECT string_agg(pg_get_triggerdef(g.oid), ' | ')
                     FROM pg_trigger g WHERE g.tgrelid = c.oid
                       AND g.tgname LIKE 'zebridge_%'), '')
    FROM pg_class c JOIN pg_namespace n ON n.oid = c.relnamespace
    WHERE n.nspname = 'public' AND c.relkind = 'r'
      -- Prefix test rather than `zebridge_is_internal_table()`: a LANGUAGE sql body is
      -- validated when it is CREATEd, so calling a function declared later in this file
      -- fails the whole script. Caught by scripts/scenarios/render.py, which applies the
      -- rendered SQL to a scratch database rather than trusting that it parses.
      AND c.relname NOT LIKE 'zebridge%'
    ORDER BY c.relname;
$$ LANGUAGE sql STABLE;


-- ---------------------------------------------------------
-- Per-tenant scoping — ONE entry point, on purpose
-- ---------------------------------------------------------
--
-- ⚠️ **Use `zebridge_scope_publication_to_one_tenant()` and nothing else** to put a tenant-scoped
-- table into a publication. A bare `ALTER PUBLICATION ... ADD TABLE` is the failure this
-- exists to prevent: it publishes the table with no row filter and no policy, and every
-- subscriber of that publication then receives every tenant's rows. Nothing errors, and
-- nothing in the bridge can detect it — the bridge is a pass-through and is meant to stay
-- one.
--
-- The function does the four things that must not drift apart, in the order they must
-- happen:
--
--   1. refuses a nullable or absent tenant column (a NULL tenant belongs to nobody, and a
--      nullable column cannot carry a replica identity)
--   2. moves the REPLICA IDENTITY to (tenant_col, <pk>) if the tenant is not already in it
--      — without this PostgreSQL refuses every UPDATE and DELETE on the table once the
--      publication carries a filter, including the application's own
--   3. enables RLS and creates two policies: writes bounded by the principal→tenant
--      mapping, reads left wide for `bridge_reader` (one session serves all tenants, so it
--      has no principal to resolve — read scoping is the publication's job)
--   4. only then adds the table to the publication, with the filter
--
-- Called once per (table, tenant): the per-table work is idempotent, the publication entry
-- is per-tenant. Each publication needs its own replication slot and its own bridge
-- instance.
--
-- ⚠️ `bridge_reader` must NOT be given BYPASSRLS to make step 3 work. That is a *role*
-- attribute covering every table in the database, and this file already grants the reader
-- SELECT on all of them — including tables added later by someone who never heard of the
-- bridge. The per-table `zb_reader_all` policy is the same capability, scoped, and it
-- fails closed for anything nobody considered.

-- Which principal may act for which tenant. Referenced by the write policy above; for
-- *writes* the bridge never reads it and never learns what a tenant is — it sets
-- `zb.principal` from the subject token it already trusts, and PostgreSQL resolves the
-- rest.
--
-- For *reads* it is a second source: this table is replicated (below) so the bridge can
-- mirror it live into `$KV.tenants.<principal>`, letting a fresh consumer holding only a
-- principal ID ask NATS for its own tenant (PROTOCOL.md "The Connection Flow", Step 0;
-- NOTES.md §1.12 part 3) instead of guessing or hardcoding it. The bridge still never
-- resolves a tenant *for routing purposes* from this table — that stays the subject
-- token's job — this is purely a client-facing lookup, mirrored from the same source of
-- truth as the write policy above so the two cannot drift apart.
-- principal → tenant, the mapping RLS resolves against — see PROTOCOL.md §7.4
CREATE TABLE IF NOT EXISTS public.zebridge_user_tenants (
    principal text NOT NULL,
    -- A tenant is a NATS subject token (§10fo): no `.`, `*`, `>` or space — the
    -- application-table guard checks a row's value, this checks the roster's, and
    -- the invite below the DBA's, so a wildcard can never become a stream name, a
    -- KV entry or a JWT tag.
    tenant_id text NOT NULL CHECK (tenant_id <> '' AND tenant_id !~ '[\x01-\x20\x7f.*>]' AND octet_length(tenant_id) <= ${MAX_TENANT_BYTES}),
    PRIMARY KEY (principal, tenant_id)
);
ALTER TABLE public.zebridge_user_tenants DROP CONSTRAINT IF EXISTS zebridge_user_tenants_tenant_id_check;
ALTER TABLE public.zebridge_user_tenants ADD CONSTRAINT zebridge_user_tenants_tenant_id_check CHECK (tenant_id <> '' AND tenant_id !~ '[\x01-\x20\x7f.*>]' AND octet_length(tenant_id) <= ${MAX_TENANT_BYTES});
GRANT SELECT ON public.zebridge_user_tenants TO ${POSTGRES_READER_USER}, ${POSTGRES_WRITER_USER};

-- A restricted leaf owns a principal namespace <leaf_id>.<device>. Register
-- its explicit tenant list before issuing leaf-scoped invites or link credentials.
-- This table is admin-owned; the bridge can only read it. A leaf id starts with
-- ${LEAF_PREFIX} (grammar.json leaf_prefix) and no plain principal may: a principal
-- named like a leaf would sit above <leaf_id>.<device> in every wildcard grant.
-- A leaf id must also never have been used as a hub principal.
CREATE TABLE IF NOT EXISTS public.zebridge_leaf_scopes (
    leaf_id text PRIMARY KEY,
    tenant_ids text[] NOT NULL CHECK (cardinality(tenant_ids) > 0
        AND array_position(tenant_ids, NULL) IS NULL
        AND array_position(tenant_ids, '') IS NULL
        AND array_to_string(tenant_ids, '') ~ '^[A-Za-z0-9_-]+$'),
    nats_url text,
    nats_ws_url text
);
ALTER TABLE public.zebridge_leaf_scopes DROP CONSTRAINT IF EXISTS zebridge_leaf_scopes_leaf_id_check;
ALTER TABLE public.zebridge_leaf_scopes ADD CONSTRAINT zebridge_leaf_scopes_leaf_id_check
    CHECK (leaf_id ~ '^${LEAF_PREFIX}[A-Za-z0-9_-]+$' AND length(leaf_id) <= 64);
GRANT SELECT ON public.zebridge_leaf_scopes TO ${POSTGRES_READER_USER}, ${POSTGRES_WRITER_USER};

CREATE OR REPLACE FUNCTION public.zebridge_check_leaf_membership(p_principal text, p_tenant text)
RETURNS void LANGUAGE plpgsql SECURITY DEFINER SET search_path = pg_catalog, public AS $$
DECLARE scope text;
BEGIN
    PERFORM pg_advisory_xact_lock(hashtextextended(split_part(p_principal, '.', 1), 0));
    IF p_principal ~ '^${LEAF_PREFIX}[A-Za-z0-9_-]+[.][A-Za-z0-9_-]+$' THEN
        scope := split_part(p_principal, '.', 1);
        PERFORM 1 FROM public.zebridge_leaf_scopes
            WHERE leaf_id = scope AND p_tenant = ANY(tenant_ids) FOR SHARE;
        IF NOT FOUND THEN
            RAISE EXCEPTION 'principal % is outside its registered leaf tenant list', p_principal
                USING ERRCODE = '23514';
        END IF;
    ELSIF p_principal !~ '^[A-Za-z0-9_-]+$' OR p_principal ~ '^${LEAF_PREFIX}' THEN
        RAISE EXCEPTION 'invalid or reserved principal namespace %', p_principal USING ERRCODE = '23514';
    END IF;
END;
$$;
REVOKE ALL ON FUNCTION public.zebridge_check_leaf_membership(text, text) FROM PUBLIC;
GRANT EXECUTE ON FUNCTION public.zebridge_check_leaf_membership(text, text) TO ${POSTGRES_WRITER_USER};

CREATE OR REPLACE FUNCTION public.zebridge_guard_leaf_membership()
RETURNS trigger LANGUAGE plpgsql SECURITY DEFINER SET search_path = pg_catalog, public AS $$
BEGIN
    PERFORM public.zebridge_check_leaf_membership(NEW.principal, NEW.tenant_id);
    RETURN NEW;
END;
$$;
DROP TRIGGER IF EXISTS zebridge_guard_leaf_membership_t ON public.zebridge_user_tenants;
CREATE TRIGGER zebridge_guard_leaf_membership_t BEFORE INSERT OR UPDATE ON public.zebridge_user_tenants
    FOR EACH ROW EXECUTE FUNCTION public.zebridge_guard_leaf_membership();

-- The keys a principal enrolled with (§10cl). /enroll used to mint from the pubkey
-- and FORGET it — which made any future hard kill (the NATS account revocation list,
-- keyed by user pubkey) impossible to retrofit: you cannot revoke a key you never
-- recorded. One row per enrollment; a principal may hold several (laptop, phone,
-- re-enrollments after expiry), so this is one-to-many and the PUBKEY is the key.
-- Written by the enroll CTE in the same transaction as the invite redemption; kept
-- after revocation (rows are the audit trail AND the future revocation list's input —
-- `bridge --revoke` does not delete them).
CREATE TABLE IF NOT EXISTS public.zebridge_principal_keys (
    user_pubkey  text PRIMARY KEY,
    principal    text NOT NULL,
    tenant_id    text NOT NULL,
    enrolled_at  timestamptz NOT NULL DEFAULT now(),
    -- Stamped by `bridge --revoke` (§10cm). PG is the source of truth for the FULL
    -- revocation map: the account JWT's `revocations` is re-derived from every row
    -- with a revoked_at, so revoking B can never silently un-revoke A.
    revoked_at   timestamptz
);
GRANT SELECT ON public.zebridge_principal_keys TO ${POSTGRES_READER_USER}, ${POSTGRES_WRITER_USER};
-- INSERT for enrollment, UPDATE for the ON CONFLICT arm (a re-enrolled key refreshes
-- its ownership row). Never DELETE: the rows outlive revocation on purpose.
GRANT INSERT, UPDATE ON public.zebridge_principal_keys TO ${POSTGRES_WRITER_USER};

-- §10kn: `bridge --revoke <p> --purge` asks the principal's devices to delete their local
-- replica and identity. A row here means the latest revocation of <p> asked for it: the
-- bridge adds `"purge": true` to the ban, and `/renew` answers a revoked key with it. A
-- plain `--revoke` deletes the row. Written only by the admin command; read by both roles.
-- By principal, not by key: a principal minted without a recorded key is purged too.
CREATE TABLE IF NOT EXISTS public.zebridge_purges (
    principal    text PRIMARY KEY,
    requested_at timestamptz NOT NULL DEFAULT now()
);
GRANT SELECT ON public.zebridge_purges TO ${POSTGRES_READER_USER}, ${POSTGRES_WRITER_USER};

-- A revoked principal is dead for good (§10dm, §10kb). `bridge --revoke` deletes its
-- mappings and stamps its keys; /enroll then refuses any invite for the name and /renew
-- refuses its keys. A mapping put back by hand would still reopen the door: writes pass
-- the tenant guard again, and once the 2 h ban on VERDICTS ages out, the revoked
-- device's unexpired JWT works until it expires. So the door itself refuses: no mapping
-- for a principal with a revoked key. The fix is a new principal name.
CREATE OR REPLACE FUNCTION public.zebridge_refuse_revoked_mapping()
RETURNS trigger AS $$
BEGIN
    IF EXISTS (SELECT 1 FROM public.zebridge_principal_keys
               WHERE principal = NEW.principal AND revoked_at IS NOT NULL) THEN
        RAISE EXCEPTION 'principal % was revoked and cannot be mapped again', NEW.principal
            USING HINT = 'A revoked name is dead for good: invite the person under a new principal.';
    END IF;
    RETURN NEW;
END;
$$ LANGUAGE plpgsql SECURITY DEFINER SET search_path = public, pg_catalog;

DROP TRIGGER IF EXISTS zebridge_refuse_revoked_mapping_t ON public.zebridge_user_tenants;
CREATE TRIGGER zebridge_refuse_revoked_mapping_t
    BEFORE INSERT OR UPDATE OF principal ON public.zebridge_user_tenants
    FOR EACH ROW EXECUTE FUNCTION public.zebridge_refuse_revoked_mapping();

-- Enrollment invites — the pump-starter (NOTES: the JWT mint flow). One row per
-- invitation: a high-entropy single-use code (122 random bits, generated here when the
-- INSERT omits it) the operator hands out out-of-band;
-- presenting it to the bridge's /enroll endpoint IS the authentication (a one-time
-- password), and the row carries everything the mint needs: principal and tenant.
-- Every invite makes a CLIENT: the bridge holds the client signing key only, and
-- services get their creds offline (--mint-responder). The principal CHECK is the naming police — the narrow waist between the
-- identity world and the routing world ([A-Za-z0-9_-] is the intersection of NATS
-- subject-token, KV-key and template-expansion alphabets, measured).
CREATE TABLE IF NOT EXISTS public.zebridge_invites (
    code       text PRIMARY KEY DEFAULT replace(gen_random_uuid()::text, '-', ''),
    principal  text NOT NULL CHECK (principal ~ '^[A-Za-z0-9_-]+$'),
    tenant_id  text NOT NULL CHECK (tenant_id <> '' AND tenant_id !~ '[\x01-\x20\x7f.*>]' AND octet_length(tenant_id) <= ${MAX_TENANT_BYTES}),
    created_at timestamptz NOT NULL DEFAULT now(),
    expires_at timestamptz NOT NULL DEFAULT now() + interval '1 days',
    used_at    timestamptz
);
-- The role column is gone: nothing read it, and a value other than 'client' was accepted
-- and ignored. An existing table loses it here.
ALTER TABLE public.zebridge_invites DROP CONSTRAINT IF EXISTS zebridge_invites_principal_check;
ALTER TABLE public.zebridge_invites ADD CONSTRAINT zebridge_invites_principal_check
    CHECK ((principal ~ '^[A-Za-z0-9_-]+$' AND principal !~ '^${LEAF_PREFIX}')
        OR principal ~ '^${LEAF_PREFIX}[A-Za-z0-9_-]+[.][A-Za-z0-9_-]+$');
DROP TRIGGER IF EXISTS zebridge_guard_leaf_membership_t ON public.zebridge_invites;
CREATE TRIGGER zebridge_guard_leaf_membership_t BEFORE INSERT OR UPDATE OF principal, tenant_id ON public.zebridge_invites
    FOR EACH ROW EXECUTE FUNCTION public.zebridge_guard_leaf_membership();
ALTER TABLE public.zebridge_invites DROP COLUMN IF EXISTS role;
ALTER TABLE public.zebridge_invites DROP CONSTRAINT IF EXISTS zebridge_invites_tenant_id_check;
ALTER TABLE public.zebridge_invites ADD CONSTRAINT zebridge_invites_tenant_id_check CHECK (tenant_id <> '' AND tenant_id !~ '[\x01-\x20\x7f.*>]' AND octet_length(tenant_id) <= ${MAX_TENANT_BYTES});
-- The defaults too: CREATE TABLE IF NOT EXISTS leaves an older table as it was, and an
-- invite inserted without a code then failed ("null value in column code").
ALTER TABLE public.zebridge_invites ALTER COLUMN code SET DEFAULT replace(gen_random_uuid()::text, '-', '');
ALTER TABLE public.zebridge_invites ALTER COLUMN expires_at SET DEFAULT now() + interval '1 days';
-- The bridge redeems invites over its WRITER connection: read the row, stamp
-- used_at, and register the principal→tenant mapping in the same transaction.
GRANT SELECT, UPDATE ON public.zebridge_invites TO ${POSTGRES_WRITER_USER};
GRANT INSERT ON public.zebridge_user_tenants TO ${POSTGRES_WRITER_USER};

-- Creating an invite without the admin — see SECURITY.md §1.6. An app's backend, after its
-- own sign-in, asks for an invite for the principal and tenant it chose; `bridge --invite`
-- calls the same function. Its login is the DBA's to create, then
-- Changing a scope cannot strand existing memberships or reuse a principal
-- namespace whose old credentials may still be valid. Retire/revoke devices first.
CREATE OR REPLACE FUNCTION public.zebridge_guard_leaf_scope()
RETURNS trigger LANGUAGE plpgsql SECURITY DEFINER SET search_path = pg_catalog, public AS $$
DECLARE lid text;
BEGIN
    lid := CASE WHEN TG_OP = 'DELETE' THEN OLD.leaf_id ELSE NEW.leaf_id END;
    PERFORM pg_advisory_xact_lock(hashtextextended(lid, 0));
    IF TG_OP <> 'DELETE' AND (
        EXISTS (SELECT 1 FROM public.zebridge_principal_keys WHERE principal = NEW.leaf_id) OR
        EXISTS (SELECT 1 FROM public.zebridge_invites WHERE principal = NEW.leaf_id) OR
        EXISTS (SELECT 1 FROM public.zebridge_user_tenants WHERE principal = NEW.leaf_id)) THEN
        RAISE EXCEPTION 'leaf id % has a hub-principal history; choose a fresh id', NEW.leaf_id USING ERRCODE = '23514';
    END IF;
    IF EXISTS (SELECT 1 FROM public.zebridge_user_tenants
        WHERE split_part(principal, '.', 1) = lid AND strpos(principal, '.') > 0
          AND (TG_OP = 'DELETE' OR NOT tenant_id = ANY(NEW.tenant_ids))) OR
        (TG_OP = 'UPDATE' AND NEW.leaf_id <> OLD.leaf_id) THEN
        RAISE EXCEPTION 'leaf scope % still has incompatible memberships; revoke them before changing it', lid USING ERRCODE = '23514';
    END IF;
    IF TG_OP = 'DELETE' THEN RETURN OLD; END IF;
    RETURN NEW;
END;
$$;
DROP TRIGGER IF EXISTS zebridge_guard_leaf_scope_t ON public.zebridge_leaf_scopes;
CREATE TRIGGER zebridge_guard_leaf_scope_t BEFORE INSERT OR UPDATE OR DELETE ON public.zebridge_leaf_scopes
    FOR EACH ROW EXECUTE FUNCTION public.zebridge_guard_leaf_scope();

-- `GRANT zebridge_inviter TO <login>`: the group role has no login and no password here.
-- It may call this function and nothing else — no table, so it cannot read codes, spend
-- one, or touch an invite it did not make. The function refuses a revoked principal
-- (revocation is for good) and a lifetime over 30 days. Revoking stays an admin act.
CREATE OR REPLACE FUNCTION public.zebridge_invite(p_principal text, p_tenant text, p_valid_for interval DEFAULT interval '1 day')
RETURNS text
LANGUAGE plpgsql
SECURITY DEFINER
SET search_path = pg_catalog, public
AS $$
DECLARE
    v_code text;
BEGIN
    IF p_valid_for IS NULL OR p_valid_for <= interval '0' OR p_valid_for > interval '30 days' THEN
        RAISE EXCEPTION 'an invite lasts more than 0 and at most 30 days, not %', p_valid_for
            USING ERRCODE = '22023';
    END IF;
    IF EXISTS (SELECT 1 FROM public.zebridge_principal_keys k
               WHERE k.principal = p_principal AND k.revoked_at IS NOT NULL) THEN
        RAISE EXCEPTION 'principal % was revoked, which is for good: invite a new principal', p_principal
            USING ERRCODE = '42501';
    END IF;
    INSERT INTO public.zebridge_invites (principal, tenant_id, expires_at)
    VALUES (p_principal, p_tenant, now() + p_valid_for)
    RETURNING code INTO v_code;
    RETURN v_code;
END;
$$;
REVOKE ALL ON FUNCTION public.zebridge_invite(text, text, interval) FROM PUBLIC;
DO $$
BEGIN
    IF NOT EXISTS (SELECT 1 FROM pg_roles WHERE rolname = 'zebridge_inviter') THEN
        CREATE ROLE zebridge_inviter NOLOGIN;
    END IF;
END;
$$;
GRANT USAGE ON SCHEMA public TO zebridge_inviter;
GRANT EXECUTE ON FUNCTION public.zebridge_invite(text, text, interval) TO zebridge_inviter;


-- ⚠️ Never published as ordinary CDC — see the matching exemption in
-- `zebridge_publication_guard()` (init.core.template.sql). The bridge special-cases this
-- relation's name in its WAL loop and diverts every row into a `$KV.tenants.<principal>`
-- put instead, exactly as it already does for `zebridge_ddl_events` → `$KV.schemas.*`.
-- Publishing it as `cdc.zebridge_user_tenants.*` would hand every client with a broad CDC
-- grant the full principal→tenant roster — the disclosure the KV bucket's principal-first
-- key order exists to prevent.
-- Attached to every ZeBridge publication that ALREADY exists; the ones created
-- later get it from `zebridge_create_publication`, which knows this table by name.
-- Between the two, the install order stops mattering — core then write then
-- create, or core then create then write, and the table lands either way. That is
-- the whole point: this file used to attach it to one name rendered from the
-- environment, so a publication made any other way silently had no tenant roster
-- and PROTOCOL Step 0 failed for every one of its clients.
--
-- A ZeBridge publication is recognised by the one table they all carry and
-- nothing else does: zebridge_ddl_events.
DO $$
DECLARE p record;
BEGIN
    FOR p IN SELECT DISTINCT pubname FROM pg_publication_tables
              WHERE schemaname = 'public' AND tablename = 'zebridge_ddl_events'
    LOOP
        IF NOT EXISTS (
            SELECT 1 FROM pg_publication_tables
            WHERE pubname = p.pubname AND schemaname = 'public'
              AND tablename = 'zebridge_user_tenants'
        ) THEN
            EXECUTE format('ALTER PUBLICATION %I ADD TABLE public.zebridge_user_tenants', p.pubname);
            RAISE NOTICE 'zebridge: attached zebridge_user_tenants to publication %', p.pubname;
        END IF;
    END LOOP;
END $$;

-- ⚠️ Dollar-quote with a bare double-dollar ONLY — never a named tag (dollar, letters,
-- dollar). This file is rendered by `envsubst`, which reads such a tag as a variable
-- reference and substitutes it away, breaking the quoting and truncating every definition
-- after it. Measured: the whole file rendered down to one surviving function, and psql
-- reported no error. A bare double-dollar is not a valid identifier, so it survives.
--
-- The same rule forces the nested DDL below to use doubled single quotes rather than a
-- second dollar-quoted string — this comment cannot even show you the syntax to avoid.
-- the only supported way to publish a tenant-scoped table — see PROTOCOL.md §7.4, NOTES.md §1.8
CREATE OR REPLACE FUNCTION public.zebridge_scope_publication_to_one_tenant(
    tbl          regclass,
    tenant_col   name,
    tenant_value text,
    publication  name
) RETURNS void AS $$
BEGIN
    IF tbl::text IN ('zebridge_ddl_events','public.zebridge_ddl_events') THEN
        RAISE EXCEPTION 'refusing to publish %: it is the bridge''s own DDL tracker', tbl;
    END IF;

    -- 1. A NULL tenant belongs to nobody, and a nullable column cannot carry a replica
    --    identity.
    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;

    -- 2. The tenant must be inside the replica identity, or PostgreSQL refuses every
    --    UPDATE and DELETE once the publication carries a filter on it — including the
    --    application's own writes.
    PERFORM public.zebridge_tenant_replica_identity(tbl, tenant_col);

    -- 3. Per-TABLE, idempotent: this runs once per (table, tenant) and the policies do not
    --    vary by tenant — the mapping table does.
    EXECUTE format('ALTER TABLE %s ENABLE ROW LEVEL SECURITY', tbl);

    IF NOT EXISTS (SELECT 1 FROM pg_policy WHERE polrelid = tbl AND polname = 'zb_tenant_write') THEN
        EXECUTE format(
            'CREATE POLICY zb_tenant_write ON %s FOR ALL TO %I'
            ' USING (%I IN (SELECT tenant_id FROM public.zebridge_user_tenants'
            '   WHERE principal = (SELECT current_setting(''zb.principal'', true))))'
            ' WITH CHECK (%I IN (SELECT tenant_id FROM public.zebridge_user_tenants'
            '   WHERE principal = (SELECT current_setting(''zb.principal'', true))))',
            tbl, 'bridge_writer', tenant_col, tenant_col);
    END IF;

    -- The reader sees every row of the tables it replicates: one session serves all
    -- tenants and has no principal to resolve. Read scoping is the publication's job.
    IF NOT EXISTS (SELECT 1 FROM pg_policy WHERE polrelid = tbl AND polname = 'zb_reader_all') THEN
        EXECUTE format('CREATE POLICY zb_reader_all ON %s FOR SELECT TO %I USING (true)',
                       tbl, 'bridge_reader');
    END IF;

    -- 4. Per-TENANT, and only now: the publication entry is what makes it a tenant stream.
    IF EXISTS (SELECT 1 FROM pg_publication_rel pr JOIN pg_publication p ON p.oid = pr.prpubid
               WHERE pr.prrelid = tbl AND p.pubname = publication) THEN
        EXECUTE format('ALTER PUBLICATION %I SET TABLE %s WHERE (%I = %L)',
                       publication, tbl, tenant_col, tenant_value);
    ELSE
        EXECUTE format('ALTER PUBLICATION %I ADD TABLE %s WHERE (%I = %L)',
                       publication, tbl, tenant_col, tenant_value);
    END IF;

    RAISE NOTICE 'published % to % for tenant %', tbl, publication, tenant_value;
END;
$$ LANGUAGE plpgsql;

-- The sweeper's reach, granted automatically — the audit below is the net, this is the fix.
--
-- Every tenant enters the system through one door: a row in `zebridge_user_tenants`
-- (a DBA's INSERT, or `/enroll` redeeming an invite — both land here). The sweeper is
-- bounded by that same table, so a new tenant used to start life UNSWEPT until someone
-- remembered the mapping; the failure is silent accumulation, and `zebridge_audit_sweeper`
-- can only report it after the fact. This trigger closes the gap at the door: mapping any
-- principal to a tenant also maps the sweeper to it.
--
--   • `ON CONFLICT DO NOTHING`, not SELECT-then-INSERT: atomic, so two concurrent
--     enrollments of the same new tenant cannot race.
--   • `WHEN (NEW.principal <> 'zb_sweeper')`: the auto-grant row must not re-fire the
--     trigger. (It would terminate anyway — the second pass conflicts and inserts
--     nothing — but a guarded trigger says the intent.)
--   • ⚠️ Never the OPEN tenant. The sweeper reaches `${OPEN_TENANT}` rows through the
--     policy's own arm (`%I = '${OPEN_TENANT}'` above) — a mapping adds nothing. And it
--     subtracts: boot derives the tenant-stream list from DISTINCT tenant_id here, and a
--     `${OPEN_TENANT}` entry makes it try to create CDC_${OPEN_TENANT}, whose subject
--     overlaps CDC_PUBLIC — the bridge refuses to start (measured 2026-09-03, and true
--     of ANY open-tenant mapping, not just the sweeper's — the reconciler now skips the
--     open tenant too, but the mapping stays pointless).
--   • 'zb_sweeper' is spelled here, matching the tombstone guard's carve-out above. An
--     operator running a renamed SWEEPER_PRINCIPAL must adjust both — the audit takes
--     the name as a parameter and will say so.
--   • Side effect, harmless: `zebridge_user_tenants` rides the publication, so the
--     bridge projects a `$KV.tenants.zb_sweeper` key like any other principal's. The
--     bucket holds ONE bare tenant string per principal, so for a many-tenant principal
--     like the sweeper it just shows whichever mapping was written last — meaningless,
--     and inert: the sweeper is not a NATS client and nothing resolves that key.
--
-- What this does NOT cover: a tenant that exists only as data (rows inserted by an admin
-- with a tenant_id nobody is mapped to). No mapping insert ever happens, so no trigger
-- fires; that remains the audit's job.
CREATE OR REPLACE FUNCTION public.zebridge_sweeper_autogrant()
RETURNS trigger AS $$
BEGIN
    INSERT INTO public.zebridge_user_tenants (principal, tenant_id)
    VALUES ('zb_sweeper', NEW.tenant_id)
    ON CONFLICT DO NOTHING;
    RETURN NULL;
END;
$$ LANGUAGE plpgsql SECURITY DEFINER SET search_path = public, pg_catalog;

DROP TRIGGER IF EXISTS zebridge_sweeper_autogrant_t ON public.zebridge_user_tenants;
CREATE TRIGGER zebridge_sweeper_autogrant_t
    AFTER INSERT ON public.zebridge_user_tenants
    FOR EACH ROW
    WHEN (NEW.principal <> 'zb_sweeper' AND NEW.tenant_id <> '${OPEN_TENANT}')
    EXECUTE FUNCTION public.zebridge_sweeper_autogrant();

-- Backfill: the trigger only guards the door from now on. Tenants already inside —
-- mapped to some principal but never to the sweeper — are exactly the ones the audit
-- would have caught too late.
INSERT INTO public.zebridge_user_tenants (principal, tenant_id)
SELECT DISTINCT 'zb_sweeper', tenant_id FROM public.zebridge_user_tenants
WHERE tenant_id <> '${OPEN_TENANT}'
ON CONFLICT DO NOTHING;

-- Answers "which tenants will never have their tombstones reaped?"
--
-- The sweeper acts as a named principal (`zb_sweeper` by default) and is therefore bounded
-- by `zebridge_user_tenants` like any other writer — which is the point: forgetting to map
-- a tenant fails **closed**, and its reach is auditable in the same table as everyone
-- else's. The earlier design gave it a policy exempting principal-less sessions, which
-- opened the whole table to anything that reached that connection without setting a
-- principal. Failing open on an omission is the wrong default for a process that deletes.
--
-- ⚠️ The sweeper cannot run this check itself. RLS hides the unmapped rows from a warning
-- query exactly as it hides them from the DELETE — measured, the same query returns 0 as
-- the sweeper and 1 as the admin. A blind spot cannot survey itself, so the audit lives
-- here, with the grant.
--
--     INSERT INTO zebridge_user_tenants (principal, tenant_id) VALUES ('zb_sweeper', '<tenant>');
-- tenants whose tombstones will never be reaped — see PROTOCOL.md §7.5
CREATE OR REPLACE FUNCTION public.zebridge_audit_sweeper(sweeper_principal text DEFAULT 'zb_sweeper'::text)
 RETURNS TABLE(tbl text, tenant_id text, verdict text)
 LANGUAGE plpgsql
AS $$
DECLARE
    r record;
BEGIN
    -- Every published table that has BOTH a tenant column and a tombstone column: those
    -- are the ones the sweeper is expected to reach. A table with no tombstone is never
    -- swept (its deletes are physical), so an unmapped tenant there costs nothing.
    FOR r IN
        SELECT c.oid::regclass::text AS relname
        FROM pg_class c
        JOIN pg_namespace n ON n.oid = c.relnamespace AND n.nspname = 'public'
        WHERE c.relkind = 'r'
          AND EXISTS (SELECT 1 FROM pg_attribute a WHERE a.attrelid = c.oid
                        AND a.attname = 'tenant_id' AND a.attnum > 0 AND NOT a.attisdropped)
          AND EXISTS (SELECT 1 FROM pg_attribute a WHERE a.attrelid = c.oid
                        AND a.attname = 'deleted_at' AND a.attnum > 0 AND NOT a.attisdropped)
    LOOP
        -- WARNING: doubled quotes, not a nested dollar-quote. A `$q` + `$` tag here is
        -- eaten by envsubst (it reads it as a variable) and truncates every definition
        -- after this point — caught by scripts/scenarios/render.py, which reported only
        -- 3 of 9 functions surviving the render.
        -- The open tenant is never "unmapped": zb_tenant_write admits it to every principal,
        -- the sweeper included, with no mapping row (and the auto-grant skips it on purpose).
        -- Counting it reported a reachable tenant as never reaped.
        RETURN QUERY EXECUTE format(
            'SELECT %L::text, t.tenant_id::text, ''UNREAPED - tombstones for this tenant' || ' accumulate; %I is not mapped to it'' FROM (SELECT DISTINCT tenant_id FROM %s'
            || ' WHERE deleted_at IS NOT NULL) t WHERE NOT EXISTS (SELECT 1 FROM'
            || ' public.zebridge_user_tenants m WHERE m.principal = %L AND m.tenant_id = t.tenant_id)'
            || ' AND t.tenant_id <> %L',
            r.relname, sweeper_principal, r.relname, sweeper_principal, '${OPEN_TENANT}');
    END LOOP;
END $$;

GRANT SELECT, INSERT, UPDATE ON public.zebridge_gc_watermark TO ${POSTGRES_WRITER_USER};

-- ─── The floor protocol: one sweeper, one floor, and writes that cannot outrun a reap ─────────────
--
-- GC is one promise about the DATABASE's history ("a write older than the floor can no longer be
-- replayed safely"), so it has one policy, one owner and one number, whatever bridges read it.
-- Advisory-lock keys, class 23106 ('ZB'): objid 1 = the sweeper's ownership (session level), objid 2 =
-- the floor (transaction level: shared for a mutation, exclusive for the sweeper's advance).

-- The sweeper's ownership: ONE active sweeper per database, a session-level lock held on the
-- sweeper's working connection. It is lost with that connection (a dead sweeper frees it, a
-- pooler that swaps the server session loses it), and `zebridge_gc_sweeper_holds` says so.
CREATE OR REPLACE FUNCTION public.zebridge_gc_sweeper_lock() RETURNS boolean
LANGUAGE sql AS $$ SELECT pg_try_advisory_lock(23106, 1) $$;

CREATE OR REPLACE FUNCTION public.zebridge_gc_sweeper_holds() RETURNS boolean
LANGUAGE sql AS $$
    SELECT EXISTS (SELECT 1 FROM pg_locks
                    WHERE locktype = 'advisory' AND database = (SELECT oid FROM pg_database WHERE datname = current_database())
                      AND classid = 23106 AND objid = 1 AND objsubid = 2 AND pid = pg_backend_pid() AND granted)
$$;

-- Step 1 of a sweep, BEFORE any row is reaped: advance the floor to the retention boundary (never
-- backwards: a longer retention leaves it where it is and it stays until time catches up) and say
-- where to reap. The exclusive floor lock waits for every in-flight mutation (each holds the shared
-- lock until it commits) and holds new ones back until this commits, so no write that passed the old
-- floor can commit after a tombstone behind the new one is gone. `reap_before` <= `floor_at` always:
-- at every committed state, what was reaped is older than the floor that was advertised.
-- The wait is bounded (`p_lock_ms`): past it the pass is skipped, not forced.
CREATE OR REPLACE FUNCTION public.zebridge_gc_advance(p_retention_secs double precision, p_threshold_ms bigint, p_lock_ms integer DEFAULT 5000)
RETURNS TABLE (floor_at timestamptz, reap_before timestamptz)
LANGUAGE plpgsql AS $$
DECLARE
    v_cut   timestamptz;
    v_floor timestamptz;
BEGIN
    -- A zero or sub-second retention reaps every tombstone, and `threshold_ms = 0` is how the producer
    -- reads "the sweeper has never run" (its checkpoint guard is then off): refused here, whoever calls,
    -- the override of the sweeper's own minute-floor included.
    IF p_threshold_ms IS NULL OR p_threshold_ms < 1000 OR p_retention_secs IS NULL OR p_retention_secs < 1 THEN
        RAISE EXCEPTION 'zebridge_gc_advance: a retention below one second is refused (threshold_ms %, % s)', p_threshold_ms, p_retention_secs
            USING ERRCODE = 'invalid_parameter_value',
                  HINT = 'Zero would reap every tombstone and read as "never swept"; GC_THRESHOLD_MS is the offline window clients are promised.';
    END IF;
    PERFORM set_config('lock_timeout', p_lock_ms::text || 'ms', true);
    PERFORM pg_advisory_xact_lock(23106, 2);
    v_cut := now() - make_interval(secs => p_retention_secs);
    -- reaped = 0: this row change is "a pass began", not a report; the pass reports with the count after.
    UPDATE public.zebridge_gc_watermark w
       SET watermark = GREATEST(w.watermark, v_cut), threshold_ms = p_threshold_ms,
           reaped = 0, swept_at = now(), updated_at = now()
     WHERE w.id = 1
    RETURNING w.watermark INTO v_floor;
    IF NOT FOUND THEN
        RAISE EXCEPTION 'the watermark row is missing (zebridge_gc_watermark id=1): re-apply init.core.template.sql';
    END IF;
    RETURN QUERY SELECT v_floor, v_cut;
END $$;

-- A mutation's side of the protocol: the shared floor lock, then the floor, IN THE SAME TRANSACTION as
-- the write it guards. The lock is taken in this function and the floor read after it, as a statement
-- of its own: a snapshot taken before the wait could miss the advance that just committed.
-- NULL = no row: nothing was ever reaped, no floor to enforce.
CREATE OR REPLACE FUNCTION public.zebridge_gc_floor_acquire() RETURNS bigint
LANGUAGE plpgsql AS $$
DECLARE v bigint;
BEGIN
    PERFORM pg_advisory_xact_lock_shared(23106, 2);
    SELECT (extract(epoch FROM watermark) * 1000000)::bigint INTO v FROM public.zebridge_gc_watermark WHERE id = 1;
    RETURN v;
END $$;

GRANT EXECUTE ON FUNCTION public.zebridge_gc_sweeper_lock() TO ${POSTGRES_WRITER_USER};
GRANT EXECUTE ON FUNCTION public.zebridge_gc_sweeper_holds() TO ${POSTGRES_WRITER_USER};
GRANT EXECUTE ON FUNCTION public.zebridge_gc_advance(double precision, bigint, integer) TO ${POSTGRES_WRITER_USER};
GRANT EXECUTE ON FUNCTION public.zebridge_gc_floor_acquire() TO ${POSTGRES_WRITER_USER};


-- ─── Mutation receipts: one durable outcome per (principal, original message ID) ─────
-- A retried write must never execute twice, however long its device stayed offline.
-- JetStream deduplicates for two minutes; a receipt is the database's own record,
-- committed in the same transaction as the write it describes.
--
-- Everything lives in zebridge_private, not public: the public DDL triggers never
-- announce its monthly partitions, the reader's public grants never reach it, no edge
-- mutation can name it, and the writer only EXECUTEs the three functions granted below.
-- The functions are SECURITY DEFINER with search_path = pg_catalog: every name in
-- them is schema-qualified.
CREATE SCHEMA IF NOT EXISTS zebridge_private;
REVOKE ALL ON SCHEMA zebridge_private FROM PUBLIC;
GRANT USAGE ON SCHEMA zebridge_private TO ${POSTGRES_WRITER_USER};

-- The first day of the UTC month holding epoch milliseconds `ms`. Date arithmetic
-- only, so it is IMMUTABLE (a timestamptz detour would depend on TimeZone).
-- One expression, no FROM: PostgreSQL inlines it (measured: per-row receipt work fell
-- from about 150 to 60 µs once these helpers lost their regex and subqueries).
CREATE OR REPLACE FUNCTION zebridge_private.month_of_ms(ms bigint)
RETURNS date LANGUAGE sql IMMUTABLE PARALLEL SAFE AS $$
    SELECT (DATE '1970-01-01' + (ms / 86400000)::int) - (extract(day FROM DATE '1970-01-01' + (ms / 86400000)::int)::int - 1)
$$;

-- A new-format ID's creation time: zb1-t<epoch-ms>-<rest>. NULL for anything else.
-- String functions, not a regex: it runs in the CHECK on every insert and update.
CREATE OR REPLACE FUNCTION zebridge_private.receipt_created_ms(message_id text)
RETURNS bigint LANGUAGE sql IMMUTABLE PARALLEL SAFE AS $$
    SELECT CASE WHEN left(message_id, 5) = 'zb1-t'
                 AND split_part(substr(message_id, 6), '-', 1) <> ''
                 AND length(split_part(substr(message_id, 6), '-', 1)) <= 15
                 AND translate(split_part(substr(message_id, 6), '-', 1), '0123456789', '') = ''
                 AND length(message_id) > 6 + length(split_part(substr(message_id, 6), '-', 1))
            THEN split_part(substr(message_id, 6), '-', 1)::bigint END
$$;

-- The partition month of a message ID, from the ID alone: a retry of the same ID always
-- addresses the same partition, so the primary key below is unique across months.
-- No zb1- prefix: a legacy ID, all in the 1900-01 sentinel month (no new-format ID can
-- fall there). zb1- but malformed: NULL, which the CHECK below refuses.
-- Not STRICT, like the two helpers it calls: PostgreSQL inlines a SQL function only when
-- its body is as strict as its declaration, and these are CASE expressions. A NULL ID
-- still yields NULL.
CREATE OR REPLACE FUNCTION zebridge_private.receipt_month(message_id text)
RETURNS date LANGUAGE sql IMMUTABLE PARALLEL SAFE AS $$
    SELECT CASE WHEN left(message_id, 4) <> 'zb1-' THEN DATE '1900-01-01'
                ELSE zebridge_private.month_of_ms(zebridge_private.receipt_created_ms(message_id)) END
$$;

-- A receipt is inserted once, complete, after its write ran:
-- outcome is NOT NULL. Concurrent duplicates are kept apart by transaction-scoped advisory
-- locks taken before the receipt is read (lock_receipts), not by a placeholder row.
-- There is no default partition: a month without one is an operational error.
CREATE TABLE IF NOT EXISTS zebridge_private.receipts (
    principal   text        NOT NULL,
    message_id  text        NOT NULL,
    id_month    date        NOT NULL,
    fingerprint bytea       NOT NULL CHECK (length(fingerprint) = 32),
    outcome     jsonb       NOT NULL,
    claimed_at  timestamptz NOT NULL DEFAULT now(),
    PRIMARY KEY (principal, message_id, id_month),
    CONSTRAINT receipts_month_from_id CHECK (id_month IS NOT DISTINCT FROM zebridge_private.receipt_month(message_id))
) PARTITION BY RANGE (id_month);
ALTER TABLE zebridge_private.receipts ALTER COLUMN outcome SET NOT NULL;
DROP TRIGGER IF EXISTS receipts_finished ON zebridge_private.receipts;
DROP FUNCTION IF EXISTS zebridge_private.receipt_finished();

-- The completion guard (receipt_policy.completion_guard): one row per transaction that has undecided writes, holding
-- the exact (principal, ID, fingerprint) set lookup_receipts left undecided (expected) and
-- the set record_receipts inserted (actual). close_guard deletes it only when the two sets
-- are identical; a deferred check refuses any COMMIT that still has its guard. Unused when
-- completion_guard is false; the caller cannot choose. Every guard error is SQLSTATE ZBG01: the
-- bridge's, never read as a refusal of the client's write.
CREATE TABLE IF NOT EXISTS zebridge_private.receipt_guards (
    txid          xid8    PRIMARY KEY,
    expected_keys text[]  NOT NULL DEFAULT '{}',
    expected_fps  bytea[] NOT NULL DEFAULT '{}',
    actual_keys   text[]  NOT NULL DEFAULT '{}',
    actual_fps    bytea[] NOT NULL DEFAULT '{}'
);
CREATE OR REPLACE FUNCTION zebridge_private.receipt_guard_closed()
RETURNS trigger LANGUAGE plpgsql SECURITY DEFINER SET search_path = pg_catalog AS $$
BEGIN
    IF EXISTS (SELECT 1 FROM zebridge_private.receipt_guards g WHERE g.txid = NEW.txid) THEN
        RAISE EXCEPTION 'a transaction with undecided writes would commit without their receipts'
            USING ERRCODE = 'ZBG01';
    END IF;
    RETURN NULL;
END;
$$;
DROP TRIGGER IF EXISTS receipt_guard_closed ON zebridge_private.receipt_guards;
CREATE CONSTRAINT TRIGGER receipt_guard_closed AFTER INSERT ON zebridge_private.receipt_guards
    DEFERRABLE INITIALLY DEFERRED FOR EACH ROW EXECUTE FUNCTION zebridge_private.receipt_guard_closed();

-- One row: the retention and the minimum accepted creation time. The minimum only ever
-- rises, so a dropped month can never become admissible again through a config change.
-- An operator changes retention_days directly; the next maintenance applies it.
CREATE TABLE IF NOT EXISTS zebridge_private.receipt_policy (
    singleton       boolean PRIMARY KEY DEFAULT true CHECK (singleton),
    retention_days  integer NOT NULL DEFAULT 180 CHECK (retention_days BETWEEN 1 AND 3650),
    min_accepted_ms bigint  NOT NULL DEFAULT 0 CHECK (min_accepted_ms >= 0),
    completion_guard boolean NOT NULL DEFAULT ${RECEIPT_COMPLETION_GUARD}
);
-- Upgrade a pre-guard database without silently changing its policy. Existing rows
-- acquire false; the administrator explicitly enables the guard after stopping the old workers.
-- Fresh installs use this binary's completion-guard default above (enabled by default).
ALTER TABLE zebridge_private.receipt_policy ADD COLUMN IF NOT EXISTS completion_guard boolean NOT NULL DEFAULT false;
ALTER TABLE zebridge_private.receipt_policy ALTER COLUMN completion_guard SET DEFAULT ${RECEIPT_COMPLETION_GUARD};
INSERT INTO zebridge_private.receipt_policy DEFAULT VALUES ON CONFLICT DO NOTHING;

CREATE OR REPLACE FUNCTION zebridge_private.receipt_policy_guard()
RETURNS trigger LANGUAGE plpgsql SET search_path = pg_catalog AS $$
BEGIN
    IF TG_OP = 'DELETE' THEN
        RAISE EXCEPTION 'the receipt policy row cannot be deleted' USING ERRCODE = '23514';
    END IF;
    IF NEW.min_accepted_ms < OLD.min_accepted_ms THEN
        RAISE EXCEPTION 'min_accepted_ms only rises (% -> %)', OLD.min_accepted_ms, NEW.min_accepted_ms
            USING ERRCODE = '23514';
    END IF;
    IF OLD.completion_guard AND NOT NEW.completion_guard THEN
        RAISE EXCEPTION 'the completion guard is never switched off' USING ERRCODE = '23514';
    END IF;
    RETURN NEW;
END;
$$;
DROP TRIGGER IF EXISTS receipt_policy_guard ON zebridge_private.receipt_policy;
CREATE TRIGGER receipt_policy_guard BEFORE UPDATE OR DELETE ON zebridge_private.receipt_policy
    FOR EACH ROW EXECUTE FUNCTION zebridge_private.receipt_policy_guard();

-- Raise the minimum to now - retention, then make sure a partition exists for every month
-- from the minimum through next month, plus the legacy month. Returns what it created.
-- The bridge runs it at startup and daily; the policy row lock serializes two bridges.
-- It never drops anything: see prune_receipts.
CREATE OR REPLACE FUNCTION zebridge_private.maintain_receipts()
RETURNS SETOF text LANGUAGE plpgsql SECURITY DEFINER SET search_path = pg_catalog AS $$
DECLARE
    now_ms   bigint := (extract(epoch FROM clock_timestamp()) * 1000)::bigint;
    pol      record;
    floor_ms bigint;
    m        date;
    last_m   date;
    part     text;
BEGIN
    SELECT * INTO pol FROM zebridge_private.receipt_policy FOR UPDATE;
    floor_ms := greatest(pol.min_accepted_ms, now_ms - pol.retention_days::bigint * 86400000);
    IF floor_ms > pol.min_accepted_ms THEN
        UPDATE zebridge_private.receipt_policy SET min_accepted_ms = floor_ms;
    END IF;
    IF to_regclass('zebridge_private.receipts_legacy') IS NULL THEN
        CREATE TABLE zebridge_private.receipts_legacy PARTITION OF zebridge_private.receipts
            FOR VALUES FROM ('1900-01-01') TO ('1900-02-01');
        RETURN NEXT 'receipts_legacy';
    END IF;
    m := zebridge_private.month_of_ms(floor_ms);
    last_m := (zebridge_private.month_of_ms(now_ms) + interval '1 month')::date;
    WHILE m <= last_m LOOP
        part := 'receipts_' || to_char(m, 'YYYY_MM');
        IF to_regclass('zebridge_private.' || part) IS NULL THEN
            -- Insert-only receipts: full pages, no room kept for updates.
            EXECUTE format('CREATE TABLE zebridge_private.%I PARTITION OF zebridge_private.receipts FOR VALUES FROM (%L) TO (%L)',
                           part, m, (m + interval '1 month')::date);
            RETURN NEXT part;
        END IF;
        m := (m + interval '1 month')::date;
    END LOOP;
END;
$$;

-- Drop whole months that ended before the minimum accepted month. Explicit, operator-run:
-- nothing calls it automatically. The legacy partition is never dropped here.
CREATE OR REPLACE FUNCTION zebridge_private.prune_receipts()
RETURNS SETOF text LANGUAGE plpgsql SECURITY DEFINER SET search_path = pg_catalog AS $$
DECLARE
    floor_m date;
    p       record;
BEGIN
    SELECT zebridge_private.month_of_ms(min_accepted_ms) INTO floor_m FROM zebridge_private.receipt_policy;
    FOR p IN SELECT c.relname FROM pg_inherits i JOIN pg_class c ON c.oid = i.inhrelid
              WHERE i.inhparent = 'zebridge_private.receipts'::regclass
                AND c.relname ~ '^receipts_[0-9]{4}_[0-9]{2}$'
                AND to_date(substr(c.relname, 10), 'YYYY_MM') < floor_m
              ORDER BY c.relname
    LOOP
        EXECUTE format('DROP TABLE zebridge_private.%I', p.relname);
        RETURN NEXT p.relname;
    END LOOP;
END;
$$;

-- Lock every (principal, ID) of a batch, in one global order (sorted lock keys across
-- principals), before any receipt is read: a concurrent duplicate waits here, then finds
-- the winner's receipt. Transaction-scoped: released at COMMIT or ROLLBACK. The key is a
-- 64-bit hash (hashtextextended, seed 0x5A42 "ZB") in the single-bigint advisory lock
-- space: a collision, with these keys or another application's bigint locks, only
-- serializes two transactions; the receipt itself is still the exact text key. The one
-- order rules out cycles among these locks only: application row locks and triggers can
-- still deadlock, and the listener's deadlock recovery still applies.
CREATE OR REPLACE FUNCTION zebridge_private.lock_receipts(p_principals text[], p_ids text[])
RETURNS void LANGUAGE plpgsql SECURITY DEFINER SET search_path = pg_catalog AS $$
DECLARE
    k bigint;
BEGIN
    FOR k IN SELECT DISTINCT hashtextextended(x.pr || chr(31) || x.id, 23106) FROM unnest(p_principals, p_ids) AS x(pr, id) ORDER BY 1 LOOP
        PERFORM pg_advisory_xact_lock(k);
    END LOOP;
END;
$$;

-- After lock_receipts: one row per position, without inserting anything. A whole batch in
-- one call, a principal per row: a function call has a fixed cost, so one call per
-- principal made a batch from many devices pay that cost once per device.
--   claimed   undecided: execute it, then record_receipts with its outcome
--   duplicate / repeat / conflict / expired / future / invalid
-- A repeat is a later position with the same (principal, ID) in the same call. With the
-- completion guard on, the undecided entries become this transaction's guard.
DROP FUNCTION IF EXISTS zebridge_private.lookup_receipts(text, text[], bytea[], bytea[]);
DROP FUNCTION IF EXISTS zebridge_private.record_receipts(text, text[], bytea[], jsonb[]);
CREATE OR REPLACE FUNCTION zebridge_private.lookup_receipts(p_principals text[], p_ids text[], p_routes bytea[], p_payloads bytea[])
RETURNS TABLE(ord integer, status text, outcome jsonb, fingerprint bytea)
LANGUAGE plpgsql SECURITY DEFINER SET search_path = pg_catalog
SET enable_hashjoin = off SET enable_mergejoin = off SET enable_seqscan = off AS $$
DECLARE
    now_ms   bigint := (extract(epoch FROM clock_timestamp()) * 1000)::bigint;
    floor_ms bigint;
    fps      bytea[];
    months   date[];
    verdicts text[];
    timing   text[];
    firsts   integer[];
    rec_fp   bytea;
    rec_outcome jsonb;
    g_keys   text[] := '{}';
    g_fps    bytea[] := '{}';
    p_guard  boolean;
BEGIN
    IF cardinality(p_principals) IS DISTINCT FROM cardinality(p_ids)
       OR cardinality(p_ids) IS DISTINCT FROM cardinality(p_routes)
       OR cardinality(p_ids) IS DISTINCT FROM cardinality(p_payloads) THEN
        RAISE EXCEPTION 'lookup_receipts: principals, ids, routes and payloads differ in length' USING ERRCODE = '22023';
    END IF;
    IF EXISTS (SELECT 1 FROM unnest(p_principals) AS pr WHERE pr IS NULL OR pr = '') THEN
        RAISE EXCEPTION 'a receipt needs an authenticated principal' USING ERRCODE = '22023';
    END IF;
    SELECT min_accepted_ms, completion_guard INTO floor_ms, p_guard FROM zebridge_private.receipt_policy;
    SELECT array_agg(i.fp ORDER BY i.n), array_agg(i.m ORDER BY i.n), array_agg(i.v ORDER BY i.n), array_agg(i.t ORDER BY i.n),
           array_agg(i.f ORDER BY i.n)
      INTO fps, months, verdicts, timing, firsts
      FROM (SELECT c.*,
                   -- The first position of this (principal, ID) among the valid candidates.
                   CASE WHEN c.v = 'candidate' THEN min(c.n) FILTER (WHERE c.v = 'candidate') OVER (PARTITION BY c.pr, c.id) END AS f
              FROM (SELECT x.n::integer AS n, x.pr, x.id,
                           sha256(int4send(length(x.r)) || x.r || x.p) AS fp,
                           zebridge_private.receipt_month(x.id) AS m,
                           CASE WHEN x.id IS NULL OR x.r IS NULL OR x.p IS NULL
                                     OR octet_length(x.id) NOT BETWEEN 1 AND 512 OR x.id ~ '[.*>\x01-\x20\x7f]'
                                     OR zebridge_private.receipt_month(x.id) IS NULL THEN 'invalid'
                                ELSE 'candidate' END AS v,
                           CASE WHEN zebridge_private.receipt_created_ms(x.id) < floor_ms THEN 'expired'
                                WHEN zebridge_private.receipt_created_ms(x.id) > now_ms + 60000 THEN 'future'
                                ELSE 'ok' END AS t
                      FROM unnest(p_principals, p_ids, p_routes, p_payloads) WITH ORDINALITY AS x(pr, id, r, p, n)) c) i;

    FOR n IN 1 .. coalesce(cardinality(p_ids), 0) LOOP
        ord := n;
        outcome := NULL;
        fingerprint := fps[n];
        IF verdicts[n] <> 'candidate' THEN
            status := verdicts[n];
        ELSIF firsts[n] < n THEN
            status := CASE WHEN fps[n] = fps[firsts[n]] THEN 'repeat' ELSE 'conflict' END;
        ELSE
            SELECT r.fingerprint, r.outcome INTO rec_fp, rec_outcome
              FROM zebridge_private.receipts r
             WHERE r.principal = p_principals[n] AND r.message_id = p_ids[n] AND r.id_month = months[n];
            IF FOUND AND rec_fp = fps[n] THEN
                status := 'duplicate';
                outcome := rec_outcome;
            ELSIF FOUND THEN
                status := 'conflict';
            ELSIF timing[n] <> 'ok' THEN
                status := timing[n];
            ELSE
                status := 'claimed';
                g_keys := g_keys || (p_principals[n] || chr(31) || p_ids[n]);
                g_fps := g_fps || fps[n];
            END IF;
        END IF;
        RETURN NEXT;
    END LOOP;
    IF p_guard AND cardinality(g_keys) > 0 THEN
        INSERT INTO zebridge_private.receipt_guards AS g (txid, expected_keys, expected_fps)
        VALUES (pg_current_xact_id(), g_keys, g_fps)
        ON CONFLICT (txid) DO UPDATE SET expected_keys = g.expected_keys || EXCLUDED.expected_keys,
                                         expected_fps = g.expected_fps || EXCLUDED.expected_fps;
    END IF;
END;
$$;

-- Insert the completed receipts of a batch, once, a principal per row. A plain INSERT: a
-- conflict raises (unique_violation), so the transaction can never commit a write without
-- its own receipt. With the completion guard on, the inserted set is added to this
-- transaction's guard.
CREATE OR REPLACE FUNCTION zebridge_private.record_receipts(p_principals text[], p_ids text[], p_fps bytea[], p_outcomes jsonb[])
RETURNS void LANGUAGE plpgsql SECURITY DEFINER SET search_path = pg_catalog AS $$
DECLARE
    p_guard boolean;
BEGIN
    SELECT completion_guard INTO p_guard FROM zebridge_private.receipt_policy;
    IF cardinality(p_principals) IS DISTINCT FROM cardinality(p_ids)
       OR cardinality(p_ids) IS DISTINCT FROM cardinality(p_fps) OR cardinality(p_ids) IS DISTINCT FROM cardinality(p_outcomes)
       OR array_position(p_outcomes, NULL) IS NOT NULL THEN
        RAISE EXCEPTION 'record_receipts: one principal, fingerprint and non-null outcome per ID' USING ERRCODE = '22023';
    END IF;
    IF EXISTS (SELECT 1 FROM unnest(p_principals) AS pr WHERE pr IS NULL OR pr = '') THEN
        RAISE EXCEPTION 'a receipt needs an authenticated principal' USING ERRCODE = '22023';
    END IF;
    INSERT INTO zebridge_private.receipts (principal, message_id, id_month, fingerprint, outcome)
    SELECT o.pr, o.id, zebridge_private.receipt_month(o.id), o.fp, o.outcome
      FROM unnest(p_principals, p_ids, p_fps, p_outcomes) AS o(pr, id, fp, outcome);
    IF p_guard THEN
        UPDATE zebridge_private.receipt_guards g
           SET actual_keys = g.actual_keys || ARRAY(SELECT o.pr || chr(31) || o.id FROM unnest(p_principals, p_ids) WITH ORDINALITY AS o(pr, id, n) ORDER BY o.n),
               actual_fps = g.actual_fps || p_fps
         WHERE g.txid = pg_current_xact_id();
        IF NOT FOUND THEN
            RAISE EXCEPTION 'record_receipts: no guard for this transaction' USING ERRCODE = 'ZBG01';
        END IF;
    END IF;
END;
$$;

-- Whether the completion guard is on: the bridge's boot check compares it with its build.
CREATE OR REPLACE FUNCTION zebridge_private.completion_guard_on()
RETURNS boolean LANGUAGE sql STABLE SECURITY DEFINER SET search_path = pg_catalog AS $$
    SELECT completion_guard FROM zebridge_private.receipt_policy
$$;

-- The completion-guard finalizer: the receipts this transaction inserted must be exactly the
-- undecided set lookup_receipts recorded, entry for entry (key and fingerprint, as
-- multisets: both sides sorted by key, then fingerprint, whatever order they arrived in;
-- an aggregate's ORDER BY 1 would sort by the constant 1, keeping arrival order). Then
-- the guard goes, and the deferred check lets the COMMIT through.
CREATE OR REPLACE FUNCTION zebridge_private.close_guard()
RETURNS void LANGUAGE plpgsql SECURITY DEFINER SET search_path = pg_catalog AS $$
DECLARE
    g zebridge_private.receipt_guards%ROWTYPE;
BEGIN
    SELECT * INTO g FROM zebridge_private.receipt_guards WHERE txid = pg_current_xact_id() FOR UPDATE;
    IF NOT FOUND THEN
        RAISE EXCEPTION 'close_guard: no guard for this transaction' USING ERRCODE = 'ZBG01';
    END IF;
    IF (SELECT coalesce(array_agg(k || ':' || encode(f, 'hex') ORDER BY k, f), '{}') FROM unnest(g.expected_keys, g.expected_fps) AS e(k, f))
       IS DISTINCT FROM
       (SELECT coalesce(array_agg(k || ':' || encode(f, 'hex') ORDER BY k, f), '{}') FROM unnest(g.actual_keys, g.actual_fps) AS a(k, f)) THEN
        RAISE EXCEPTION 'close_guard: the receipts recorded are not the undecided writes of this transaction'
            USING ERRCODE = 'ZBG01';
    END IF;
    DELETE FROM zebridge_private.receipt_guards WHERE txid = g.txid;
END;
$$;

DO $$
DECLARE
    f regprocedure;
BEGIN
    FOR f IN SELECT p.oid::regprocedure FROM pg_proc p JOIN pg_namespace n ON n.oid = p.pronamespace
              WHERE n.nspname = 'zebridge_private'
    LOOP
        EXECUTE format('REVOKE ALL ON FUNCTION %s FROM PUBLIC', f);
    END LOOP;
END $$;
GRANT EXECUTE ON FUNCTION zebridge_private.maintain_receipts() TO ${POSTGRES_WRITER_USER};
GRANT EXECUTE ON FUNCTION zebridge_private.lock_receipts(text[], text[]) TO ${POSTGRES_WRITER_USER};
GRANT EXECUTE ON FUNCTION zebridge_private.lookup_receipts(text[], text[], bytea[], bytea[]) TO ${POSTGRES_WRITER_USER};
GRANT EXECUTE ON FUNCTION zebridge_private.record_receipts(text[], text[], bytea[], jsonb[]) TO ${POSTGRES_WRITER_USER};
GRANT EXECUTE ON FUNCTION zebridge_private.close_guard() TO ${POSTGRES_WRITER_USER};
GRANT EXECUTE ON FUNCTION zebridge_private.completion_guard_on() TO ${POSTGRES_WRITER_USER};
-- No partition is created here: the bridge runs maintain_receipts at startup. A recovery
-- applies this file before `pg_restore --clean`, which cannot drop a partition's
-- inherited primary key; with no partitions present it restores the dumped ones intact.


-- ─── The writer's EXECUTE, and the same rules for this file's functions ───────────────
-- init.core took EXECUTE from PUBLIC and gave it to the reader; the writer needs it too
-- (it applies mutations, the sweeper connects as it), and the functions created here get
-- the same search_path and revocations. The write policies are rewritten as the reader's.
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, %I', f, '${POSTGRES_READER_USER}', '${POSTGRES_WRITER_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;
    FOR t IN SELECT pol.polname, pol.polrelid::regclass AS rel,
                    pg_get_expr(pol.polqual, pol.polrelid) AS q, pg_get_expr(pol.polwithcheck, pol.polrelid) AS c
               FROM pg_policy pol WHERE pol.polname = 'zb_tenant_write'
                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) WITH CHECK (%s)', t.polname, t.rel,
            regexp_replace(t.q, 'current_setting\(''([a-z_.]+)''::text, true\)', '(SELECT current_setting(''\1''::text, true))', 'g'),
            regexp_replace(t.c, 'current_setting\(''([a-z_.]+)''::text, true\)', '(SELECT current_setting(''\1''::text, true))', 'g'));
    END LOOP;
END $$;

-- ── A client write cannot reach a bridge-owned table, whatever the bridge let through ────────────────
--
-- A client's write reaches PostgreSQL as this role with `zb.principal` set to the client
-- (set_config(…, true): transaction-local, from the NATS subject). The sweeper sets it to `zb_sweeper`
-- for its session. Nothing else the bridge does to its own tables sets it: enrollment, the producer,
-- `--revoke`, the DDL trigger, a DBA's psql all write with it unset. So a write to a `zebridge_` table
-- under any other principal is a client's, and this refuses it.
--
-- Until now the only guard in front of `zebridge_gc_watermark` and its neighbours was
-- `mutation_listener.isForbiddenTable()`: with it disabled a client moved the GC floor (the writer holds
-- UPDATE on the table for the sweeper, and the floor's own trigger only refuses a LOWER one). This is the
-- layer behind it, in the database. It is not a defence against a compromised bridge, which can set
-- `zb.principal` to anything.
--
-- Statement-level, so it fires once per statement whatever the row count (a bulk write to
-- `zebridge_generations` pays one call) and also catches TRUNCATE. `zebridge_private` (the receipts) is
-- left out on purpose: the receipts are written inside the client's own transaction, with the principal set.
-- Attached by name prefix, like the revoke loop above, so a table a later release adds is covered; the
-- prefix is reserved for the bridge (OPERATIONS "Reserved names"). `scripts/scenarios/bridge_table_guard.py`
-- checks every `zebridge_` table has it.
CREATE OR REPLACE FUNCTION public.zebridge_refuse_client_write()
RETURNS trigger AS $$
DECLARE
    who text := coalesce(current_setting('zb.principal', true), '');
BEGIN
    IF who <> '' AND who <> 'zb_sweeper' THEN
        RAISE EXCEPTION 'refusing a client write to %.%: it is a bridge-owned table', TG_TABLE_SCHEMA, TG_TABLE_NAME
            USING ERRCODE = '42501',
                  HINT = 'A write carrying a client principal (zb.principal) never reaches a zebridge_ table: only the bridge, the sweeper and a DBA write them.';
    END IF;
    RETURN NULL;
END;
$$ LANGUAGE plpgsql;

DO $$
DECLARE
    t text;
BEGIN
    FOR t IN SELECT c.relname FROM pg_class c JOIN pg_namespace n ON n.oid = c.relnamespace
              WHERE n.nspname = 'public' AND c.relkind = 'r' AND c.relname LIKE 'zebridge\_%'
    LOOP
        EXECUTE format('DROP TRIGGER IF EXISTS zebridge_refuse_client_write_t ON public.%I', t);
        EXECUTE format('CREATE TRIGGER zebridge_refuse_client_write_t BEFORE INSERT OR UPDATE OR DELETE OR TRUNCATE ON public.%I '
                       'FOR EACH STATEMENT EXECUTE FUNCTION public.zebridge_refuse_client_write()', t);
    END LOOP;
END $$;