DataOps-platform/migrations/versions/20260723_150_rule_artifact_handoff_state.py

"""Upgrade artifact handoff from old-140 rows to a durable state machine."""

from alembic import op

revision = "20260723_150"
down_revision = "20260723_140"
branch_labels = None
depends_on = None


def upgrade() -> None:
    op.execute(
        """
        ALTER TABLE public.rule_run_artifacts
            ADD COLUMN binding_hash CHAR(64),
            ADD COLUMN handoff_status VARCHAR(20) NOT NULL
                DEFAULT 'pending'
                CHECK (handoff_status IN ('pending','ready','failed')),
            ADD COLUMN ready_at TIMESTAMPTZ,
            ADD COLUMN failed_at TIMESTAMPTZ,
            ADD COLUMN failure_code VARCHAR(100),
            ADD COLUMN updated_at TIMESTAMPTZ NOT NULL
                DEFAULT CURRENT_TIMESTAMP;

        UPDATE public.rule_run_artifacts a
        SET binding_hash = (
                SELECT binding_hash
                FROM public.dataflow_dataset_bindings b
                WHERE b.id = a.binding_id
            ),
            handoff_status = 'ready',
            ready_at = a.created_at,
            updated_at = a.created_at;

        DO $$
        BEGIN
            IF EXISTS (
                SELECT 1
                FROM public.rule_run_artifacts
                WHERE binding_hash IS NULL
            ) THEN
                RAISE EXCEPTION
                    'artifact handoff migration found unattested bindings';
            END IF;
            IF EXISTS (
                SELECT 1
                FROM public.rule_run_artifacts
                GROUP BY correlation_id, binding_id, artifact_kind
                HAVING COUNT(*) > 1
            ) THEN
                RAISE EXCEPTION
                    'artifact handoff migration found conflicting old rows';
            END IF;
        END
        $$;

        ALTER TABLE public.rule_run_artifacts
            ALTER COLUMN binding_hash SET NOT NULL;

        DO $$
        DECLARE
            old_constraint TEXT;
        BEGIN
            SELECT c.conname
            INTO old_constraint
            FROM pg_constraint c
            WHERE c.conrelid = 'public.rule_run_artifacts'::regclass
              AND c.contype = 'u'
              AND pg_get_constraintdef(c.oid) =
                  'UNIQUE (correlation_id, binding_id, artifact_digest)';
            IF old_constraint IS NULL THEN
                RAISE EXCEPTION
                    'old-140 artifact uniqueness constraint was not found';
            END IF;
            EXECUTE format(
                'ALTER TABLE public.rule_run_artifacts DROP CONSTRAINT %I',
                old_constraint
            );
        END
        $$;

        ALTER TABLE public.rule_run_artifacts
            ADD CONSTRAINT rule_run_artifacts_handoff_key
            UNIQUE (correlation_id, binding_id, artifact_kind);

        CREATE INDEX idx_rule_run_artifacts_reconcile
            ON public.rule_run_artifacts
            (handoff_status, updated_at, id);
        """
    )


def downgrade() -> None:
    raise RuntimeError(
        "artifact handoff state is forward-only and cannot downgrade "
        "without violating durable publication evidence"
    )