DataOps-platform/migrations/versions/20260729_350_device_observability.py

"""Add canonical device operational events and evidence relations."""

from alembic import op

revision = "20260729_350"
down_revision = "20260729_340"
branch_labels = None
depends_on = None


def upgrade() -> None:
    op.execute(
        """
        CREATE TABLE public.device_operational_events (
            uid UUID PRIMARY KEY,
            source_uid UUID NOT NULL
                REFERENCES public.ingestion_sources(uid) ON DELETE RESTRICT,
            source_entity VARCHAR(300) NOT NULL,
            source_code VARCHAR(300) NOT NULL,
            event_type VARCHAR(30) NOT NULL
                CHECK (
                    event_type IN (
                        'alarm','fault','maintenance','downtime'
                    )
                ),
            asset_uid UUID NOT NULL
                REFERENCES public.device_assets(uid) ON DELETE RESTRICT,
            component_uid UUID
                REFERENCES public.device_assets(uid) ON DELETE RESTRICT,
            title VARCHAR(500) NOT NULL,
            severity VARCHAR(20) NOT NULL
                CHECK (severity IN ('info','warning','error','critical')),
            status VARCHAR(20) NOT NULL
                CHECK (status IN ('observed','resolved')),
            occurred_at TIMESTAMPTZ NOT NULL,
            ended_at TIMESTAMPTZ
                CHECK (ended_at IS NULL OR ended_at >= occurred_at),
            evidence_refs JSONB NOT NULL,
            content_hash CHAR(64) NOT NULL,
            created_by UUID NOT NULL
                REFERENCES public.users(id) ON DELETE RESTRICT,
            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
            UNIQUE (source_uid, source_entity, source_code)
        );
        CREATE INDEX idx_device_operational_event_asset_time
            ON public.device_operational_events(
                asset_uid, occurred_at DESC, uid
            );
        CREATE INDEX idx_device_operational_event_worklist
            ON public.device_operational_events(
                event_type, severity, status, occurred_at DESC
            );

        CREATE TABLE public.device_evidence_relations (
            uid UUID PRIMARY KEY,
            from_kind VARCHAR(30) NOT NULL
                CHECK (
                    from_kind IN ('asset','event','quality_issue')
                ),
            from_uid UUID NOT NULL,
            relation_type VARCHAR(40) NOT NULL
                CHECK (
                    relation_type IN (
                        'part_of','occurred_on','indicates','triggered',
                        'addresses','impacts','evidences','related_to'
                    )
                ),
            to_kind VARCHAR(30) NOT NULL
                CHECK (
                    to_kind IN ('asset','event','quality_issue')
                ),
            to_uid UUID NOT NULL,
            evidence_refs JSONB NOT NULL,
            source VARCHAR(30) NOT NULL
                CHECK (source IN ('imported','manual','derived')),
            created_by UUID NOT NULL
                REFERENCES public.users(id) ON DELETE RESTRICT,
            created_at TIMESTAMPTZ NOT NULL DEFAULT CURRENT_TIMESTAMP,
            CHECK (NOT (from_kind = to_kind AND from_uid = to_uid)),
            UNIQUE (
                from_kind, from_uid, relation_type, to_kind, to_uid
            )
        );
        CREATE INDEX idx_device_evidence_relation_from
            ON public.device_evidence_relations(
                from_kind, from_uid, created_at, uid
            );
        CREATE INDEX idx_device_evidence_relation_to
            ON public.device_evidence_relations(
                to_kind, to_uid, created_at, uid
            );
        """
    )


def downgrade() -> None:
    # Operational evidence is retained during application rollback.
    pass