-- liquibase formatted sql

-- changeset worm-audit:1
CREATE TABLE IF NOT EXISTS public.worm_audit_chain (
   id bigint PRIMARY KEY,
   kind text NOT NULL,
   audit_id bigint NOT NULL,
   detail_ids text NOT NULL,
   prev_hash text NOT NULL,
   link_hash text NOT NULL,
   chained_at timestamptz NOT NULL,
   CHECK (id > 0),
   CHECK (kind IN ('AUDIT', 'LATE_DETAIL')),
   CHECK (prev_hash ~ '^[0-9a-f]{64}$' AND link_hash ~ '^[0-9a-f]{64}$'),
   CHECK (detail_ids = '' OR (detail_ids ~ '^(0|[1-9][0-9]*)(,(0|[1-9][0-9]*))*$' AND cardinality(string_to_array(detail_ids, ',')) <= 250))
);

CREATE TABLE IF NOT EXISTS public.worm_audit_checkpoint (
   id bigint GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
   from_link_id bigint,
   to_link_id bigint,
   link_count bigint,
   head_hash text,
   merkle_root text,
   algorithm text,
   key_id text,
   public_key text,
   signature text,
   created_at timestamptz
);

CREATE TABLE IF NOT EXISTS public.worm_audit_export (
   id bigint GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
   from_link_id bigint,
   to_link_id bigint,
   from_checkpoint_id bigint,
   to_checkpoint_id bigint,
   bucket_alias text,
   object_prefix text,
   object_versions text,
   manifest_key text,
   manifest_sha256 text,
   prev_manifest_sha256 text,
   object_count bigint,
   byte_count bigint,
   lock_mode text,
   retain_until timestamptz,
   sealed_at timestamptz
);

CREATE TABLE IF NOT EXISTS public.worm_audit_verification (
   id bigint GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
   kind text,
   "trigger" text,
   started_at timestamptz,
   finished_at timestamptz,
   from_link_id bigint,
   to_link_id bigint,
   links_checked bigint,
   outcome text,
   break_count bigint,
   first_break_link_id bigint,
   first_break_audit_id bigint,
   first_break_reason text,
   export_id bigint,
   run_by text
);

CREATE TABLE IF NOT EXISTS public.worm_audit_purge (
   id bigint GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY,
   requested_at timestamptz,
   requested_by text,
   cutoff timestamptz,
   retention_interval text,
   outcome text,
   reason text,
   audits_deleted bigint,
   details_deleted bigint,
   links_deleted bigint,
   anchor_link_id bigint,
   anchor_hash text
);

CREATE UNIQUE INDEX IF NOT EXISTS worm_audit_chain_audit ON public.worm_audit_chain(audit_id) WHERE kind = 'AUDIT';
CREATE INDEX IF NOT EXISTS worm_audit_chain_audit_all ON public.worm_audit_chain(audit_id);

-- changeset worm-audit:2 splitStatements:false
CREATE OR REPLACE FUNCTION public.worm_audit_reject() RETURNS trigger
LANGUAGE plpgsql SET search_path = pg_catalog, public, pg_temp SET timezone = 'UTC' AS $$
DECLARE
   old_time timestamptz;
   function_owner name;
   stack text;
   row_identity text := 'all';
BEGIN
   IF TG_OP = 'DELETE' AND current_setting('worm_audit.purging', true) = 'on' THEN
      SELECT pg_get_userbyid(proowner) INTO function_owner FROM pg_proc
         WHERE oid = 'public.worm_audit_purge(timestamptz)'::regprocedure;
      GET DIAGNOSTICS stack = PG_CONTEXT;
      IF current_user = function_owner AND position('function worm_audit_purge(timestamp with time zone)' in stack) > 0 THEN
         IF TG_RELID = '"audit"'::regclass THEN
            old_time := OLD.timestamp;
         ELSIF TG_RELID = '"audit_detail"'::regclass THEN
            SELECT timestamp INTO old_time FROM "audit" WHERE id = OLD.audit_id;
         ELSIF TG_RELID = 'public.worm_audit_chain'::regclass THEN
            SELECT timestamp INTO old_time FROM "audit" WHERE id = OLD.audit_id;
         END IF;
         IF old_time < now() - interval '6 years' THEN
            RETURN OLD;
         END IF;
      END IF;
   END IF;
   IF TG_LEVEL = 'ROW' THEN
      row_identity := OLD.id::text;
   END IF;
   RAISE EXCEPTION USING ERRCODE = 'P0001',
      MESSAGE = format('worm_audit: audit rows are write-once; %s of "%s" is not allowed (id=%s). Expired rows are removed only by worm_audit_purge().',
         TG_OP, TG_TABLE_NAME, row_identity),
      HINT = 'see qbit-worm-audit docs/DESIGN.md section 4';
