DataOps-platform/migrations/versions/20260802_440_product_governance.py

"""Add governed data products, applications, contracts and certificates."""

from alembic import op

revision = "20260802_440"
down_revision = "20260802_430"
branch_labels = None
depends_on = None


def upgrade() -> None:
    op.execute(
        """
        CREATE TABLE public.governed_data_products (
            uid UUID PRIMARY KEY,
            legacy_product_id INTEGER NOT NULL UNIQUE
                REFERENCES public.data_products(id) ON DELETE RESTRICT,
            product_code VARCHAR(120) NOT NULL UNIQUE,
            name VARCHAR(300) NOT NULL,
            product_type VARCHAR(30) NOT NULL CHECK (
                product_type IN ('database','api','file','data_product')
            ),
            owner_uid UUID NOT NULL REFERENCES public.users(id),
            business_domain_uid UUID NOT NULL,
            description VARCHAR(2000) NOT NULL,
            quality_target NUMERIC(6,2) NOT NULL CHECK (
                quality_target >= 0 AND quality_target <= 100
            ),
            sla JSONB NOT NULL,
            status VARCHAR(20) NOT NULL CHECK (
                status IN ('draft','in_review','active','suspended','retired')
            ),
            current_version INTEGER NOT NULL DEFAULT 1 CHECK (current_version > 0),
            created_by UUID NOT NULL REFERENCES public.users(id),
            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
            updated_by UUID NOT NULL REFERENCES public.users(id),
            updated_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
            retired_at TIMESTAMPTZ,
            CHECK (jsonb_typeof(sla) = 'object')
        );
        CREATE INDEX idx_governed_product_domain_status
            ON public.governed_data_products(
                business_domain_uid, status, updated_at DESC
            );
        CREATE INDEX idx_governed_product_owner_status
            ON public.governed_data_products(owner_uid, status, updated_at DESC);

        CREATE TABLE public.data_product_governance_events (
            uid UUID PRIMARY KEY,
            product_uid UUID NOT NULL
                REFERENCES public.governed_data_products(uid) ON DELETE CASCADE,
            product_version INTEGER NOT NULL CHECK (product_version > 0),
            action VARCHAR(50) NOT NULL,
            actor_uid UUID NOT NULL REFERENCES public.users(id),
            payload JSONB NOT NULL DEFAULT '{}'::jsonb,
            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
            CHECK (jsonb_typeof(payload) = 'object')
        );
        CREATE INDEX idx_product_governance_timeline
            ON public.data_product_governance_events(product_uid, created_at, uid);

        CREATE TABLE public.data_product_applications (
            uid UUID PRIMARY KEY,
            application_code VARCHAR(40) NOT NULL UNIQUE,
            title VARCHAR(300) NOT NULL,
            source_type VARCHAR(30) NOT NULL CHECK (
                source_type IN ('database','api','file','data_product')
            ),
            source_ref JSONB NOT NULL,
            business_domain_uid UUID NOT NULL,
            purpose VARCHAR(1000) NOT NULL,
            requested_fields JSONB NOT NULL,
            status VARCHAR(30) NOT NULL CHECK (
                status IN (
                    'draft','pending_approval','approved','rejected',
                    'fulfilled','cancelled'
                )
            ),
            approval_task_uid UUID REFERENCES public.governance_tasks(uid),
            product_uid UUID REFERENCES public.governed_data_products(uid),
            current_version INTEGER NOT NULL DEFAULT 1 CHECK (current_version > 0),
            created_by UUID NOT NULL REFERENCES public.users(id),
            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
            updated_by UUID NOT NULL REFERENCES public.users(id),
            updated_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
            CHECK (jsonb_typeof(source_ref) = 'object'),
            CHECK (jsonb_typeof(requested_fields) = 'array')
        );
        CREATE INDEX idx_product_application_worklist
            ON public.data_product_applications(
                created_by, status, updated_at DESC
            );
        CREATE INDEX idx_product_application_domain
            ON public.data_product_applications(
                business_domain_uid, status, updated_at DESC
            );

        CREATE TABLE public.data_product_application_events (
            uid UUID PRIMARY KEY,
            application_uid UUID NOT NULL
                REFERENCES public.data_product_applications(uid) ON DELETE CASCADE,
            application_version INTEGER NOT NULL CHECK (application_version > 0),
            action VARCHAR(50) NOT NULL,
            actor_uid UUID NOT NULL REFERENCES public.users(id),
            payload JSONB NOT NULL DEFAULT '{}'::jsonb,
            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
            CHECK (jsonb_typeof(payload) = 'object')
        );
        CREATE INDEX idx_product_application_timeline
            ON public.data_product_application_events(
                application_uid, created_at, uid
            );

        CREATE TABLE public.data_product_contracts (
            uid UUID PRIMARY KEY,
            product_uid UUID NOT NULL UNIQUE
                REFERENCES public.governed_data_products(uid) ON DELETE RESTRICT,
            contract_code VARCHAR(180) NOT NULL UNIQUE,
            status VARCHAR(20) NOT NULL CHECK (
                status IN ('draft','active','terminated')
            ),
            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_by UUID NOT NULL REFERENCES public.users(id),
            updated_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
            terminated_at TIMESTAMPTZ,
            termination_reason VARCHAR(1000)
        );

        CREATE TABLE public.data_product_contract_versions (
            uid UUID PRIMARY KEY,
            contract_uid UUID NOT NULL
                REFERENCES public.data_product_contracts(uid) ON DELETE RESTRICT,
            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,
            compatibility JSONB 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 (contract_uid, version),
            UNIQUE (contract_uid, content_hash),
            CHECK (jsonb_typeof(definition) = 'object'),
            CHECK (jsonb_typeof(compatibility) = 'object')
        );
        ALTER TABLE public.data_product_contracts
            ADD CONSTRAINT data_product_contract_active_version_fk
            FOREIGN KEY (active_version_uid)
            REFERENCES public.data_product_contract_versions(uid);
        CREATE UNIQUE INDEX uq_data_product_contract_published_version
            ON public.data_product_contract_versions(contract_uid)
            WHERE status = 'published';

        CREATE TABLE public.data_product_certificates (
            uid UUID PRIMARY KEY,
            certificate_code VARCHAR(50) NOT NULL UNIQUE,
            product_uid UUID NOT NULL
                REFERENCES public.governed_data_products(uid) ON DELETE RESTRICT,
            contract_uid UUID NOT NULL
                REFERENCES public.data_product_contracts(uid) ON DELETE RESTRICT,
            contract_version INTEGER NOT NULL CHECK (contract_version > 0),
            approval_task_uid UUID NOT NULL REFERENCES public.governance_tasks(uid),
            evidence_snapshot JSONB NOT NULL,
            status VARCHAR(20) NOT NULL CHECK (
                status IN ('qualified','unqualified')
            ),
            content_hash CHAR(64) NOT NULL,
            issued_by UUID NOT NULL REFERENCES public.users(id),
            issued_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
            UNIQUE (product_uid, content_hash),
            CHECK (jsonb_typeof(evidence_snapshot) = 'object')
        );
        CREATE INDEX idx_product_certificate_latest
            ON public.data_product_certificates(product_uid, issued_at DESC);

        CREATE TABLE public.data_product_feedback (
            uid UUID PRIMARY KEY,
            product_uid UUID NOT NULL
                REFERENCES public.governed_data_products(uid) ON DELETE CASCADE,
            category VARCHAR(30) NOT NULL CHECK (
                category IN (
                    'quality','freshness','usability',
                    'documentation','service'
                )
            ),
            rating INTEGER NOT NULL CHECK (rating BETWEEN 1 AND 5),
            summary VARCHAR(300) NOT NULL,
            details VARCHAR(2000) NOT NULL,
            status VARCHAR(20) NOT NULL CHECK (
                status IN ('open','triaged','in_progress','resolved','closed')
            ),
            assignee_uid UUID REFERENCES public.users(id),
            resolution VARCHAR(2000),
            evidence_refs JSONB NOT NULL DEFAULT '[]'::jsonb,
            current_version INTEGER NOT NULL DEFAULT 1 CHECK (current_version > 0),
            created_by UUID NOT NULL REFERENCES public.users(id),
            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
            updated_by UUID NOT NULL REFERENCES public.users(id),
            updated_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
            closed_at TIMESTAMPTZ,
            CHECK (jsonb_typeof(evidence_refs) = 'array')
        );
        CREATE INDEX idx_product_feedback_worklist
            ON public.data_product_feedback(
                product_uid, status, updated_at DESC
            );
        CREATE INDEX idx_product_feedback_assignee
            ON public.data_product_feedback(
                assignee_uid, status, updated_at DESC
            );
        """
    )


def downgrade() -> None:
    raise RuntimeError(
        "governed product contracts, certificates and feedback are retained; "
        "downgrade requires an approved archival migration"
    )