LegacyHUB/app/indexing/projection.py

"""Purge LegacyHUB search projections by ``asset_id`` (10 §6 / §8.4).

When an owner module deletes an asset (tombstone event) LegacyHUB removes its
projection - chunks plus OpenSearch and Qdrant entries - for that asset's
documents. A legal hold blocks the purge (10 §6: legal_hold overrides deletion
and applies to index chunks/embeddings).
"""

from __future__ import annotations

import uuid
from dataclasses import dataclass, field
from typing import Protocol

from sqlalchemy import delete, select

from app.db.audit import record_audit
from app.db.models import AssetIngestJob, Chunk
from app.db.session import session_scope
from app.indexing import opensearch_client, qdrant_client
from app.logging_config import get_logger

logger = get_logger(__name__)


@dataclass
class AssetPurgeTarget:
    found: bool
    document_ids: list[str] = field(default_factory=list)
    legal_hold: bool = False


class AssetProjectionRepository(Protocol):
    def find_asset_documents(self, asset_id: str) -> AssetPurgeTarget: ...

    def purge_chunks(self, asset_id: str, document_ids: list[str]) -> None: ...


class SqlAlchemyAssetProjectionRepository:
    def find_asset_documents(self, asset_id: str) -> AssetPurgeTarget:
        with session_scope() as db:
            jobs = (
                db.execute(
                    select(AssetIngestJob)
                    .where(AssetIngestJob.asset_id == asset_id)
                    .order_by(AssetIngestJob.created_at.desc())
                )
                .scalars()
                .all()
            )
            if not jobs:
                return AssetPurgeTarget(found=False)
            legal_hold = bool((jobs[0].manifest_json or {}).get("retention", {}).get("legal_hold"))
            document_ids = sorted({str(j.document_id) for j in jobs})
            return AssetPurgeTarget(found=True, document_ids=document_ids, legal_hold=legal_hold)

    def purge_chunks(self, asset_id: str, document_ids: list[str]) -> None:
        with session_scope() as db:
            for did in document_ids:
                db.execute(delete(Chunk).where(Chunk.document_id == uuid.UUID(did)))
            record_audit(
                db,
                action="asset.projection.purged",
                entity_type="asset",
                entity_id=asset_id,
                details={"documents": document_ids},
            )


def _default_search_purge(document_ids: list[str]) -> None:
    for did in document_ids:
        try:
            opensearch_client.delete_by_document(did)
        except Exception as exc:  # noqa: BLE001
            logger.warning("projection.os_purge_failed", document_id=did, error=str(exc))
        try:
            qdrant_client.delete_by_document(did)
        except Exception as exc:  # noqa: BLE001
            logger.warning("projection.qdrant_purge_failed", document_id=did, error=str(exc))


def purge_projection_by_asset_id(
    asset_id: str,
    *,
    repository: AssetProjectionRepository | None = None,
    search_purge=None,
) -> dict[str, object]:
    """Remove the search projection for ``asset_id``. Blocked by a legal hold."""
    repo = repository or SqlAlchemyAssetProjectionRepository()
    target = repo.find_asset_documents(asset_id)
    if not target.found:
        return {"asset_id": asset_id, "purged": False, "reason": "unknown_asset"}
    if target.legal_hold:
        logger.info("projection.purge_blocked_legal_hold", asset_id=asset_id)
        return {"asset_id": asset_id, "purged": False, "reason": "legal_hold"}

    repo.purge_chunks(asset_id, target.document_ids)
    (search_purge or _default_search_purge)(target.document_ids)
    logger.info("projection.purged", asset_id=asset_id, documents=len(target.document_ids))
    return {"asset_id": asset_id, "purged": True, "documents": target.document_ids}


__all__ = ["AssetPurgeTarget", "purge_projection_by_asset_id"]