END;
$$;

-- changeset worm-audit:3 splitStatements:false
CREATE OR REPLACE FUNCTION public.worm_audit_lookup_reject() RETURNS trigger
LANGUAGE plpgsql SET search_path = pg_catalog, public, pg_temp AS $$
BEGIN
   IF TG_OP = 'UPDATE' AND NEW.id IS NOT DISTINCT FROM OLD.id AND NEW.name IS NOT DISTINCT FROM OLD.name THEN
      RETURN NEW;
   END IF;
   RAISE EXCEPTION USING ERRCODE = 'P0001',
      MESSAGE = format('worm_audit: audit rows are write-once; %s of "%s" is not allowed (id=%s). Expired rows are removed only by worm_audit_purge().',
         TG_OP, TG_TABLE_NAME, OLD.id),
      HINT = 'see qbit-worm-audit docs/DESIGN.md section 4';
END;
$$;

-- changeset worm-audit:4 splitStatements:false
DO $$
BEGIN
   IF NOT EXISTS (SELECT FROM pg_roles WHERE rolname = 'worm_audit_purger') THEN
      CREATE ROLE worm_audit_purger NOLOGIN;
   END IF;
END;
$$;

-- changeset worm-audit:5 splitStatements:false
CREATE OR REPLACE FUNCTION public.worm_audit_purge(p_cutoff timestamptz) RETURNS void
LANGUAGE plpgsql SECURITY DEFINER SET search_path = pg_catalog, public, pg_temp SET timezone = 'UTC' AS $$
DECLARE
   boundary bigint;
   smaller bigint;
   anchor text;
   audits_count bigint;
   details_count bigint;
   links_count bigint;
BEGIN
   IF p_cutoff IS NULL OR p_cutoff > now() - interval '6 years' THEN
      INSERT INTO public.worm_audit_purge(requested_at, requested_by, cutoff, retention_interval, outcome, reason)
         VALUES(now(), session_user, p_cutoff, '6 years', 'REFUSED', 'Cutoff is within retention or null');
      RETURN;
   END IF;
   -- Same lock as the Java chainer; table locks also exclude concurrent audit inserts.
   PERFORM pg_advisory_xact_lock(86753090421);
   LOCK TABLE "audit", "audit_detail", public.worm_audit_chain IN SHARE ROW EXCLUSIVE MODE;
   SELECT coalesce(min(c.id) - 1, (SELECT max(id) FROM public.worm_audit_chain)) INTO boundary
      FROM public.worm_audit_chain c LEFT JOIN "audit" a ON a.id = c.audit_id
      WHERE a.id IS NULL OR a.timestamp IS NULL OR a.timestamp >= p_cutoff;
   -- Shrink to a prefix closed over every audit's links AND all its stored details.
   LOOP
      SELECT min(c.id) - 1 INTO smaller FROM public.worm_audit_chain c
         WHERE c.id <= boundary AND (
            EXISTS (SELECT FROM public.worm_audit_chain later WHERE later.audit_id = c.audit_id AND later.id > boundary)
            OR EXISTS (SELECT FROM "audit_detail" d WHERE d.audit_id = c.audit_id AND NOT EXISTS (
               SELECT FROM public.worm_audit_chain covered WHERE covered.audit_id = c.audit_id
                  AND d.id::text = ANY(string_to_array(covered.detail_ids, ',')))));
      EXIT WHEN smaller IS NULL;
      boundary := smaller;
   END LOOP;
   SELECT link_hash INTO anchor FROM public.worm_audit_chain WHERE id = boundary;
   IF anchor IS NULL THEN
      INSERT INTO public.worm_audit_purge(requested_at, requested_by, cutoff, retention_interval, outcome, reason)
         VALUES(now(), session_user, p_cutoff, '6 years', 'REFUSED', 'No complete expired chain prefix');
      RETURN;
   END IF;
   PERFORM set_config('worm_audit.purging', 'on', true);
   DELETE FROM "audit_detail" d WHERE EXISTS
      (SELECT FROM public.worm_audit_chain c WHERE c.audit_id = d.audit_id AND c.id <= boundary);
   GET DIAGNOSTICS details_count = ROW_COUNT;
   -- Keep ids locally because links must be deleted before their audit headers.
   WITH removed AS (DELETE FROM public.worm_audit_chain WHERE id <= boundary RETURNING audit_id),
      removed_audits AS (DELETE FROM "audit" WHERE id IN (SELECT audit_id FROM removed) RETURNING id)
      SELECT (SELECT count(*) FROM removed), (SELECT count(*) FROM removed_audits) INTO links_count, audits_count;
   PERFORM set_config('worm_audit.purging', 'off', true);
   INSERT INTO public.worm_audit_purge(requested_at, requested_by, cutoff, retention_interval, outcome,
      audits_deleted, details_deleted, links_deleted, anchor_link_id, anchor_hash)
      VALUES(now(), session_user, p_cutoff, '6 years', 'PURGED', audits_count, details_count, links_count, boundary, anchor);
