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

"""Expose persisted, versioned allocation replay through the runtime gateway."""

from alembic import op

revision = "20260818_548"
down_revision = "20260818_547"
branch_labels = None
depends_on = None


def upgrade() -> None:
    op.get_bind().exec_driver_sql(r'''
    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; month_start timestamptz;
    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;
      month_start := (p_payload->>'window'||'-01T00:00:00Z')::timestamptz;
      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 AND effective_start<=month_start AND effective_end>month_start;
      IF NOT FOUND THEN RAISE EXCEPTION 'metering_allocation_rule_not_found'; END IF;
      SELECT COALESCE(sum(quantity_micros),0) INTO 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';
      WITH weights AS (SELECT a.target_ref,a.weight_micros,(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|'||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,'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_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("REVOKE EXECUTE ON FUNCTION public.metering_showback_allocation_replay(jsonb) FROM dataops_app_runtime; DROP FUNCTION public.metering_showback_allocation_replay(jsonb);")