"""Fence provision provider calls behind a durable database-time outbox lease.""" from alembic import op revision = "20260811_494" down_revision = "20260811_493" branch_labels = None depends_on = None def upgrade() -> None: op.execute(r""" /* Keep every pre-494 operation intact and add a deliberately narrower provision lifecycle. The restricted runtime role owns no table DML. */ ALTER FUNCTION public.trusted_delivery_runtime_write(text,jsonb) RENAME TO trusted_delivery_runtime_write_validated493; REVOKE ALL ON FUNCTION public.trusted_delivery_runtime_write_validated493(text,jsonb) FROM PUBLIC, dataops_app_runtime; CREATE OR REPLACE FUNCTION public.trusted_delivery_provision_payload_is_closed( p_operation text, p_payload jsonb ) RETURNS boolean LANGUAGE plpgsql IMMUTABLE AS $closed$ DECLARE v_required text[]; v_key text; v_value jsonb; BEGIN IF jsonb_typeof(p_payload)<>'object' THEN RETURN false; END IF; v_required:=CASE p_operation WHEN 'claim_provision' THEN ARRAY['uid','worker'] WHEN 'fail_provision' THEN ARRAY['uid','worker','fence'] WHEN 'complete_provision' THEN ARRAY['grant_uid','uid','worker','fence','receipt_uid','receipt_code','response_digest','diff_digest'] ELSE NULL END; IF v_required IS NULL OR NOT p_payload ?& v_required OR p_payload-v_required<>'{}'::jsonb THEN RETURN false; END IF; FOR v_key,v_value IN SELECT key,value FROM jsonb_each(p_payload) LOOP IF v_key='fence' THEN IF jsonb_typeof(v_value)<>'number' THEN RETURN false; END IF; ELSIF jsonb_typeof(v_value)<>'string' OR octet_length(v_value #>> '{}')>512 THEN RETURN false; END IF; END LOOP; IF EXISTS(SELECT 1 FROM jsonb_each_text(p_payload) e(key,value) WHERE e.key IN ('uid','grant_uid','receipt_uid') AND e.value !~* '^[0-9a-f]{8}-[0-9a-f]{4}-[1-8][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$') THEN RETURN false; END IF; IF EXISTS(SELECT 1 FROM jsonb_each_text(p_payload) e(key,value) WHERE e.key IN ('response_digest','diff_digest') AND e.value !~ '^[0-9a-f]{64}$') THEN RETURN false; END IF; RETURN true; END $closed$; CREATE OR REPLACE FUNCTION public.trusted_delivery_runtime_provision_write( p_operation text,p_payload jsonb ) RETURNS jsonb LANGUAGE plpgsql SECURITY DEFINER SET search_path=pg_catalog,public AS $provision$ DECLARE r record; v_receipt record; BEGIN IF NOT public.trusted_delivery_provision_payload_is_closed(p_operation,p_payload) THEN RAISE EXCEPTION 'trusted-delivery runtime write is not permitted'; END IF; IF p_operation='claim_provision' THEN SELECT rcp.status,rcp.receipt_code,rcp.response_digest,rcp.diff_digest INTO v_receipt FROM public.trusted_delivery_deliveries d JOIN public.trusted_delivery_receipts rcp ON rcp.delivery_uid=d.uid WHERE d.uid=(p_payload->>'uid')::uuid AND d.operation='provision'; IF FOUND THEN RETURN jsonb_build_object('state','completed','receipt',jsonb_build_object( 'status',v_receipt.status,'receipt_code',v_receipt.receipt_code, 'response_digest',v_receipt.response_digest,'diff_digest',v_receipt.diff_digest)); END IF; UPDATE public.trusted_delivery_deliveries SET status='processing', lease_owner=p_payload->>'worker',lease_fence=lease_fence+1, lease_expires_at=clock_timestamp()+interval '300 seconds',updated_at=clock_timestamp() WHERE uid=(p_payload->>'uid')::uuid AND operation='provision' AND ( (status='pending' AND next_attempt_at<=clock_timestamp()) OR (status='processing' AND lease_expires_at<=clock_timestamp()) ) RETURNING uid::text,grant_uid::text,lease_fence,lease_owner,lease_expires_at INTO r; IF NOT FOUND THEN RETURN jsonb_build_object('state','leased'); END IF; RETURN jsonb_build_object('state','claimed','uid',r.uid,'grant_uid',r.grant_uid, 'lease_fence',r.lease_fence,'lease_owner',r.lease_owner,'lease_expires_at',r.lease_expires_at); ELSIF p_operation='fail_provision' THEN UPDATE public.trusted_delivery_deliveries SET status='pending',lease_owner=NULL, lease_expires_at=NULL,next_attempt_at=clock_timestamp(),updated_at=clock_timestamp() WHERE uid=(p_payload->>'uid')::uuid AND operation='provision' AND status='processing' AND lease_owner=p_payload->>'worker' AND lease_fence=(p_payload->>'fence')::bigint AND lease_expires_at>clock_timestamp() RETURNING uid::text INTO r; IF NOT FOUND THEN RAISE EXCEPTION 'trusted delivery lease conflict'; END IF; RETURN jsonb_build_object('uid',r.uid,'status','pending'); ELSIF p_operation='complete_provision' THEN UPDATE public.trusted_delivery_deliveries SET status='applied',lease_owner=NULL, lease_expires_at=NULL,updated_at=clock_timestamp() WHERE uid=(p_payload->>'uid')::uuid AND grant_uid=(p_payload->>'grant_uid')::uuid AND operation='provision' AND status='processing' AND lease_owner=p_payload->>'worker' AND lease_fence=(p_payload->>'fence')::bigint AND lease_expires_at>clock_timestamp() RETURNING uid::text,grant_uid::text,request_digest,status INTO r; IF NOT FOUND THEN RAISE EXCEPTION 'trusted delivery lease conflict'; END IF; INSERT INTO public.trusted_delivery_receipts(uid,delivery_uid,grant_uid,status,receipt_code,request_digest,response_digest,diff_digest) VALUES((p_payload->>'receipt_uid')::uuid,r.uid::uuid,r.grant_uid::uuid,'applied', p_payload->>'receipt_code',r.request_digest,p_payload->>'response_digest',p_payload->>'diff_digest'); UPDATE public.trusted_delivery_grants SET reason_code='provision_applied', current_version=current_version+1,updated_at=clock_timestamp() WHERE uid=r.grant_uid::uuid AND status='active' RETURNING uid::text,status,reason_code,current_version INTO r; IF NOT FOUND THEN RAISE EXCEPTION 'trusted delivery grant version conflict'; END IF; RETURN jsonb_build_object('delivery_uid',p_payload->>'uid','status','applied', 'grant_uid',r.uid,'grant_status',r.status,'reason_code',r.reason_code, 'current_version',r.current_version); END IF; RAISE EXCEPTION 'trusted-delivery runtime write is not permitted'; END $provision$; CREATE OR REPLACE FUNCTION public.trusted_delivery_runtime_write( p_operation text,p_payload jsonb ) RETURNS jsonb LANGUAGE plpgsql SECURITY DEFINER SET search_path=pg_catalog,public AS $wrapper$ BEGIN IF p_operation IN ('claim_provision','fail_provision','complete_provision') THEN RETURN public.trusted_delivery_runtime_provision_write(p_operation,p_payload); END IF; RETURN public.trusted_delivery_runtime_write_validated493(p_operation,p_payload); END $wrapper$; REVOKE ALL ON FUNCTION public.trusted_delivery_runtime_provision_write(text,jsonb) FROM PUBLIC; REVOKE ALL ON FUNCTION public.trusted_delivery_runtime_write(text,jsonb) FROM PUBLIC; GRANT EXECUTE ON FUNCTION public.trusted_delivery_runtime_write(text,jsonb) TO dataops_app_runtime; ALTER FUNCTION public.trusted_delivery_provision_payload_is_closed(text,jsonb) OWNER TO dataops_trusted_delivery_writer; ALTER FUNCTION public.trusted_delivery_runtime_provision_write(text,jsonb) OWNER TO dataops_trusted_delivery_writer; ALTER FUNCTION public.trusted_delivery_runtime_write(text,jsonb) OWNER TO dataops_trusted_delivery_writer; """) def downgrade() -> None: op.execute(""" REVOKE EXECUTE ON FUNCTION public.trusted_delivery_runtime_write(text,jsonb) FROM dataops_app_runtime; DROP FUNCTION public.trusted_delivery_runtime_write(text,jsonb); DROP FUNCTION public.trusted_delivery_runtime_provision_write(text,jsonb); ALTER FUNCTION public.trusted_delivery_runtime_write_validated493(text,jsonb) RENAME TO trusted_delivery_runtime_write; GRANT EXECUTE ON FUNCTION public.trusted_delivery_runtime_write(text,jsonb) TO dataops_app_runtime; DROP FUNCTION public.trusted_delivery_provision_payload_is_closed(text,jsonb); """)