DataOps-platform/migrations/versions/20260718_50_datasource_credentials.py

"""Add encrypted, versioned external data-source credentials."""

from alembic import op


revision = "20260718_50"
down_revision = "20260716_40"
branch_labels = None
depends_on = None


def upgrade() -> None:
    op.execute(
        """
        CREATE TABLE IF NOT EXISTS public.datasource_credentials (
            id UUID PRIMARY KEY,
            data_source_uid UUID NOT NULL,
            credential_version INTEGER NOT NULL
                CHECK (credential_version > 0),
            encrypted_payload BYTEA NOT NULL,
            nonce BYTEA NOT NULL CHECK (octet_length(nonce) = 12),
            key_version VARCHAR(40) NOT NULL,
            status VARCHAR(20) NOT NULL DEFAULT 'active'
                CHECK (status IN ('active', 'retired', 'revoked')),
            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
            retired_at TIMESTAMPTZ,
            UNIQUE (data_source_uid, credential_version)
        );

        CREATE UNIQUE INDEX IF NOT EXISTS uq_datasource_credential_active
            ON public.datasource_credentials(data_source_uid)
            WHERE status = 'active';
        CREATE INDEX IF NOT EXISTS idx_datasource_credential_lookup
            ON public.datasource_credentials(
                data_source_uid, credential_version, status
            );

        CREATE TABLE IF NOT EXISTS public.datasource_credential_audit_events (
            id BIGSERIAL PRIMARY KEY,
            data_source_uid UUID NOT NULL,
            credential_version INTEGER,
            actor_uid UUID REFERENCES public.users(id) ON DELETE SET NULL,
            event_type VARCHAR(60) NOT NULL,
            success BOOLEAN NOT NULL,
            safe_detail VARCHAR(500),
            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP
        );
        CREATE INDEX IF NOT EXISTS idx_datasource_credential_audit_source
            ON public.datasource_credential_audit_events(
                data_source_uid, created_at DESC
            );
        CREATE INDEX IF NOT EXISTS idx_datasource_credential_audit_actor
            ON public.datasource_credential_audit_events(
                actor_uid, created_at DESC
            );
        """
    )


def downgrade() -> None:
    # Credential and audit history remain available during application rollback.
    pass