DataOps-platform/migrations/versions/20260811_494_trusted_delivery_provision_lease.py

"""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);
    """)