DataOps-platform/app/core/knowledge/audit.py

from __future__ import annotations

from dataclasses import dataclass

from sqlalchemy import text


@dataclass(frozen=True)
class AuditFinding:
    finding_type: str
    count: int
    severity: str


def run_canonical_audit(session) -> tuple[AuditFinding, ...]:
    checks = {
        "active_document_without_chunks": """
            SELECT COUNT(*) FROM public.governance_documents d
            WHERE d.status = 'active' AND NOT EXISTS (
                SELECT 1 FROM public.governance_chunks c WHERE c.document_id = d.id
            )
        """,
        "active_chunk_without_embedding": """
            SELECT COUNT(*) FROM public.governance_chunks c
            JOIN public.governance_documents d ON d.id = c.document_id
            WHERE d.status = 'active' AND NOT EXISTS (
                SELECT 1 FROM public.knowledge_chunk_embeddings e
                JOIN public.knowledge_embedding_profiles p ON p.id = e.profile_id
                WHERE e.chunk_id = c.id AND p.status = 'active'
            )
        """,
        "active_point_without_document": """
            SELECT COUNT(*) FROM public.knowledge_points p
            WHERE p.status = 'active' AND NOT EXISTS (
                SELECT 1 FROM public.governance_documents d
                WHERE d.object_uid = p.source_uid AND d.status = 'active'
            )
        """,
        "ready_projection_hash_mismatch": """
            SELECT COUNT(*) FROM public.knowledge_index_projections projection
            JOIN public.governance_documents d ON d.id = projection.document_id
            WHERE projection.status = 'ready'
              AND projection.content_hash <> d.content_hash
        """,
        "expired_cache_dependency": """
            SELECT COUNT(*) FROM public.knowledge_cache_dependencies
            WHERE expires_at <= CURRENT_TIMESTAMP
        """,
    }
    findings = []
    for finding_type, statement in checks.items():
        count = int(session.execute(text(statement)).scalar_one())
        if count:
            findings.append(AuditFinding(finding_type, count, "error"))
    return tuple(findings)