"""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; """ )