DataOps-platform/migrations/versions/20260723_110_ai_data_rules.py

"""Add governed AI data rules, standards, production lines, and deployments."""

from alembic import op


revision = "20260723_110"
down_revision = "20260722_110"
branch_labels = None
depends_on = None


def upgrade() -> None:
    op.execute(
        """
        CREATE TABLE public.data_rules (
            id UUID PRIMARY KEY,
            rule_uid UUID NOT NULL UNIQUE,
            name VARCHAR(200) NOT NULL,
            category VARCHAR(50) NOT NULL,
            owner_uid UUID REFERENCES public.users(id) ON DELETE SET NULL,
            status VARCHAR(20) NOT NULL DEFAULT 'active'
                CHECK (status IN ('active','deprecated')),
            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
            updated_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP
        );

        CREATE TABLE public.data_rule_versions (
            id UUID PRIMARY KEY,
            rule_uid UUID NOT NULL REFERENCES public.data_rules(rule_uid)
                ON DELETE RESTRICT,
            version_no INTEGER NOT NULL CHECK (version_no > 0),
            source_text TEXT NOT NULL,
            source_language VARCHAR(20) NOT NULL DEFAULT 'zh-CN',
            rule_spec JSONB NOT NULL,
            spec_hash CHAR(64) NOT NULL,
            generated_kind VARCHAR(30) NOT NULL DEFAULT 'rulespec'
                CHECK (generated_kind IN (
                    'rulespec','cel','sql','polars','generated_python'
                )),
            status VARCHAR(20) NOT NULL DEFAULT 'draft'
                CHECK (status IN (
                    'draft','validated','published','deprecated','rejected'
                )),
            created_by UUID REFERENCES public.users(id) ON DELETE SET NULL,
            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
            published_at TIMESTAMPTZ,
            UNIQUE (rule_uid, version_no),
            UNIQUE (rule_uid, spec_hash)
        );
        CREATE INDEX idx_data_rule_versions_status
            ON public.data_rule_versions(rule_uid, status, version_no DESC);

        CREATE TABLE public.rule_generation_runs (
            id UUID PRIMARY KEY,
            rule_version_id UUID REFERENCES public.data_rule_versions(id)
                ON DELETE SET NULL,
            authoring_surface VARCHAR(30) NOT NULL
                CHECK (authoring_surface IN ('data_standard','data_flow')),
            source_text_hash CHAR(64) NOT NULL,
            model_provider VARCHAR(80) NOT NULL,
            model_name VARCHAR(120) NOT NULL,
            prompt_version VARCHAR(80) NOT NULL,
            schema_version VARCHAR(40) NOT NULL,
            context_hash CHAR(64) NOT NULL,
            candidate_hash CHAR(64) NOT NULL,
            confidence NUMERIC(5,4) NOT NULL
                CHECK (confidence >= 0 AND confidence <= 1),
            assumptions JSONB NOT NULL DEFAULT '[]'::jsonb,
            ambiguities JSONB NOT NULL DEFAULT '[]'::jsonb,
            repair_attempts INTEGER NOT NULL DEFAULT 0
                CHECK (repair_attempts >= 0 AND repair_attempts <= 2),
            decision VARCHAR(30) NOT NULL CHECK (decision IN (
                'ready','clarification_required','rejected','failed'
            )),
            decision_detail JSONB NOT NULL DEFAULT '{}'::jsonb,
            correlation_id UUID NOT NULL,
            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP
        );
        CREATE INDEX idx_rule_generation_correlation
            ON public.rule_generation_runs(correlation_id, created_at);

        CREATE TABLE public.data_standards (
            id UUID PRIMARY KEY,
            standard_uid UUID NOT NULL UNIQUE,
            name VARCHAR(200) NOT NULL,
            owner_uid UUID REFERENCES public.users(id) ON DELETE SET NULL,
            status VARCHAR(20) NOT NULL DEFAULT 'active'
                CHECK (status IN ('active','deprecated')),
            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
            updated_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP
        );

        CREATE TABLE public.data_standard_versions (
            id UUID PRIMARY KEY,
            standard_uid UUID NOT NULL
                REFERENCES public.data_standards(standard_uid)
                ON DELETE RESTRICT,
            version_no INTEGER NOT NULL CHECK (version_no > 0),
            source_text TEXT NOT NULL,
            standard_spec JSONB NOT NULL,
            spec_hash CHAR(64) NOT NULL,
            scope JSONB NOT NULL,
            status VARCHAR(20) NOT NULL DEFAULT 'draft'
                CHECK (status IN (
                    'draft','validated','published','deprecated','rejected'
                )),
            created_by UUID REFERENCES public.users(id) ON DELETE SET NULL,
            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
            published_at TIMESTAMPTZ,
            UNIQUE (standard_uid, version_no),
            UNIQUE (standard_uid, spec_hash)
        );
        CREATE INDEX idx_data_standard_versions_status
            ON public.data_standard_versions(
                standard_uid, status, version_no DESC
            );

        CREATE TABLE public.standard_rule_bindings (
            id UUID PRIMARY KEY,
            standard_version_id UUID NOT NULL
                REFERENCES public.data_standard_versions(id)
                ON DELETE RESTRICT,
            clause_id VARCHAR(100) NOT NULL,
            rule_version_id UUID NOT NULL
                REFERENCES public.data_rule_versions(id)
                ON DELETE RESTRICT,
            severity VARCHAR(20) NOT NULL CHECK (
                severity IN ('info','warning','error','critical')
            ),
            exception_policy VARCHAR(20) NOT NULL CHECK (
                exception_policy IN ('reject','quarantine','warn','fail')
            ),
            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
            UNIQUE (standard_version_id, clause_id)
        );
        CREATE INDEX idx_standard_rule_binding_rule
            ON public.standard_rule_bindings(rule_version_id);

        CREATE TABLE public.dataflow_versions (
            id UUID PRIMARY KEY,
            dataflow_uid UUID NOT NULL,
            version_no INTEGER NOT NULL CHECK (version_no > 0),
            name VARCHAR(200) NOT NULL,
            source_text TEXT,
            dataflow_spec JSONB NOT NULL,
            input_schema_hashes JSONB NOT NULL,
            output_schema_hash CHAR(64) NOT NULL,
            package JSONB,
            package_hash CHAR(64),
            status VARCHAR(20) NOT NULL DEFAULT 'draft'
                CHECK (status IN (
                    'draft','validated','released','deprecated','rejected'
                )),
            created_by UUID REFERENCES public.users(id) ON DELETE SET NULL,
            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
            released_at TIMESTAMPTZ,
            UNIQUE (dataflow_uid, version_no)
        );
        CREATE INDEX idx_dataflow_versions_status
            ON public.dataflow_versions(
                dataflow_uid, status, version_no DESC
            );

        CREATE TABLE public.dataflow_component_bindings (
            id UUID PRIMARY KEY,
            dataflow_version_id UUID NOT NULL
                REFERENCES public.dataflow_versions(id)
                ON DELETE RESTRICT,
            component_id VARCHAR(100) NOT NULL,
            component_kind VARCHAR(30) NOT NULL CHECK (
                component_kind IN (
                    'standard.enforce','rule.apply','quality.check'
                )
            ),
            standard_version_id UUID
                REFERENCES public.data_standard_versions(id)
                ON DELETE RESTRICT,
            rule_version_id UUID
                REFERENCES public.data_rule_versions(id)
                ON DELETE RESTRICT,
            stage VARCHAR(30) NOT NULL,
            order_no INTEGER NOT NULL CHECK (order_no >= 0),
            idempotency JSONB,
            provenance JSONB NOT NULL DEFAULT '{}'::jsonb,
            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
            UNIQUE (dataflow_version_id, component_id),
            CHECK (
                (
                    component_kind = 'standard.enforce'
                    AND standard_version_id IS NOT NULL
                    AND rule_version_id IS NULL
                )
                OR (
                    component_kind IN ('rule.apply','quality.check')
                    AND standard_version_id IS NULL
                    AND rule_version_id IS NOT NULL
                )
            )
        );
        CREATE INDEX idx_dataflow_component_standard
            ON public.dataflow_component_bindings(standard_version_id)
            WHERE standard_version_id IS NOT NULL;
        CREATE INDEX idx_dataflow_component_rule
            ON public.dataflow_component_bindings(rule_version_id)
            WHERE rule_version_id IS NOT NULL;

        CREATE TABLE public.rule_execution_plans (
            id UUID PRIMARY KEY,
            component_binding_id UUID NOT NULL
                REFERENCES public.dataflow_component_bindings(id)
                ON DELETE RESTRICT,
            backend VARCHAR(30) NOT NULL CHECK (
                backend IN (
                    'sql_pushdown','polars_batch','quality_check',
                    'generated_python','external_adapter'
                )
            ),
            compiler_version VARCHAR(80) NOT NULL,
            plan JSONB NOT NULL,
            plan_hash CHAR(64) NOT NULL,
            schema_hashes JSONB NOT NULL,
            status VARCHAR(20) NOT NULL DEFAULT 'compiled'
                CHECK (status IN ('compiled','published','revoked')),
            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
            UNIQUE (component_binding_id, plan_hash)
        );

        CREATE TABLE public.rule_artifacts (
            id UUID PRIMARY KEY,
            rule_version_id UUID NOT NULL
                REFERENCES public.data_rule_versions(id)
                ON DELETE RESTRICT,
            kind VARCHAR(30) NOT NULL CHECK (
                kind IN ('cel','sql','polars','python','test','manifest')
            ),
            uri VARCHAR(1000) NOT NULL,
            artifact_digest CHAR(64) NOT NULL,
            signature VARCHAR(2000),
            runtime_digest CHAR(64),
            dependency_manifest JSONB NOT NULL DEFAULT '{}'::jsonb,
            status VARCHAR(20) NOT NULL DEFAULT 'built'
                CHECK (status IN ('built','signed','published','revoked')),
            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
            UNIQUE (rule_version_id, kind, artifact_digest)
        );

        CREATE TABLE public.dataflow_deployments (
            id UUID PRIMARY KEY,
            dataflow_version_id UUID NOT NULL
                REFERENCES public.dataflow_versions(id)
                ON DELETE RESTRICT,
            environment VARCHAR(20) NOT NULL CHECK (
                environment IN ('development','test','production')
            ),
            workflow_version_id UUID
                REFERENCES public.dataflow_workflow_versions(id)
                ON DELETE SET NULL,
            schedule_plan_id UUID
                REFERENCES public.workflow_schedules(id)
                ON DELETE SET NULL,
            deployment_config JSONB NOT NULL DEFAULT '{}'::jsonb,
            status VARCHAR(30) NOT NULL DEFAULT 'draft' CHECK (
                status IN (
                    'draft','disabled','canary','active','superseded',
                    'rolled_back','failed'
                )
            ),
            activated_by UUID REFERENCES public.users(id) ON DELETE SET NULL,
            activated_at TIMESTAMPTZ,
            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
            updated_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
            UNIQUE (dataflow_version_id, environment)
        );
        CREATE UNIQUE INDEX uq_dataflow_deployment_active_environment
            ON public.dataflow_deployments(dataflow_version_id, environment)
            WHERE status = 'active';

        CREATE TABLE public.rule_runs (
            id UUID PRIMARY KEY,
            workflow_run_id UUID REFERENCES public.workflow_runs(id)
                ON DELETE SET NULL,
            deployment_id UUID NOT NULL
                REFERENCES public.dataflow_deployments(id)
                ON DELETE RESTRICT,
            component_binding_id UUID NOT NULL
                REFERENCES public.dataflow_component_bindings(id)
                ON DELETE RESTRICT,
            rule_version_id UUID NOT NULL
                REFERENCES public.data_rule_versions(id)
                ON DELETE RESTRICT,
            plan_hash CHAR(64) NOT NULL,
            rows_in BIGINT NOT NULL DEFAULT 0 CHECK (rows_in >= 0),
            rows_out BIGINT NOT NULL DEFAULT 0 CHECK (rows_out >= 0),
            rows_rejected BIGINT NOT NULL DEFAULT 0 CHECK (rows_rejected >= 0),
            rows_quarantined BIGINT NOT NULL DEFAULT 0 CHECK (
                rows_quarantined >= 0
            ),
            status VARCHAR(30) NOT NULL CHECK (
                status IN (
                    'queued','running','success','failed','unknown','cancelled'
                )
            ),
            timings JSONB NOT NULL DEFAULT '{}'::jsonb,
            correlation_id UUID NOT NULL,
            started_at TIMESTAMPTZ,
            finished_at TIMESTAMPTZ,
            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP
        );
        CREATE INDEX idx_rule_runs_correlation
            ON public.rule_runs(correlation_id, created_at);

        CREATE TABLE public.rule_violation_samples (
            id UUID PRIMARY KEY,
            rule_run_id UUID NOT NULL REFERENCES public.rule_runs(id)
                ON DELETE CASCADE,
            artifact_ref VARCHAR(1000) NOT NULL,
            sample_count INTEGER NOT NULL CHECK (
                sample_count >= 0 AND sample_count <= 1000
            ),
            redaction_policy VARCHAR(100) NOT NULL,
            expires_at TIMESTAMPTZ NOT NULL,
            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP
        );
        """
    )


def downgrade() -> None:
    # Published standards, rules, production lines, deployments, and run evidence
    # are intentionally retained during application rollback.
    pass