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

"""Persist P3-WP06 subscription governance and fenced delivery queue."""

from alembic import op

revision = "20260811_483"
down_revision = "20260811_482"
branch_labels = None
depends_on = None


def upgrade() -> None:
    op.execute(
        """
        CREATE TABLE public.trusted_delivery_subscriptions (
          uid UUID PRIMARY KEY,
          grant_uid UUID NOT NULL REFERENCES public.trusted_delivery_grants(uid),
          asset_uid UUID NOT NULL,
          trigger JSONB NOT NULL CHECK(jsonb_typeof(trigger)='object'),
          purpose VARCHAR(100) NOT NULL,
          expires_at TIMESTAMPTZ NOT NULL,
          idempotency_key VARCHAR(160) NOT NULL UNIQUE,
          request_digest CHAR(64) NOT NULL CHECK(request_digest ~ '^[0-9a-f]{64}$'),
          status VARCHAR(20) NOT NULL CHECK(status IN ('draft','active','paused','terminated','expired','reclaimed')),
          current_version INTEGER NOT NULL DEFAULT 1 CHECK(current_version>0),
          created_by UUID NOT NULL REFERENCES public.users(id),
          created_at TIMESTAMPTZ NOT NULL DEFAULT clock_timestamp(),
          updated_at TIMESTAMPTZ NOT NULL DEFAULT clock_timestamp(),
          CHECK((trigger->>'kind'='schedule' AND jsonb_typeof(trigger->'schedule_ref')='string'
                 AND trigger=jsonb_build_object('kind','schedule','schedule_ref',trigger->'schedule_ref'))
             OR (trigger->>'kind'='event' AND jsonb_typeof(trigger->'event_type')='string'
                 AND trigger=jsonb_build_object('kind','event','event_type',trigger->'event_type')))
        );
        CREATE INDEX ix_trusted_delivery_subscription_expiry ON public.trusted_delivery_subscriptions(status,expires_at);
        CREATE TABLE public.trusted_delivery_subscription_deliveries (
          uid UUID PRIMARY KEY,
          subscription_uid UUID NOT NULL REFERENCES public.trusted_delivery_subscriptions(uid),
          trigger_ref VARCHAR(160) NOT NULL,
          event_digest CHAR(64) NOT NULL CHECK(event_digest ~ '^[0-9a-f]{64}$'),
          idempotency_key VARCHAR(320) NOT NULL UNIQUE,
          request_digest CHAR(64) NOT NULL CHECK(request_digest ~ '^[0-9a-f]{64}$'),
          status VARCHAR(20) NOT NULL CHECK(status IN ('pending','processing','delivered','dead_letter','compensated')),
          attempt_count INTEGER NOT NULL DEFAULT 0 CHECK(attempt_count BETWEEN 0 AND 3),
          next_attempt_at TIMESTAMPTZ NOT NULL DEFAULT clock_timestamp(),
          lease_owner VARCHAR(80),
          lease_fence BIGINT NOT NULL DEFAULT 0 CHECK(lease_fence>=0),
          lease_expires_at TIMESTAMPTZ,
          last_reason_code VARCHAR(80),
          compensation_reason_code VARCHAR(64),
          compensation_receipt_code VARCHAR(64),
          created_at TIMESTAMPTZ NOT NULL DEFAULT clock_timestamp(),
          updated_at TIMESTAMPTZ NOT NULL DEFAULT clock_timestamp(),
          CHECK((status='processing')=(lease_owner IS NOT NULL AND lease_expires_at IS NOT NULL)),
          CHECK((status='compensated')=(compensation_reason_code IS NOT NULL AND compensation_receipt_code IS NOT NULL))
        );
        CREATE INDEX ix_trusted_delivery_subscription_claim ON public.trusted_delivery_subscription_deliveries(status,next_attempt_at);
        """
    )


def downgrade() -> None:
    op.execute(
        """DO $$ BEGIN
          IF EXISTS(SELECT 1 FROM public.trusted_delivery_subscription_deliveries)
             OR EXISTS(SELECT 1 FROM public.trusted_delivery_subscriptions) THEN
            RAISE EXCEPTION 'downgrade requires approved subscription archival migration';
          END IF;
        END $$;
        DROP TABLE public.trusted_delivery_subscription_deliveries;
        DROP TABLE public.trusted_delivery_subscriptions;"""
    )