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

"""Add generic semantic governance, field mappings and publication audit."""

from alembic import op

revision = "20260731_390"
down_revision = "20260730_380"
branch_labels = None
depends_on = None


def upgrade() -> None:
    op.execute(
        """
        CREATE TABLE public.semantic_assets (
            uid UUID PRIMARY KEY,
            asset_kind VARCHAR(30) NOT NULL CHECK (
                asset_kind IN ('business_term','code_set','metric')
            ),
            code VARCHAR(120) NOT NULL,
            name VARCHAR(300) NOT NULL,
            owner_uid UUID NOT NULL REFERENCES public.users(id),
            business_domain_uid UUID NOT NULL,
            status VARCHAR(20) NOT NULL CHECK (
                status IN (
                    'draft','in_review','approved','published',
                    'superseded','retired'
                )
            ),
            current_version INTEGER NOT NULL CHECK (current_version > 0),
            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 (asset_kind, code)
        );

        CREATE TABLE public.semantic_asset_versions (
            uid UUID PRIMARY KEY,
            asset_uid UUID NOT NULL REFERENCES public.semantic_assets(uid),
            version INTEGER NOT NULL CHECK (version > 0),
            status VARCHAR(20) NOT NULL CHECK (
                status IN (
                    'draft','in_review','approved','published',
                    'superseded','retired'
                )
            ),
            content_hash CHAR(64) NOT NULL,
            definition JSONB NOT NULL,
            change_reason VARCHAR(1000) NOT NULL,
            created_by UUID NOT NULL REFERENCES public.users(id),
            rollback_from_version INTEGER,
            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
            published_at TIMESTAMPTZ,
            UNIQUE (asset_uid, version),
            CHECK (jsonb_typeof(definition) = 'object'),
            CHECK (
                rollback_from_version IS NULL
                OR rollback_from_version > 0
            )
        );

        CREATE TABLE public.semantic_asset_reviews (
            uid UUID PRIMARY KEY,
            asset_uid UUID NOT NULL REFERENCES public.semantic_assets(uid),
            version INTEGER NOT NULL CHECK (version > 0),
            decision VARCHAR(20) NOT NULL
                CHECK (decision IN ('approve','reject')),
            reason VARCHAR(1000) NOT NULL,
            reviewer_uid UUID NOT NULL REFERENCES public.users(id),
            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
            UNIQUE (asset_uid, version),
            FOREIGN KEY (asset_uid, version)
                REFERENCES public.semantic_asset_versions(asset_uid, version)
        );

        CREATE TABLE public.semantic_asset_links (
            uid UUID PRIMARY KEY,
            asset_uid UUID NOT NULL REFERENCES public.semantic_assets(uid),
            version INTEGER NOT NULL CHECK (version > 0),
            link_type VARCHAR(30) NOT NULL CHECK (
                link_type IN ('asset','data_element','standard','code_value')
            ),
            target_type VARCHAR(60) NOT NULL,
            target_uid VARCHAR(200) NOT NULL,
            target_key VARCHAR(200),
            metadata JSONB NOT NULL DEFAULT '{}'::jsonb,
            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
            FOREIGN KEY (asset_uid, version)
                REFERENCES public.semantic_asset_versions(asset_uid, version),
            UNIQUE (
                asset_uid, version, link_type, target_type,
                target_uid, target_key
            ),
            CHECK (jsonb_typeof(metadata) = 'object')
        );

        CREATE TABLE public.data_element_field_mappings (
            uid UUID PRIMARY KEY,
            asset_uid UUID NOT NULL
                REFERENCES public.active_metadata_assets(uid),
            field_name VARCHAR(200) NOT NULL,
            data_element_uid UUID NOT NULL REFERENCES public.data_elements(uid),
            data_element_version INTEGER NOT NULL CHECK (
                data_element_version > 0
            ),
            owner_uid UUID NOT NULL REFERENCES public.users(id),
            status VARCHAR(20) NOT NULL
                CHECK (status IN ('published','retired')),
            evidence JSONB NOT NULL DEFAULT '{}'::jsonb,
            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 (asset_uid, field_name, data_element_uid),
            CHECK (jsonb_typeof(evidence) = 'object')
        );

        CREATE TABLE public.semantic_publication_audits (
            uid UUID PRIMARY KEY,
            target_type VARCHAR(40) NOT NULL CHECK (
                target_type IN (
                    'semantic_asset','data_standard','field_mapping'
                )
            ),
            target_uid UUID NOT NULL,
            target_version INTEGER NOT NULL CHECK (target_version > 0),
            action VARCHAR(40) NOT NULL CHECK (
                action IN (
                    'created','revised','submitted','approved','rejected',
                    'published','rolled_back','mapped','retired'
                )
            ),
            actor_uid UUID NOT NULL REFERENCES public.users(id),
            detail JSONB NOT NULL DEFAULT '{}'::jsonb,
            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
            CHECK (jsonb_typeof(detail) = 'object')
        );

        CREATE INDEX idx_semantic_assets_kind_status
            ON public.semantic_assets(asset_kind, status, updated_at DESC);
        CREATE INDEX idx_semantic_versions_asset
            ON public.semantic_asset_versions(asset_uid, version DESC);
        CREATE INDEX idx_semantic_links_target
            ON public.semantic_asset_links(target_type, target_uid, target_key);
        CREATE UNIQUE INDEX uq_published_physical_field_mapping
            ON public.data_element_field_mappings(asset_uid, field_name)
            WHERE status = 'published';
        CREATE INDEX idx_field_mappings_element
            ON public.data_element_field_mappings(
                data_element_uid, status, updated_at DESC
            );
        CREATE INDEX idx_semantic_audits_target
            ON public.semantic_publication_audits(
                target_type, target_uid, target_version
            );
        """
    )


def downgrade() -> None:
    raise RuntimeError(
        "semantic governance history is append-only; "
        "downgrade requires an approved archival migration"
    )