DataOps-platform/migrations/versions/20260731_400_quality_operations.py

"""Add generic deterministic quality profiling, findings and SLA events."""

from alembic import op

revision = "20260731_400"
down_revision = "20260731_390"
branch_labels = None
depends_on = None


def upgrade() -> None:
    op.execute(
        """
        CREATE TABLE public.quality_templates (
            uid UUID PRIMARY KEY,
            code VARCHAR(120) NOT NULL UNIQUE,
            name VARCHAR(300) NOT NULL,
            owner_uid UUID NOT NULL REFERENCES public.users(id),
            status VARCHAR(20) NOT NULL CHECK (
                status IN ('draft','published','retired')
            ),
            current_version INTEGER NOT NULL CHECK (current_version > 0),
            active_version_uid UUID,
            created_by UUID NOT NULL REFERENCES public.users(id),
            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
            updated_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP
        );

        CREATE TABLE public.quality_template_versions (
            uid UUID PRIMARY KEY,
            template_uid UUID NOT NULL
                REFERENCES public.quality_templates(uid),
            version INTEGER NOT NULL CHECK (version > 0),
            status VARCHAR(20) NOT NULL CHECK (
                status IN ('draft','published','superseded')
            ),
            definition JSONB NOT NULL,
            content_hash CHAR(64) NOT NULL,
            created_by UUID NOT NULL REFERENCES public.users(id),
            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
            published_by UUID REFERENCES public.users(id),
            published_at TIMESTAMPTZ,
            UNIQUE (template_uid, version),
            UNIQUE (template_uid, content_hash),
            CHECK (jsonb_typeof(definition) = 'object')
        );

        ALTER TABLE public.quality_templates
            ADD CONSTRAINT fk_quality_template_active_version
            FOREIGN KEY (active_version_uid)
            REFERENCES public.quality_template_versions(uid);

        CREATE UNIQUE INDEX uq_quality_template_published_version
            ON public.quality_template_versions(template_uid)
            WHERE status = 'published';

        CREATE TABLE public.quality_profile_runs (
            uid UUID PRIMARY KEY,
            template_uid UUID NOT NULL
                REFERENCES public.quality_templates(uid),
            template_version_uid UUID NOT NULL
                REFERENCES public.quality_template_versions(uid),
            template_hash CHAR(64) NOT NULL,
            asset_uid UUID NOT NULL
                REFERENCES public.active_metadata_assets(uid),
            source_uid UUID NOT NULL
                REFERENCES public.ingestion_sources(uid),
            business_domain_uid VARCHAR(200),
            batch_key VARCHAR(160) NOT NULL,
            status VARCHAR(20) NOT NULL CHECK (status IN ('success','failed')),
            row_count INTEGER NOT NULL CHECK (row_count > 0),
            score NUMERIC(6,2) NOT NULL CHECK (
                score >= 0 AND score <= 100
            ),
            source_observed_at TIMESTAMPTZ NOT NULL,
            previous_run_uid UUID
                REFERENCES public.quality_profile_runs(uid),
            comparison JSONB NOT NULL,
            profile JSONB NOT NULL,
            field_bindings JSONB NOT NULL,
            finding_count INTEGER NOT NULL CHECK (finding_count >= 0),
            deterministic BOOLEAN NOT NULL DEFAULT TRUE CHECK (deterministic),
            created_by UUID NOT NULL REFERENCES public.users(id),
            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
            UNIQUE (asset_uid, batch_key),
            CHECK (jsonb_typeof(comparison) = 'object'),
            CHECK (jsonb_typeof(profile) = 'object'),
            CHECK (jsonb_typeof(field_bindings) = 'object')
        );

        CREATE TABLE public.quality_profile_metrics (
            uid UUID PRIMARY KEY,
            run_uid UUID NOT NULL
                REFERENCES public.quality_profile_runs(uid) ON DELETE CASCADE,
            field_name VARCHAR(200) NOT NULL,
            metric_value JSONB NOT NULL,
            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
            UNIQUE (run_uid, field_name),
            CHECK (jsonb_typeof(metric_value) = 'object')
        );

        CREATE TABLE public.quality_findings (
            uid UUID PRIMARY KEY,
            run_uid UUID NOT NULL
                REFERENCES public.quality_profile_runs(uid) ON DELETE CASCADE,
            asset_uid UUID NOT NULL
                REFERENCES public.active_metadata_assets(uid),
            finding_type VARCHAR(30) NOT NULL CHECK (
                finding_type IN (
                    'completeness','uniqueness','pattern','duplicate',
                    'outlier','volume','distribution','schema','freshness'
                )
            ),
            field_name VARCHAR(200),
            severity VARCHAR(20) NOT NULL CHECK (
                severity IN ('info','warning','error','critical')
            ),
            status VARCHAR(20) NOT NULL CHECK (
                status IN ('open','acknowledged','resolved')
            ),
            actual JSONB NOT NULL,
            expected JSONB NOT NULL,
            recurrence_key CHAR(64) NOT NULL,
            occurrence_number INTEGER NOT NULL CHECK (occurrence_number > 0),
            evidence JSONB NOT NULL,
            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
            CHECK (jsonb_typeof(actual) IN ('object','number','string')),
            CHECK (jsonb_typeof(expected) IN ('object','number','string')),
            CHECK (jsonb_typeof(evidence) = 'object')
        );

        CREATE TABLE public.quality_sla_events (
            uid UUID PRIMARY KEY,
            run_uid UUID NOT NULL
                REFERENCES public.quality_profile_runs(uid) ON DELETE CASCADE,
            asset_uid UUID NOT NULL
                REFERENCES public.active_metadata_assets(uid),
            sla_type VARCHAR(30) NOT NULL CHECK (
                sla_type IN ('freshness','quality_score')
            ),
            status VARCHAR(20) NOT NULL CHECK (
                status IN ('met','violated','recovered')
            ),
            severity VARCHAR(20) NOT NULL CHECK (
                severity IN ('info','warning','error','critical')
            ),
            actual NUMERIC NOT NULL,
            threshold NUMERIC NOT NULL,
            owner_uid UUID NOT NULL REFERENCES public.users(id),
            escalation_level INTEGER NOT NULL CHECK (
                escalation_level BETWEEN 0 AND 3
            ),
            evidence JSONB NOT NULL,
            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
            UNIQUE (run_uid, sla_type),
            CHECK (jsonb_typeof(evidence) = 'object')
        );

        CREATE INDEX idx_quality_runs_asset_created
            ON public.quality_profile_runs(asset_uid, created_at DESC);
        CREATE INDEX idx_quality_runs_domain_created
            ON public.quality_profile_runs(
                business_domain_uid, created_at DESC
            );
        CREATE INDEX idx_quality_findings_recurrence
            ON public.quality_findings(
                recurrence_key, occurrence_number DESC
            );
        CREATE INDEX idx_quality_findings_asset_status
            ON public.quality_findings(asset_uid, status, created_at DESC);
        CREATE INDEX idx_quality_sla_asset_status
            ON public.quality_sla_events(
                asset_uid, sla_type, status, created_at DESC
            );
        """
    )


def downgrade() -> None:
    raise RuntimeError(
        "quality operation evidence is append-only; "
        "downgrade requires an approved archival migration"
    )