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

"""Add the unified governance approval, task and notification work center."""

from alembic import op


revision = "20260802_430"
down_revision = "20260801_420"
branch_labels = None
depends_on = None


def upgrade() -> None:
    op.execute(
        """
        CREATE TABLE public.governance_workflows (
            uid UUID PRIMARY KEY,
            code VARCHAR(120) NOT NULL UNIQUE,
            name VARCHAR(300) NOT NULL,
            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.governance_workflow_versions (
            uid UUID PRIMARY KEY,
            workflow_uid UUID NOT NULL REFERENCES public.governance_workflows(uid),
            version INTEGER NOT NULL CHECK (version > 0),
            status VARCHAR(20) NOT NULL CHECK (
                status IN ('draft','published','superseded')
            ),
            definition 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 (workflow_uid, version),
            CHECK (jsonb_typeof(definition) = 'object')
        );
        ALTER TABLE public.governance_workflows
            ADD CONSTRAINT governance_workflow_active_version_fk
            FOREIGN KEY (active_version_uid)
            REFERENCES public.governance_workflow_versions(uid);
        CREATE UNIQUE INDEX uq_governance_workflow_published_version
            ON public.governance_workflow_versions(workflow_uid)
            WHERE status = 'published';

        CREATE TABLE public.governance_tasks (
            uid UUID PRIMARY KEY,
            task_code VARCHAR(40) NOT NULL UNIQUE,
            workflow_uid UUID NOT NULL REFERENCES public.governance_workflows(uid),
            workflow_version INTEGER NOT NULL CHECK (workflow_version > 0),
            task_type VARCHAR(40) NOT NULL CHECK (
                task_type IN (
                    'approval','quality_issue','semantic_governance',
                    'data_product_approval','agent_approval',
                    'governance_work_order','release','high_risk'
                )
            ),
            subject_type VARCHAR(40) NOT NULL CHECK (
                subject_type IN (
                    'quality_issue','semantic_governance','data_product','agent'
                )
            ),
            subject_uid VARCHAR(200) NOT NULL,
            source_type VARCHAR(80) NOT NULL,
            source_uid VARCHAR(200) NOT NULL,
            title VARCHAR(300) NOT NULL,
            description VARCHAR(2000) NOT NULL,
            business_domain_uid VARCHAR(200),
            priority VARCHAR(20) NOT NULL CHECK (
                priority IN ('low','medium','high','critical')
            ),
            status VARCHAR(30) NOT NULL CHECK (
                status IN (
                    'pending','in_progress','pending_review','approved',
                    'rejected','closed','reopened','cancelled'
                )
            ),
            assignee_uid UUID NOT NULL REFERENCES public.users(id),
            due_at TIMESTAMPTZ NOT NULL,
            escalation_level INTEGER NOT NULL DEFAULT 0 CHECK (
                escalation_level BETWEEN 0 AND 10
            ),
            current_version INTEGER NOT NULL DEFAULT 1 CHECK (
                current_version > 0
            ),
            route_snapshot JSONB NOT NULL,
            context JSONB NOT NULL DEFAULT '{}'::jsonb,
            source_state_unchanged BOOLEAN NOT NULL DEFAULT TRUE CHECK (
                source_state_unchanged
            ),
            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(route_snapshot) = 'object'),
            CHECK (jsonb_typeof(context) = 'object')
        );
        CREATE UNIQUE INDEX uq_governance_task_active_source
            ON public.governance_tasks(source_type, source_uid)
            WHERE status NOT IN ('closed','cancelled');
        CREATE INDEX idx_governance_task_worklist
            ON public.governance_tasks(
                assignee_uid, status, priority, due_at, created_at DESC
            );
        CREATE INDEX idx_governance_task_domain
            ON public.governance_tasks(
                business_domain_uid, status, due_at
            );

        CREATE TABLE public.governance_task_participants (
            uid UUID PRIMARY KEY,
            task_uid UUID NOT NULL
                REFERENCES public.governance_tasks(uid) ON DELETE CASCADE,
            user_uid UUID NOT NULL REFERENCES public.users(id),
            sequence INTEGER NOT NULL CHECK (sequence > 0),
            status VARCHAR(20) NOT NULL CHECK (
                status IN ('pending','decided','transferred')
            ),
            transferred_from_uid UUID REFERENCES public.users(id),
            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
            UNIQUE (task_uid, user_uid)
        );
        CREATE INDEX idx_governance_task_participant_user
            ON public.governance_task_participants(user_uid, status, task_uid);

        CREATE TABLE public.governance_task_reviews (
            uid UUID PRIMARY KEY,
            task_uid UUID NOT NULL
                REFERENCES public.governance_tasks(uid) ON DELETE CASCADE,
            reviewer_uid UUID NOT NULL REFERENCES public.users(id),
            decision VARCHAR(20) NOT NULL CHECK (
                decision IN ('approve','reject')
            ),
            reason VARCHAR(1000) NOT NULL,
            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
            UNIQUE (task_uid, reviewer_uid)
        );

        CREATE TABLE public.governance_task_comments (
            uid UUID PRIMARY KEY,
            task_uid UUID NOT NULL
                REFERENCES public.governance_tasks(uid) ON DELETE CASCADE,
            content VARCHAR(4000) NOT NULL,
            mentions JSONB NOT NULL DEFAULT '[]'::jsonb,
            created_by UUID NOT NULL REFERENCES public.users(id),
            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
            CHECK (jsonb_typeof(mentions) = 'array')
        );

        CREATE TABLE public.governance_task_attachments (
            uid UUID PRIMARY KEY,
            task_uid UUID NOT NULL
                REFERENCES public.governance_tasks(uid) ON DELETE CASCADE,
            name VARCHAR(300) NOT NULL,
            storage_ref VARCHAR(1000) 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,
            UNIQUE (task_uid, content_hash)
        );

        CREATE TABLE public.governance_task_events (
            uid UUID PRIMARY KEY,
            task_uid UUID NOT NULL
                REFERENCES public.governance_tasks(uid) ON DELETE CASCADE,
            task_version INTEGER NOT NULL CHECK (task_version > 0),
            action VARCHAR(40) NOT NULL,
            actor_uid UUID NOT NULL REFERENCES public.users(id),
            before_state JSONB NOT NULL DEFAULT '{}'::jsonb,
            after_state JSONB NOT NULL,
            payload JSONB NOT NULL DEFAULT '{}'::jsonb,
            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
            UNIQUE (task_uid, task_version, action),
            CHECK (jsonb_typeof(before_state) = 'object'),
            CHECK (jsonb_typeof(after_state) = 'object'),
            CHECK (jsonb_typeof(payload) = 'object')
        );
        CREATE INDEX idx_governance_task_event_timeline
            ON public.governance_task_events(task_uid, created_at, uid);

        CREATE TABLE public.governance_notification_templates (
            uid UUID PRIMARY KEY,
            code VARCHAR(120) NOT NULL,
            channel VARCHAR(20) NOT NULL CHECK (
                channel IN ('in_app','email')
            ),
            subject_template VARCHAR(500) NOT NULL,
            body_template VARCHAR(4000) NOT NULL,
            status VARCHAR(20) NOT NULL CHECK (
                status IN ('active','retired')
            ),
            current_version INTEGER NOT NULL DEFAULT 1 CHECK (
                current_version > 0
            ),
            created_by UUID NOT NULL REFERENCES public.users(id),
            updated_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 (code, channel)
        );

        CREATE TABLE public.governance_notification_preferences (
            user_uid UUID PRIMARY KEY REFERENCES public.users(id),
            enabled_channels JSONB NOT NULL DEFAULT '["in_app","email"]'::jsonb,
            subscribed_events JSONB NOT NULL DEFAULT '[]'::jsonb,
            quiet_hours JSONB NOT NULL DEFAULT '{}'::jsonb,
            revision INTEGER NOT NULL DEFAULT 1 CHECK (revision > 0),
            updated_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
            CHECK (jsonb_typeof(enabled_channels) = 'array'),
            CHECK (jsonb_typeof(subscribed_events) = 'array'),
            CHECK (jsonb_typeof(quiet_hours) = 'object')
        );

        CREATE TABLE public.governance_notifications (
            uid UUID PRIMARY KEY,
            event_key VARCHAR(500) NOT NULL UNIQUE,
            event_type VARCHAR(80) NOT NULL,
            recipient_uid UUID NOT NULL REFERENCES public.users(id),
            channel VARCHAR(20) NOT NULL CHECK (
                channel IN ('in_app','email')
            ),
            subject VARCHAR(500) NOT NULL,
            body VARCHAR(4000) NOT NULL,
            related_task_uid UUID
                REFERENCES public.governance_tasks(uid) ON DELETE CASCADE,
            status VARCHAR(20) NOT NULL CHECK (
                status IN (
                    'pending','processing','delivered','suppressed','dead_letter'
                )
            ),
            attempts INTEGER NOT NULL DEFAULT 0 CHECK (attempts >= 0),
            max_attempts INTEGER NOT NULL DEFAULT 5 CHECK (max_attempts > 0),
            next_attempt_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
            last_error VARCHAR(1000),
            delivered_at TIMESTAMPTZ,
            read_at TIMESTAMPTZ,
            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP
        );
        CREATE INDEX idx_governance_notification_inbox
            ON public.governance_notifications(
                recipient_uid, channel, read_at, created_at DESC
            );
        CREATE INDEX idx_governance_notification_delivery
            ON public.governance_notifications(
                channel, status, next_attempt_at
            );

        CREATE TABLE public.governance_notification_attempts (
            uid UUID PRIMARY KEY,
            notification_uid UUID NOT NULL
                REFERENCES public.governance_notifications(uid) ON DELETE CASCADE,
            attempt INTEGER NOT NULL CHECK (attempt > 0),
            status VARCHAR(20) NOT NULL CHECK (
                status IN ('pending','delivered','dead_letter')
            ),
            safe_error VARCHAR(1000),
            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
            UNIQUE (notification_uid, attempt)
        );
        """
    )


def downgrade() -> None:
    raise RuntimeError(
        "workflow, task and notification evidence is retained; "
        "downgrade requires an approved archival migration"
    )