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

"""Add canonical SLO policies, alerts and evidence-bound data incidents."""

from alembic import op

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


def upgrade() -> None:
    op.execute(
        """
        CREATE TABLE public.data_slo_policies (
            uid UUID PRIMARY KEY,
            code VARCHAR(120) NOT NULL UNIQUE,
            name VARCHAR(300) NOT NULL,
            sli_type VARCHAR(30) NOT NULL CHECK (
                sli_type IN ('freshness','completeness','quality','delivery')
            ),
            scope_type VARCHAR(40) NOT NULL,
            scope_uid VARCHAR(500) NOT NULL,
            operator VARCHAR(2) NOT NULL CHECK (operator IN ('>=','<=')),
            target NUMERIC NOT NULL,
            window_seconds INTEGER NOT NULL CHECK (
                window_seconds BETWEEN 60 AND 31536000
            ),
            owner_uid UUID NOT NULL REFERENCES public.users(id),
            status VARCHAR(20) NOT NULL CHECK (
                status IN ('active','retired')
            ),
            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,
            UNIQUE (sli_type, scope_type, scope_uid)
        );

        CREATE TABLE public.data_incidents (
            uid UUID PRIMARY KEY,
            code VARCHAR(40) NOT NULL UNIQUE,
            dedup_key CHAR(64) NOT NULL,
            title VARCHAR(300) NOT NULL,
            severity VARCHAR(20) NOT NULL CHECK (
                severity IN ('info','warning','error','critical')
            ),
            status VARCHAR(20) NOT NULL CHECK (
                status IN ('open','monitoring','closed')
            ),
            owner_uid UUID NOT NULL REFERENCES public.users(id),
            escalation_level INTEGER NOT NULL CHECK (
                escalation_level BETWEEN 1 AND 3
            ),
            first_detected_at TIMESTAMPTZ NOT NULL,
            last_observed_at TIMESTAMPTZ NOT NULL,
            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,
            closed_by UUID REFERENCES public.users(id),
            closed_at TIMESTAMPTZ
        );

        CREATE UNIQUE INDEX uq_data_incident_open_dedup
            ON public.data_incidents(dedup_key)
            WHERE status <> 'closed';

        CREATE TABLE public.data_observability_alerts (
            uid UUID PRIMARY KEY,
            dedup_key CHAR(64) NOT NULL,
            incident_uid UUID NOT NULL REFERENCES public.data_incidents(uid),
            layer VARCHAR(20) NOT NULL CHECK (
                layer IN ('service','task','data','capacity')
            ),
            sli_type VARCHAR(30) NOT NULL CHECK (
                sli_type IN ('freshness','completeness','quality','delivery')
            ),
            title VARCHAR(300) NOT NULL,
            severity VARCHAR(20) NOT NULL CHECK (
                severity IN ('info','warning','error','critical')
            ),
            status VARCHAR(20) NOT NULL CHECK (
                status IN ('open','suppressed','recovered')
            ),
            occurrence_count INTEGER NOT NULL CHECK (occurrence_count > 0),
            escalation_level INTEGER NOT NULL CHECK (
                escalation_level BETWEEN 1 AND 3
            ),
            owner_uid UUID NOT NULL REFERENCES public.users(id),
            first_observed_at TIMESTAMPTZ NOT NULL,
            last_observed_at TIMESTAMPTZ NOT NULL,
            suppressed_until TIMESTAMPTZ,
            suppression_reason VARCHAR(500),
            delivery_status VARCHAR(30) NOT NULL CHECK (
                delivery_status IN (
                    'pending','suppressed','acknowledged','recovery_pending'
                )
            ),
            delivery_receipt JSONB NOT NULL DEFAULT '{}'::jsonb,
            evidence JSONB NOT NULL,
            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
            updated_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
            CHECK (jsonb_typeof(delivery_receipt) = 'object'),
            CHECK (jsonb_typeof(evidence) = 'object')
        );

        CREATE UNIQUE INDEX uq_data_alert_open_dedup
            ON public.data_observability_alerts(dedup_key)
            WHERE status <> 'recovered';

        CREATE TABLE public.data_observability_source_events (
            uid UUID PRIMARY KEY,
            source_event_key VARCHAR(500) NOT NULL UNIQUE,
            source_type VARCHAR(80) NOT NULL,
            source_uid VARCHAR(500) NOT NULL,
            status VARCHAR(20) NOT NULL CHECK (
                status IN ('violated','recovered')
            ),
            signal JSONB NOT NULL,
            processed_at TIMESTAMPTZ NOT NULL,
            CHECK (jsonb_typeof(signal) = 'object')
        );

        CREATE TABLE public.data_incident_impacts (
            uid BIGSERIAL PRIMARY KEY,
            incident_uid UUID NOT NULL
                REFERENCES public.data_incidents(uid) ON DELETE CASCADE,
            target_type VARCHAR(40) NOT NULL CHECK (
                target_type IN (
                    'asset','data_product','business_domain',
                    'user_group','service'
                )
            ),
            target_uid VARCHAR(500) NOT NULL,
            label VARCHAR(300) NOT NULL,
            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
            UNIQUE (incident_uid, target_type, target_uid)
        );

        CREATE TABLE public.data_incident_timeline (
            uid UUID PRIMARY KEY,
            incident_uid UUID NOT NULL
                REFERENCES public.data_incidents(uid) ON DELETE CASCADE,
            action VARCHAR(40) NOT NULL,
            actor_uid UUID NOT NULL REFERENCES public.users(id),
            evidence JSONB NOT NULL,
            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
            CHECK (jsonb_typeof(evidence) = 'object')
        );

        CREATE TABLE public.data_incident_postmortems (
            uid UUID PRIMARY KEY,
            incident_uid UUID NOT NULL UNIQUE
                REFERENCES public.data_incidents(uid) ON DELETE CASCADE,
            root_cause TEXT NOT NULL,
            user_impact TEXT NOT NULL,
            corrective_actions JSONB NOT NULL,
            closure_evidence JSONB NOT NULL,
            created_by UUID NOT NULL REFERENCES public.users(id),
            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
            CHECK (jsonb_typeof(corrective_actions) = 'array'),
            CHECK (jsonb_array_length(corrective_actions) > 0),
            CHECK (jsonb_typeof(closure_evidence) = 'array'),
            CHECK (jsonb_array_length(closure_evidence) > 0)
        );

        CREATE INDEX idx_data_alert_layer_status
            ON public.data_observability_alerts(
                layer, status, severity, last_observed_at DESC
            );
        CREATE INDEX idx_data_incident_status_updated
            ON public.data_incidents(status, updated_at DESC);
        CREATE INDEX idx_data_incident_timeline_created
            ON public.data_incident_timeline(
                incident_uid, created_at, uid
            );
        """
    )


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