DataOps-platform/migrations/versions/20260809_477_edge_gateway_control_plane.py

"""Enterprise edge-gateway control-plane identity and durable ledgers."""

from alembic import op

revision = "20260809_477"
down_revision = "20260802_476"
branch_labels = None
depends_on = None


def upgrade() -> None:
    op.execute(r"""
    DO $$
    BEGIN
      IF NOT EXISTS (SELECT 1 FROM pg_roles WHERE rolname='dataops_app_runtime') OR
         NOT EXISTS (SELECT 1 FROM pg_roles WHERE rolname='dataops_edge_evidence_owner') THEN
        RAISE EXCEPTION 'edge database roles are not pre-provisioned; run the privileged role-init step';
      END IF;
      IF EXISTS (
        SELECT 1 FROM pg_roles
         WHERE rolname IN ('dataops_app_runtime','dataops_edge_evidence_owner')
           AND (rolcanlogin OR rolsuper OR rolcreatedb OR rolcreaterole OR rolreplication)
      ) THEN
        RAISE EXCEPTION 'edge database roles violate the NOLOGIN least-privilege contract';
      END IF;
    END $$;
    CREATE OR REPLACE FUNCTION public.edge_hosts_valid(hosts TEXT[], require_nonempty BOOLEAN)
    RETURNS BOOLEAN LANGUAGE plpgsql IMMUTABLE STRICT AS $$
    DECLARE host TEXT;
    BEGIN
      IF cardinality(hosts)>32 OR (require_nonempty AND cardinality(hosts)=0) OR
         cardinality(hosts)<>(SELECT count(DISTINCT item) FROM unnest(hosts) item) THEN
        RETURN FALSE;
      END IF;
      FOREACH host IN ARRAY hosts LOOP
        IF host<>lower(host) OR host !~ '^[a-z0-9]([a-z0-9.-]*[a-z0-9])?$' OR
           length(host)>253 OR host LIKE '%..%' OR host LIKE '%:%' OR
           host LIKE '%/%' OR host LIKE '%@%' THEN
          RETURN FALSE;
        END IF;
      END LOOP;
      RETURN TRUE;
    END $$;
    CREATE TABLE public.edge_gateway_enrollments (
      uid UUID PRIMARY KEY, gateway_id UUID NOT NULL UNIQUE,
      gateway_name VARCHAR(200) NOT NULL, environment VARCHAR(20) NOT NULL,
      network_zone VARCHAR(120) NOT NULL, policy_digest CHAR(64) NOT NULL,
      expected_certificate_sha256 CHAR(64) NOT NULL,
      allowed_control_hosts TEXT[] NOT NULL, allowed_proxy_hosts TEXT[] NOT NULL,
      token_hash CHAR(64) NOT NULL UNIQUE, status VARCHAR(20) NOT NULL DEFAULT 'pending',
      expires_at TIMESTAMPTZ NOT NULL, consumed_at TIMESTAMPTZ,
      created_by UUID NOT NULL, created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
      CONSTRAINT ck_edge_enrollment_environment CHECK (environment IN ('development','staging','production')),
      CONSTRAINT ck_edge_enrollment_status CHECK (status IN ('pending','consumed','revoked','expired')),
      CONSTRAINT ck_edge_enrollment_digest CHECK (
        policy_digest ~ '^[0-9a-f]{64}$' AND token_hash ~ '^[0-9a-f]{64}$' AND
        expected_certificate_sha256 ~ '^[0-9a-f]{64}$'),
      CONSTRAINT ck_edge_enrollment_ttl CHECK (expires_at > created_at AND expires_at <= created_at + INTERVAL '24 hours'),
      CONSTRAINT ck_edge_enrollment_consumed CHECK ((status='consumed') = (consumed_at IS NOT NULL)),
      CONSTRAINT ck_edge_enrollment_hosts CHECK (
        public.edge_hosts_valid(allowed_control_hosts,TRUE) AND
        public.edge_hosts_valid(allowed_proxy_hosts,FALSE))
    );
    CREATE INDEX ix_edge_enrollment_pending ON public.edge_gateway_enrollments(token_hash,expires_at) WHERE status='pending';

    CREATE OR REPLACE FUNCTION public.protect_edge_enrollment_identity()
    RETURNS trigger LANGUAGE plpgsql AS $$
    BEGIN
      IF NEW.uid<>OLD.uid OR NEW.gateway_id<>OLD.gateway_id OR
         NEW.gateway_name<>OLD.gateway_name OR NEW.environment<>OLD.environment OR
         NEW.network_zone<>OLD.network_zone OR NEW.policy_digest<>OLD.policy_digest OR
         NEW.expected_certificate_sha256<>OLD.expected_certificate_sha256 OR
         NEW.allowed_control_hosts<>OLD.allowed_control_hosts OR
         NEW.allowed_proxy_hosts<>OLD.allowed_proxy_hosts OR NEW.token_hash<>OLD.token_hash OR
         NEW.expires_at<>OLD.expires_at OR NEW.created_by<>OLD.created_by OR
         NEW.created_at<>OLD.created_at THEN
        RAISE EXCEPTION 'edge enrollment identity binding is immutable';
      END IF;
      IF NOT (
        (OLD.status='pending' AND NEW.status IN ('pending','consumed','revoked','expired')) OR
        (OLD.status<>'pending' AND NEW.status=OLD.status)
      ) THEN
        RAISE EXCEPTION 'edge enrollment status transition is invalid';
      END IF;
      RETURN NEW;
    END $$;
    CREATE TRIGGER trg_edge_enrollment_identity_immutable
      BEFORE UPDATE ON public.edge_gateway_enrollments FOR EACH ROW
      EXECUTE FUNCTION public.protect_edge_enrollment_identity();

    CREATE TABLE public.edge_gateways (
      uid UUID PRIMARY KEY, name VARCHAR(200) NOT NULL, environment VARCHAR(20) NOT NULL,
      network_zone VARCHAR(120) NOT NULL, status VARCHAR(20) NOT NULL,
      current_generation INTEGER NOT NULL, policy_digest CHAR(64) NOT NULL,
      allowed_control_hosts TEXT[] NOT NULL, allowed_proxy_hosts TEXT[] NOT NULL,
      last_heartbeat_at TIMESTAMPTZ, runtime_version VARCHAR(80),
      safe_health_summary JSONB NOT NULL DEFAULT '{}'::jsonb,
      created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
      updated_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP, revoked_at TIMESTAMPTZ,
      CONSTRAINT ck_edge_gateway_environment CHECK (environment IN ('development','staging','production')),
      CONSTRAINT ck_edge_gateway_status CHECK (status IN ('pending','active','offline','revoked')),
      CONSTRAINT ck_edge_gateway_generation CHECK (current_generation > 0),
      CONSTRAINT ck_edge_gateway_policy CHECK (policy_digest ~ '^[0-9a-f]{64}$'),
      CONSTRAINT ck_edge_gateway_summary CHECK (jsonb_typeof(safe_health_summary)='object'),
      CONSTRAINT ck_edge_gateway_hosts CHECK (
        public.edge_hosts_valid(allowed_control_hosts,TRUE) AND
        public.edge_hosts_valid(allowed_proxy_hosts,FALSE)),
      CONSTRAINT fk_edge_gateway_enrollment FOREIGN KEY (uid)
        REFERENCES public.edge_gateway_enrollments(gateway_id)
    );
    CREATE INDEX ix_edge_gateways_environment_status ON public.edge_gateways(environment,status);

    CREATE TABLE public.edge_gateway_credentials (
      uid UUID PRIMARY KEY, gateway_id UUID NOT NULL REFERENCES public.edge_gateways(uid),
      generation INTEGER NOT NULL, credential_hash CHAR(64) NOT NULL UNIQUE,
      certificate_sha256 CHAR(64) NOT NULL, status VARCHAR(20) NOT NULL,
      expires_at TIMESTAMPTZ NOT NULL, created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
      revoked_at TIMESTAMPTZ, rotated_from_uid UUID REFERENCES public.edge_gateway_credentials(uid),
      CONSTRAINT uq_edge_credential_generation UNIQUE(gateway_id,generation),
      CONSTRAINT ck_edge_credential_status CHECK (status IN ('active','rotated','revoked','expired')),
      CONSTRAINT ck_edge_credential_generation CHECK (generation > 0),
      CONSTRAINT ck_edge_credential_hashes CHECK (credential_hash ~ '^[0-9a-f]{64}$' AND certificate_sha256 ~ '^[0-9a-f]{64}$'),
      CONSTRAINT ck_edge_credential_expiry CHECK (expires_at > created_at AND expires_at <= created_at + INTERVAL '370 days')
    );
    CREATE INDEX ix_edge_credential_auth ON public.edge_gateway_credentials(credential_hash,status,expires_at);
    CREATE UNIQUE INDEX uq_edge_one_active_credential
      ON public.edge_gateway_credentials(gateway_id) WHERE status='active';
    CREATE TABLE public.edge_gateway_rotation_requests (
      gateway_id UUID NOT NULL REFERENCES public.edge_gateways(uid),
      request_id VARCHAR(255) NOT NULL, request_digest CHAR(64) NOT NULL,
      generation INTEGER NOT NULL, created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
      PRIMARY KEY(gateway_id,request_id),
      CONSTRAINT ck_edge_rotation_digest CHECK (request_digest ~ '^[0-9a-f]{64}$')
    );

    CREATE OR REPLACE FUNCTION public.protect_edge_gateway_identity()
    RETURNS trigger LANGUAGE plpgsql AS $$
    BEGIN
      IF NEW.uid<>OLD.uid OR NEW.name<>OLD.name OR NEW.environment<>OLD.environment OR
         NEW.network_zone<>OLD.network_zone OR NEW.policy_digest<>OLD.policy_digest OR
         NEW.allowed_control_hosts<>OLD.allowed_control_hosts OR
         NEW.allowed_proxy_hosts<>OLD.allowed_proxy_hosts THEN
        RAISE EXCEPTION 'registered edge gateway identity binding is immutable';
      END IF;
      IF OLD.status='revoked' AND NEW.status<>'revoked' THEN
        RAISE EXCEPTION 'revoked edge gateway is terminal';
      END IF;
      IF NEW.current_generation<OLD.current_generation OR
         NEW.current_generation>OLD.current_generation+1 THEN
        RAISE EXCEPTION 'edge gateway generation transition is invalid';
      END IF;
      IF NOT (
        (OLD.status='pending' AND NEW.status IN ('pending','active','revoked')) OR
        (OLD.status='active' AND NEW.status IN ('active','offline','revoked')) OR
        (OLD.status='offline' AND NEW.status IN ('active','offline','revoked')) OR
        (OLD.status='revoked' AND NEW.status='revoked')
      ) THEN
        RAISE EXCEPTION 'edge gateway status transition is invalid';
      END IF;
      RETURN NEW;
    END $$;
    CREATE TRIGGER trg_edge_gateway_identity_immutable
      BEFORE UPDATE ON public.edge_gateways FOR EACH ROW
      EXECUTE FUNCTION public.protect_edge_gateway_identity();

    CREATE OR REPLACE FUNCTION public.protect_edge_credential_identity()
    RETURNS trigger LANGUAGE plpgsql AS $$
    BEGIN
      IF NEW.uid<>OLD.uid OR NEW.gateway_id<>OLD.gateway_id OR
         NEW.generation<>OLD.generation OR NEW.credential_hash<>OLD.credential_hash OR
         NEW.certificate_sha256<>OLD.certificate_sha256 OR
         NEW.created_at<>OLD.created_at OR NEW.expires_at<>OLD.expires_at THEN
        RAISE EXCEPTION 'edge credential identity binding is immutable';
      END IF;
      IF NOT (
        (OLD.status='active' AND NEW.status IN ('active','rotated','revoked','expired')) OR
        (OLD.status<>'active' AND NEW.status=OLD.status)
      ) THEN
        RAISE EXCEPTION 'edge credential status transition is invalid';
      END IF;
      RETURN NEW;
    END $$;
    CREATE TRIGGER trg_edge_credential_identity_immutable
      BEFORE UPDATE ON public.edge_gateway_credentials FOR EACH ROW
      EXECUTE FUNCTION public.protect_edge_credential_identity();

    CREATE OR REPLACE FUNCTION public.validate_edge_gateway_binding(p_gateway UUID)
    RETURNS void LANGUAGE plpgsql AS $$
    DECLARE g public.edge_gateways%ROWTYPE;
    BEGIN
      SELECT * INTO g FROM public.edge_gateways WHERE uid=p_gateway;
      IF NOT FOUND OR g.status='revoked' THEN RETURN; END IF;
      IF g.status IN ('active','offline') AND NOT EXISTS (
        SELECT 1 FROM public.edge_gateway_credentials c
         WHERE c.gateway_id=g.uid AND c.generation=g.current_generation
           AND c.status='active' AND c.expires_at>CURRENT_TIMESTAMP
      ) THEN
        RAISE EXCEPTION 'active edge gateway requires its exact active credential binding';
      END IF;
      IF g.status IN ('active','offline') AND NOT EXISTS (
        SELECT 1 FROM public.edge_gateway_enrollments e
        JOIN public.edge_gateway_credentials c ON c.gateway_id=g.uid AND c.generation=1
         WHERE e.gateway_id=g.uid AND e.status='consumed'
           AND e.gateway_name=g.name AND e.environment=g.environment
           AND e.network_zone=g.network_zone AND e.policy_digest=g.policy_digest
           AND e.allowed_control_hosts=g.allowed_control_hosts
           AND e.allowed_proxy_hosts=g.allowed_proxy_hosts
           AND e.expected_certificate_sha256=c.certificate_sha256
      ) THEN
        RAISE EXCEPTION 'active edge gateway does not match its consumed enrollment binding';
      END IF;
      IF EXISTS (
        SELECT 1 FROM public.edge_gateway_credentials c
        JOIN public.edge_gateway_enrollments e ON e.gateway_id=c.gateway_id
         WHERE c.gateway_id=g.uid AND c.generation=1
           AND c.certificate_sha256<>e.expected_certificate_sha256
      ) THEN
        RAISE EXCEPTION 'initial edge certificate does not match enrollment binding';
      END IF;
    END $$;
    CREATE OR REPLACE FUNCTION public.enforce_edge_gateway_binding()
    RETURNS trigger LANGUAGE plpgsql AS $$
    BEGIN
      IF TG_OP='UPDATE' AND OLD.status IN ('active','offline') AND NEW.status='pending' THEN
        RAISE EXCEPTION 'registered edge gateway cannot return to pending status';
      END IF;
      PERFORM public.validate_edge_gateway_binding(COALESCE(NEW.uid,OLD.uid));
      RETURN NULL;
    END $$;
    CREATE CONSTRAINT TRIGGER trg_edge_gateway_binding
      AFTER INSERT OR UPDATE ON public.edge_gateways DEFERRABLE INITIALLY DEFERRED
      FOR EACH ROW EXECUTE FUNCTION public.enforce_edge_gateway_binding();
    CREATE OR REPLACE FUNCTION public.enforce_edge_credential_binding()
    RETURNS trigger LANGUAGE plpgsql AS $$
    DECLARE target UUID;
    BEGIN
      target := COALESCE(NEW.gateway_id,OLD.gateway_id);
      PERFORM public.validate_edge_gateway_binding(target);
      IF TG_OP <> 'DELETE' AND NEW.status='active' AND NOT EXISTS (
        SELECT 1 FROM public.edge_gateways g WHERE g.uid=NEW.gateway_id
         AND g.status IN ('active','offline') AND g.current_generation=NEW.generation
      ) THEN
        RAISE EXCEPTION 'active credential must be the exact current gateway binding';
      END IF;
      RETURN NULL;
    END $$;
    CREATE CONSTRAINT TRIGGER trg_edge_credential_binding
      AFTER INSERT OR UPDATE OR DELETE ON public.edge_gateway_credentials
      DEFERRABLE INITIALLY DEFERRED FOR EACH ROW
      EXECUTE FUNCTION public.enforce_edge_credential_binding();

    CREATE TABLE public.edge_gateway_tasks (
      task_id VARCHAR(255) PRIMARY KEY, gateway_id UUID NOT NULL REFERENCES public.edge_gateways(uid),
      idempotency_key VARCHAR(255) NOT NULL, contract JSONB NOT NULL, contract_digest CHAR(64) NOT NULL,
      status VARCHAR(24) NOT NULL, lease_token_hash CHAR(64), lease_expires_at TIMESTAMPTZ,
      cancel_requested_at TIMESTAMPTZ, result_summary JSONB,
      created_by UUID NOT NULL, created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
      updated_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
      CONSTRAINT uq_edge_task_idempotency UNIQUE(gateway_id,idempotency_key),
      CONSTRAINT uq_edge_task_gateway UNIQUE(task_id,gateway_id),
      CONSTRAINT ck_edge_task_status CHECK (status IN ('pending','leased','cancel_requested','completed','failed','cancelled')),
      CONSTRAINT ck_edge_task_contract CHECK (
        jsonb_typeof(contract)='object' AND contract_digest ~ '^[0-9a-f]{64}$' AND
        contract->>'task_id'=task_id AND contract->>'gateway_id'=gateway_id::text AND
        contract->>'idempotency_key'=idempotency_key),
      CONSTRAINT ck_edge_task_lease CHECK (
        (status='leased' AND lease_token_hash IS NOT NULL AND lease_expires_at IS NOT NULL) OR
        (status='cancel_requested' AND ((lease_token_hash IS NULL AND lease_expires_at IS NULL) OR
          (lease_token_hash IS NOT NULL AND lease_expires_at IS NOT NULL))) OR
        (status NOT IN ('leased','cancel_requested') AND lease_token_hash IS NULL AND lease_expires_at IS NULL)),
      CONSTRAINT ck_edge_task_result CHECK (result_summary IS NULL OR jsonb_typeof(result_summary)='object')
    );
    CREATE INDEX ix_edge_task_pull ON public.edge_gateway_tasks(gateway_id,status,created_at);

    CREATE TABLE public.edge_gateway_events (
      event_id VARCHAR(80) PRIMARY KEY, gateway_id UUID NOT NULL REFERENCES public.edge_gateways(uid),
      task_id VARCHAR(255) NOT NULL,
      classification VARCHAR(40) NOT NULL, contract JSONB NOT NULL, contract_digest CHAR(64) NOT NULL,
      ack JSONB NOT NULL, received_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
      CONSTRAINT ck_edge_event_classification CHECK (classification IN ('desensitized_metadata','statistics','lineage','evidence','health_summary','diagnostic_summary')),
      CONSTRAINT fk_edge_event_task_gateway FOREIGN KEY (task_id,gateway_id)
        REFERENCES public.edge_gateway_tasks(task_id,gateway_id),
      CONSTRAINT ck_edge_event_contract CHECK (
        jsonb_typeof(contract)='object' AND jsonb_typeof(ack)='object' AND
        contract_digest ~ '^[0-9a-f]{64}$' AND contract->>'event_id'=event_id AND
        contract->>'task_id'=task_id AND contract->>'gateway_id'=gateway_id::text AND
        contract->>'classification'=classification)
    );
    CREATE INDEX ix_edge_event_gateway_received ON public.edge_gateway_events(gateway_id,received_at DESC);

    CREATE TABLE public.edge_gateway_releases (
      uid UUID PRIMARY KEY, gateway_id UUID NOT NULL REFERENCES public.edge_gateways(uid),
      version VARCHAR(80) NOT NULL, artifact_digest CHAR(64) NOT NULL,
      rollback_version VARCHAR(80) NOT NULL, status VARCHAR(24) NOT NULL,
      request_id VARCHAR(255) NOT NULL, request_digest CHAR(64) NOT NULL,
      safe_summary JSONB NOT NULL DEFAULT '{}'::jsonb, created_by UUID NOT NULL,
      created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP, acknowledged_at TIMESTAMPTZ,
      CONSTRAINT uq_edge_release_version UNIQUE(gateway_id,version),
      CONSTRAINT uq_edge_release_request UNIQUE(gateway_id,request_id),
      CONSTRAINT ck_edge_release_status CHECK (status IN ('offered','accepted','installed','failed','rolled_back')),
      CONSTRAINT ck_edge_release_digest CHECK (artifact_digest ~ '^[0-9a-f]{64}$' AND request_digest ~ '^[0-9a-f]{64}$'),
      CONSTRAINT ck_edge_release_summary CHECK (jsonb_typeof(safe_summary)='object')
    );

    CREATE TABLE public.edge_gateway_audits (
      uid UUID PRIMARY KEY, gateway_id UUID REFERENCES public.edge_gateways(uid),
      event_type VARCHAR(80) NOT NULL, actor_uid UUID, success BOOLEAN NOT NULL,
      safe_detail JSONB NOT NULL, created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
      CONSTRAINT ck_edge_audit_detail CHECK (jsonb_typeof(safe_detail)='object')
    );
    CREATE INDEX ix_edge_audit_gateway_created ON public.edge_gateway_audits(gateway_id,created_at DESC);
    CREATE TABLE public.edge_gateway_failure_windows (
      failure_fingerprint CHAR(64) NOT NULL, window_started_at TIMESTAMPTZ NOT NULL,
      gateway_id UUID REFERENCES public.edge_gateways(uid), event_type VARCHAR(80) NOT NULL,
      occurrence_count INTEGER NOT NULL, first_seen_at TIMESTAMPTZ NOT NULL,
      last_seen_at TIMESTAMPTZ NOT NULL,
      PRIMARY KEY(failure_fingerprint,window_started_at),
      CONSTRAINT ck_edge_failure_count CHECK (occurrence_count BETWEEN 1 AND 1000000)
    );
    ALTER TABLE public.edge_gateway_audits ADD COLUMN failure_fingerprint CHAR(64);
    ALTER TABLE public.edge_gateway_audits ADD COLUMN window_started_at TIMESTAMPTZ;
    CREATE UNIQUE INDEX uq_edge_failure_audit_window
      ON public.edge_gateway_audits(failure_fingerprint,window_started_at)
      WHERE success=FALSE;
    CREATE OR REPLACE FUNCTION public.protect_edge_audit_append_only()
    RETURNS trigger LANGUAGE plpgsql AS $$
    BEGIN
      IF TG_OP='INSERT' AND current_user='dataops_edge_evidence_owner' THEN
        RETURN NEW;
      END IF;
      RAISE EXCEPTION 'edge gateway audit ledger is owner-written and append-only';
    END $$;
    CREATE TRIGGER trg_edge_audit_append_only
      BEFORE INSERT OR UPDATE OR DELETE ON public.edge_gateway_audits
      FOR EACH ROW EXECUTE FUNCTION public.protect_edge_audit_append_only();
    CREATE TRIGGER trg_edge_audit_no_truncate
      BEFORE TRUNCATE ON public.edge_gateway_audits
      FOR EACH STATEMENT EXECUTE FUNCTION public.protect_edge_audit_append_only();
    CREATE OR REPLACE FUNCTION public.protect_edge_failure_window()
    RETURNS trigger LANGUAGE plpgsql AS $$
    BEGIN
      IF TG_OP IN ('DELETE','TRUNCATE') THEN
        RAISE EXCEPTION 'edge gateway failure evidence cannot be removed';
      END IF;
      IF current_user<>'dataops_edge_evidence_owner' THEN
        RAISE EXCEPTION 'edge gateway failure evidence requires the controlled writer';
      END IF;
      IF TG_OP='INSERT' THEN RETURN NEW; END IF;
      IF NEW.failure_fingerprint<>OLD.failure_fingerprint OR
         NEW.window_started_at<>OLD.window_started_at OR
         NEW.gateway_id IS DISTINCT FROM OLD.gateway_id OR
         NEW.event_type<>OLD.event_type OR NEW.first_seen_at<>OLD.first_seen_at THEN
        RAISE EXCEPTION 'edge gateway failure window identity is immutable';
      END IF;
      IF NEW.occurrence_count<>OLD.occurrence_count+1 OR
         NEW.last_seen_at<OLD.last_seen_at OR NEW.last_seen_at<NEW.first_seen_at THEN
        RAISE EXCEPTION 'edge gateway failure window update is invalid';
      END IF;
      RETURN NEW;
    END $$;
    CREATE TRIGGER trg_edge_failure_window_protected
      BEFORE INSERT OR UPDATE OR DELETE ON public.edge_gateway_failure_windows
      FOR EACH ROW EXECUTE FUNCTION public.protect_edge_failure_window();
    CREATE TRIGGER trg_edge_failure_window_no_truncate
      BEFORE TRUNCATE ON public.edge_gateway_failure_windows
      FOR EACH STATEMENT EXECUTE FUNCTION public.protect_edge_failure_window();
    CREATE OR REPLACE FUNCTION public.record_edge_gateway_audit(
      p_uid UUID,p_gateway UUID,p_event_type VARCHAR(80),p_actor UUID,
      p_success BOOLEAN,p_detail JSONB
    ) RETURNS void LANGUAGE plpgsql SECURITY DEFINER
      SET search_path=pg_catalog AS $$
    BEGIN
      IF p_uid IS NULL OR p_event_type NOT IN (
        'enrollment_created','gateway_registered','credential_rotated','gateway_revoked',
        'heartbeat_accepted','task_issued','task_cancel_requested','task_outcome_recorded',
        'event_accepted','release_offered','release_acknowledged'
      ) OR jsonb_typeof(p_detail)<>'object' OR octet_length(p_detail::text)>8192 OR
         (p_gateway IS NOT NULL AND NOT EXISTS (
           SELECT 1 FROM public.edge_gateways WHERE uid=p_gateway
         )) THEN
        RAISE EXCEPTION 'edge gateway audit evidence is invalid';
      END IF;
      INSERT INTO public.edge_gateway_audits
        (uid,gateway_id,event_type,actor_uid,success,safe_detail)
      VALUES (p_uid,p_gateway,p_event_type,p_actor,p_success,p_detail);
    END $$;
    CREATE OR REPLACE FUNCTION public.record_edge_gateway_failure(
      p_audit_uid UUID,p_fingerprint CHAR(64),p_gateway UUID,
      p_event_type VARCHAR(80),p_reason_code VARCHAR(64)
    ) RETURNS TABLE(
      failure_fingerprint CHAR(64), window_started_at TIMESTAMPTZ,
      occurrence_count INTEGER, gateway_id UUID, event_type VARCHAR(80)
    ) LANGUAGE plpgsql SECURITY DEFINER
      SET search_path=pg_catalog AS $$
    DECLARE
      target_fingerprint CHAR(64) := p_fingerprint;
      target_gateway UUID := p_gateway;
      target_event_type VARCHAR(80) := p_event_type;
      target_reason_code VARCHAR(64) := p_reason_code;
      target_window TIMESTAMPTZ := date_trunc('minute',CURRENT_TIMESTAMP);
      total_windows INTEGER;
    BEGIN
      IF p_audit_uid IS NULL OR p_fingerprint !~ '^[0-9a-f]{64}$' OR
         p_event_type NOT IN (
           'untrusted_request_rejected','policy_rejected','binding_rejected',
           'idempotency_conflict','release_rejected'
         ) OR p_reason_code !~ '^[a-z][a-z0-9_]{0,63}$' OR
         (p_gateway IS NOT NULL AND NOT EXISTS (
           SELECT 1 FROM public.edge_gateways WHERE uid=p_gateway
         )) THEN
        RAISE EXCEPTION 'edge gateway failure evidence is invalid';
      END IF;
      PERFORM pg_advisory_xact_lock(4774096);
      SELECT count(*) INTO total_windows FROM public.edge_gateway_failure_windows;
      IF NOT EXISTS (
        SELECT 1 FROM public.edge_gateway_failure_windows AS existing
         WHERE existing.failure_fingerprint=target_fingerprint
           AND existing.window_started_at=target_window
      ) AND total_windows>=4095 THEN
        target_fingerprint := '16b5bae3613ee26975528e149d401a46f1d659405cfe0148262caec288dfb365';
        target_gateway := NULL;
        target_event_type := 'untrusted_request_rejected';
        target_reason_code := 'capacity_overflow';
        target_window := '1970-01-01 00:00:00+00'::timestamptz;
      END IF;
      RETURN QUERY
        INSERT INTO public.edge_gateway_failure_windows AS windows
          (failure_fingerprint,window_started_at,gateway_id,event_type,
           occurrence_count,first_seen_at,last_seen_at)
        VALUES (target_fingerprint,target_window,target_gateway,target_event_type,
                1,CURRENT_TIMESTAMP,CURRENT_TIMESTAMP)
        ON CONFLICT ON CONSTRAINT edge_gateway_failure_windows_pkey DO UPDATE
          SET occurrence_count=windows.occurrence_count+1,
              last_seen_at=CURRENT_TIMESTAMP
        RETURNING windows.failure_fingerprint,windows.window_started_at,
                  windows.occurrence_count,windows.gateway_id,windows.event_type;
      IF NOT EXISTS (
        SELECT 1 FROM public.edge_gateway_audits AS existing_audit
         WHERE existing_audit.failure_fingerprint=target_fingerprint
           AND existing_audit.window_started_at=target_window
           AND existing_audit.success=FALSE
      ) THEN
        INSERT INTO public.edge_gateway_audits
          (uid,gateway_id,event_type,actor_uid,success,safe_detail,
           failure_fingerprint,window_started_at)
        VALUES (p_audit_uid,target_gateway,target_event_type,NULL,FALSE,
                jsonb_build_object('reason_code',target_reason_code),
                target_fingerprint,target_window);
      END IF;
    END $$;
    ALTER TABLE public.edge_gateway_audits OWNER TO dataops_edge_evidence_owner;
    ALTER TABLE public.edge_gateway_failure_windows OWNER TO dataops_edge_evidence_owner;
    ALTER FUNCTION public.protect_edge_audit_append_only() OWNER TO dataops_edge_evidence_owner;
    ALTER FUNCTION public.protect_edge_failure_window() OWNER TO dataops_edge_evidence_owner;
    ALTER FUNCTION public.record_edge_gateway_audit(UUID,UUID,VARCHAR(80),UUID,BOOLEAN,JSONB)
      OWNER TO dataops_edge_evidence_owner;
    ALTER FUNCTION public.record_edge_gateway_failure(UUID,CHAR(64),UUID,VARCHAR(80),VARCHAR(64))
      OWNER TO dataops_edge_evidence_owner;
    GRANT SELECT ON public.edge_gateways TO dataops_edge_evidence_owner;
    REVOKE ALL ON public.edge_gateway_audits,public.edge_gateway_failure_windows FROM PUBLIC;
    REVOKE ALL ON FUNCTION public.record_edge_gateway_audit(UUID,UUID,VARCHAR(80),UUID,BOOLEAN,JSONB) FROM PUBLIC;
    REVOKE ALL ON FUNCTION public.record_edge_gateway_failure(UUID,CHAR(64),UUID,VARCHAR(80),VARCHAR(64)) FROM PUBLIC;
    GRANT USAGE ON SCHEMA public TO dataops_app_runtime;
    GRANT SELECT,INSERT,UPDATE,DELETE ON
      public.edge_gateway_enrollments,public.edge_gateways,
      public.edge_gateway_credentials,public.edge_gateway_rotation_requests,
      public.edge_gateway_tasks,public.edge_gateway_events,public.edge_gateway_releases
      TO dataops_app_runtime;
    ALTER DEFAULT PRIVILEGES IN SCHEMA public
      GRANT SELECT,INSERT,UPDATE,DELETE ON TABLES TO dataops_app_runtime;
    ALTER DEFAULT PRIVILEGES IN SCHEMA public
      GRANT USAGE,SELECT,UPDATE ON SEQUENCES TO dataops_app_runtime;
    REVOKE INSERT,UPDATE,DELETE,TRUNCATE ON
      public.edge_gateway_audits,public.edge_gateway_failure_windows
      FROM dataops_app_runtime;
    GRANT SELECT ON public.edge_gateway_audits,public.edge_gateway_failure_windows
      TO dataops_app_runtime;
    GRANT EXECUTE ON FUNCTION public.record_edge_gateway_audit(UUID,UUID,VARCHAR(80),UUID,BOOLEAN,JSONB)
      TO dataops_app_runtime;
    GRANT EXECUTE ON FUNCTION public.record_edge_gateway_failure(UUID,CHAR(64),UUID,VARCHAR(80),VARCHAR(64))
      TO dataops_app_runtime;
    """)


