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

"""Create the local-only, gateway-bound P3-WP12 metering baseline."""

from alembic import op

revision = "20260818_546"
down_revision = "20260818_545"
branch_labels = None
depends_on = None


_TABLES = (
    "metering_scope_grants",
    "metering_events",
    "metering_allocation_rules",
    "metering_allocations",
    "metering_budgets",
    "metering_alert_outbox",
    "metering_reconciliation_reports",
    "metering_audit_events",
    "metering_runtime_claims",
    "metering_runtime_leases",
    "metering_showback_configuration",
)


def upgrade() -> None:
    op.get_bind().exec_driver_sql(
        r'''
        DO $$ BEGIN
          IF NOT EXISTS (SELECT 1 FROM pg_roles WHERE rolname='dataops_tenant_foundation_owner')
             OR NOT EXISTS (SELECT 1 FROM pg_roles WHERE rolname='dataops_app_runtime')
             OR NOT EXISTS (SELECT 1 FROM pg_roles WHERE rolname='dataops_bi_ai_catalog_control')
             OR NOT EXISTS (SELECT 1 FROM pg_roles WHERE rolname='dataops_bi_ai_catalog_issuer') THEN
            RAISE EXCEPTION 'WP12 roles must be provisioned by role-init';
          END IF;
        END $$;
        CREATE TABLE public.metering_scope_grants (
          grant_uid uuid PRIMARY KEY DEFAULT gen_random_uuid(),
          principal_ref text NOT NULL CHECK(principal_ref ~ '^[A-Za-z0-9_.:-]{1,120}$'),
          tenant_ref text NOT NULL CHECK(tenant_ref ~ '^[a-z][a-z0-9-]{0,62}$'),
          domain_ref text NOT NULL CHECK(domain_ref ~ '^[a-z][a-z0-9-]{0,62}$'),
          role_name text NOT NULL CHECK(role_name IN ('viewer','operator','admin')),
          active boolean NOT NULL DEFAULT true,
          created_at timestamptz NOT NULL DEFAULT clock_timestamp(),
          revoked_at timestamptz,
          UNIQUE(principal_ref,tenant_ref,domain_ref,role_name)
        );
        CREATE TABLE public.metering_showback_configuration (
          configuration_uid boolean PRIMARY KEY DEFAULT true CHECK(configuration_uid),
          schema_version integer NOT NULL DEFAULT 1 CHECK(schema_version=1),
          mode text NOT NULL DEFAULT 'ENGINEERING_EVIDENCE_ONLY' CHECK(mode='ENGINEERING_EVIDENCE_ONLY'),
          enterprise_cost_input text NOT NULL DEFAULT 'TBD_EXTERNAL' CHECK(enterprise_cost_input='TBD_EXTERNAL'),
          owner_ref text NOT NULL DEFAULT 'enterprise_finance_owner' CHECK(owner_ref='enterprise_finance_owner'),
          chargeback_enabled boolean NOT NULL DEFAULT false CHECK(chargeback_enabled=false),
          provider_enabled boolean NOT NULL DEFAULT false CHECK(provider_enabled=false),
          created_at timestamptz NOT NULL DEFAULT clock_timestamp()
        );
        INSERT INTO public.metering_showback_configuration DEFAULT VALUES;
        CREATE TABLE public.metering_runtime_leases (
          tenant_ref text PRIMARY KEY CHECK(tenant_ref ~ '^[a-z][a-z0-9-]{0,62}$'),
          lease_owner text, lease_token uuid, lease_fence bigint NOT NULL DEFAULT 0 CHECK(lease_fence>=0),
          lease_expires_at timestamptz, updated_at timestamptz NOT NULL DEFAULT clock_timestamp()
        );
        CREATE TABLE public.metering_runtime_claims (
          claim_uid uuid PRIMARY KEY DEFAULT gen_random_uuid(), action_name text NOT NULL CHECK(action_name IN ('lease','record','read')),
          principal_ref text NOT NULL, tenant_ref text NOT NULL, domain_ref text NOT NULL,
          request_digest char(64) NOT NULL CHECK(request_digest ~ '^[0-9a-f]{64}$'),
          expires_at timestamptz NOT NULL, consumed_at timestamptz, created_at timestamptz NOT NULL DEFAULT clock_timestamp()
        );
        CREATE TABLE public.metering_events (
          event_uid text PRIMARY KEY CHECK(event_uid ~ '^[A-Za-z0-9][A-Za-z0-9._:-]{0,119}$'),
          tenant_ref text NOT NULL CHECK(tenant_ref ~ '^[a-z][a-z0-9-]{0,62}$'),
          domain_ref text NOT NULL CHECK(domain_ref ~ '^[a-z][a-z0-9-]{0,62}$'),
          department_ref text NOT NULL CHECK(department_ref ~ '^[a-z][a-z0-9-]{0,62}$'),
          project_ref text NOT NULL CHECK(project_ref ~ '^[a-z][a-z0-9-]{0,62}$'),
          cost_center_ref text NOT NULL CHECK(cost_center_ref ~ '^[a-z][a-z0-9-]{0,62}$'),
          event_kind text NOT NULL CHECK(event_kind IN ('query','api','file','subscription','storage','compute','task','model_call')),
          occurred_at timestamptz NOT NULL, window_start timestamptz NOT NULL, window_end timestamptz NOT NULL CHECK(window_start<window_end),
          quantity_micros bigint NOT NULL CHECK(quantity_micros>=0), unit text NOT NULL CHECK(unit IN ('bytes','kb','mb','gb','seconds','milliseconds','requests','tasks','tokens')),
          idempotency_key text NOT NULL CHECK(idempotency_key ~ '^[A-Za-z0-9][A-Za-z0-9._:-]{0,119}$'),
          evidence_digest char(64) NOT NULL CHECK(evidence_digest ~ '^[0-9a-f]{64}$'),
          evidence_reference text NOT NULL CHECK(evidence_reference ~ '^local-fixture://wp12/v1(/[A-Za-z0-9._:-]{1,80})?$'),
          request_digest char(64) NOT NULL CHECK(request_digest ~ '^[0-9a-f]{64}$'),
          correction_of text REFERENCES public.metering_events(event_uid) ON DELETE RESTRICT,
          lease_fence bigint NOT NULL CHECK(lease_fence>=0), created_at timestamptz NOT NULL DEFAULT clock_timestamp(),
          UNIQUE(tenant_ref,idempotency_key)
        );
        CREATE TABLE public.metering_allocation_rules (
          rule_uid text NOT NULL CHECK(rule_uid ~ '^[A-Za-z0-9][A-Za-z0-9._:-]{0,119}$'), rule_version integer NOT NULL CHECK(rule_version>0),
          tenant_ref text NOT NULL, domain_ref text NOT NULL, department_ref text NOT NULL, project_ref text NOT NULL, cost_center_ref text NOT NULL,
          effective_start timestamptz NOT NULL, effective_end timestamptz NOT NULL CHECK(effective_start<effective_end),
          rule_digest char(64) NOT NULL CHECK(rule_digest ~ '^[0-9a-f]{64}$'), created_at timestamptz NOT NULL DEFAULT clock_timestamp(),
          PRIMARY KEY(rule_uid,rule_version)
        );
        CREATE TABLE public.metering_allocations (
          rule_uid text NOT NULL, rule_version integer NOT NULL, target_ref text NOT NULL CHECK(target_ref ~ '^[a-z][a-z0-9-]{0,62}$'),
          weight_micros integer NOT NULL CHECK(weight_micros>=0 AND weight_micros<=1000000),
          PRIMARY KEY(rule_uid,rule_version,target_ref),
          FOREIGN KEY(rule_uid,rule_version) REFERENCES public.metering_allocation_rules(rule_uid,rule_version) ON DELETE RESTRICT
        );
        CREATE TABLE public.metering_budgets (
          budget_uid text PRIMARY KEY CHECK(budget_uid ~ '^[A-Za-z0-9][A-Za-z0-9._:-]{0,119}$'), tenant_ref text NOT NULL, domain_ref text NOT NULL,
          department_ref text NOT NULL, project_ref text NOT NULL, cost_center_ref text NOT NULL, window_key char(7) NOT NULL CHECK(window_key ~ '^[0-9]{4}-(0[1-9]|1[0-2])$'),
          limit_micros bigint NOT NULL CHECK(limit_micros>=0), threshold_micros bigint NOT NULL CHECK(threshold_micros>=0 AND threshold_micros<=limit_micros),
          created_at timestamptz NOT NULL DEFAULT clock_timestamp()
        );
        CREATE TABLE public.metering_alert_outbox (
          alert_uid uuid PRIMARY KEY DEFAULT gen_random_uuid(), budget_uid text NOT NULL REFERENCES public.metering_budgets(budget_uid) ON DELETE RESTRICT,
          window_key char(7) NOT NULL, threshold_micros bigint NOT NULL, payload_digest char(64) NOT NULL CHECK(payload_digest ~ '^[0-9a-f]{64}$'),
          provider_state text NOT NULL DEFAULT 'disabled' CHECK(provider_state='disabled'), created_at timestamptz NOT NULL DEFAULT clock_timestamp(),
          UNIQUE(budget_uid,window_key,threshold_micros)
        );
        CREATE TABLE public.metering_reconciliation_reports (
          report_uid uuid PRIMARY KEY DEFAULT gen_random_uuid(), tenant_ref text NOT NULL, domain_ref text NOT NULL, window_key char(7) NOT NULL,
          source_micros bigint NOT NULL CHECK(source_micros>=0), allocated_micros bigint NOT NULL CHECK(allocated_micros>=0), difference_micros bigint NOT NULL,
          report_digest char(64) NOT NULL CHECK(report_digest ~ '^[0-9a-f]{64}$'), created_at timestamptz NOT NULL DEFAULT clock_timestamp(),
          UNIQUE(tenant_ref,domain_ref,window_key,report_digest)
        );
        CREATE TABLE public.metering_audit_events (
          audit_uid uuid PRIMARY KEY DEFAULT gen_random_uuid(), tenant_ref text NOT NULL, domain_ref text NOT NULL,
          event_type text NOT NULL CHECK(event_type IN ('metering.recorded','metering.corrected','metering.lease.claimed','showback.read','budget.alert.created')),
          actor_ref text NOT NULL, payload_digest char(64) NOT NULL CHECK(payload_digest ~ '^[0-9a-f]{64}$'), lease_fence bigint, created_at timestamptz NOT NULL DEFAULT clock_timestamp()
        );
        DO $$ DECLARE item text; BEGIN
          FOREACH item IN ARRAY ARRAY['metering_scope_grants','metering_showback_configuration','metering_runtime_leases','metering_runtime_claims',
            'metering_events','metering_allocation_rules','metering_allocations','metering_budgets','metering_alert_outbox','metering_reconciliation_reports','metering_audit_events'] LOOP
            EXECUTE 'ALTER TABLE public.' || quote_ident(item) || ' OWNER TO dataops_tenant_foundation_owner';
          END LOOP;
        END $$;

        CREATE FUNCTION public.metering_showback_issue_claim(p_action text,p_principal text,p_body jsonb)
        RETURNS jsonb LANGUAGE plpgsql SECURITY DEFINER SET search_path=pg_catalog,public AS $claim$
        DECLARE v_tenant text; v_domain text; v_role text; v_claim uuid;
        BEGIN
          IF NOT pg_has_role(session_user,'dataops_bi_ai_catalog_issuer','MEMBER') OR p_action NOT IN ('lease','record','read')
             OR jsonb_typeof(p_body)<>'object' OR p_principal !~ '^[A-Za-z0-9_.:-]{1,120}$' THEN RAISE EXCEPTION 'metering_claim_denied'; END IF;
          SELECT tenant_ref,domain_ref,role_name INTO v_tenant,v_domain,v_role FROM public.metering_scope_grants
            WHERE principal_ref=p_principal AND active AND revoked_at IS NULL ORDER BY CASE role_name WHEN 'admin' THEN 1 WHEN 'operator' THEN 2 ELSE 3 END LIMIT 1;
          IF NOT FOUND OR (p_action IN ('lease','record') AND v_role NOT IN ('operator','admin')) THEN RAISE EXCEPTION 'metering_claim_denied'; END IF;
          INSERT INTO public.metering_runtime_claims(action_name,principal_ref,tenant_ref,domain_ref,request_digest,expires_at)
            VALUES(p_action,p_principal,v_tenant,v_domain,encode(sha256(convert_to(p_action||'|'||p_body::text,'utf8')),'hex'),clock_timestamp()+interval '30 seconds') RETURNING claim_uid INTO v_claim;
          RETURN jsonb_build_object('request_claim',v_claim);
        END; $claim$;
        ALTER FUNCTION public.metering_showback_issue_claim(text,text,jsonb) OWNER TO dataops_tenant_foundation_owner;

        CREATE FUNCTION public.metering_showback_runtime_write(p_action text,p_payload jsonb)
        RETURNS jsonb LANGUAGE plpgsql SECURITY DEFINER SET search_path=pg_catalog,public AS $write$
        DECLARE v_claim record; v_fence bigint; v_digest text; v_existing text;
        BEGIN
          IF p_action='chargeback' THEN RAISE EXCEPTION 'chargeback_disabled'; END IF;
          IF NOT pg_has_role(session_user,'dataops_app_runtime','MEMBER') OR p_action NOT IN ('claim_lease','record_event') OR jsonb_typeof(p_payload)<>'object' THEN RAISE EXCEPTION 'metering_runtime_denied'; END IF;
          IF NOT (p_payload ? 'request_claim') OR p_payload->>'request_claim' !~ '^[0-9a-f-]{36}$' THEN RAISE EXCEPTION 'metering_payload_closed'; END IF;
          SELECT * INTO v_claim FROM public.metering_runtime_claims WHERE claim_uid=(p_payload->>'request_claim')::uuid AND consumed_at IS NULL AND expires_at>clock_timestamp() FOR UPDATE;
          IF NOT FOUND OR v_claim.action_name<>(CASE WHEN p_action='claim_lease' THEN 'lease' ELSE 'record' END) THEN RAISE EXCEPTION 'metering_claim_denied'; END IF;
          IF p_action='claim_lease' THEN
            IF p_payload-ARRAY['request_claim','lease_owner','lease_token']<>'{}'::jsonb OR p_payload->>'lease_owner' !~ '^[A-Za-z0-9_.:-]{1,120}$' OR p_payload->>'lease_token' !~ '^[0-9a-f-]{36}$' THEN RAISE EXCEPTION 'metering_payload_closed'; END IF;
            INSERT INTO public.metering_runtime_leases(tenant_ref) VALUES(v_claim.tenant_ref) ON CONFLICT DO NOTHING;
            UPDATE public.metering_runtime_leases SET lease_owner=p_payload->>'lease_owner',lease_token=(p_payload->>'lease_token')::uuid,lease_fence=lease_fence+1,lease_expires_at=clock_timestamp()+interval '30 seconds',updated_at=clock_timestamp()
              WHERE tenant_ref=v_claim.tenant_ref AND (lease_expires_at IS NULL OR lease_expires_at<=clock_timestamp()) RETURNING lease_fence INTO v_fence;
            IF NOT FOUND THEN RAISE EXCEPTION 'metering_lease_unavailable'; END IF;
            UPDATE public.metering_runtime_claims SET consumed_at=clock_timestamp() WHERE claim_uid=v_claim.claim_uid;
            INSERT INTO public.metering_audit_events(tenant_ref,domain_ref,event_type,actor_ref,payload_digest,lease_fence) VALUES(v_claim.tenant_ref,v_claim.domain_ref,'metering.lease.claimed',v_claim.principal_ref,encode(sha256(convert_to(p_payload::text,'utf8')),'hex'),v_fence);
            RETURN jsonb_build_object('lease_fence',v_fence);
          END IF;
          IF p_payload-ARRAY['request_claim','lease_owner','lease_token','lease_fence','event_uid','event_kind','occurred_at','window_start','window_end','quantity_micros','unit','idempotency_key','evidence','mapping','correction_of']<>'{}'::jsonb
             OR NOT (p_payload ?& ARRAY['request_claim','lease_owner','lease_token','lease_fence','event_uid','event_kind','occurred_at','window_start','window_end','quantity_micros','unit','idempotency_key','evidence','mapping'])
             OR jsonb_typeof(p_payload->'evidence')<>'object' OR jsonb_typeof(p_payload->'mapping')<>'object' OR jsonb_typeof(p_payload->'quantity_micros')<>'number'
             OR (p_payload->'evidence')-ARRAY['digest','reference']<>'{}'::jsonb OR (p_payload->'mapping')-ARRAY['department','business_domain','project','cost_center']<>'{}'::jsonb
             OR p_payload->>'event_kind' NOT IN ('query','api','file','subscription','storage','compute','task','model_call')
             OR p_payload->>'event_uid' !~ '^[A-Za-z0-9][A-Za-z0-9._:-]{0,119}$' OR p_payload->>'idempotency_key' !~ '^[A-Za-z0-9][A-Za-z0-9._:-]{0,119}$'
             OR p_payload->>'quantity_micros' !~ '^[0-9]{1,18}$' OR p_payload->>'unit' NOT IN ('bytes','kb','mb','gb','seconds','milliseconds','requests','tasks','tokens')
             OR p_payload->'evidence'->>'digest' !~ '^[0-9a-f]{64}$' OR p_payload->'evidence'->>'reference' !~ '^local-fixture://wp12/v1(/[A-Za-z0-9._:-]{1,80})?$'
             OR p_payload->'mapping'->>'department' !~ '^[a-z][a-z0-9-]{0,62}$' OR p_payload->'mapping'->>'business_domain'<>v_claim.domain_ref
             OR p_payload->'mapping'->>'project' !~ '^[a-z][a-z0-9-]{0,62}$' OR p_payload->'mapping'->>'cost_center' !~ '^[a-z][a-z0-9-]{0,62}$' THEN RAISE EXCEPTION 'metering_payload_closed'; END IF;
          SELECT lease_fence INTO v_fence FROM public.metering_runtime_leases WHERE tenant_ref=v_claim.tenant_ref AND lease_owner=p_payload->>'lease_owner' AND lease_token=(p_payload->>'lease_token')::uuid AND lease_expires_at>clock_timestamp() FOR UPDATE;
          IF NOT FOUND OR v_fence<>(p_payload->>'lease_fence')::bigint THEN RAISE EXCEPTION 'metering_stale_fence'; END IF;
          -- Claims and leases are one-use transport fences, not event
          -- identity.  Exact replay of the same immutable event may obtain a
          -- fresh claim and a newer lease fence after a process restart.
          v_digest:=encode(sha256(convert_to((p_payload-ARRAY['request_claim','lease_owner','lease_token','lease_fence'])::text,'utf8')),'hex');
          SELECT request_digest INTO v_existing FROM public.metering_events WHERE tenant_ref=v_claim.tenant_ref AND idempotency_key=p_payload->>'idempotency_key' FOR UPDATE;
          IF FOUND THEN
            IF v_existing<>v_digest THEN RAISE EXCEPTION 'metering_replay_conflict'; END IF;
            UPDATE public.metering_runtime_claims SET consumed_at=clock_timestamp() WHERE claim_uid=v_claim.claim_uid;
            RETURN jsonb_build_object('event_uid',(SELECT event_uid FROM public.metering_events WHERE tenant_ref=v_claim.tenant_ref AND idempotency_key=p_payload->>'idempotency_key'),'replay',true,'lease_fence',v_fence);
          END IF;
          IF p_payload ? 'correction_of' AND NOT EXISTS(SELECT 1 FROM public.metering_events WHERE event_uid=p_payload->>'correction_of' AND tenant_ref=v_claim.tenant_ref) THEN RAISE EXCEPTION 'metering_correction_missing'; END IF;
          INSERT INTO public.metering_events(event_uid,tenant_ref,domain_ref,department_ref,project_ref,cost_center_ref,event_kind,occurred_at,window_start,window_end,quantity_micros,unit,idempotency_key,evidence_digest,evidence_reference,request_digest,correction_of,lease_fence)
          VALUES(p_payload->>'event_uid',v_claim.tenant_ref,v_claim.domain_ref,p_payload->'mapping'->>'department',p_payload->'mapping'->>'project',p_payload->'mapping'->>'cost_center',p_payload->>'event_kind',(p_payload->>'occurred_at')::timestamptz,(p_payload->>'window_start')::timestamptz,(p_payload->>'window_end')::timestamptz,(p_payload->>'quantity_micros')::bigint,p_payload->>'unit',p_payload->>'idempotency_key',p_payload->'evidence'->>'digest',p_payload->'evidence'->>'reference',v_digest,p_payload->>'correction_of',v_fence);
          UPDATE public.metering_runtime_claims SET consumed_at=clock_timestamp() WHERE claim_uid=v_claim.claim_uid;
          INSERT INTO public.metering_audit_events(tenant_ref,domain_ref,event_type,actor_ref,payload_digest,lease_fence) VALUES(v_claim.tenant_ref,v_claim.domain_ref,CASE WHEN p_payload ? 'correction_of' THEN 'metering.corrected' ELSE 'metering.recorded' END,v_claim.principal_ref,v_digest,v_fence);
          RETURN jsonb_build_object('event_uid',p_payload->>'event_uid','replay',false,'lease_fence',v_fence,'persisted_before_ack',true);
        END; $write$;
        ALTER FUNCTION public.metering_showback_runtime_write(text,jsonb) OWNER TO dataops_tenant_foundation_owner;

        CREATE FUNCTION public.metering_showback_runtime_read(p_action text,p_payload jsonb)
        RETURNS jsonb LANGUAGE plpgsql SECURITY DEFINER SET search_path=pg_catalog,public AS $read$
        DECLARE v_claim record; v_window text; v_source bigint; v_allocated bigint; v_report jsonb;
        BEGIN
          IF NOT pg_has_role(session_user,'dataops_app_runtime','MEMBER') OR p_action NOT IN ('showback','reconciliation','audit') OR jsonb_typeof(p_payload)<>'object'
             OR p_payload-ARRAY['request_claim','window']<>'{}'::jsonb OR p_payload->>'request_claim' !~ '^[0-9a-f-]{36}$' OR p_payload->>'window' !~ '^[0-9]{4}-(0[1-9]|1[0-2])$' THEN RAISE EXCEPTION 'metering_read_denied'; END IF;
          SELECT * INTO v_claim FROM public.metering_runtime_claims WHERE claim_uid=(p_payload->>'request_claim')::uuid AND action_name='read' AND consumed_at IS NULL AND expires_at>clock_timestamp() FOR UPDATE;
          IF NOT FOUND THEN RAISE EXCEPTION 'metering_claim_denied'; END IF;
          v_window:=p_payload->>'window';
          IF p_action='audit' THEN
            v_report:=COALESCE((SELECT jsonb_agg(jsonb_build_object('event_type',event_type,'payload_digest',payload_digest,'lease_fence',lease_fence) ORDER BY created_at) FROM (SELECT * FROM public.metering_audit_events WHERE tenant_ref=v_claim.tenant_ref AND domain_ref=v_claim.domain_ref ORDER BY created_at DESC LIMIT 100) a),'[]'::jsonb);
          ELSE
            SELECT COALESCE(sum(quantity_micros),0) INTO v_source FROM public.metering_events WHERE tenant_ref=v_claim.tenant_ref AND domain_ref=v_claim.domain_ref AND to_char(window_start AT TIME ZONE 'UTC','YYYY-MM')=v_window;
            v_allocated:=v_source;
            INSERT INTO public.metering_reconciliation_reports(tenant_ref,domain_ref,window_key,source_micros,allocated_micros,difference_micros,report_digest)
              VALUES(v_claim.tenant_ref,v_claim.domain_ref,v_window,v_source,v_allocated,v_source-v_allocated,encode(sha256(convert_to(v_claim.tenant_ref||'|'||v_claim.domain_ref||'|'||v_window||'|'||v_source::text,'utf8')),'hex')) ON CONFLICT DO NOTHING;
            v_report:=jsonb_build_object('window',v_window,'source_micros',v_source,'allocated_micros',v_allocated,'difference_micros',v_source-v_allocated,'mode','ENGINEERING_EVIDENCE_ONLY','chargeback_enabled',false);
          END IF;
          UPDATE public.metering_runtime_claims SET consumed_at=clock_timestamp() WHERE claim_uid=v_claim.claim_uid;
          INSERT INTO public.metering_audit_events(tenant_ref,domain_ref,event_type,actor_ref,payload_digest) VALUES(v_claim.tenant_ref,v_claim.domain_ref,'showback.read',v_claim.principal_ref,encode(sha256(convert_to(p_action||'|'||p_payload::text,'utf8')),'hex'));
          RETURN v_report;
        END; $read$;
        ALTER FUNCTION public.metering_showback_runtime_read(text,jsonb) OWNER TO dataops_tenant_foundation_owner;

        REVOKE ALL ON TABLE public.metering_events FROM PUBLIC,dataops_app,dataops_app_runtime;
        REVOKE ALL ON TABLE public.metering_scope_grants,public.metering_showback_configuration,public.metering_runtime_leases,public.metering_runtime_claims,public.metering_allocation_rules,public.metering_allocations,public.metering_budgets,public.metering_alert_outbox,public.metering_reconciliation_reports,public.metering_audit_events FROM PUBLIC,dataops_app,dataops_app_runtime;
        REVOKE ALL ON FUNCTION public.metering_showback_issue_claim(text,text,jsonb),public.metering_showback_runtime_write(text,jsonb),public.metering_showback_runtime_read(text,jsonb) FROM PUBLIC,dataops_app,dataops_app_runtime;
        GRANT USAGE ON SCHEMA public TO dataops_app_runtime,dataops_bi_ai_catalog_control;
        GRANT EXECUTE ON FUNCTION public.metering_showback_issue_claim(text,text,jsonb) TO dataops_bi_ai_catalog_control;
        GRANT EXECUTE ON FUNCTION public.metering_showback_runtime_write(text,jsonb),public.metering_showback_runtime_read(text,jsonb) TO dataops_app_runtime;
        '''
    )


