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

"""Add enterprise identity, OIDC, lifecycle, session and emergency ledgers."""

from alembic import op

revision = "20260802_470"
down_revision = "20260802_460"
branch_labels = None
depends_on = None


def upgrade() -> None:
    op.execute(
        """
        CREATE TABLE public.identity_idp_config_versions (
            uid UUID PRIMARY KEY, provider_uid UUID NOT NULL, version INTEGER NOT NULL CHECK (version > 0),
            status VARCHAR(20) NOT NULL CHECK (status IN ('draft','active','disabled','superseded')),
            issuer TEXT NOT NULL, client_id VARCHAR(300) NOT NULL,
            secret_ref VARCHAR(160) NOT NULL CHECK (secret_ref ~ '^env:[A-Z][A-Z0-9_]*$'),
            authorization_endpoint TEXT NOT NULL, token_endpoint TEXT NOT NULL, jwks_uri TEXT NOT NULL,
            redirect_uris JSONB NOT NULL CHECK (jsonb_typeof(redirect_uris) = 'array'),
            algorithms JSONB NOT NULL CHECK (jsonb_typeof(algorithms) = 'array'),
            mapping_version VARCHAR(120) NOT NULL, config JSONB NOT NULL DEFAULT '{}'::jsonb,
            created_by UUID NOT NULL REFERENCES public.users(id), created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
            activated_at TIMESTAMPTZ, disabled_at TIMESTAMPTZ,
            UNIQUE (provider_uid, version), CHECK (jsonb_typeof(config) = 'object')
        );
        CREATE UNIQUE INDEX uq_identity_active_provider ON public.identity_idp_config_versions(provider_uid) WHERE status = 'active';
        CREATE INDEX idx_identity_idp_status_time ON public.identity_idp_config_versions(status, created_at DESC);

        CREATE TABLE public.identity_oidc_flows (
            uid UUID PRIMARY KEY, state_hash CHAR(64) NOT NULL UNIQUE, provider_uid UUID NOT NULL,
            provider_version INTEGER NOT NULL, nonce_hash CHAR(64) NOT NULL, pkce_verifier_hash CHAR(64) NOT NULL,
            redirect_uri TEXT NOT NULL, status VARCHAR(20) NOT NULL DEFAULT 'pending' CHECK (status IN ('pending','consumed','expired','failed')),
            expires_at TIMESTAMPTZ NOT NULL, consumed_at TIMESTAMPTZ, created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
            CHECK (expires_at > created_at)
        );
        CREATE INDEX idx_identity_oidc_flow_expiry ON public.identity_oidc_flows(status, expires_at);

        CREATE TABLE public.enterprise_identity_links (
            uid UUID PRIMARY KEY, provider_uid UUID NOT NULL, enterprise_subject VARCHAR(500) NOT NULL,
            user_uid UUID NOT NULL REFERENCES public.users(id), username VARCHAR(64) NOT NULL,
            display_name VARCHAR(300) NOT NULL, department_external_id VARCHAR(300) NOT NULL,
            groups JSONB NOT NULL DEFAULT '[]'::jsonb, role_names JSONB NOT NULL DEFAULT '[]'::jsonb,
            authorization_scope JSONB NOT NULL DEFAULT '{}'::jsonb, mapping_version VARCHAR(120) NOT NULL,
            claims_digest CHAR(64) NOT NULL, status VARCHAR(20) NOT NULL CHECK (status IN ('active','disabled','left')),
            token_version INTEGER NOT NULL DEFAULT 1 CHECK (token_version > 0),
            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP, updated_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
            UNIQUE (provider_uid, enterprise_subject), UNIQUE (provider_uid, user_uid),
            CHECK (jsonb_typeof(groups) = 'array'), CHECK (jsonb_typeof(role_names) = 'array'),
            CHECK (jsonb_typeof(authorization_scope) = 'object')
        );
        CREATE INDEX idx_enterprise_identity_status ON public.enterprise_identity_links(status, updated_at DESC);

        CREATE TABLE public.identity_organization_nodes (
            uid UUID PRIMARY KEY, provider_uid UUID NOT NULL, external_id VARCHAR(300) NOT NULL,
            node_type VARCHAR(20) NOT NULL CHECK (node_type IN ('department','group')),
            parent_external_id VARCHAR(300), display_name VARCHAR(300) NOT NULL,
            status VARCHAR(20) NOT NULL CHECK (status IN ('active','disabled')),
            attributes JSONB NOT NULL DEFAULT '{}'::jsonb, updated_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
            UNIQUE (provider_uid, node_type, external_id), CHECK (jsonb_typeof(attributes) = 'object')
        );
        CREATE INDEX idx_identity_org_parent ON public.identity_organization_nodes(provider_uid, parent_external_id);

        CREATE TABLE public.identity_directory_events (
            uid UUID PRIMARY KEY, provider_uid UUID NOT NULL, source VARCHAR(80) NOT NULL, source_event_id VARCHAR(300) NOT NULL,
            cursor_value VARCHAR(500) NOT NULL, cursor_sequence BIGINT NOT NULL CHECK(cursor_sequence > 0),
            event_type VARCHAR(20) NOT NULL CHECK (event_type IN ('JOINER','MOVER','LEAVER','DISABLE','RESTORE','DEPARTMENT','GROUP')),
            enterprise_subject VARCHAR(500), payload_digest CHAR(64) NOT NULL,
            status VARCHAR(20) NOT NULL CHECK (status IN ('applied','rejected','conflict')),
            safe_result JSONB NOT NULL DEFAULT '{}'::jsonb, processed_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
            UNIQUE (provider_uid, source, source_event_id), UNIQUE(provider_uid,source,cursor_sequence),
            CHECK (jsonb_typeof(safe_result) = 'object')
        );
        CREATE INDEX idx_identity_directory_cursor ON public.identity_directory_events(provider_uid,source,cursor_sequence DESC);
        CREATE TABLE public.identity_directory_checkpoints (
            provider_uid UUID NOT NULL,source VARCHAR(80) NOT NULL,cursor_value VARCHAR(500) NOT NULL,
            cursor_sequence BIGINT NOT NULL CHECK(cursor_sequence > 0),updated_at TIMESTAMPTZ NOT NULL,
            PRIMARY KEY(provider_uid,source)
        );
        CREATE TABLE public.identity_directory_conflicts (
            uid UUID PRIMARY KEY,provider_uid UUID NOT NULL,source VARCHAR(80) NOT NULL,source_event_id VARCHAR(300) NOT NULL,
            cursor_value VARCHAR(500) NOT NULL,cursor_sequence BIGINT NOT NULL,payload_digest CHAR(64) NOT NULL,
            reason VARCHAR(120) NOT NULL,created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP
        );

        CREATE TABLE public.identity_sessions (
            uid UUID PRIMARY KEY, family_uid UUID NOT NULL, identity_link_uid UUID REFERENCES public.enterprise_identity_links(uid),
            user_uid UUID NOT NULL REFERENCES public.users(id), identity_source VARCHAR(20) NOT NULL CHECK (identity_source IN ('oidc','local','emergency')),
            token_version INTEGER NOT NULL CHECK (token_version > 0), refresh_hash CHAR(64) NOT NULL UNIQUE,
            status VARCHAR(20) NOT NULL CHECK (status IN ('active','rotated','revoked','expired')),
            rotated_from_uid UUID REFERENCES public.identity_sessions(uid), revoke_reason VARCHAR(120), risk_reason VARCHAR(300),
            emergency_request_uid UUID,
            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP, last_seen_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
            expires_at TIMESTAMPTZ NOT NULL, rotated_at TIMESTAMPTZ, revoked_at TIMESTAMPTZ, CHECK (expires_at > created_at),
            CHECK ((identity_source='emergency') = (emergency_request_uid IS NOT NULL))
        );
        CREATE INDEX idx_identity_session_user_status ON public.identity_sessions(user_uid, status, created_at DESC);
        CREATE INDEX idx_identity_session_family ON public.identity_sessions(family_uid, created_at);

        CREATE TABLE public.identity_exchange_codes (
            uid UUID PRIMARY KEY, code_hash CHAR(64) NOT NULL UNIQUE, session_uid UUID NOT NULL REFERENCES public.identity_sessions(uid),
            redirect_uri TEXT NOT NULL, status VARCHAR(20) NOT NULL CHECK (status IN ('pending','consumed','expired')),
            expires_at TIMESTAMPTZ NOT NULL, consumed_at TIMESTAMPTZ, created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
            CHECK (expires_at > created_at)
        );
        CREATE INDEX idx_identity_exchange_expiry ON public.identity_exchange_codes(status, expires_at);

        CREATE TABLE public.identity_emergency_requests (
            uid UUID PRIMARY KEY, requester_uid UUID NOT NULL REFERENCES public.users(id),
            emergency_account_uid UUID NOT NULL REFERENCES public.users(id), reason TEXT NOT NULL,
            status VARCHAR(20) NOT NULL CHECK (status IN ('pending','active','closed','expired','reviewed','rejected')),
            starts_at TIMESTAMPTZ NOT NULL, expires_at TIMESTAMPTZ NOT NULL,
            activated_at TIMESTAMPTZ, closed_at TIMESTAMPTZ, reviewer_uid UUID REFERENCES public.users(id),
            review_outcome TEXT, created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
            CHECK (requester_uid <> emergency_account_uid), CHECK (expires_at > starts_at)
        );
        CREATE INDEX idx_identity_emergency_status_expiry ON public.identity_emergency_requests(status, expires_at);
        CREATE TABLE public.identity_emergency_approvals (
            request_uid UUID NOT NULL REFERENCES public.identity_emergency_requests(uid),
            approver_uid UUID NOT NULL REFERENCES public.users(id), approved_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
            PRIMARY KEY (request_uid, approver_uid)
        );
        ALTER TABLE public.identity_sessions ADD CONSTRAINT fk_identity_session_emergency
          FOREIGN KEY(emergency_request_uid) REFERENCES public.identity_emergency_requests(uid);

        CREATE TABLE public.identity_audit_events (
            uid UUID PRIMARY KEY, event_type VARCHAR(80) NOT NULL, outcome VARCHAR(20) NOT NULL CHECK (outcome IN ('success','failure','denied','warning')),
            actor_uid UUID REFERENCES public.users(id), provider_uid UUID, enterprise_subject_digest CHAR(64),
            resource_type VARCHAR(80) NOT NULL, resource_uid VARCHAR(300), correlation_uid UUID,
            safe_detail JSONB NOT NULL DEFAULT '{}'::jsonb, created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
            CHECK (jsonb_typeof(safe_detail) = 'object')
        );
        CREATE INDEX idx_identity_audit_time ON public.identity_audit_events(created_at DESC, uid DESC);
        CREATE INDEX idx_identity_audit_actor ON public.identity_audit_events(actor_uid, created_at DESC);
        """
    )


def downgrade() -> None:
    raise RuntimeError("Forward-only migration: enterprise identity and authentication evidence cannot be removed")