Coverage for src/local_deep_research/research_library/deletion/utils/cascade_helper.py: 92%
184 statements
« prev ^ index » next coverage.py v7.16.0, created at 2026-09-06 15:42 +0000
1"""
2Cascade helper for deletion operations.
4Handles cleanup of related records that don't have proper FK constraints:
5- DocumentChunk (source_id has no FK constraint)
6- FAISS index files
7- Filesystem files
8"""
10import time
11from pathlib import Path
12from typing import Dict, List, Optional, Any, Union
14from loguru import logger
15from sqlalchemy.orm import Session
17from ....constants import FILE_PATH_SENTINELS
18from ....database.models.library import (
19 Document,
20 DocumentBlob,
21 DocumentChunk,
22 DocumentCollection,
23 RAGIndex,
24)
25from ....database.models.download_tracker import DownloadTracker
27# A <name>.*.tmp younger than this is assumed to be a concurrent writer's live
28# persist() temp (not yet os.replace()d into place); deleting it would crash
29# that write. Matches FaissVectorStore.persist()'s own stale-temp sweep so both
30# paths agree on when a temp is certainly crash-orphaned rather than in-flight.
31_STALE_TMP_AGE_SECONDS = 3600
34class CascadeHelper:
35 """Helper class for cleaning up related records during deletion."""
37 @staticmethod
38 def delete_document_chunks(
39 session: Session,
40 document_id: str,
41 collection_name: Optional[str] = None,
42 ) -> int:
43 """
44 Delete DocumentChunks for a document.
46 Since DocumentChunk.source_id has no FK constraint, we must manually
47 clean up chunks when deleting a document.
49 Args:
50 session: Database session
51 document_id: The document ID to delete chunks for
52 collection_name: Optional collection name to limit deletion scope
54 Returns:
55 Number of chunks deleted
56 """
57 query = session.query(DocumentChunk).filter(
58 DocumentChunk.source_id == document_id,
59 DocumentChunk.source_type == "document",
60 )
62 if collection_name:
63 query = query.filter(
64 DocumentChunk.collection_name == collection_name
65 )
67 count = query.delete(synchronize_session=False)
68 logger.debug(
69 f"Deleted {count} chunks for document {document_id[:8]}..."
70 + (f" in collection {collection_name}" if collection_name else "")
71 )
72 return count
74 @staticmethod
75 def delete_collection_chunks(
76 session: Session,
77 collection_name: str,
78 ) -> int:
79 """
80 Delete all DocumentChunks for a collection.
82 Args:
83 session: Database session
84 collection_name: The collection name (e.g., "collection_<uuid>")
86 Returns:
87 Number of chunks deleted
88 """
89 count = (
90 session.query(DocumentChunk)
91 .filter_by(collection_name=collection_name)
92 .delete(synchronize_session=False)
93 )
94 logger.debug(f"Deleted {count} chunks for collection {collection_name}")
95 return count
97 @staticmethod
98 def get_document_blob_size(session: Session, document_id: str) -> int:
99 """
100 Get the size of a document's blob in bytes.
102 Args:
103 session: Database session
104 document_id: The document ID
106 Returns:
107 Size in bytes, or 0 if no blob exists
108 """
109 blob = (
110 session.query(DocumentBlob)
111 .filter_by(document_id=document_id)
112 .first()
113 )
114 if blob and blob.pdf_binary:
115 return len(blob.pdf_binary)
116 return 0
118 @staticmethod
119 def delete_document_blob(session: Session, document_id: str) -> int:
120 """
121 Delete a document's blob record.
123 Note: This is typically handled by CASCADE, but can be called explicitly
124 for blob-only deletion.
126 Args:
127 session: Database session
128 document_id: The document ID
130 Returns:
131 Size of deleted blob in bytes
132 """
133 blob = (
134 session.query(DocumentBlob)
135 .filter_by(document_id=document_id)
136 .first()
137 )
138 if blob:
139 size = len(blob.pdf_binary) if blob.pdf_binary else 0
140 session.delete(blob)
141 logger.debug(
142 f"Deleted blob for document {document_id[:8]}... ({size} bytes)"
143 )
144 return size
145 return 0
147 @staticmethod
148 def _resolved_containment_ok(
149 path: Path, allowed_root: Optional[Union[str, Path]]
150 ) -> bool:
151 """Return True if `path`'s fully-resolved real location stays under
152 `allowed_root`.
154 `Path.resolve()` follows every symlink in the chain -- not just the
155 final component -- so this also catches an escape via a symlinked
156 *ancestor* directory (e.g. a library subfolder swapped for a symlink
157 pointing outside the library root between validation and delete).
159 When `allowed_root` is None, no containment boundary is enforced
160 here; callers that can't cheaply supply a root still get the
161 unconditional is_symlink() check the callers of this helper perform
162 on the leaf file itself.
163 """
164 if allowed_root is None:
165 return True
166 try:
167 resolved_path = path.resolve(strict=False)
168 resolved_root = Path(allowed_root).resolve(strict=False)
169 except OSError:
170 # Can't resolve (e.g. a broken component) -- refuse rather than
171 # risk unlinking something outside the intended root.
172 return False
173 return resolved_path.is_relative_to(resolved_root)
175 @staticmethod
176 def delete_filesystem_file(
177 file_path: Optional[str],
178 allowed_root: Optional[Union[str, Path]] = None,
179 ) -> bool:
180 """
181 Delete a file from the filesystem.
183 Args:
184 file_path: Path to the file (can be relative or absolute)
185 allowed_root: If provided, the file's fully-resolved real path
186 must be contained within this directory or the delete is
187 refused. Defense-in-depth against a symlink planted at (or
188 swapped in for) `file_path` redirecting the unlink to an
189 arbitrary file elsewhere on disk -- see #5481.
191 Returns:
192 True if file was deleted, False otherwise
193 """
194 if not file_path:
195 return False
197 # Skip special path markers
198 if file_path in FILE_PATH_SENTINELS:
199 return False
201 try:
202 path = Path(file_path)
204 # Refuse to unlink through a symlink. This check is unconditional
205 # (doesn't require allowed_root) -- is_file()/unlink() both
206 # follow symlinks, so without it a symlink planted at this exact
207 # path would redirect the delete to whatever it points at.
208 if path.is_symlink():
209 logger.warning(
210 f"Refusing to delete symlinked filesystem file: {file_path}"
211 )
212 return False
214 if not CascadeHelper._resolved_containment_ok(path, allowed_root):
215 logger.warning(
216 f"Refusing to delete filesystem file outside allowed root: {file_path}"
217 )
218 return False
220 if path.is_file():
221 path.unlink()
222 logger.debug(f"Deleted filesystem file: {file_path}")
223 return True
224 except Exception:
225 logger.exception(f"Failed to delete filesystem file: {file_path}")
226 return False
228 @staticmethod
229 def delete_faiss_index_files(
230 index_path: Optional[str],
231 allowed_root: Optional[Union[str, Path]] = None,
232 ) -> bool:
233 """
234 Delete FAISS index files.
236 FAISS stores indices as .faiss and .pkl files.
238 Args:
239 index_path: Path to the FAISS index file (without extension)
240 allowed_root: If provided, the resolved index family's parent
241 directory must be contained within this directory or the
242 delete is refused. See `delete_filesystem_file` for the
243 symlink-escape rationale.
245 Returns:
246 True if files were deleted, False otherwise
247 """
248 if not index_path:
249 return False
251 try:
252 path = Path(index_path)
254 if not CascadeHelper._resolved_containment_ok(path, allowed_root):
255 logger.warning(
256 f"Refusing to delete FAISS index outside allowed root: {index_path}"
257 )
258 return False
260 deleted_any = False
262 # FAISS index file
263 faiss_file = path.with_suffix(".faiss")
264 if faiss_file.is_symlink():
265 logger.warning(
266 f"Refusing to delete symlinked FAISS index file: {faiss_file}"
267 )
268 elif faiss_file.is_file():
269 faiss_file.unlink()
270 logger.debug(f"Deleted FAISS index file: {faiss_file}")
271 deleted_any = True
273 # Pickle file for metadata (legacy pre-cutover format)
274 pkl_file = path.with_suffix(".pkl")
275 if pkl_file.is_symlink(): 275 ↛ 276line 275 didn't jump to line 276 because the condition on line 275 was never true
276 logger.warning(
277 f"Refusing to delete symlinked FAISS pkl file: {pkl_file}"
278 )
279 elif pkl_file.is_file():
280 pkl_file.unlink()
281 logger.debug(f"Deleted FAISS pkl file: {pkl_file}")
282 deleted_any = True
284 # Migration sidecar (.idmap.json) — the text-free position->uuid map
285 # phase-1 writes and phase-2 consumes. If a collection is deleted
286 # during the .pkl -> .idmap.json cutover window (phase-1 ran, phase-2
287 # hasn't), the sidecar would otherwise be orphaned on disk forever.
288 idmap_file = path.with_suffix(".idmap.json")
289 if idmap_file.is_symlink(): 289 ↛ 290line 289 didn't jump to line 290 because the condition on line 289 was never true
290 logger.warning(
291 f"Refusing to delete symlinked FAISS idmap sidecar: {idmap_file}"
292 )
293 elif idmap_file.is_file():
294 idmap_file.unlink()
295 logger.debug(f"Deleted FAISS idmap sidecar: {idmap_file}")
296 deleted_any = True
298 # Sweep crash-orphaned temp files (persist writes <name>.*.tmp) and
299 # quarantined siblings (<name>.corrupt-<ns>) for the whole family,
300 # so deleting a collection doesn't leave them on disk forever — a
301 # leftover .pkl.corrupt-* is quarantined PLAINTEXT.
302 parent = path.parent
303 now = time.time()
304 for base in (faiss_file, pkl_file, idmap_file):
305 # .tmp files are AGE-GATED: FaissVectorStore.persist() writes to
306 # a <name>.*.tmp then os.replace()s it into place, so a live temp
307 # is a concurrent writer's not-yet-renamed file — unlinking it
308 # would crash that persist with FileNotFoundError. Only sweep
309 # temps old enough to be certainly crash-orphaned (mirrors
310 # persist()'s own _STALE_TMP_AGE_SECONDS sweep). A younger temp is
311 # left; a later delete/persist sweeps it once it ages out.
312 for stray in parent.glob(f"{base.name}.*.tmp"):
313 try:
314 if stray.is_symlink(): 314 ↛ 315line 314 didn't jump to line 315 because the condition on line 314 was never true
315 logger.warning(
316 f"Refusing to remove symlinked stray index file: {stray}"
317 )
318 continue
319 if now - stray.stat().st_mtime < _STALE_TMP_AGE_SECONDS:
320 continue
321 stray.unlink()
322 logger.debug(f"Removed stray temp index file: {stray}")
323 deleted_any = True
324 except FileNotFoundError:
325 # Raced a concurrent writer's os.replace() — already gone.
326 continue
327 except OSError:
328 logger.warning(
329 f"Could not remove stray index file {stray}"
330 )
331 # .corrupt-* are quarantine artefacts (and .pkl.corrupt-* is
332 # PLAINTEXT) — always safe and required to remove, no age gate.
333 for stray in parent.glob(f"{base.name}.corrupt-*"):
334 try:
335 if stray.is_symlink(): 335 ↛ 336line 335 didn't jump to line 336 because the condition on line 335 was never true
336 logger.warning(
337 f"Refusing to remove symlinked stray index file: {stray}"
338 )
339 continue
340 stray.unlink()
341 logger.debug(f"Removed quarantined index file: {stray}")
342 deleted_any = True
343 except FileNotFoundError:
344 continue
345 except OSError:
346 logger.warning(
347 f"Could not remove stray index file {stray}"
348 )
350 return deleted_any
351 except Exception:
352 logger.exception(f"Failed to delete FAISS files for: {index_path}")
353 return False
355 @staticmethod
356 def delete_rag_indices_for_collection(
357 session: Session,
358 collection_name: str,
359 *,
360 unlink_files: bool = True,
361 ) -> Dict[str, Any]:
362 """
363 Delete RAGIndex records (and, by default, their FAISS files) for a
364 collection.
366 Args:
367 session: Database session
368 collection_name: The collection name (e.g., "collection_<uuid>")
369 unlink_files: When True (default) the on-disk .faiss/.pkl/.idmap
370 files are unlinked here. When False, ONLY the RAGIndex DB rows
371 are staged for deletion and the file paths are returned under
372 ``index_paths`` — the caller is responsible for unlinking them
373 AFTER its transaction commits. Deleting files in-transaction is
374 unsafe when more DB work follows before the commit: a later
375 rollback restores the RAGIndex rows but cannot restore the
376 already-unlinked files, leaving the collection pointing at a
377 missing index (silently-broken search).
379 Returns:
380 Dict with deletion results (includes ``index_paths`` when
381 ``unlink_files`` is False).
382 """
383 indices = (
384 session.query(RAGIndex)
385 .filter_by(collection_name=collection_name)
386 .all()
387 )
389 deleted_indices = 0
390 deleted_files = 0
391 index_paths: List[str] = []
393 # Only needed on the unlink_files=True path (below) -- computed
394 # lazily so the unlink_files=False path (all 3 current external
395 # callers) doesn't pay for it. Same containment root those callers
396 # pass to their own post-commit delete_faiss_index_files() call
397 # (see collection_deletion.py, web/routers/rag.py), computed here too so
398 # unlink_files=True gets the same symlinked-ancestor defense --
399 # currently unreachable (all callers pass unlink_files=False) but
400 # kept in sync to avoid a latent gap if that changes.
401 #
402 # NOTE: this is the COARSE shared rag_indices/ root, not the tighter
403 # per-user rag_indices/<sha256(user)>/ subdir. It is intentionally
404 # NOT narrowed: pre-per-user-scoping (legacy) indexes live DIRECTLY
405 # in this shared root (see
406 # library_rag_service._migrate_legacy_index_files), so a per-user
407 # allowed_root would refuse to delete a legacy-layout index and
408 # orphan its files. Kept consistent with the external callers above.
409 rag_indices_root = None
410 if unlink_files:
411 from ....config.paths import get_cache_directory
413 rag_indices_root = get_cache_directory() / "rag_indices"
415 for index in indices:
416 path = str(index.index_path)
417 if unlink_files:
418 if CascadeHelper.delete_faiss_index_files(
419 path, allowed_root=rag_indices_root
420 ):
421 deleted_files += 1
422 else:
423 index_paths.append(path)
425 session.delete(index)
426 deleted_indices += 1
428 logger.debug(
429 f"Deleted {deleted_indices} RAGIndex records and {deleted_files} "
430 f"FAISS files for collection {collection_name}"
431 )
433 return {
434 "deleted_indices": deleted_indices,
435 "deleted_files": deleted_files,
436 "index_paths": index_paths,
437 }
439 @staticmethod
440 def update_download_tracker(
441 session: Session,
442 document: Document,
443 ) -> bool:
444 """
445 Update DownloadTracker when a document is deleted.
447 The FK has SET NULL, but we also need to update is_downloaded flag.
449 Args:
450 session: Database session
451 document: The document being deleted
453 Returns:
454 True if tracker was updated
455 """
456 if not document.original_url:
457 return False
459 # Get URL hash using the same method as library_service
460 from ...utils import get_url_hash
462 try:
463 url_hash = get_url_hash(str(document.original_url))
464 tracker = (
465 session.query(DownloadTracker)
466 .filter_by(url_hash=url_hash)
467 .first()
468 )
470 if tracker:
471 tracker.is_downloaded = False # type: ignore[assignment]
472 tracker.file_path = None # type: ignore[assignment]
473 logger.debug(
474 f"Updated DownloadTracker for document {document.id[:8]}..."
475 )
476 return True
477 except Exception:
478 logger.exception("Failed to update DownloadTracker")
479 return False
481 @staticmethod
482 def count_document_in_collections(
483 session: Session,
484 document_id: str,
485 ) -> int:
486 """
487 Count how many collections a document is in.
489 Args:
490 session: Database session
491 document_id: The document ID
493 Returns:
494 Number of collections the document is in
495 """
496 return (
497 session.query(DocumentCollection)
498 .filter_by(document_id=document_id)
499 .count()
500 )
502 @staticmethod
503 def get_document_collections(
504 session: Session,
505 document_id: str,
506 ) -> List[str]:
507 """
508 Get all collection IDs a document belongs to.
510 Args:
511 session: Database session
512 document_id: The document ID
514 Returns:
515 List of collection IDs
516 """
517 doc_collections = (
518 session.query(DocumentCollection.collection_id)
519 .filter_by(document_id=document_id)
520 .all()
521 )
522 return [dc.collection_id for dc in doc_collections]
524 @staticmethod
525 def delete_document_completely(
526 session: Session,
527 document_id: str,
528 ) -> bool:
529 """
530 Delete a document and all related records using query-based deletes.
532 This avoids ORM cascade issues where SQLAlchemy tries to set
533 DocumentBlob.document_id to NULL (which fails because it's a PK).
535 Deletes in order:
536 1. DocumentBlob
537 2. DocumentCollection links
538 3. Document itself
540 Note: DocumentChunks should be deleted separately before calling this,
541 as they may need collection-specific handling.
543 Args:
544 session: Database session
545 document_id: The document ID to delete
547 Returns:
548 True if document was deleted
549 """
550 # Delete blob (has document_id as PK, can't be nulled by cascade)
551 session.query(DocumentBlob).filter_by(document_id=document_id).delete(
552 synchronize_session=False
553 )
555 # Delete collection links
556 session.query(DocumentCollection).filter_by(
557 document_id=document_id
558 ).delete(synchronize_session=False)
560 # Delete document itself
561 deleted = (
562 session.query(Document)
563 .filter_by(id=document_id)
564 .delete(synchronize_session=False)
565 )
567 if deleted:
568 logger.debug(f"Deleted document {document_id[:8]}... completely")
570 return deleted > 0