END;
$$;

-- changeset worm-audit:6
DROP TRIGGER IF EXISTS worm_audit_row_guard ON "audit";
CREATE TRIGGER worm_audit_row_guard BEFORE UPDATE OR DELETE ON "audit"
   FOR EACH ROW EXECUTE FUNCTION public.worm_audit_reject();
DROP TRIGGER IF EXISTS worm_audit_truncate_guard ON "audit";
CREATE TRIGGER worm_audit_truncate_guard BEFORE TRUNCATE ON "audit"
   FOR EACH STATEMENT EXECUTE FUNCTION public.worm_audit_reject();
REVOKE UPDATE, DELETE, TRUNCATE ON "audit" FROM "app";
GRANT SELECT, INSERT ON "audit" TO "app";

DROP TRIGGER IF EXISTS worm_audit_row_guard ON "audit_detail";
CREATE TRIGGER worm_audit_row_guard BEFORE UPDATE OR DELETE ON "audit_detail"
   FOR EACH ROW EXECUTE FUNCTION public.worm_audit_reject();
DROP TRIGGER IF EXISTS worm_audit_truncate_guard ON "audit_detail";
CREATE TRIGGER worm_audit_truncate_guard BEFORE TRUNCATE ON "audit_detail"
   FOR EACH STATEMENT EXECUTE FUNCTION public.worm_audit_reject();
REVOKE UPDATE, DELETE, TRUNCATE ON "audit_detail" FROM "app";
GRANT SELECT, INSERT ON "audit_detail" TO "app";

DROP TRIGGER IF EXISTS worm_audit_row_guard ON public.worm_audit_chain;
CREATE TRIGGER worm_audit_row_guard BEFORE UPDATE OR DELETE ON public.worm_audit_chain
   FOR EACH ROW EXECUTE FUNCTION public.worm_audit_reject();
DROP TRIGGER IF EXISTS worm_audit_truncate_guard ON public.worm_audit_chain;
CREATE TRIGGER worm_audit_truncate_guard BEFORE TRUNCATE ON public.worm_audit_chain
   FOR EACH STATEMENT EXECUTE FUNCTION public.worm_audit_reject();
REVOKE UPDATE, DELETE, TRUNCATE ON public.worm_audit_chain FROM "app";
GRANT SELECT, INSERT ON public.worm_audit_chain TO "app";

DROP TRIGGER IF EXISTS worm_audit_row_guard ON public.worm_audit_checkpoint;
CREATE TRIGGER worm_audit_row_guard BEFORE UPDATE OR DELETE ON public.worm_audit_checkpoint
   FOR EACH ROW EXECUTE FUNCTION public.worm_audit_reject();
DROP TRIGGER IF EXISTS worm_audit_truncate_guard ON public.worm_audit_checkpoint;
CREATE TRIGGER worm_audit_truncate_guard BEFORE TRUNCATE ON public.worm_audit_checkpoint
   FOR EACH STATEMENT EXECUTE FUNCTION public.worm_audit_reject();
REVOKE UPDATE, DELETE, TRUNCATE ON public.worm_audit_checkpoint FROM "app";
GRANT SELECT, INSERT ON public.worm_audit_checkpoint TO "app";

DROP TRIGGER IF EXISTS worm_audit_row_guard ON public.worm_audit_export;
CREATE TRIGGER worm_audit_row_guard BEFORE UPDATE OR DELETE ON public.worm_audit_export
   FOR EACH ROW EXECUTE FUNCTION public.worm_audit_reject();
DROP TRIGGER IF EXISTS worm_audit_truncate_guard ON public.worm_audit_export;
CREATE TRIGGER worm_audit_truncate_guard BEFORE TRUNCATE ON public.worm_audit_export
   FOR EACH STATEMENT EXECUTE FUNCTION public.worm_audit_reject();
REVOKE UPDATE, DELETE, TRUNCATE ON public.worm_audit_export FROM "app";
GRANT SELECT, INSERT ON public.worm_audit_export TO "app";

