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

from __future__ import annotations

from sqlalchemy import text


def knowledge_status(session) -> dict:
    change_sets = session.execute(
        text(
            """
            SELECT status, COUNT(*)
            FROM public.knowledge_change_sets
            GROUP BY status
            """
        )
    ).all()
    projections = session.execute(
        text(
            """
            SELECT engine, status, COUNT(*)
            FROM public.knowledge_index_projections
            GROUP BY engine, status
            """
        )
    ).all()
    freshness = session.execute(
        text(
            """
            SELECT COUNT(*) FILTER (WHERE status = 'active') AS active_documents,
                   MAX(source_updated_at) FILTER (WHERE status = 'active') AS latest_source
            FROM public.governance_documents
            """
        )
    ).one()
    return {
        "change_sets": {row[0]: row[1] for row in change_sets},
        "projections": {f"{row[0]}:{row[1]}": row[2] for row in projections},
        "active_documents": freshness[0],
        "latest_source_updated_at": (
            freshness[1].isoformat() if freshness[1] else None
        ),
    }


def list_change_sets(session, *, limit: int = 50) -> list[dict]:
    rows = session.execute(
        text(
            """
            SELECT id::text, source_type, source_uid::text, source_revision,
                   change_type, added_count, modified_count, deleted_count,
                   impacted_point_count, impact_truncated, status, target_generation,
                   last_error, created_at, activated_at
            FROM public.knowledge_change_sets
            ORDER BY created_at DESC
            LIMIT :limit
            """
        ),
        {"limit": limit},
    ).mappings()
    return [dict(row) for row in rows]


def change_set_detail(session, change_set_id: str) -> dict | None:
    change_set = (
        session.execute(
            text(
                """
            SELECT id::text, source_type, source_uid::text, source_revision,
                   change_type, source_snapshot_hash, added_count, modified_count,
                   deleted_count, impacted_point_count, impact_truncated, status,
                   target_generation, last_error, created_at, updated_at, activated_at
            FROM public.knowledge_change_sets
            WHERE id = CAST(:id AS uuid)
            """
            ),
            {"id": change_set_id},
        )
        .mappings()
        .one_or_none()
    )
    if change_set is None:
        return None
    items = session.execute(
        text(
            """
            SELECT point_key, change_kind, old_point_revision, new_point_revision,
                   old_content_hash, new_content_hash, caused_by_point_key,
                   propagation_hop, canonical_status, embedding_status,
                   cache_status, lightrag_status, attempts, last_error
            FROM public.knowledge_change_items
            WHERE change_set_id = CAST(:id AS uuid)
            ORDER BY point_key
            """
        ),
        {"id": change_set_id},
    ).mappings()
    return {**dict(change_set), "items": [dict(row) for row in items]}


def retry_change_set(session, change_set_id: str) -> bool:
    updated = session.execute(
        text(
            """
            UPDATE public.knowledge_change_sets
            SET status = 'pending', last_error = NULL, updated_at = CURRENT_TIMESTAMP
            WHERE id = CAST(:id AS uuid) AND status IN ('failed','degraded')
            """
        ),
        {"id": change_set_id},
    ).rowcount
    if updated:
        session.execute(
            text(
                """
                UPDATE public.knowledge_index_projections
                SET status = 'pending', last_error = NULL, updated_at = CURRENT_TIMESTAMP
                WHERE change_set_id = CAST(:id AS uuid) AND status IN ('failed','unverified')
                """
            ),
            {"id": change_set_id},
        )
    return updated == 1


def rollback_change_set(session, change_set_id: str) -> bool:
    current = session.execute(
        text(
            """
            SELECT d.id::text, d.object_uid::text, d.object_version
            FROM public.governance_documents d
            JOIN public.knowledge_change_sets cs ON cs.id = d.change_set_id
            WHERE cs.id = CAST(:id AS uuid)
              AND cs.status = 'canonical_active'
              AND d.status = 'active'
            FOR UPDATE
            """
        ),
        {"id": change_set_id},
    ).one_or_none()
    if current is None:
        return False
    previous = session.execute(
        text(
            """
            SELECT id::text, object_version
            FROM public.governance_documents
            WHERE object_uid = CAST(:uid AS uuid)
              AND object_version < :version
            ORDER BY object_version DESC
            LIMIT 1
            FOR UPDATE
            """
        ),
        {"uid": current[1], "version": current[2]},
    ).one_or_none()
    if previous is None:
        return False
    session.execute(
        text(
            """
            UPDATE public.governance_documents SET status = 'superseded'
            WHERE id = CAST(:id AS uuid);
            UPDATE public.governance_documents SET status = 'active'
            WHERE id = CAST(:previous_id AS uuid);
            UPDATE public.knowledge_points SET status = 'superseded',
                valid_to = CURRENT_TIMESTAMP
            WHERE source_uid = CAST(:uid AS uuid) AND status = 'active';
            WITH restore AS (
                SELECT DISTINCT ON (point_key) id
                FROM public.knowledge_points
                WHERE source_uid = CAST(:uid AS uuid)
                  AND source_revision <= :previous_version
                  AND status <> 'deleted'
                ORDER BY point_key, point_revision DESC
            )
            UPDATE public.knowledge_points p
            SET status = 'active', valid_to = NULL, activated_at = CURRENT_TIMESTAMP
            FROM restore WHERE p.id = restore.id;
            UPDATE public.knowledge_change_sets
            SET status = 'rolled_back', updated_at = CURRENT_TIMESTAMP
            WHERE id = CAST(:change_set_id AS uuid);
            UPDATE public.knowledge_index_projections
            SET status = 'deleting', updated_at = CURRENT_TIMESTAMP
            WHERE document_id = CAST(:id AS uuid) AND status <> 'deleted';
            """
        ),
        {
            "id": current[0],
            "previous_id": previous[0],
            "uid": current[1],
            "previous_version": previous[1],
            "change_set_id": change_set_id,
        },
    )
    return True