qbit-worm-audit/docs/ddl/worm-audit-postgres.sql

-- 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);