DROP TRIGGER IF EXISTS worm_audit_row_guard ON public.worm_audit_verification;
CREATE TRIGGER worm_audit_row_guard BEFORE UPDATE OR DELETE ON public.worm_audit_verification
   FOR EACH ROW EXECUTE FUNCTION public.worm_audit_reject();
DROP TRIGGER IF EXISTS worm_audit_truncate_guard ON public.worm_audit_verification;
CREATE TRIGGER worm_audit_truncate_guard BEFORE TRUNCATE ON public.worm_audit_verification
   FOR EACH STATEMENT EXECUTE FUNCTION public.worm_audit_reject();
REVOKE UPDATE, DELETE, TRUNCATE ON public.worm_audit_verification FROM "app";
GRANT SELECT, INSERT ON public.worm_audit_verification TO "app";

DROP TRIGGER IF EXISTS worm_audit_row_guard ON public.worm_audit_purge;
CREATE TRIGGER worm_audit_row_guard BEFORE UPDATE OR DELETE ON public.worm_audit_purge
   FOR EACH ROW EXECUTE FUNCTION public.worm_audit_reject();
DROP TRIGGER IF EXISTS worm_audit_truncate_guard ON public.worm_audit_purge;
CREATE TRIGGER worm_audit_truncate_guard BEFORE TRUNCATE ON public.worm_audit_purge
   FOR EACH STATEMENT EXECUTE FUNCTION public.worm_audit_reject();
REVOKE UPDATE, DELETE, TRUNCATE ON public.worm_audit_purge FROM "app";
GRANT SELECT, INSERT ON public.worm_audit_purge TO "app";

DROP TRIGGER IF EXISTS worm_audit_row_guard ON "audit_table";
CREATE TRIGGER worm_audit_row_guard BEFORE UPDATE OR DELETE ON "audit_table"
   FOR EACH ROW EXECUTE FUNCTION public.worm_audit_lookup_reject();
DROP TRIGGER IF EXISTS worm_audit_truncate_guard ON "audit_table";
CREATE TRIGGER worm_audit_truncate_guard BEFORE TRUNCATE ON "audit_table"
   FOR EACH STATEMENT EXECUTE FUNCTION public.worm_audit_reject();
REVOKE UPDATE, DELETE, TRUNCATE ON "audit_table" FROM "app";
GRANT SELECT, INSERT ON "audit_table" TO "app";

DROP TRIGGER IF EXISTS worm_audit_row_guard ON "audit_user";
CREATE TRIGGER worm_audit_row_guard BEFORE UPDATE OR DELETE ON "audit_user"
   FOR EACH ROW EXECUTE FUNCTION public.worm_audit_lookup_reject();
DROP TRIGGER IF EXISTS worm_audit_truncate_guard ON "audit_user";
CREATE TRIGGER worm_audit_truncate_guard BEFORE TRUNCATE ON "audit_user"
   FOR EACH STATEMENT EXECUTE FUNCTION public.worm_audit_reject();
REVOKE UPDATE, DELETE, TRUNCATE ON "audit_user" FROM "app";
GRANT SELECT, INSERT ON "audit_user" TO "app";

-- Lookups may refresh display metadata, but never identities or names.
GRANT UPDATE (label, create_date, modify_date) ON "audit_table" TO "app";
GRANT UPDATE (create_date, modify_date) ON "audit_user" TO "app";

REVOKE ALL ON FUNCTION public.worm_audit_purge(timestamptz) FROM PUBLIC;
REVOKE ALL ON FUNCTION public.worm_audit_purge(timestamptz) FROM "app";
GRANT EXECUTE ON FUNCTION public.worm_audit_purge(timestamptz) TO worm_audit_purger;

-- changeset worm-audit:7 splitStatements:false
DO $$
DECLARE
   table_name text;
   sequence_name text;
BEGIN
   FOREACH table_name IN ARRAY ARRAY['"audit"', '"audit_detail"', '"audit_table"', '"audit_user"',
      'public.worm_audit_checkpoint', 'public.worm_audit_export', 'public.worm_audit_verification', 'public.worm_audit_purge'] LOOP
      sequence_name := pg_get_serial_sequence(table_name, 'id');
      IF sequence_name IS NOT NULL THEN
         EXECUTE format('GRANT USAGE ON SEQUENCE %s TO "app"', sequence_name);
      END IF;
   END LOOP;
END;
$$;

