DataOps-platform/migrations/versions/20260716_20_workflow_versions.py

"""Add governed DataFlow to n8n Workflow version mappings."""

from alembic import op


revision = "20260716_20"
down_revision = "20260716_10"
branch_labels = None
depends_on = None


def upgrade() -> None:
    op.execute(
        """
        CREATE TABLE IF NOT EXISTS public.dataflow_workflow_versions (
            id UUID PRIMARY KEY,
            dataflow_uid UUID NOT NULL,
            environment VARCHAR(20) NOT NULL
                CHECK (environment IN ('development', 'test', 'production')),
            version_no INTEGER NOT NULL CHECK (version_no > 0),
            n8n_workflow_id VARCHAR(100) NOT NULL,
            n8n_workflow_name VARCHAR(200),
            definition_hash CHAR(64) NOT NULL,
            definition_snapshot JSONB NOT NULL,
            status VARCHAR(30) NOT NULL DEFAULT 'draft'
                CHECK (status IN ('draft', 'validating', 'active',
                    'superseded', 'activation_failed')),
            activation_error VARCHAR(1000),
            activation_correlation_id UUID,
            created_by UUID REFERENCES public.users(id) ON DELETE SET NULL,
            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
            activated_by UUID REFERENCES public.users(id) ON DELETE SET NULL,
            activated_at TIMESTAMPTZ,
            updated_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
            UNIQUE (dataflow_uid, environment, version_no),
            UNIQUE (dataflow_uid, environment, definition_hash)
        );
        CREATE UNIQUE INDEX IF NOT EXISTS uq_dataflow_workflow_active_environment
            ON public.dataflow_workflow_versions(dataflow_uid, environment)
            WHERE status = 'active';
        CREATE INDEX IF NOT EXISTS idx_workflow_versions_n8n
            ON public.dataflow_workflow_versions(n8n_workflow_id);
        """
    )


def downgrade() -> None:
    pass