def downgrade() -> None:
    op.execute("""
    DO $$
    DECLARE table_name TEXT; has_data BOOLEAN;
    BEGIN
      FOREACH table_name IN ARRAY ARRAY[
        'edge_gateway_enrollments','edge_gateways','edge_gateway_credentials',
        'edge_gateway_tasks','edge_gateway_events','edge_gateway_releases',
        'edge_gateway_audits','edge_gateway_failure_windows','edge_gateway_rotation_requests'
      ] LOOP
        IF to_regclass('public.' || table_name) IS NOT NULL THEN
          EXECUTE format('SELECT EXISTS (SELECT 1 FROM public.%I)',table_name) INTO has_data;
          IF has_data THEN
            RAISE EXCEPTION 'edge gateway downgrade rejected: clear Task2 business and audit data first';
          END IF;
        END IF;
      END LOOP;
    END $$;
    DROP TRIGGER IF EXISTS trg_edge_audit_append_only ON public.edge_gateway_audits;
    DROP TRIGGER IF EXISTS trg_edge_audit_no_truncate ON public.edge_gateway_audits;
    DROP TRIGGER IF EXISTS trg_edge_failure_window_protected ON public.edge_gateway_failure_windows;
    DROP TRIGGER IF EXISTS trg_edge_failure_window_no_truncate ON public.edge_gateway_failure_windows;
    DROP FUNCTION IF EXISTS public.record_edge_gateway_failure(UUID,CHAR(64),UUID,VARCHAR(80),VARCHAR(64));
    DROP FUNCTION IF EXISTS public.record_edge_gateway_audit(UUID,UUID,VARCHAR(80),UUID,BOOLEAN,JSONB);
    DROP FUNCTION IF EXISTS public.protect_edge_failure_window();
    DROP FUNCTION IF EXISTS public.protect_edge_audit_append_only();
    DROP TABLE IF EXISTS public.edge_gateway_failure_windows;
    DROP TABLE IF EXISTS public.edge_gateway_audits;
    DROP TABLE IF EXISTS public.edge_gateway_events;
    DROP TABLE IF EXISTS public.edge_gateway_releases;
    DROP TABLE IF EXISTS public.edge_gateway_tasks;
    DROP TABLE IF EXISTS public.edge_gateway_rotation_requests;
    DROP TRIGGER IF EXISTS trg_edge_credential_identity_immutable ON public.edge_gateway_credentials;
    DROP TRIGGER IF EXISTS trg_edge_gateway_identity_immutable ON public.edge_gateways;
    DROP TRIGGER IF EXISTS trg_edge_credential_binding ON public.edge_gateway_credentials;
    DROP TRIGGER IF EXISTS trg_edge_gateway_binding ON public.edge_gateways;
    DROP FUNCTION IF EXISTS public.enforce_edge_credential_binding();
    DROP FUNCTION IF EXISTS public.enforce_edge_gateway_binding();
    DROP FUNCTION IF EXISTS public.validate_edge_gateway_binding(UUID);
    DROP FUNCTION IF EXISTS public.protect_edge_credential_identity();
    DROP FUNCTION IF EXISTS public.protect_edge_gateway_identity();
    DROP TABLE IF EXISTS public.edge_gateway_credentials;
    DROP TABLE IF EXISTS public.edge_gateways;
    DROP TRIGGER IF EXISTS trg_edge_enrollment_identity_immutable ON public.edge_gateway_enrollments;
    DROP TABLE IF EXISTS public.edge_gateway_enrollments;
    DROP FUNCTION IF EXISTS public.protect_edge_enrollment_identity();
    DROP FUNCTION IF EXISTS public.edge_hosts_valid(TEXT[],BOOLEAN);
    """)