-- changeset worm-audit:8 splitStatements:false
CREATE OR REPLACE FUNCTION public.worm_audit_detail_coverage() RETURNS trigger
LANGUAGE plpgsql SET search_path = pg_catalog, public, pg_temp AS $$
BEGIN
   -- Serialize even direct SQL appends; the GIN index makes overlap checks indexed.
   PERFORM pg_advisory_xact_lock(86753090421);
   IF NEW.detail_ids <> '' AND (
      EXISTS (SELECT FROM unnest(string_to_array(NEW.detail_ids, ',')) AS d(id) GROUP BY d.id HAVING count(*) > 1)
      OR EXISTS (SELECT FROM public.worm_audit_chain c
         WHERE string_to_array(c.detail_ids, ',') && string_to_array(NEW.detail_ids, ','))) THEN
      RAISE EXCEPTION USING ERRCODE = '23505', MESSAGE = 'worm_audit: audit detail is already covered by the chain';
   END IF;
   RETURN NEW;
END;
$$;

-- changeset worm-audit:9
CREATE INDEX IF NOT EXISTS worm_audit_chain_detail_coverage ON public.worm_audit_chain USING gin (string_to_array(detail_ids, ','));
DROP TRIGGER IF EXISTS worm_audit_detail_coverage_guard ON public.worm_audit_chain;
CREATE TRIGGER worm_audit_detail_coverage_guard BEFORE INSERT ON public.worm_audit_chain
   FOR EACH ROW EXECUTE FUNCTION public.worm_audit_detail_coverage();

-- changeset worm-audit:10
-- Editable singleton: deliberately excluded from append-only triggers.
CREATE TABLE IF NOT EXISTS public.worm_audit_settings (
   id integer PRIMARY KEY CHECK (id = 1),
   audit_retention_years integer NOT NULL DEFAULT 6 CHECK (audit_retention_years BETWEEN 1 AND 100),
   export_lock_mode text NOT NULL DEFAULT 'GOVERNANCE' CHECK (export_lock_mode IN ('GOVERNANCE', 'COMPLIANCE')),
   export_retention_days integer NOT NULL DEFAULT 30 CHECK (export_retention_days BETWEEN 1 AND 36500),
   checkpoint_every_links integer NOT NULL DEFAULT 10000 CHECK (checkpoint_every_links > 0),
   checkpoint_every_minutes integer NOT NULL DEFAULT 60 CHECK (checkpoint_every_minutes > 0),
   chain_interval_seconds integer NOT NULL DEFAULT 30 CHECK (chain_interval_seconds > 0),
   settle_horizon_seconds integer NOT NULL DEFAULT 60 CHECK (settle_horizon_seconds >= 0),
   late_detail_lookback_hours integer NOT NULL DEFAULT 24 CHECK (late_detail_lookback_hours > 0),
   export_cron text NOT NULL DEFAULT '0 0 2 * * ?',
   verify_cron text NOT NULL DEFAULT '0 0 3 * * ?',
   notes text,
   modify_date timestamptz NOT NULL DEFAULT now(),
   modified_by text NOT NULL DEFAULT current_user
);
REVOKE DELETE, TRUNCATE ON public.worm_audit_settings FROM "app";
GRANT SELECT, INSERT, UPDATE ON public.worm_audit_settings TO "app";
INSERT INTO public.worm_audit_settings
   (id, audit_retention_years, export_lock_mode, export_retention_days, checkpoint_every_links,
    checkpoint_every_minutes, chain_interval_seconds, settle_horizon_seconds, late_detail_lookback_hours,
    export_cron, verify_cron)
SELECT 1, 6, 'GOVERNANCE', 30, 10000, 60, 30, 60, 24, '0 0 2 * * ?', '0 0 3 * * ?'
WHERE NOT EXISTS (SELECT 1 FROM public.worm_audit_settings WHERE id = 1);

-- changeset worm-audit:11
-- comment: Index the foreign key columns added by worm-audit:12 (worm_audit_chain.audit_id already leads worm_audit_chain_audit_all).
CREATE INDEX IF NOT EXISTS worm_audit_export_from_checkpoint ON public.worm_audit_export(from_checkpoint_id);
CREATE INDEX IF NOT EXISTS worm_audit_export_to_checkpoint ON public.worm_audit_export(to_checkpoint_id);
CREATE INDEX IF NOT EXISTS worm_audit_verification_export ON public.worm_audit_verification(export_id);

