from __future__ import annotations from dataclasses import dataclass, field from typing import Any from bson import ObjectId from app.core.logging import bind_log_context, get_audit_logger from app.individual_ingestion.domain.ports import IndividualUploadInvalidationPort BASE_DOCUMENT_EXCLUDED_TYPES = { "epicrisis", "epicrisis_case_cache", "rips_case_payload", "rips_operational_template", "rda_case_artifact", "rda_case_job_status", } CASE_DERIVED_TYPES = { "epicrisis", "epicrisis_case_cache", "rips_case_payload", "rda_case_artifact", "rda_case_job_status", } class CaseDeletionError(Exception): def __init__(self, message: str, *, status_code: int) -> None: super().__init__(message) self.status_code = status_code @dataclass(slots=True) class CaseDeletionService: mongo_analyses: Any mongo_database: Any individual_upload_invalidation: IndividualUploadInvalidationPort | None = None audit_logger: Any = field(init=False) case_runtime_collection: Any = field(init=False) def __post_init__(self) -> None: self.audit_logger = get_audit_logger() self.case_runtime_collection = self.mongo_database["processing_batch_cases"] def delete_document(self, *, username: str, document_id: str) -> dict[str, Any]: normalized_document_id = str(document_id or "").strip() if not ObjectId.is_valid(normalized_document_id): raise CaseDeletionError("ID de documento no válido.", status_code=404) document = self.mongo_analyses.collection.find_one({"_id": ObjectId(normalized_document_id)}) if not document or str(document.get("usuario") or "").strip() != username: raise CaseDeletionError("Documento no encontrado.", status_code=404) document_type = str(document.get("tipo_documento") or "").strip() if document_type in BASE_DOCUMENT_EXCLUDED_TYPES: raise CaseDeletionError( "Este endpoint solo permite borrar documentos base del caso.", status_code=400, ) case_key = str(document.get("case_key") or "").strip() bind_log_context(username=username, case_key=case_key, file_id=normalized_document_id) invalidated_uploads = self._invalidate_document_upload( username=username, clinical_document_id=normalized_document_id, batch_file_id=str(document.get("batch_file_id") or ""), document_id=normalized_document_id, case_key=case_key, reason="document_deleted", ) self.mongo_analyses.collection.delete_one({"_id": ObjectId(normalized_document_id), "usuario": username}) remaining_documents = self._count_case_base_documents(username=username, case_key=case_key) derived_deleted = 0 case_deleted = remaining_documents == 0 and bool(case_key) if case_deleted: invalidated_uploads += self._invalidate_case_uploads( username=username, case_key=case_key, reason="case_deleted", ) derived_deleted = self._cleanup_case_derivatives(username=username, case_key=case_key) self.audit_logger.business_event( event_type="case.document_deleted", action="delete_case_document", outcome="success", service="case_deletion_service", resource={ "case_key": case_key, "document_id": normalized_document_id, "document_type": document_type, }, metrics={ "remaining_documents": remaining_documents, "derived_artifacts_deleted": derived_deleted, "case_deleted": case_deleted, "invalidated_uploads": invalidated_uploads, }, ) return { "message": "Documento eliminado correctamente.", "deleted_document_id": normalized_document_id, "case_key": case_key, "remaining_documents": remaining_documents, "case_deleted": case_deleted, "invalidated_uploads": invalidated_uploads, } def purge_document_record(self, *, username: str, document_id: str) -> bool: """Elimina un documento puntual por ID para rollback interno sin validar tipo.""" normalized_document_id = str(document_id or "").strip() if not ObjectId.is_valid(normalized_document_id): return False result = self.mongo_analyses.collection.delete_one( {"_id": ObjectId(normalized_document_id), "usuario": username} ) return bool(getattr(result, "deleted_count", 0)) def delete_case(self, *, username: str, case_key: str) -> dict[str, Any]: normalized_case_key = str(case_key or "").strip() if not normalized_case_key: raise CaseDeletionError("El case_key es obligatorio.", status_code=400) base_documents = list(self._list_case_base_documents(username=username, case_key=normalized_case_key)) if not base_documents: raise CaseDeletionError("Caso no encontrado.", status_code=404) bind_log_context(username=username, case_key=normalized_case_key) invalidated_uploads = self._invalidate_case_uploads( username=username, case_key=normalized_case_key, reason="case_deleted", ) base_document_ids = [item["_id"] for item in base_documents if item.get("_id")] deleted_documents = 0 if base_document_ids: result = self.mongo_analyses.collection.delete_many( {"_id": {"$in": base_document_ids}, "usuario": username} ) deleted_documents = int(getattr(result, "deleted_count", 0) or 0) deleted_derived_artifacts = self._cleanup_case_derivatives( username=username, case_key=normalized_case_key, ) self.audit_logger.business_event( event_type="case.deleted", action="delete_case", outcome="success", service="case_deletion_service", resource={"case_key": normalized_case_key}, metrics={ "deleted_documents": deleted_documents, "deleted_derived_artifacts": deleted_derived_artifacts, "invalidated_uploads": invalidated_uploads, }, ) return { "message": "Caso eliminado correctamente.", "case_key": normalized_case_key, "deleted_documents": deleted_documents, "deleted_derived_artifacts": deleted_derived_artifacts, "invalidated_uploads": invalidated_uploads, } def purge_documents_by_batch_file_ids(self, *, username: str, batch_file_ids: list[str]) -> int: normalized_ids = sorted( {str(value or "").strip() for value in batch_file_ids if str(value or "").strip()} ) if not normalized_ids: return 0 result = self.mongo_analyses.collection.delete_many( {"usuario": username, "batch_file_id": {"$in": normalized_ids}} ) return int(getattr(result, "deleted_count", 0) or 0) def _invalidate_document_upload( self, *, username: str, clinical_document_id: str, batch_file_id: str, document_id: str, case_key: str, reason: str, ) -> int: if self.individual_upload_invalidation is None: return 0 invalidated = self.individual_upload_invalidation.invalidate_by_clinical_document_id( username=username, clinical_document_id=clinical_document_id, reason=reason, document_id=document_id, case_key=case_key, deleted_by=username, ) if batch_file_id: invalidated += self.individual_upload_invalidation.invalidate_by_batch_file_id( username=username, batch_file_id=batch_file_id, reason=reason, deleted_by=username, ) return invalidated def _invalidate_case_uploads(self, *, username: str, case_key: str, reason: str) -> int: if self.individual_upload_invalidation is None: return 0 return self.individual_upload_invalidation.invalidate_by_case_key( username=username, case_key=case_key, reason=reason, deleted_by=username, ) def _list_case_base_documents(self, *, username: str, case_key: str) -> list[dict[str, Any]]: documents = self.mongo_analyses.collection.find( { "usuario": username, "case_key": case_key, "tipo_documento": {"$nin": sorted(BASE_DOCUMENT_EXCLUDED_TYPES)}, } ) return list(documents) def _count_case_base_documents(self, *, username: str, case_key: str) -> int: if not case_key: return 0 count_documents = getattr(self.mongo_analyses.collection, "count_documents", None) query = { "usuario": username, "case_key": case_key, "tipo_documento": {"$nin": sorted(BASE_DOCUMENT_EXCLUDED_TYPES)}, } if callable(count_documents): return int(count_documents(query) or 0) return len(self._list_case_base_documents(username=username, case_key=case_key)) def _cleanup_case_derivatives(self, *, username: str, case_key: str) -> int: if not case_key: return 0 analysis_result = self.mongo_analyses.collection.delete_many( { "usuario": username, "case_key": case_key, "tipo_documento": {"$in": sorted(CASE_DERIVED_TYPES)}, } ) runtime_result = self.case_runtime_collection.delete_many( {"usuario": username, "case_key": case_key} ) return int(getattr(analysis_result, "deleted_count", 0) or 0) + int( getattr(runtime_result, "deleted_count", 0) or 0 )