DataOps-platform/migrations/versions/20260818_558_plugin_review_generation_gate.py

"""Bind approvals to the current independent plugin review generation."""

from alembic import op

revision = "20260818_558"
down_revision = "20260818_557"
branch_labels = None
depends_on = None

_FACT_TABLES = (
    "plugin_registry_versions", "plugin_approvals", "plugin_runtime_leases",
    "plugin_runs", "plugin_dead_letters", "plugin_audit_outbox",
    "plugin_request_claims", "plugin_runtime_breakers", "plugin_recovery_claims",
)


def upgrade() -> None:
    op.get_bind().exec_driver_sql(r'''
    ALTER TABLE public.plugin_registry_versions
      ADD COLUMN submitter_ref text CHECK(submitter_ref ~ '^[A-Za-z0-9_.:-]{1,120}$'),
      ADD COLUMN review_generation bigint NOT NULL DEFAULT 0 CHECK(review_generation >= 0);
    UPDATE public.plugin_registry_versions r
      SET submitter_ref=(SELECT a.actor_ref FROM public.plugin_audit_outbox a
                         WHERE a.plugin_uid=r.plugin_uid AND a.version=r.version
                           AND a.event_type='registered'
                         ORDER BY a.created_at ASC LIMIT 1)
      WHERE submitter_ref IS NULL;
    ALTER TABLE public.plugin_approvals
      ADD COLUMN review_generation bigint NOT NULL DEFAULT 0 CHECK(review_generation >= 0),
      ADD COLUMN reviewed_at timestamptz;
    DO $drop_old_approval_unique$
    DECLARE constraint_name text;
    BEGIN
      SELECT c.conname INTO constraint_name
        FROM pg_constraint c
       WHERE c.conrelid='public.plugin_approvals'::regclass AND c.contype='u'
        AND (SELECT array_agg(a.attname::text ORDER BY k.ordinality)
                FROM unnest(c.conkey) WITH ORDINALITY AS k(attnum,ordinality)
                JOIN pg_attribute a ON a.attrelid=c.conrelid AND a.attnum=k.attnum)
             = ARRAY['plugin_uid','version','action_name','actor_ref','tenant_ref','domain_ref','manifest_digest']::text[];
      IF constraint_name IS NOT NULL THEN
        EXECUTE format('ALTER TABLE public.plugin_approvals DROP CONSTRAINT %%I', constraint_name);
      END IF;
    END $drop_old_approval_unique$;
    ALTER TABLE public.plugin_approvals ADD CONSTRAINT plugin_approvals_review_generation_unique UNIQUE(plugin_uid,version,action_name,actor_ref,tenant_ref,domain_ref,manifest_digest,review_generation);

    CREATE FUNCTION public.plugin_platform_control_v4(p jsonb) RETURNS jsonb
    LANGUAGE plpgsql SECURITY DEFINER SET search_path=pg_catalog,public AS $control$
    DECLARE r record; a record; result jsonb; expected_action text;
    BEGIN
      IF NOT pg_has_role(session_user,'dataops_plugin_platform_control','MEMBER') OR jsonb_typeof(p)<>'object' THEN RAISE EXCEPTION 'plugin_control_denied'; END IF;
      IF p->>'action'='register' THEN
        result:=public.plugin_platform_control_v3(p);
        UPDATE public.plugin_registry_versions
          SET submitter_ref=p->>'actor_ref',review_actor=NULL,reviewed_at=NULL,review_generation=0
          WHERE plugin_uid=p->>'plugin_uid' AND version=p->>'version';
        RETURN result;
      ELSIF p->>'action'='review' THEN
        SELECT * INTO r FROM public.plugin_registry_versions WHERE plugin_uid=p->>'plugin_uid' AND version=p->>'version' FOR UPDATE;
        IF NOT FOUND OR r.state<>'draft' OR r.submitter_ref IS NULL OR r.submitter_ref=p->>'actor_ref' THEN RAISE EXCEPTION 'plugin_review_independence_denied'; END IF;
        result:=public.plugin_platform_control_v3(p);
        UPDATE public.plugin_registry_versions SET review_generation=review_generation+1
          WHERE plugin_uid=p->>'plugin_uid' AND version=p->>'version';
        RETURN result;
      ELSIF p->>'action'='issue_approval' THEN
        IF p-ARRAY['action','plugin_uid','version','actor_ref','approval_action','tenant_ref','domain_ref','expires_in_seconds']<>'{}'::jsonb THEN RAISE EXCEPTION 'plugin_approval_closed'; END IF;
        SELECT * INTO r FROM public.plugin_registry_versions WHERE plugin_uid=p->>'plugin_uid' AND version=p->>'version' FOR UPDATE;
        IF NOT FOUND OR r.state<>'reviewed' OR r.submitter_ref IS NULL OR r.review_actor IS NULL OR r.submitter_ref=r.review_actor OR r.review_actor=p->>'actor_ref' THEN RAISE EXCEPTION 'plugin_approval_review_denied'; END IF;
        result:=public.plugin_platform_control_v3(p);
        UPDATE public.plugin_approvals SET review_generation=r.review_generation,reviewed_at=r.reviewed_at
          WHERE approval_uid=(result->>'approval_uid')::uuid;
        RETURN result;
      ELSIF p->>'action'='transition' THEN
        IF p-ARRAY['action','plugin_uid','version','actor_ref','tenant_ref','domain_ref','approval_uid','target_state','expected_fence','incident_uid']<>'{}'::jsonb THEN RAISE EXCEPTION 'plugin_transition_closed'; END IF;
        expected_action:=CASE p->>'target_state' WHEN 'approved' THEN 'approve' WHEN 'active' THEN 'activate' WHEN 'rolled_back' THEN 'rollback' ELSE p->>'target_state' END;
        SELECT * INTO r FROM public.plugin_registry_versions WHERE plugin_uid=p->>'plugin_uid' AND version=p->>'version' FOR UPDATE;
        SELECT * INTO a FROM public.plugin_approvals WHERE approval_uid=(p->>'approval_uid')::uuid FOR UPDATE;
        IF NOT FOUND OR r.submitter_ref IS NULL OR r.review_actor IS NULL OR r.submitter_ref=r.review_actor
          OR a.plugin_uid<>r.plugin_uid OR a.version<>r.version OR a.actor_ref<>p->>'actor_ref'
          OR a.reviewer_ref<>r.review_actor OR a.review_generation<>r.review_generation
          OR a.reviewed_at IS NULL OR r.reviewed_at IS NULL OR a.reviewed_at<>r.reviewed_at
          OR a.created_at<r.reviewed_at OR a.manifest_digest<>r.manifest_digest
          OR a.action_name<>expected_action OR a.tenant_ref<>p->>'tenant_ref' OR a.domain_ref<>p->>'domain_ref'
          OR a.consumed_at IS NOT NULL OR a.expires_at<=clock_timestamp() THEN RAISE EXCEPTION 'plugin_approval_review_binding_denied'; END IF;
        RETURN public.plugin_platform_control_v3(p);
      END IF;
      RETURN public.plugin_platform_control_v3(p);
    END $control$;
    ALTER FUNCTION public.plugin_platform_control_v4(jsonb) OWNER TO dataops_plugin_platform_owner;
    REVOKE ALL ON FUNCTION public.plugin_platform_control_v3(jsonb) FROM PUBLIC,dataops_app,dataops_app_runtime,dataops_plugin_platform_control;
    GRANT EXECUTE ON FUNCTION public.plugin_platform_control_v4(jsonb) TO dataops_plugin_platform_control;
    ''')