-- changeset worm-audit:12 splitStatements:false
-- comment: Foreign keys, ON DELETE RESTRICT only: nothing cascades into or out of audit evidence. A chain link needs its audit header (the purge deletes links before headers); a sealed export needs its signed checkpoints and an export verification its export (neither is ever deleted). Links referenced by checkpoints, exports, verifications and purges are not foreign keys: the purge deletes the sealed prefix those rows outlive. Rows whose parent is already missing are evidence and are never changed: they are copied to qqq_fk_quarantine (action KEPT) and the constraint is left NOT VALID (enforced for new rows only).
DO $$
DECLARE
   fk record;
   orphans bigint;
BEGIN
   CREATE TABLE IF NOT EXISTS qqq_fk_quarantine (id BIGINT GENERATED BY DEFAULT AS IDENTITY PRIMARY KEY, table_name VARCHAR(100) NOT NULL, column_name VARCHAR(100) NOT NULL, row_id VARCHAR(100), missing_parent_id VARCHAR(100), action VARCHAR(20) NOT NULL, row_data TEXT, changeset VARCHAR(200), quarantined_at TIMESTAMP DEFAULT CURRENT_TIMESTAMP);
   FOR fk IN SELECT * FROM (VALUES
      ('fk_worm_audit_chain_audit_id', 'worm_audit_chain', 'audit_id', '"audit"'),
      ('fk_worm_audit_export_from_checkpoint_id', 'worm_audit_export', 'from_checkpoint_id', 'public.worm_audit_checkpoint'),
      ('fk_worm_audit_export_to_checkpoint_id', 'worm_audit_export', 'to_checkpoint_id', 'public.worm_audit_checkpoint'),
      ('fk_worm_audit_verification_export_id', 'worm_audit_verification', 'export_id', 'public.worm_audit_export')
   ) AS t(name, child, col, parent) LOOP
      IF NOT EXISTS (SELECT FROM pg_constraint WHERE conname = fk.name) THEN
         EXECUTE format('INSERT INTO qqq_fk_quarantine (table_name, column_name, row_id, missing_parent_id, action, row_data, changeset)'
            || ' SELECT %L, %L, c.id::text, c.%I::text, ''KEPT'', row_to_json(c)::text, ''worm-audit:12'' FROM public.%I c'
            || ' WHERE c.%I IS NOT NULL AND NOT EXISTS (SELECT FROM %s p WHERE p.id = c.%I)', fk.child, fk.col, fk.col, fk.child, fk.col, fk.parent, fk.col);
         GET DIAGNOSTICS orphans = ROW_COUNT;
         EXECUTE format('ALTER TABLE public.%I ADD CONSTRAINT %I FOREIGN KEY (%I) REFERENCES %s (id) ON DELETE RESTRICT NOT VALID', fk.child, fk.name, fk.col, fk.parent);
         IF orphans = 0 THEN
            EXECUTE format('ALTER TABLE public.%I VALIDATE CONSTRAINT %I', fk.child, fk.name);
         END IF;
      END IF;
   END LOOP;
END;
$$;

-- changeset worm-audit:13 splitStatements:false
-- comment: Normalized detail coverage. One insert-only row per audit_detail id covered by a chain link, written by the coverage trigger in the same statement as the link, so the chain writer probes a primary key instead of the GIN array index. Append-only like the chain, with no foreign keys (the owner cannot UPDATE WORM tables, so FK checks would be denied). The advisory lock keeps a running chainer from appending between the backfill and the new trigger body.
DO $$
BEGIN
   PERFORM pg_advisory_xact_lock(86753090421);
END;
$$;
CREATE TABLE IF NOT EXISTS public.worm_audit_chain_detail (
   detail_id bigint PRIMARY KEY,
   chain_id bigint NOT NULL,
   CHECK (detail_id >= 0),
   CHECK (chain_id > 0)
);
CREATE INDEX IF NOT EXISTS worm_audit_chain_detail_chain ON public.worm_audit_chain_detail(chain_id);
-- The first link wins if a tampered chain covers a detail twice; the verifier still reports the duplicate from the chain itself.
INSERT INTO public.worm_audit_chain_detail(detail_id, chain_id)
   SELECT x.id::bigint, c.id FROM public.worm_audit_chain c CROSS JOIN LATERAL unnest(string_to_array(c.detail_ids, ',')) AS x(id)
   WHERE c.detail_ids <> '' ORDER BY c.id
   ON CONFLICT (detail_id) DO NOTHING;

CREATE OR REPLACE FUNCTION public.worm_audit_reject() RETURNS trigger
LANGUAGE plpgsql SET search_path = pg_catalog, public, pg_temp SET timezone = 'UTC' AS $fn$
DECLARE
   old_time timestamptz;
   function_owner name;
   stack text;
   row_identity text := 'all';
