Coverage for src/local_deep_research/research_library/deletion/services/collection_deletion.py: 98%
140 statements
« prev ^ index » next coverage.py v7.16.2, created at 2026-10-22 21:20 +0000
1"""
2Collection deletion service.
4Handles:
5- Full collection deletion with proper cleanup
6- Documents are preserved but unlinked
7- RAG index and chunks are deleted
8"""
10from typing import Dict, List, Any
12from loguru import logger
14from ....database.models.library import (
15 Collection,
16 Document,
17 DocumentCollection,
18 DocumentChunk,
19 CollectionFolder,
20 RAGIndex,
21 RagDocumentStatus,
22 SourceType,
23)
24from ....database.session_context import get_user_db_session
25from ....utilities.formatting import short_id
26from ..utils.cascade_helper import (
27 CascadeHelper,
28 RemoteVectorStoreDropError,
29 read_vector_cleanup_settings_snapshot,
30 remote_drop_failure_message,
31)
33# System collection types that must never be deleted via the public API.
34# These collections hold first-class user data whose lifecycle is owned by
35# other subsystems (Library, Research History, Notes); allowing them to be
36# deleted via the generic collection-delete path triggers cascade orphan
37# deletion of every document inside, which for Notes means total data loss
38# of notes, versions, links, and synthesis sources.
39# NOTE: 'zotero' is deliberately NOT protected — unlike the singleton system
40# collections above, users create/remove Zotero-synced collections and the
41# generic delete path is their only UI for it. Whether deleting a Zotero
42# collection should also purge its synced documents is a separate question
43# for the Zotero sync owner to decide, not something to block wholesale here.
44PROTECTED_COLLECTION_TYPES = frozenset(
45 {"default_library", "research_history", "notes"}
46)
49def _user_settings_snapshot(session) -> Dict[str, Any]:
50 """The user's settings snapshot, read through the deletion's own session.
52 The RAG-index cascade needs it to resolve the user's vector-store provider
53 and its connection config (#5765); without it the provider silently falls
54 back to the global default. Read through ``session`` (the caller's own
55 per-user session), so this is safe off the request thread.
57 Strictly read-only: ``seed_defaults=False`` because the default
58 constructor seeds an EMPTY settings table from the defaults files and
59 COMMITS on ``session`` -- mid-deletion, outside the deletion's own
60 rollback. An empty table still reads as all-defaults (the manager merges
61 its in-memory defaults). Fails closed on an unreadable snapshot once a
62 server-backed store exists (see read_vector_cleanup_settings_snapshot).
63 """
64 from ....settings.manager import SettingsManager
66 return read_vector_cleanup_settings_snapshot(
67 SettingsManager(session, seed_defaults=False)
68 )
71def _remove_unfinished_rebuilds(username: str, collection_id: str) -> None:
72 """Remove interrupted rebuild stages; the deletion has already committed.
74 A failure only leaves those stages behind, so it is logged without a
75 traceback rather than reported as a failed deletion.
76 """
77 from ...services.collection_rebuild import remove_unfinished_rebuilds
79 try:
80 remove_unfinished_rebuilds(
81 username, collection_id, lambda: get_user_db_session(username)
82 )
83 except Exception:
84 logger.warning(
85 "Could not remove unfinished rebuilds of collection {}",
86 short_id(collection_id),
87 )
90def _remote_drop_failure_result(
91 collection_id: str, exc: RemoteVectorStoreDropError, outcome: str
92) -> Dict[str, Any]:
93 """Result dict for a deletion aborted by a failed remote drop."""
94 return {
95 "deleted": False,
96 "collection_id": collection_id,
97 "error": remote_drop_failure_message(exc, outcome),
98 # Stable machine-readable keys so callers (the router) map this shape
99 # to a status code without pattern matching the human-readable
100 # "error" text (#5765): ``vector_store_error`` marks the shape and
101 # ``retryable`` says whether a retry can succeed (503) or not (500).
102 "vector_store_error": True,
103 "retryable": exc.retryable,
104 }
107class CollectionDeletionService:
108 """Service for collection deletion operations."""
110 def __init__(self, username: str):
111 """
112 Initialize collection deletion service.
114 Args:
115 username: Username for database session
116 """
117 self.username = username
119 def delete_collection(
120 self, collection_id: str, delete_orphaned_documents: bool = True
121 ) -> Dict[str, Any]:
122 """
123 Delete a collection and clean up all related data.
125 By default, orphaned documents (not in any other collection) are deleted.
126 Set delete_orphaned_documents=False to preserve all documents.
128 The following are deleted:
129 - DocumentChunks for this collection
130 - Vector data (FAISS index files, or the provider's remote
131 collection — dropped before the RAGIndex rows, see #5765)
132 - RAGIndex records
133 - CollectionFolder records (CASCADE)
134 - DocumentCollection links (CASCADE)
135 - RagDocumentStatus records (CASCADE)
136 - Orphaned documents (if delete_orphaned_documents=True)
138 Args:
139 collection_id: ID of the collection to delete
140 delete_orphaned_documents: If True, delete documents not in any
141 other collection after unlinking
143 Returns:
144 Dict with deletion details:
145 {
146 "deleted": True/False,
147 "collection_id": str,
148 "collection_name": str,
149 "chunks_deleted": int,
150 "documents_unlinked": int,
151 "indices_deleted": int,
152 "folders_deleted": int,
153 "orphaned_documents_deleted": int,
154 "error": str (if failed)
155 }
156 """
157 with get_user_db_session(self.username) as session:
158 try:
159 # Get collection
160 collection = session.get(Collection, collection_id)
161 if not collection:
162 return {
163 "deleted": False,
164 "collection_id": collection_id,
165 "error": "Collection not found",
166 }
168 if collection.collection_type in PROTECTED_COLLECTION_TYPES:
169 logger.warning(
170 "Refused to delete protected system collection "
171 f"{short_id(collection_id)} (type={collection.collection_type})"
172 )
173 return {
174 "deleted": False,
175 "collection_id": collection_id,
176 "collection_name": collection.name,
177 "collection_type": collection.collection_type,
178 "error": (
179 "Cannot delete system collection "
180 f"'{collection.name}' (type={collection.collection_type}). "
181 "This collection holds first-class user data."
182 ),
183 }
185 collection_name = f"collection_{collection_id}"
186 result = {
187 "deleted": False,
188 "collection_id": collection_id,
189 "collection_name": collection.name,
190 "chunks_deleted": 0,
191 "documents_unlinked": 0,
192 "indices_deleted": 0,
193 "folders_deleted": 0,
194 "orphaned_documents_deleted": 0,
195 }
197 # 1. Get document IDs BEFORE deleting links (for orphan check)
198 doc_ids_in_collection = [
199 dc.document_id
200 for dc in session.query(DocumentCollection)
201 .filter_by(collection_id=collection_id)
202 .all()
203 ]
204 result["documents_unlinked"] = len(doc_ids_in_collection)
206 # 2. Delete RAGIndex records (DB only). The on-disk FAISS files
207 # are unlinked AFTER the commit below — steps 3-8 still do DB
208 # work that could raise and roll back, and a rollback restores
209 # the RAGIndex rows but not files already deleted from disk.
210 #
211 # A SERVER-BACKED store is the opposite case and is handled
212 # inside the helper: its remote collection is dropped BEFORE
213 # the rows go, because the collection name is derived from
214 # RAGIndex.index_hash. Deferring that to post-commit like the
215 # files would leave the vectors unaddressable (#5765). A
216 # failed drop raises RemoteVectorStoreDropError, caught below.
217 #
218 # This runs BEFORE the first write of this transaction (the
219 # chunk delete below) so a remote drop's network call is not
220 # made while holding the user's database write lock. The
221 # user's settings snapshot selects the provider and carries
222 # its connection config.
223 rag_result = CascadeHelper.delete_rag_indices_for_collection(
224 session,
225 collection_name,
226 unlink_files=False,
227 username=self.username,
228 settings_snapshot=_user_settings_snapshot(session),
229 )
230 result["indices_deleted"] = rag_result["deleted_indices"]
231 faiss_paths_to_unlink = rag_result["index_paths"]
233 # 3. Delete DocumentChunks for this collection
234 result["chunks_deleted"] = (
235 CascadeHelper.delete_collection_chunks(
236 session, collection_name
237 )
238 )
239 # Orphaned documents hard-deleted below may still have chunk
240 # rows/vectors in OTHER collections (a partial reindex can strand
241 # them). Collect them here and purge those after the commit —
242 # delete_document_completely does not touch DocumentChunk.
243 orphaned_doc_ids: List[str] = []
245 # 4. Count folders before deletion
246 result["folders_deleted"] = (
247 session.query(CollectionFolder)
248 .filter_by(collection_id=collection_id)
249 .count()
250 )
252 # 5. Delete DocumentCollection links explicitly before collection
253 session.query(DocumentCollection).filter_by(
254 collection_id=collection_id
255 ).delete(synchronize_session=False)
257 # 6. Delete linked folders explicitly
258 session.query(CollectionFolder).filter_by(
259 collection_id=collection_id
260 ).delete(synchronize_session=False)
262 # 7. Delete the collection itself
263 session.delete(collection)
265 # 8. Delete orphaned documents if requested
266 if delete_orphaned_documents:
267 # Notes must never be hard-deleted via the orphan
268 # cascade: they live behind their own deletion API
269 # (DELETE /api/notes/<id>) and remain discoverable
270 # via list_notes(), which filters by source_type_id
271 # rather than collection membership. This mirrors the
272 # note-skip in document_deletion.py's orphan path. The
273 # check is kept local (a single SourceType lookup) to
274 # avoid importing the notes-services package into the
275 # deletion package, matching document_deletion's own
276 # local _is_note_document.
277 note_source = (
278 session.query(SourceType).filter_by(name="note").first()
279 )
280 for doc_id in doc_ids_in_collection:
281 # Check if document is in any other collection
282 remaining = (
283 session.query(DocumentCollection)
284 .filter_by(document_id=doc_id)
285 .count()
286 )
287 if remaining == 0:
288 if note_source is not None:
289 document = session.get(Document, doc_id)
290 if (
291 document is not None
292 and document.source_type_id
293 == note_source.id
294 ):
295 logger.info(
296 f"Note {short_id(doc_id)} orphaned by "
297 "collection deletion — skipping orphan "
298 "delete; use DELETE /api/notes/<id> "
299 "to remove."
300 )
301 continue
302 # Document is orphaned - delete it
303 CascadeHelper.delete_document_completely(
304 session, doc_id
305 )
306 orphaned_doc_ids.append(doc_id)
307 result["orphaned_documents_deleted"] += 1
308 logger.info(
309 f"Deleted orphaned document {short_id(doc_id)}"
310 )
312 session.commit()
314 from ....web_search_engines.search_engines_config import (
315 invalidate_collection_engines_cache,
316 )
318 invalidate_collection_engines_cache(self.username)
320 result["deleted"] = True
321 logger.info(
322 f"Deleted collection {short_id(collection_id)} "
323 f"({result['collection_name']}): {result['chunks_deleted']} chunks, "
324 f"{result['documents_unlinked']} documents unlinked, "
325 f"{result['orphaned_documents_deleted']} orphaned deleted"
326 )
328 except RemoteVectorStoreDropError as exc:
329 # A remote collection could not be dropped, so its vectors are
330 # still there. Roll back so the RAGIndex rows — the only handle
331 # to those vectors — survive, and report the deletion as
332 # FAILED. Completing it here would be the one outcome that
333 # cannot be undone or retried (#5765). Collections dropped
334 # before the failing one stay dropped (exc.dropped); the
335 # message says so rather than claiming nothing changed.
336 #
337 # Logged from the error's sanitized fields, at WARNING and
338 # with no traceback: a traceback (logger.exception) would
339 # add frame state, and the backend client's error text can
340 # carry a URI or token.
341 drop_detail = exc.log_detail()
342 logger.warning(
343 f"Aborted deletion of collection {collection_id}: could "
344 f"not drop its vector-store collection: {drop_detail}"
345 )
346 session.rollback()
347 return _remote_drop_failure_result(
348 collection_id, exc, "the collection was not deleted"
349 )
350 except Exception:
351 logger.exception(f"Failed to delete collection {collection_id}")
352 session.rollback()
353 return {
354 "deleted": False,
355 "collection_id": collection_id,
356 "error": "Failed to delete collection",
357 }
359 # Post-commit best-effort cleanup, OUTSIDE the deletion transaction and
360 # its try/except: the DB delete is already durable, so a failure here
361 # must NOT be reported as a failed collection delete — it isn't one. It
362 # would only leave orphaned files/vectors, logged for ops visibility.
363 if faiss_paths_to_unlink:
364 from ....config.paths import get_cache_directory
366 # NOTE: allowed_root is the COARSE shared rag_indices/ root, not
367 # the tighter per-user rag_indices/<sha256(user)>/ subdir. It
368 # cannot be narrowed to the per-user subdir without risking
369 # refused deletes: pre-per-user-scoping (legacy) indexes live
370 # DIRECTLY in this shared root (see
371 # library_rag_service._migrate_legacy_index_files), so a per-user
372 # allowed_root would reject a legacy-layout RAGIndex.index_path
373 # and orphan its files. The containment check still blocks
374 # symlink / ancestor escapes outside rag_indices/ entirely.
375 rag_indices_root = get_cache_directory() / "rag_indices"
376 for faiss_path in faiss_paths_to_unlink:
377 CascadeHelper.delete_faiss_index_files(
378 faiss_path, allowed_root=rag_indices_root
379 )
381 # Before the orphan purge below, so it never sees staging chunk rows.
382 _remove_unfinished_rebuilds(self.username, collection_id)
384 # Purge each hard-deleted orphan's vectors + chunk rows in its OTHER
385 # collections (this collection's chunks were removed in step 2). Reuses
386 # DocumentDeletionService's per-collection vector purge + chunk backstop.
387 if orphaned_doc_ids:
388 from .document_deletion import DocumentDeletionService
389 from ....database.models.library import DocumentChunk
391 deleter = DocumentDeletionService(self.username)
392 for doc_id in orphaned_doc_ids:
393 try:
394 with get_user_db_session(self.username) as purge_session:
395 coll_rows = (
396 purge_session.query(DocumentChunk.collection_name)
397 .filter_by(source_type="document", source_id=doc_id)
398 .distinct()
399 .all()
400 )
401 collection_ids = [
402 row[0].removeprefix("collection_")
403 for row in coll_rows
404 if row[0]
405 ]
406 deleter._purge_document_rag(
407 doc_id, collection_ids, full_delete=True
408 )
409 except Exception:
410 logger.opt(exception=True).error(
411 "Failed to purge orphaned document {} after collection "
412 "delete (delete already committed)",
413 doc_id,
414 )
416 return result
418 def delete_collection_index_only(
419 self, collection_id: str
420 ) -> Dict[str, Any]:
421 """
422 Delete only the RAG index for a collection, keeping the collection itself.
424 This is useful for rebuilding an index from scratch.
426 Args:
427 collection_id: ID of the collection
429 Returns:
430 Dict with deletion details
431 """
432 with get_user_db_session(self.username) as session:
433 try:
434 # Verify collection exists
435 collection = session.get(Collection, collection_id)
436 if not collection:
437 return {
438 "deleted": False,
439 "collection_id": collection_id,
440 "error": "Collection not found",
441 }
443 collection_name = f"collection_{collection_id}"
444 result = {
445 "deleted": False,
446 "collection_id": collection_id,
447 "chunks_deleted": 0,
448 "indices_deleted": 0,
449 "documents_reset": 0,
450 }
452 # 1. Delete RAGIndex records (DB only). Files are unlinked after
453 # the commit — steps 2-5 do more DB work that could roll back,
454 # and a rollback can't restore files already removed from disk.
455 # A server-backed store's remote collection is dropped inside
456 # the helper BEFORE the rows go, and before this
457 # transaction's first write (see delete_collection).
458 rag_result = CascadeHelper.delete_rag_indices_for_collection(
459 session,
460 collection_name,
461 unlink_files=False,
462 username=self.username,
463 settings_snapshot=_user_settings_snapshot(session),
464 )
465 result["indices_deleted"] = rag_result["deleted_indices"]
466 faiss_paths_to_unlink = rag_result["index_paths"]
468 # 2. Delete DocumentChunks
469 result["chunks_deleted"] = (
470 CascadeHelper.delete_collection_chunks(
471 session, collection_name
472 )
473 )
475 # 3. Reset DocumentCollection indexed status
476 result["documents_reset"] = (
477 session.query(DocumentCollection)
478 .filter_by(collection_id=collection_id)
479 .update({"indexed": False, "chunk_count": 0})
480 )
482 # 4. Delete RagDocumentStatus for this collection
483 session.query(RagDocumentStatus).filter_by(
484 collection_id=collection_id
485 ).delete(synchronize_session=False)
487 # 5. Reset collection embedding info
488 collection.embedding_model = None
489 collection.embedding_model_type = None
490 collection.embedding_dimension = None
491 collection.chunk_size = None
492 collection.chunk_overlap = None
493 collection.rebuild_pending_since = None
495 session.commit()
497 # DB delete durable — now unlink the FAISS files (best-effort).
498 if faiss_paths_to_unlink:
499 from ....config.paths import get_cache_directory
501 # NOTE: allowed_root is the COARSE shared rag_indices/
502 # root, not the per-user rag_indices/<sha256(user)>/
503 # subdir -- see delete_collection above for why it can't
504 # be narrowed (legacy indexes live directly in the shared
505 # root, so a per-user scope would orphan them).
506 rag_indices_root = get_cache_directory() / "rag_indices"
507 for faiss_path in faiss_paths_to_unlink:
508 CascadeHelper.delete_faiss_index_files(
509 faiss_path, allowed_root=rag_indices_root
510 )
512 result["deleted"] = True
514 logger.info(
515 f"Deleted index for collection {short_id(collection_id)}: "
516 f"{result['chunks_deleted']} chunks, "
517 f"{result['documents_reset']} documents reset"
518 )
520 except RemoteVectorStoreDropError as exc:
521 # See delete_collection: keep the RAGIndex rows so the remote
522 # collection stays addressable, and report failure. Logged
523 # the same way (sanitized fields, no traceback).
524 drop_detail = exc.log_detail()
525 logger.warning(
526 f"Aborted index deletion for collection {collection_id}: "
527 f"could not drop its vector-store collection: {drop_detail}"
528 )
529 session.rollback()
530 return _remote_drop_failure_result(
531 collection_id, exc, "the index was not deleted"
532 )
533 except Exception:
534 logger.exception(
535 f"Failed to delete index for collection {collection_id}"
536 )
537 session.rollback()
538 return {
539 "deleted": False,
540 "collection_id": collection_id,
541 "error": "Failed to delete collection index",
542 }
544 # After the commit and outside its scope, as in delete_collection.
545 _remove_unfinished_rebuilds(self.username, collection_id)
546 return result
548 def get_deletion_preview(self, collection_id: str) -> Dict[str, Any]:
549 """
550 Get a preview of what will be deleted.
552 Useful for showing the user what will happen before confirming.
554 Args:
555 collection_id: ID of the collection
557 Returns:
558 Dict with preview information
559 """
560 with get_user_db_session(self.username) as session:
561 collection = session.get(Collection, collection_id)
562 if not collection:
563 return {"found": False, "collection_id": collection_id}
565 collection_name = f"collection_{collection_id}"
567 # Count documents
568 documents_count = (
569 session.query(DocumentCollection)
570 .filter_by(collection_id=collection_id)
571 .count()
572 )
574 # Count chunks
575 chunks_count = (
576 session.query(DocumentChunk)
577 .filter_by(collection_name=collection_name)
578 .count()
579 )
581 # Count folders
582 folders_count = (
583 session.query(CollectionFolder)
584 .filter_by(collection_id=collection_id)
585 .count()
586 )
588 # Check for RAG index
589 has_index = (
590 session.query(RAGIndex)
591 .filter_by(collection_name=collection_name)
592 .first()
593 is not None
594 )
596 return {
597 "found": True,
598 "collection_id": collection_id,
599 "name": collection.name,
600 "description": collection.description,
601 "is_default": collection.is_default,
602 "documents_count": documents_count,
603 "chunks_count": chunks_count,
604 "folders_count": folders_count,
605 "has_rag_index": has_index,
606 "embedding_model": collection.embedding_model,
607 }