def downgrade() -> None:
    bind = op.get_bind()
    checks = " OR ".join(f"EXISTS(SELECT 1 FROM public.{table} LIMIT 1)" for table in _FACT_TABLES)
    if bind.exec_driver_sql(f"SELECT {checks}").scalar():
        raise RuntimeError("downgrade refused: review-bound WP13 facts are nonempty")
    bind.exec_driver_sql('''
    REVOKE ALL ON FUNCTION public.plugin_platform_control_v4(jsonb) FROM PUBLIC,dataops_app,dataops_app_runtime,dataops_plugin_platform_control;
    DROP FUNCTION public.plugin_platform_control_v4(jsonb);
    ALTER TABLE public.plugin_approvals DROP CONSTRAINT plugin_approvals_review_generation_unique;
    ALTER TABLE public.plugin_approvals DROP COLUMN reviewed_at,DROP COLUMN review_generation;
    ALTER TABLE public.plugin_approvals ADD CONSTRAINT plugin_approvals_plugin_uid_version_action_name_actor_ref_tenant_ref_domain_ref_manifest_digest_key UNIQUE(plugin_uid,version,action_name,actor_ref,tenant_ref,domain_ref,manifest_digest);
    ALTER TABLE public.plugin_registry_versions DROP COLUMN review_generation,DROP COLUMN submitter_ref;
    GRANT EXECUTE ON FUNCTION public.plugin_platform_control_v3(jsonb) TO dataops_plugin_platform_control;
    ''')