BEGIN
   IF TG_OP = 'DELETE' AND current_setting('worm_audit.purging', true) = 'on' THEN
      SELECT pg_get_userbyid(proowner) INTO function_owner FROM pg_proc
         WHERE oid = 'public.worm_audit_purge(timestamptz)'::regprocedure;
      GET DIAGNOSTICS stack = PG_CONTEXT;
      IF current_user = function_owner AND position('function worm_audit_purge(timestamp with time zone)' in stack) > 0 THEN
         IF TG_RELID = '"audit"'::regclass THEN
            old_time := OLD.timestamp;
         ELSIF TG_RELID = '"audit_detail"'::regclass THEN
            SELECT timestamp INTO old_time FROM "audit" WHERE id = OLD.audit_id;
         ELSIF TG_RELID = 'public.worm_audit_chain'::regclass THEN
            SELECT timestamp INTO old_time FROM "audit" WHERE id = OLD.audit_id;
         ELSIF TG_RELID = 'public.worm_audit_chain_detail'::regclass THEN
            SELECT a.timestamp INTO old_time FROM public.worm_audit_chain c JOIN "audit" a ON a.id = c.audit_id WHERE c.id = OLD.chain_id;
         END IF;
         IF old_time < now() - interval '6 years' THEN
            RETURN OLD;
         END IF;
      END IF;
   END IF;
   IF TG_LEVEL = 'ROW' THEN
      -- Record fields resolve per statement, so each table's key is read in its own branch.
      IF TG_RELID = 'public.worm_audit_chain_detail'::regclass THEN
         row_identity := OLD.detail_id::text;
      ELSE
         row_identity := OLD.id::text;
      END IF;
   END IF;
   RAISE EXCEPTION USING ERRCODE = 'P0001',
      MESSAGE = format('worm_audit: audit rows are write-once; %s of "%s" is not allowed (id=%s). Expired rows are removed only by worm_audit_purge().',
         TG_OP, TG_TABLE_NAME, row_identity),
      HINT = 'see qbit-worm-audit docs/DESIGN.md section 4';
END;
$fn$;

CREATE OR REPLACE FUNCTION public.worm_audit_detail_coverage() RETURNS trigger
LANGUAGE plpgsql SET search_path = pg_catalog, public, pg_temp AS $fn$
BEGIN
   -- Serialize even direct SQL appends, then record coverage in the same statement as the link.
   -- The primary key rejects a detail covered twice, within this link or by an earlier one. The
   -- previous EXISTS overlap check ran as a cached generic plan that chose a sequential scan of
   -- the whole chain for every inserted link.
   PERFORM pg_advisory_xact_lock(86753090421);
   IF NEW.detail_ids = '' THEN
      RETURN NEW;
   END IF;
   BEGIN
      INSERT INTO public.worm_audit_chain_detail(detail_id, chain_id)
         SELECT d, NEW.id FROM unnest(string_to_array(NEW.detail_ids, ',')::bigint[]) AS d;
   EXCEPTION WHEN unique_violation THEN
      RAISE EXCEPTION USING ERRCODE = '23505', MESSAGE = 'worm_audit: audit detail is already covered by the chain';
   END;
   RETURN NEW;
END;
$fn$;

-- The array index worm_audit_chain_detail_coverage is no longer used by this release. It stays until
-- a later migration drops it, so a previous-release writer still running during a rolling deploy keeps its indexed probe.

DROP TRIGGER IF EXISTS worm_audit_row_guard ON public.worm_audit_chain_detail;
CREATE TRIGGER worm_audit_row_guard BEFORE UPDATE OR DELETE ON public.worm_audit_chain_detail
   FOR EACH ROW EXECUTE FUNCTION public.worm_audit_reject();
DROP TRIGGER IF EXISTS worm_audit_truncate_guard ON public.worm_audit_chain_detail;
CREATE TRIGGER worm_audit_truncate_guard BEFORE TRUNCATE ON public.worm_audit_chain_detail
   FOR EACH STATEMENT EXECUTE FUNCTION public.worm_audit_reject();
REVOKE UPDATE, DELETE, TRUNCATE ON public.worm_audit_chain_detail FROM "app";
GRANT SELECT, INSERT ON public.worm_audit_chain_detail TO "app";

-- Same function as worm-audit:5, except that coverage rows are locked and purged before their links.
-- The migrating role may have revoked its own EXECUTE (a runtime role that owns the schema, as on CNPG);
-- PostgreSQL then refuses CREATE OR REPLACE. Grant it back for this statement only, then restore.
DO $$
BEGIN
   IF NOT has_function_privilege(current_user, 'public.worm_audit_purge(timestamptz)', 'EXECUTE') THEN
      EXECUTE format('GRANT EXECUTE ON FUNCTION public.worm_audit_purge(timestamptz) TO %I', current_user);
      PERFORM set_config('worm_audit.purge_regrant', 'on', true);
   END IF;