def downgrade() -> None:
    bind = op.get_bind()
    # A fresh schema contains one immutable disabled configuration row.  It is
    # not a metering fact and must not make an otherwise clean revision
    # impossible to roll back; all persisted usage, allocation, alert, report,
    # audit, claim, lease and scope evidence remains a hard downgrade fence.
    checks = " OR ".join(
        f"EXISTS(SELECT 1 FROM public.{table} LIMIT 1)"
        for table in _TABLES
        if table != "metering_showback_configuration"
    )
    if bind.exec_driver_sql(f"SELECT {checks}").scalar():
        raise RuntimeError("downgrade refused: WP12 metering facts are nonempty")
    bind.exec_driver_sql(
        "REVOKE EXECUTE ON FUNCTION public.metering_showback_issue_claim(text,text,jsonb),public.metering_showback_runtime_write(text,jsonb),public.metering_showback_runtime_read(text,jsonb) FROM dataops_bi_ai_catalog_control,dataops_app_runtime; "
        "DROP FUNCTION IF EXISTS public.metering_showback_runtime_read(text,jsonb); "
        "DROP FUNCTION IF EXISTS public.metering_showback_runtime_write(text,jsonb); "
        "DROP FUNCTION IF EXISTS public.metering_showback_issue_claim(text,text,jsonb); "
        "DROP TABLE IF EXISTS public.metering_audit_events,public.metering_reconciliation_reports,public.metering_alert_outbox,public.metering_budgets,public.metering_allocations,public.metering_allocation_rules,public.metering_events,public.metering_runtime_claims,public.metering_runtime_leases,public.metering_showback_configuration,public.metering_scope_grants;"
    )