DataOps-platform/migrations/versions/20260724_240_data_factory_deployments.py

"""Add recoverable immutable Data Factory deployment lifecycle."""

from alembic import op


revision = "20260724_240"
down_revision = "20260724_230"
branch_labels = None
depends_on = None


def upgrade() -> None:
    op.execute(
        """
        ALTER TABLE public.dataflow_deployments
            ADD COLUMN dataflow_uid UUID,
            ADD COLUMN package JSONB,
            ADD COLUMN package_hash CHAR(64),
            ADD COLUMN standard_version_ids JSONB NOT NULL DEFAULT '[]'::jsonb,
            ADD COLUMN rule_version_ids JSONB NOT NULL DEFAULT '[]'::jsonb,
            ADD COLUMN binding_snapshot JSONB,
            ADD COLUMN binding_hash CHAR(64),
            ADD COLUMN schema_snapshots JSONB,
            ADD COLUMN schema_snapshot_hash CHAR(64),
            ADD COLUMN physical_plan_hashes JSONB NOT NULL DEFAULT '[]'::jsonb,
            ADD COLUMN workflow_spec JSONB,
            ADD COLUMN workflow_spec_hash CHAR(64),
            ADD COLUMN schedule_snapshot JSONB,
            ADD COLUMN schedule_hash CHAR(64),
            ADD COLUMN engine_namespace VARCHAR(200),
            ADD COLUMN engine_definition_id VARCHAR(200),
            ADD COLUMN engine_revision VARCHAR(200),
            ADD COLUMN engine_definition_hash CHAR(64),
            ADD COLUMN deploy_receipt JSONB NOT NULL DEFAULT '{}'::jsonb,
            ADD COLUMN canary_evidence_id UUID,
            ADD COLUMN previous_active_deployment_id UUID
                REFERENCES public.dataflow_deployments(id) ON DELETE RESTRICT,
            ADD COLUMN created_by UUID REFERENCES public.users(id) ON DELETE SET NULL,
            ADD COLUMN create_reason VARCHAR(1000),
            ADD COLUMN create_idempotency_key VARCHAR(200),
            ADD COLUMN correlation_id UUID,
            ADD COLUMN lock_version INTEGER NOT NULL DEFAULT 0,
            ADD COLUMN failure_code VARCHAR(100),
            ADD COLUMN superseded_at TIMESTAMPTZ,
            ADD COLUMN rolled_back_at TIMESTAMPTZ;

        UPDATE public.dataflow_deployments d
        SET dataflow_uid = v.dataflow_uid
        FROM public.dataflow_versions v
        WHERE v.id = d.dataflow_version_id
          AND d.dataflow_uid IS NULL;

        ALTER TABLE public.dataflow_deployments
            ADD CONSTRAINT ck_dataflow_deployment_lock_version
                CHECK (lock_version >= 0),
            ADD CONSTRAINT ck_dataflow_deployment_hash_shape
                CHECK (
                    (package_hash IS NULL OR package_hash ~ '^[0-9a-f]{64}$')
                    AND (binding_hash IS NULL OR binding_hash ~ '^[0-9a-f]{64}$')
                    AND (
                        schema_snapshot_hash IS NULL
                        OR schema_snapshot_hash ~ '^[0-9a-f]{64}$'
                    )
                    AND (
                        workflow_spec_hash IS NULL
                        OR workflow_spec_hash ~ '^[0-9a-f]{64}$'
                    )
                    AND (
                        schedule_hash IS NULL
                        OR schedule_hash ~ '^[0-9a-f]{64}$'
                    )
                    AND (
                        engine_definition_hash IS NULL
                        OR engine_definition_hash ~ '^[0-9a-f]{64}$'
                    )
                ),
            ADD CONSTRAINT ck_dataflow_deployment_snapshot_complete
                CHECK (
                    package IS NULL
                    OR (
                        dataflow_uid IS NOT NULL
                        AND package_hash IS NOT NULL
                        AND binding_snapshot IS NOT NULL
                        AND binding_hash IS NOT NULL
                        AND schema_snapshots IS NOT NULL
                        AND schema_snapshot_hash IS NOT NULL
                        AND jsonb_array_length(physical_plan_hashes) > 0
                        AND workflow_spec IS NOT NULL
                        AND workflow_spec_hash IS NOT NULL
                        AND schedule_snapshot IS NOT NULL
                        AND schedule_hash IS NOT NULL
                        AND created_by IS NOT NULL
                        AND create_reason IS NOT NULL
                        AND create_idempotency_key IS NOT NULL
                        AND correlation_id IS NOT NULL
                    )
                );

        ALTER TABLE public.dataflow_deployments
            DROP CONSTRAINT IF EXISTS
                dataflow_deployments_dataflow_version_id_environment_key;
        DROP INDEX IF EXISTS public.uq_dataflow_deployment_active_environment;
        CREATE UNIQUE INDEX uq_dataflow_deployment_active_environment
            ON public.dataflow_deployments(dataflow_uid, environment)
            WHERE status = 'active';
        CREATE UNIQUE INDEX uq_dataflow_deployment_create_idempotency
            ON public.dataflow_deployments(
                created_by, environment, create_idempotency_key
            )
            WHERE create_idempotency_key IS NOT NULL;
        CREATE INDEX idx_dataflow_deployment_factory_list
            ON public.dataflow_deployments(
                environment, updated_at DESC, dataflow_uid
            )
            WHERE package IS NOT NULL;

        CREATE TABLE public.dataflow_canary_evidence (
            id UUID PRIMARY KEY,
            deployment_id UUID NOT NULL
                REFERENCES public.dataflow_deployments(id) ON DELETE RESTRICT,
            execution_id VARCHAR(200) NOT NULL,
            status VARCHAR(20) NOT NULL CHECK (
                status IN ('started','passed','failed','unknown')
            ),
            package_hash CHAR(64) NOT NULL,
            binding_hash CHAR(64) NOT NULL,
            schema_snapshot_hash CHAR(64) NOT NULL,
            physical_plan_hashes JSONB NOT NULL,
            workflow_spec_hash CHAR(64) NOT NULL,
            schedule_hash CHAR(64) NOT NULL,
            engine_definition_hash CHAR(64) NOT NULL,
            verified_by UUID NOT NULL REFERENCES public.users(id) ON DELETE RESTRICT,
            engine_response JSONB NOT NULL DEFAULT '{}'::jsonb,
            expires_at TIMESTAMPTZ NOT NULL,
            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
            UNIQUE (deployment_id, execution_id)
        );
        CREATE INDEX idx_dataflow_canary_activation_gate
            ON public.dataflow_canary_evidence(
                deployment_id, status, expires_at DESC
            );

        ALTER TABLE public.dataflow_deployments
            ADD CONSTRAINT fk_dataflow_deployment_canary_evidence
            FOREIGN KEY (canary_evidence_id)
            REFERENCES public.dataflow_canary_evidence(id)
            ON DELETE RESTRICT;

        CREATE TABLE public.dataflow_deployment_operations (
            id UUID PRIMARY KEY,
            deployment_id UUID NOT NULL
                REFERENCES public.dataflow_deployments(id) ON DELETE RESTRICT,
            action VARCHAR(30) NOT NULL CHECK (
                action IN (
                    'deploy_disabled','run_canary','activate','rollback'
                )
            ),
            idempotency_key VARCHAR(200) NOT NULL,
            status VARCHAR(20) NOT NULL CHECK (
                status IN ('claimed','completed','failed','unknown')
            ),
            actor_uid UUID NOT NULL REFERENCES public.users(id) ON DELETE RESTRICT,
            correlation_id UUID NOT NULL,
            reason VARCHAR(1000) NOT NULL,
            request JSONB NOT NULL DEFAULT '{}'::jsonb,
            request_hash CHAR(64) NOT NULL,
            result JSONB,
            result_hash CHAR(64),
            error_code VARCHAR(100),
            started_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
            completed_at TIMESTAMPTZ,
            failed_at TIMESTAMPTZ,
            updated_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
            UNIQUE (deployment_id, action, idempotency_key),
            CHECK (request_hash ~ '^[0-9a-f]{64}$'),
            CHECK (
                (status = 'completed' AND result IS NOT NULL
                    AND result_hash ~ '^[0-9a-f]{64}$')
                OR status <> 'completed'
            )
        );
        CREATE INDEX idx_dataflow_deployment_operation_recovery
            ON public.dataflow_deployment_operations(
                status, updated_at, deployment_id
            )
            WHERE status IN ('claimed','failed','unknown');

        CREATE TABLE public.dataflow_deployment_transitions (
            id UUID PRIMARY KEY,
            deployment_id UUID NOT NULL
                REFERENCES public.dataflow_deployments(id) ON DELETE RESTRICT,
            operation_id UUID
                REFERENCES public.dataflow_deployment_operations(id)
                ON DELETE RESTRICT,
            from_status VARCHAR(30),
            to_status VARCHAR(30) NOT NULL,
            actor_uid UUID NOT NULL REFERENCES public.users(id) ON DELETE RESTRICT,
            reason VARCHAR(1000) NOT NULL,
            correlation_id UUID NOT NULL,
            detail JSONB NOT NULL DEFAULT '{}'::jsonb,
            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP
        );
        CREATE INDEX idx_dataflow_deployment_transition_history
            ON public.dataflow_deployment_transitions(
                deployment_id, created_at DESC
            );
        """
    )


def downgrade() -> None:
    raise RuntimeError(
        "Data Factory deployment evidence is forward-only and cannot downgrade"
    )