DataOps-platform/migrations/versions/20260719_90_workflow_migration_control.py

"""Add V55 dual-run, reconciliation, and cutover control records."""

from alembic import op

revision = "20260719_90"
down_revision = "20260719_80"
branch_labels = None
depends_on = None


def upgrade() -> None:
    op.execute(
        """
        CREATE TABLE public.workflow_migration_states (
            id UUID PRIMARY KEY,
            dataflow_uid UUID NOT NULL,
            environment VARCHAR(20) NOT NULL
                CHECK (environment IN ('development', 'test', 'production')),
            mode VARCHAR(50) NOT NULL CHECK (mode IN (
                'n8n_primary',
                'n8n_primary_kestra_shadow',
                'kestra_primary_n8n_standby',
                'kestra_primary'
            )),
            status VARCHAR(30) NOT NULL DEFAULT 'stable'
                CHECK (status IN (
                    'stable', 'transition_pending', 'blocked', 'retired'
                )),
            n8n_definition_id VARCHAR(255),
            kestra_definition_id VARCHAR(255),
            migration_batch VARCHAR(50),
            observation_until TIMESTAMPTZ,
            metadata JSONB NOT NULL DEFAULT '{}'::jsonb,
            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
            updated_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
            UNIQUE (dataflow_uid, environment)
        );
        CREATE INDEX idx_workflow_migration_mode
            ON public.workflow_migration_states(
                environment, mode, status, updated_at DESC
            );

        CREATE TABLE public.workflow_dual_runs (
            id UUID PRIMARY KEY,
            migration_state_id UUID NOT NULL
                REFERENCES public.workflow_migration_states(id)
                ON DELETE CASCADE,
            input_hash CHAR(64) NOT NULL,
            n8n_execution_id VARCHAR(255),
            kestra_execution_id VARCHAR(255),
            n8n_status VARCHAR(30) NOT NULL,
            kestra_status VARCHAR(30) NOT NULL,
            isolation_mode VARCHAR(30) NOT NULL
                CHECK (isolation_mode IN ('read_only', 'isolated_target')),
            formal_engine_changed BOOLEAN NOT NULL DEFAULT FALSE,
            correlation_id UUID NOT NULL,
            safe_metrics JSONB NOT NULL DEFAULT '{}'::jsonb,
            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP
        );
        CREATE INDEX idx_workflow_dual_runs_state_created
            ON public.workflow_dual_runs(
                migration_state_id, created_at DESC
            );

        CREATE TABLE public.workflow_reconciliation_reports (
            id UUID PRIMARY KEY,
            migration_state_id UUID NOT NULL
                REFERENCES public.workflow_migration_states(id)
                ON DELETE CASCADE,
            dual_run_id UUID
                REFERENCES public.workflow_dual_runs(id)
                ON DELETE SET NULL,
            status VARCHAR(20) NOT NULL
                CHECK (status IN ('passed', 'failed', 'blocked')),
            policy JSONB NOT NULL DEFAULT '{}'::jsonb,
            safe_metrics JSONB NOT NULL DEFAULT '{}'::jsonb,
            differences JSONB NOT NULL DEFAULT '[]'::jsonb,
            correlation_id UUID NOT NULL,
            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP
        );
        CREATE INDEX idx_workflow_reconciliation_state_created
            ON public.workflow_reconciliation_reports(
                migration_state_id, status, created_at DESC
            );

        CREATE TABLE public.workflow_cutover_operations (
            id UUID PRIMARY KEY,
            migration_state_id UUID NOT NULL
                REFERENCES public.workflow_migration_states(id)
                ON DELETE CASCADE,
            idempotency_key VARCHAR(500) NOT NULL,
            source_mode VARCHAR(50) NOT NULL,
            target_mode VARCHAR(50) NOT NULL,
            status VARCHAR(30) NOT NULL
                CHECK (status IN (
                    'claimed', 'succeeded', 'rolled_back', 'failed'
                )),
            gates JSONB NOT NULL DEFAULT '{}'::jsonb,
            result JSONB NOT NULL DEFAULT '{}'::jsonb,
            safe_error VARCHAR(1000),
            correlation_id UUID NOT NULL,
            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
            updated_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
            UNIQUE (migration_state_id, idempotency_key)
        );
        CREATE INDEX idx_workflow_cutover_state_created
            ON public.workflow_cutover_operations(
                migration_state_id, status, created_at DESC
            );
        """
    )


def downgrade() -> None:
    op.execute(
        """
        DROP TABLE IF EXISTS public.workflow_cutover_operations;
        DROP TABLE IF EXISTS public.workflow_reconciliation_reports;
        DROP TABLE IF EXISTS public.workflow_dual_runs;
        DROP TABLE IF EXISTS public.workflow_migration_states;
        """
    )