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

"""Add the durable single-use Runner task execution ledger."""

from alembic import op


revision = "20260719_70"
down_revision = "20260718_60"
branch_labels = None
depends_on = None


def upgrade() -> None:
    op.execute(
        """
        CREATE TABLE public.runner_task_executions (
            token_jti UUID PRIMARY KEY,
            task_uid UUID NOT NULL,
            dataflow_uid UUID NOT NULL,
            workflow_version INTEGER NOT NULL CHECK (workflow_version > 0),
            correlation_id UUID NOT NULL,
            node_id VARCHAR(100) NOT NULL,
            node_type VARCHAR(100) NOT NULL,
            data_source_uid UUID,
            idempotency_key VARCHAR(500),
            status VARCHAR(20) NOT NULL
                CHECK (status IN ('running', 'success', 'failed', 'unknown')),
            commit_outcome VARCHAR(30) NOT NULL
                CHECK (commit_outcome IN (
                    'not_applicable', 'not_committed', 'committed', 'unknown'
                )),
            safe_detail VARCHAR(500),
            expires_at TIMESTAMPTZ NOT NULL,
            started_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
            finished_at TIMESTAMPTZ,
            UNIQUE (task_uid)
        );
        CREATE INDEX idx_runner_task_executions_correlation
            ON public.runner_task_executions(correlation_id, started_at DESC);
        CREATE INDEX idx_runner_task_executions_datasource
            ON public.runner_task_executions(data_source_uid, started_at DESC)
            WHERE data_source_uid IS NOT NULL;
        CREATE INDEX idx_runner_task_executions_expiry
            ON public.runner_task_executions(expires_at);
        """
    )

def downgrade() -> None:
    op.execute(
        "DROP TABLE IF EXISTS public.runner_task_executions;"
    )