END;
$$;
CREATE OR REPLACE FUNCTION public.worm_audit_purge(p_cutoff timestamptz) RETURNS void
LANGUAGE plpgsql SECURITY DEFINER SET search_path = pg_catalog, public, pg_temp SET timezone = 'UTC' AS $fn$
DECLARE
   boundary bigint;
   smaller bigint;
   anchor text;
   audits_count bigint;
   details_count bigint;
   links_count bigint;
BEGIN
   IF p_cutoff IS NULL OR p_cutoff > now() - interval '6 years' THEN
      INSERT INTO public.worm_audit_purge(requested_at, requested_by, cutoff, retention_interval, outcome, reason)
         VALUES(now(), session_user, p_cutoff, '6 years', 'REFUSED', 'Cutoff is within retention or null');
      RETURN;
   END IF;
   -- Same lock as the Java chainer; table locks also exclude concurrent audit inserts.
   PERFORM pg_advisory_xact_lock(86753090421);
   LOCK TABLE "audit", "audit_detail", public.worm_audit_chain, public.worm_audit_chain_detail IN SHARE ROW EXCLUSIVE MODE;
   SELECT coalesce(min(c.id) - 1, (SELECT max(id) FROM public.worm_audit_chain)) INTO boundary
      FROM public.worm_audit_chain c LEFT JOIN "audit" a ON a.id = c.audit_id
      WHERE a.id IS NULL OR a.timestamp IS NULL OR a.timestamp >= p_cutoff;
   -- Shrink to a prefix closed over every audit's links AND all its stored details.
   LOOP
      SELECT min(c.id) - 1 INTO smaller FROM public.worm_audit_chain c
         WHERE c.id <= boundary AND (
            EXISTS (SELECT FROM public.worm_audit_chain later WHERE later.audit_id = c.audit_id AND later.id > boundary)
            OR EXISTS (SELECT FROM "audit_detail" d WHERE d.audit_id = c.audit_id AND NOT EXISTS (
               SELECT FROM public.worm_audit_chain covered WHERE covered.audit_id = c.audit_id
                  AND d.id::text = ANY(string_to_array(covered.detail_ids, ',')))));
      EXIT WHEN smaller IS NULL;
      boundary := smaller;
   END LOOP;
   SELECT link_hash INTO anchor FROM public.worm_audit_chain WHERE id = boundary;
   IF anchor IS NULL THEN
      INSERT INTO public.worm_audit_purge(requested_at, requested_by, cutoff, retention_interval, outcome, reason)
         VALUES(now(), session_user, p_cutoff, '6 years', 'REFUSED', 'No complete expired chain prefix');
      RETURN;
   END IF;
   PERFORM set_config('worm_audit.purging', 'on', true);
   DELETE FROM "audit_detail" d WHERE EXISTS
      (SELECT FROM public.worm_audit_chain c WHERE c.audit_id = d.audit_id AND c.id <= boundary);
   GET DIAGNOSTICS details_count = ROW_COUNT;
   -- Coverage rows go before their links: the guard finds their age through the link.
   DELETE FROM public.worm_audit_chain_detail WHERE chain_id <= boundary;
   -- Keep ids locally because links must be deleted before their audit headers.
   WITH removed AS (DELETE FROM public.worm_audit_chain WHERE id <= boundary RETURNING audit_id),
      removed_audits AS (DELETE FROM "audit" WHERE id IN (SELECT audit_id FROM removed) RETURNING id)
      SELECT (SELECT count(*) FROM removed), (SELECT count(*) FROM removed_audits) INTO links_count, audits_count;
   PERFORM set_config('worm_audit.purging', 'off', true);
   INSERT INTO public.worm_audit_purge(requested_at, requested_by, cutoff, retention_interval, outcome,
      audits_deleted, details_deleted, links_deleted, anchor_link_id, anchor_hash)
      VALUES(now(), session_user, p_cutoff, '6 years', 'PURGED', audits_count, details_count, links_count, boundary, anchor);
END;
$fn$;
DO $$
BEGIN
   IF current_setting('worm_audit.purge_regrant', true) = 'on' THEN
      EXECUTE format('REVOKE EXECUTE ON FUNCTION public.worm_audit_purge(timestamptz) FROM %I', current_user);
      PERFORM set_config('worm_audit.purge_regrant', 'off', true);
   END IF;
END;
$$;