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

"""Unify event-time allocation replay and revoke superseded gateway paths."""

from alembic import op

revision = "20260818_550"
down_revision = "20260818_549"
branch_labels = None
depends_on = None


def upgrade() -> None:
    op.get_bind().exec_driver_sql(r'''
    DO $preflight$
    BEGIN
      IF EXISTS(SELECT 1 FROM public.metering_events e JOIN public.metering_events o ON o.event_uid=e.correction_of WHERE e.correction_of IS NOT NULL AND (e.tenant_ref<>o.tenant_ref OR e.domain_ref<>o.domain_ref OR o.correction_of IS NOT NULL OR e.event_kind<>o.event_kind OR e.unit<>o.unit OR e.window_start<>o.window_start OR e.window_end<>o.window_end OR e.department_ref<>o.department_ref OR e.project_ref<>o.project_ref OR e.cost_center_ref<>o.cost_center_ref)) OR EXISTS(SELECT 1 FROM public.metering_events WHERE evidence_reference !~ '^local-fixture://wp12/v1(/[a-z][a-z0-9-]{0,62})?$' OR evidence_reference ~ '(sql|script|select|insert|update|delete|drop|exec|curl|wget|token|secret|credential|password)') OR EXISTS(SELECT 1 FROM public.metering_allocation_rules a JOIN public.metering_allocation_rules b ON (a.rule_uid,a.rule_version)<>(b.rule_uid,b.rule_version) AND a.tenant_ref=b.tenant_ref AND a.domain_ref=b.domain_ref AND a.department_ref=b.department_ref AND a.project_ref=b.project_ref AND a.cost_center_ref=b.cost_center_ref AND tstzrange(a.effective_start,a.effective_end,'[)') && tstzrange(b.effective_start,b.effective_end,'[)')) OR EXISTS(SELECT 1 FROM public.metering_allocations WHERE weight_micros<=0) OR EXISTS(SELECT 1 FROM public.metering_allocations GROUP BY rule_uid,rule_version HAVING sum(weight_micros)<>1000000) THEN RAISE EXCEPTION 'upgrade refused: WP12 integrity preflight failed'; END IF;
    END; $preflight$;
    ALTER FUNCTION public.metering_showback_allocation_replay(jsonb) RENAME TO metering_showback_allocation_replay_legacy;
    CREATE FUNCTION public.metering_showback_allocation_replay(p_payload jsonb) RETURNS jsonb LANGUAGE plpgsql SECURITY DEFINER SET search_path=pg_catalog,public AS $replay$
    DECLARE c record; rule record; source_micros bigint; result_allocations jsonb; outside_count bigint;
    BEGIN
      IF NOT pg_has_role(session_user,'dataops_app_runtime','MEMBER') OR jsonb_typeof(p_payload)<>'object' OR p_payload-ARRAY['request_claim','window','rule_uid','rule_version']<>'{}'::jsonb OR p_payload->>'request_claim' !~ '^[0-9a-f-]{36}$' OR p_payload->>'window' !~ '^[0-9]{4}-(0[1-9]|1[0-2])$' OR p_payload->>'rule_uid' !~ '^[A-Za-z0-9][A-Za-z0-9._:-]{0,119}$' OR p_payload->>'rule_version' !~ '^[1-9][0-9]{0,6}$' THEN RAISE EXCEPTION 'metering_allocation_replay_denied'; END IF;
      SELECT * INTO c 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;
      SELECT * INTO rule FROM public.metering_allocation_rules WHERE rule_uid=p_payload->>'rule_uid' AND rule_version=(p_payload->>'rule_version')::integer AND tenant_ref=c.tenant_ref AND domain_ref=c.domain_ref;
      IF NOT FOUND THEN RAISE EXCEPTION 'metering_allocation_rule_not_found'; END IF;
      SELECT count(*) FILTER (WHERE window_start<rule.effective_start OR window_start>=rule.effective_end),COALESCE(sum(quantity_micros),0) INTO outside_count,source_micros FROM public.metering_events WHERE tenant_ref=c.tenant_ref AND domain_ref=c.domain_ref AND department_ref=rule.department_ref AND project_ref=rule.project_ref AND cost_center_ref=rule.cost_center_ref AND to_char(window_start AT TIME ZONE 'UTC','YYYY-MM')=p_payload->>'window';
      IF outside_count<>0 THEN RAISE EXCEPTION 'metering_allocation_replay_coverage_invalid'; END IF;
      WITH weights AS (SELECT a.target_ref,(source_micros*a.weight_micros/1000000)::bigint AS base_micros FROM public.metering_allocations a WHERE a.rule_uid=rule.rule_uid AND a.rule_version=rule.rule_version), totals AS (SELECT COALESCE(sum(base_micros),0) AS base_total,min(target_ref) AS residual_target FROM weights) SELECT COALESCE(jsonb_agg(jsonb_build_object('target',w.target_ref,'quantity_micros',w.base_micros+CASE WHEN w.target_ref=t.residual_target THEN source_micros-t.base_total ELSE 0 END) ORDER BY w.target_ref),'[]'::jsonb) INTO result_allocations FROM weights w CROSS JOIN totals t;
      UPDATE public.metering_runtime_claims SET consumed_at=clock_timestamp() WHERE claim_uid=c.claim_uid;
      INSERT INTO public.metering_audit_events(tenant_ref,domain_ref,event_type,actor_ref,payload_digest) VALUES(c.tenant_ref,c.domain_ref,'showback.read',c.principal_ref,encode(sha256(convert_to('allocation_replay_v2|'||p_payload::text,'utf8')),'hex'));
      RETURN jsonb_build_object('rule_uid',rule.rule_uid,'rule_version',rule.rule_version,'window',p_payload->>'window','mapping',jsonb_build_object('department',rule.department_ref,'business_domain',rule.domain_ref,'project',rule.project_ref,'cost_center',rule.cost_center_ref),'rule_digest',rule.rule_digest,'source_micros',source_micros,'allocated_micros',source_micros,'difference_micros',0,'allocations',result_allocations,'coverage','complete-event-time','mode','ENGINEERING_EVIDENCE_ONLY');
    END; $replay$;
    ALTER FUNCTION public.metering_showback_allocation_replay(jsonb) OWNER TO dataops_tenant_foundation_owner;
    REVOKE ALL ON FUNCTION public.metering_showback_runtime_read_legacy(text,jsonb),public.metering_showback_control_write_legacy(text,text,jsonb),public.metering_showback_allocation_replay_legacy(jsonb) FROM PUBLIC,dataops_app,dataops_app_runtime,dataops_bi_ai_catalog_control;
    REVOKE ALL ON FUNCTION public.metering_showback_allocation_replay(jsonb) FROM PUBLIC,dataops_app,dataops_bi_ai_catalog_control;
    GRANT EXECUTE ON FUNCTION public.metering_showback_allocation_replay(jsonb) TO dataops_app_runtime;
    ''')


def downgrade() -> None:
    op.execute("DROP FUNCTION public.metering_showback_allocation_replay(jsonb); ALTER FUNCTION public.metering_showback_allocation_replay_legacy(jsonb) RENAME TO metering_showback_allocation_replay; GRANT EXECUTE ON FUNCTION public.metering_showback_allocation_replay(jsonb) TO dataops_app_runtime;")