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

"""Make provision intent durable and singular before provider I/O."""

from alembic import op

revision = "20260811_493"
down_revision = "20260811_492"
branch_labels = None
depends_on = None


def upgrade() -> None:
    op.execute(r"""
    /* One grant represents one target-provider entitlement.  A new HTTP key
       therefore cannot manufacture a second adapter side effect. */
    CREATE UNIQUE INDEX IF NOT EXISTS uq_trusted_delivery_one_provision_per_grant
      ON public.trusted_delivery_deliveries(grant_uid) WHERE operation='provision';

    ALTER FUNCTION public.trusted_delivery_runtime_write(text,jsonb)
      RENAME TO trusted_delivery_runtime_write_validated492;
    REVOKE ALL ON FUNCTION public.trusted_delivery_runtime_write_validated492(text,jsonb)
      FROM PUBLIC, dataops_app_runtime;
    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 $outbox$
    DECLARE v_existing record;
    BEGIN
      IF NOT public.trusted_delivery_runtime_payload_is_closed(p_operation,p_payload) THEN
        RAISE EXCEPTION 'trusted-delivery runtime write is not permitted';
      END IF;
      IF p_operation='enqueue_delivery' THEN
        SELECT uid::text,grant_uid::text,request_digest,status,lease_fence,operation INTO v_existing
          FROM public.trusted_delivery_deliveries WHERE idempotency_key=p_payload->>'idempotency_key' FOR UPDATE;
        IF FOUND THEN
          IF v_existing.request_digest<>p_payload->>'request_digest' THEN
            RAISE EXCEPTION 'trusted delivery idempotency conflict';
          END IF;
          RETURN jsonb_build_object('uid',v_existing.uid,'status',v_existing.status,'lease_fence',v_existing.lease_fence);
        END IF;
        IF p_payload->>'operation'='provision' AND EXISTS(
          SELECT 1 FROM public.trusted_delivery_deliveries
           WHERE grant_uid=(p_payload->>'grant_uid')::uuid AND operation='provision'
        ) THEN RAISE EXCEPTION 'trusted delivery provision already exists for grant'; END IF;
      END IF;
      RETURN public.trusted_delivery_runtime_write_validated492(p_operation,p_payload);
    END $outbox$;
    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_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);
    ALTER FUNCTION public.trusted_delivery_runtime_write_validated492(text,jsonb)
      RENAME TO trusted_delivery_runtime_write;
    GRANT EXECUTE ON FUNCTION public.trusted_delivery_runtime_write(text,jsonb) TO dataops_app_runtime;
    DROP INDEX IF EXISTS public.uq_trusted_delivery_one_provision_per_grant;
    """)