"""
Library RAG Service

Handles indexing and searching library documents using RAG:
- Index text documents into vector database
- Chunk documents for semantic search
- Generate embeddings using local models
- Manage FAISS indices per research
- Track RAG status in library
"""

import threading
import time
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple

from langchain_core.documents import Document as LangchainDocument
from loguru import logger
from sqlalchemy import func

from ...config.paths import get_cache_directory
from ...database.models.library import (
    Document,
    DocumentChunk,
    DocumentCollection,
    Collection,
    RAGIndex,
    RagDocumentStatus,
    EmbeddingProvider,
)
from ...database.session_context import get_user_db_session, safe_rollback
from ...utilities.type_utils import to_bool
from ..utils import ensure_in_collection
from ...embeddings.splitters import get_text_splitter
from ...web_search_engines.engines.local_embedding_manager import (
    LocalEmbeddingManager,
)
from ...security.file_integrity import FileIntegrityManager, FAISSIndexVerifier
import hashlib
from faiss import IndexFlatL2, IndexFlatIP, IndexHNSWFlat
from langchain_community.vectorstores import FAISS
from langchain_community.docstore.in_memory import InMemoryDocstore
from .faiss_safe_load import safe_load_faiss


# Module-level locks serialise the FAISS save+record critical section and the
# load_or_create verify→quarantine→build sequence per (username, index_path).
# MUST be module-level: each auto-index / scheduler / search worker constructs
# its own LibraryRAGService, so an instance-scoped lock would coordinate
# nothing. Pattern is adapted from web/queue/processor_v2._user_critical_locks
# — instance-scoped→module-scoped, username-only-key→(username, path)-key.
# See #4197 for the race this guards (concurrent save_local interleaves bytes,
# producing checksum_mismatch → destructive unlink, lost data).
_faiss_write_locks: Dict[Tuple[str, str], threading.Lock] = {}
_faiss_write_locks_lock = threading.Lock()

# Hard cap on suffix-increment retries when generating the .corrupt-<ns> path.
# Normal case is one attempt — same-ns collisions only happen if the user
# manually created such files. 32 is a safety bound that converts a deadlock
# into a loud OSError.
_QUARANTINE_SUFFIX_RETRY_CAP = 32

# After a successful quarantine, keep at most this many older .corrupt-*
# files per base path (per side: .faiss.corrupt-* and .pkl.corrupt-* are
# counted independently). Prevents unbounded disk growth on systems that
# experience recurring corruption while preserving recent diagnostic
# artefacts. Keeping 5 means the user has the last 5 corruption events
# to inspect / submit with a bug report; anything older is dropped.
_QUARANTINE_KEEP_RECENT = 5


def _get_faiss_write_lock(username: str, index_path: str) -> threading.Lock:
    """Return the lock for ``(username, index_path)``, creating it on first
    access. Key is normalised via ``Path.resolve()`` to match
    ``FileIntegrityManager._normalize_path`` so writers/readers/quarantine
    all agree on identity.
    """
    key = (username, str(Path(index_path).resolve()))
    with _faiss_write_locks_lock:
        lock = _faiss_write_locks.get(key)
        if lock is None:
            lock = threading.Lock()
            _faiss_write_locks[key] = lock
        return lock


def pop_faiss_locks_for_user(username: str) -> None:
    """Remove all FAISS-write locks belonging to ``username``.

    Called from the user-close paths (connection_cleanup) so the dict
    doesn't grow one entry per (user × collection) across the process
    lifetime. Safe to call while a lock is held: the holder keeps using
    its local reference, and the next access lazily creates a fresh
    lock. Same semantics as ``pop_user_critical_lock`` in
    web/queue/processor_v2.py.
    """
    with _faiss_write_locks_lock:
        stale = [k for k in _faiss_write_locks if k[0] == username]
        for k in stale:
            _faiss_write_locks.pop(k, None)


class LibraryRAGService:
    """Service for managing RAG indexing of library documents."""

    def __init__(
        self,
        username: str,
        embedding_model: str = "all-MiniLM-L6-v2",
        embedding_provider: str = "sentence_transformers",
        chunk_size: int = 1000,
        chunk_overlap: int = 200,
        splitter_type: str = "recursive",
        text_separators: Optional[list] = None,
        distance_metric: str = "cosine",
        normalize_vectors: bool = True,
        index_type: str = "flat",
        embedding_manager: Optional["LocalEmbeddingManager"] = None,
        db_password: Optional[str] = None,
    ):
        """
        Initialize library RAG service for a user.

        Args:
            username: Username for database access
            embedding_model: Name of the embedding model to use
            embedding_provider: Provider type ('sentence_transformers' or 'ollama')
            chunk_size: Size of text chunks for splitting
            chunk_overlap: Overlap between consecutive chunks
            splitter_type: Type of splitter ('recursive', 'token', 'sentence', 'semantic')
            text_separators: List of text separators for chunking (default: ["\n\n", "\n", ". ", " ", ""])
            distance_metric: Distance metric ('cosine', 'l2', or 'dot_product')
            normalize_vectors: Whether to normalize vectors with L2
            index_type: FAISS index type ('flat', 'hnsw', or 'ivf')
            embedding_manager: Optional pre-constructed LocalEmbeddingManager for testing/flexibility
            db_password: Optional database password for background thread access
        """
        self.username = username
        self._db_password = db_password  # Can be used for thread access
        # Initialize optional attributes to None before they're set below
        # This allows the db_password setter to check them without hasattr
        self.embedding_manager = None
        self.integrity_manager = None
        self.embedding_model = embedding_model
        self.embedding_provider = embedding_provider
        self.chunk_size = chunk_size
        self.chunk_overlap = chunk_overlap
        self.splitter_type = splitter_type
        self.text_separators = (
            text_separators
            if text_separators is not None
            else ["\n\n", "\n", ". ", " ", ""]
        )
        self.distance_metric = distance_metric
        # Ensure normalize_vectors is always a proper boolean
        self.normalize_vectors = to_bool(normalize_vectors, default=True)
        self.index_type = index_type

        # Emit the active configuration so users can confirm their
        # UI-configured embedding settings are being honored (regression
        # signal for #3453).
        logger.info(
            f"RAG service initialized for user={username}: "
            f"provider={embedding_provider} model={embedding_model} "
            f"chunk_size={chunk_size} chunk_overlap={chunk_overlap} "
            f"splitter={splitter_type} index_type={index_type}"
        )

        # Use provided embedding manager or create a new one
        # (Must be created before text splitter for semantic chunking)
        # Track ownership so close() only tears down the manager when we
        # constructed it — a caller-supplied manager stays under caller
        # control (test fixtures, multi-service callers reusing one manager).
        self._owns_embedding_manager = embedding_manager is None
        if embedding_manager is not None:
            self.embedding_manager = embedding_manager
        else:
            # Initialize embedding manager with library collection
            # Load the complete user settings snapshot from database using the proper method
            from ...settings.manager import SettingsManager

            # Use proper database session for SettingsManager
            # Note: using _db_password (backing field) directly here because the
            # db_password property setter propagates to embedding_manager/integrity_manager,
            # which are still None at this point in __init__.
            with get_user_db_session(username, self._db_password) as session:
                settings_manager = SettingsManager(session)
                settings_snapshot = settings_manager.get_settings_snapshot()

            # Add the specific settings needed for this RAG service
            settings_snapshot.update(
                {
                    "_username": username,
                    "embeddings.provider": embedding_provider,
                    f"embeddings.{embedding_provider}.model": embedding_model,
                    "local_search_chunk_size": chunk_size,
                    "local_search_chunk_overlap": chunk_overlap,
                }
            )

            # Egress policy pre-flight at the constructor boundary so
            # every direct ``LibraryRAGService(...)`` construction site
            # is covered, not just the factory. Skipped when an
            # ``embedding_manager`` is injected (tests / advanced flows)
            # — those callers vouch for the manager themselves.
            from ...security.egress.policy import (
                Decision,
                PolicyDeniedError,
                context_from_snapshot,
                evaluate_embeddings,
                resolve_run_primary_engine,
            )

            # Build the context from the ACTUAL scope (not a hardcoded
            # BOTH) so PRIVATE_ONLY forces local embeddings even when the
            # raw embeddings.require_local flag is at its default False —
            # context_from_snapshot applies that coupling. Resolve the primary
            # via the shared helper (single source of truth) instead of the old
            # search.tool + searxng fallback, which was a fail-OPEN: a missing
            # primary defaulted to the public searxng so the scope relaxed and a
            # cloud embedder could be admitted. A missing primary now raises ->
            # fail closed via the ValueError handler below.
            try:
                primary = resolve_run_primary_engine(settings_snapshot)
                policy_ctx = context_from_snapshot(
                    settings_snapshot, primary, username=username
                )
            except PolicyDeniedError:
                raise
            except ValueError as exc:
                raise PolicyDeniedError(
                    Decision(False, "invalid_policy_config"),
                    target=embedding_provider,
                ) from exc
            if policy_ctx.require_local_embeddings:
                decision = evaluate_embeddings(
                    embedding_provider,
                    policy_ctx,
                    settings_snapshot=settings_snapshot,
                )
                if not decision.allowed:
                    logger.bind(policy_audit=True).warning(
                        "LibraryRAGService refused by egress policy",
                        provider=embedding_provider,
                        reason=decision.reason,
                    )
                    raise PolicyDeniedError(decision, target=embedding_provider)

            self.embedding_manager = LocalEmbeddingManager(
                embedding_model=embedding_model,
                embedding_model_type=embedding_provider,
                settings_snapshot=settings_snapshot,
            )

        # Initialize text splitter based on type
        # (Must be created AFTER embedding_manager for semantic chunking)
        self.text_splitter = get_text_splitter(
            splitter_type=self.splitter_type,
            chunk_size=self.chunk_size,
            chunk_overlap=self.chunk_overlap,
            text_separators=self.text_separators,
            embeddings=self.embedding_manager.embeddings
            if self.splitter_type == "semantic"
            else None,
        )

        # Initialize or load FAISS index for library collection
        self.faiss_index = None
        self.rag_index_record = None

        # Initialize file integrity manager for FAISS indexes
        self.integrity_manager = FileIntegrityManager(
            username, password=self._db_password
        )
        self.integrity_manager.register_verifier(FAISSIndexVerifier())

        self._closed = False

    def close(self):
        """Release embedding model and index resources."""
        if self._closed:
            return
        self._closed = True

        # Release embedding manager (which in turn closes the underlying
        # OllamaEmbeddings httpx clients — see LocalEmbeddingManager.close).
        # Only when we own it; caller-supplied managers stay under caller
        # control to avoid double-close / use-after-close.
        if self.embedding_manager is not None:
            if self._owns_embedding_manager:
                self.embedding_manager.close()
            self.embedding_manager = None

        # Clear FAISS index
        if self.faiss_index is not None:
            self.faiss_index = None

        # Clear other resources
        self.rag_index_record = None
        self.integrity_manager = None
        self.text_splitter = None

    def __enter__(self):
        """Enter context manager."""
        return self

    def __exit__(self, exc_type, exc_val, exc_tb):
        """Exit context manager, ensuring cleanup."""
        self.close()
        return False

    @property
    def db_password(self):
        """Get database password."""
        return self._db_password

    @db_password.setter
    def db_password(self, value):
        """Set database password and propagate to embedding manager and integrity manager."""
        self._db_password = value
        if self.embedding_manager:
            self.embedding_manager.db_password = value
        if self.integrity_manager:
            self.integrity_manager.password = value

    def _get_index_hash(
        self,
        collection_name: str,
        embedding_model: str,
        embedding_model_type: str,
    ) -> str:
        """Generate hash for index identification."""
        hash_input = (
            f"{collection_name}:{embedding_model}:{embedding_model_type}"
        )
        return hashlib.sha256(hash_input.encode()).hexdigest()

    def _get_index_path(self, index_hash: str) -> Path:
        """Get path for FAISS index file."""
        # Store in centralized cache directory (respects LDR_DATA_DIR)
        cache_dir = get_cache_directory() / "rag_indices"
        cache_dir.mkdir(parents=True, exist_ok=True)
        return cache_dir / f"{index_hash}.faiss"

    @staticmethod
    def _deduplicate_chunks(
        chunks: List[LangchainDocument],
        chunk_ids: List[str],
        existing_ids: Optional[set] = None,
    ) -> Tuple[List[LangchainDocument], List[str]]:
        """Deduplicate chunks by ID within a batch, optionally excluding existing IDs."""
        seen_ids: set = set()
        new_chunks: List[LangchainDocument] = []
        new_ids: List[str] = []
        for chunk, chunk_id in zip(chunks, chunk_ids):
            if chunk_id not in seen_ids and (
                existing_ids is None or chunk_id not in existing_ids
            ):
                new_chunks.append(chunk)
                new_ids.append(chunk_id)
                seen_ids.add(chunk_id)
        return new_chunks, new_ids

    def _get_or_create_rag_index(self, collection_id: str) -> RAGIndex:
        """Get or create RAGIndex record for the current configuration."""
        with get_user_db_session(self.username, self.db_password) as session:
            # Use collection_<uuid> format
            collection_name = f"collection_{collection_id}"
            index_hash = self._get_index_hash(
                collection_name, self.embedding_model, self.embedding_provider
            )

            # Try to get existing index
            rag_index = (
                session.query(RAGIndex).filter_by(index_hash=index_hash).first()
            )

            if not rag_index:
                # Create new index record
                index_path = self._get_index_path(index_hash)

                # Get embedding dimension by embedding a test string
                test_embedding = self.embedding_manager.embeddings.embed_query(
                    "test"
                )
                embedding_dim = len(test_embedding)

                rag_index = RAGIndex(
                    collection_name=collection_name,
                    embedding_model=self.embedding_model,
                    embedding_model_type=EmbeddingProvider(
                        self.embedding_provider
                    ),
                    embedding_dimension=embedding_dim,
                    index_path=str(index_path),
                    index_hash=index_hash,
                    chunk_size=self.chunk_size,
                    chunk_overlap=self.chunk_overlap,
                    splitter_type=self.splitter_type,
                    text_separators=self.text_separators,
                    distance_metric=self.distance_metric,
                    normalize_vectors=self.normalize_vectors,
                    index_type=self.index_type,
                    chunk_count=0,
                    total_documents=0,
                    status="active",
                    is_current=True,
                )
                session.add(rag_index)
                session.commit()
                session.refresh(rag_index)
                logger.info(f"Created new RAG index: {index_hash}")

            return rag_index

    def _quarantine_corrupt_index(self, index_path: Path, reason: str) -> None:
        """Rename a corrupted FAISS index and its ``.pkl`` companion to
        ``<path>.corrupt-<ns>`` instead of deleting them.

        Preserves user data for inspection/recovery. The
        dimension-mismatch branch elsewhere in this method intentionally
        *deletes* — that case rebuilds from scratch and the old bytes
        are unreadable with the new model. This helper is for the two
        "transient or unknown failure" branches (verify_file said no,
        load_local raised) where the on-disk bytes may still be
        usable by a human.

        Raises ``OSError`` on rename failure (disk full, read-only fs,
        permission denied). Re-raising prevents silent data loss: if we
        swallowed the error, the next ``save_local`` would truncate the
        corrupt bytes anyway. Caller paths log the exception via the
        broader try/except around their indexer call.

        # TODO(#4197-followup): FileIntegrityRecord.consecutive_failures
        # is not reset by the next record_file call, leaking failure
        # counts across recovery cycles. Orthogonal to this fix.
        """
        ns = time.time_ns()
        pkl_path = index_path.with_suffix(".pkl")

        # Same-nanosecond collisions essentially can't happen between
        # concurrent threads (we hold the per-path lock), but a user
        # could manually create such files. Loop with a hard cap so a
        # weird state surfaces as a clean OSError rather than a hang.
        suffix_n = 0
        faiss_target = Path(f"{index_path}.corrupt-{ns}")
        pkl_target = Path(f"{pkl_path}.corrupt-{ns}")
        while faiss_target.exists() or pkl_target.exists():
            suffix_n += 1
            if suffix_n > _QUARANTINE_SUFFIX_RETRY_CAP:
                raise OSError(
                    f"Quarantine path collisions exceeded "
                    f"{_QUARANTINE_SUFFIX_RETRY_CAP} retries for "
                    f"{index_path}"
                )
            faiss_target = Path(f"{index_path}.corrupt-{ns}-{suffix_n}")
            pkl_target = Path(f"{pkl_path}.corrupt-{ns}-{suffix_n}")

        try:
            index_path.rename(faiss_target)
            logger.warning(
                f"Quarantined corrupted FAISS index to {faiss_target} "
                f"(reason: {reason}). Searches against this collection "
                f"will return empty results until the index is rebuilt. "
                f"Document chunks are preserved in the database — "
                f"trigger 'Re-index Collection' (or set "
                f"research_library.auto_index_enabled=true and re-run "
                f"indexing) to recover."
            )
            if pkl_path.exists():
                pkl_path.rename(pkl_target)
                logger.info(f"Quarantined companion PKL to {pkl_target}")
            else:
                # Missing .pkl is recoverable (FAISS may have crashed
                # mid-write between .faiss and .pkl). Not re-raised:
                # the .faiss is already preserved, fresh build will
                # write a fresh .pkl.
                logger.warning(
                    f"PKL companion missing at {pkl_path}; only "
                    f".faiss quarantined."
                )
        except OSError:
            # Disk-full, read-only fs, or permission error. Re-raise
            # so the caller surfaces a real failure instead of letting
            # the next save_local truncate the corrupt bytes.
            logger.exception(
                f"Failed to quarantine corrupted index at {index_path}"
            )
            raise

        # Best-effort retention sweep. The quarantine itself succeeded
        # above; failing to prune older files is not a correctness
        # issue, just a disk-usage one — log and move on.
        self._prune_old_quarantined_files(index_path)

    @staticmethod
    def _corrupt_sort_key(path: Path) -> Tuple[int, int]:
        """Extract ``(ns, suffix_n)`` from a ``.corrupt-<ns>[-<n>]``
        filename so retention can sort by the monotonic nanosecond
        suffix the quarantine path embeds — *not* by ``st_mtime``.

        Filesystem timestamp granularity is sometimes 1s or 2s
        (FAT32/ext3/SMB shares), making mtime ordering non-deterministic
        when multiple quarantines happen within one tick. The
        ``.corrupt-<ns>`` suffix carries the original ``time.time_ns()``
        from the quarantine and is reliable across all filesystems.

        Returns ``(-1, -1)`` for malformed names (manually-placed
        files) so they sort below any real entry under ``reverse=True``
        — i.e., they get pruned first.
        """
        name = path.name
        marker = ".corrupt-"
        idx = name.rfind(marker)
        if idx == -1:
            return (-1, -1)
        tail = name[idx + len(marker) :]
        parts = tail.split("-")
        try:
            ns = int(parts[0])
        except ValueError:
            return (-1, -1)
        suffix_n = 0
        if len(parts) > 1:
            try:
                suffix_n = int(parts[-1])
            except ValueError:
                # Unknown trailing component — keep the file but at the
                # base ns ordering.
                pass
        return (ns, suffix_n)

    def _prune_old_quarantined_files(self, index_path: Path) -> None:
        """Keep only the ``_QUARANTINE_KEEP_RECENT`` most-recent
        ``.corrupt-*`` files for ``index_path`` and its ``.pkl``
        companion. Sweeps the two sides independently — pairs share
        the same ``-<ns>`` suffix so they're ordered identically.

        Ordering uses the embedded ``-<ns>`` from the quarantine
        filename, not ``st_mtime``: file systems with 1-2s timestamp
        granularity can otherwise produce non-deterministic retention
        on bursts.

        Best-effort: logs and swallows any error so a sweep failure
        never propagates back into the indexing path.
        """
        parent = index_path.parent
        pkl_path = index_path.with_suffix(".pkl")

        for base in (index_path, pkl_path):
            pattern = f"{base.name}.corrupt-*"
            try:
                # Sort newest-first by the embedded -<ns>; everything
                # past the keep window is stale.
                candidates = sorted(
                    parent.glob(pattern),
                    key=self._corrupt_sort_key,
                    reverse=True,
                )
            except OSError:
                logger.warning(
                    f"Failed to enumerate {pattern} in {parent} for "
                    f"quarantine retention sweep"
                )
                continue

            for stale in candidates[_QUARANTINE_KEEP_RECENT:]:
                try:
                    stale.unlink()
                    logger.info(f"Pruned old quarantined file: {stale}")
                except OSError:
                    logger.warning(
                        f"Failed to prune quarantined file {stale}; continuing"
                    )

    def _merge_and_persist_locked(
        self,
        index_path: Path,
        chunks_to_add: list,
        embedding_ids: list,
        force_reindex: bool = False,
    ) -> Dict[str, int]:
        """Read-modify-write the on-disk FAISS index under the
        ``(username, index_path)`` write lock so concurrent workers
        don't lose each other's embeddings.

        Without this, two workers indexing different documents into
        the same collection both load on-disk state X into memory,
        each calls ``add_documents`` on their own private FAISS object
        (worker A: ``X+docA``, worker B: ``X+docB``), then they save
        in sequence — last writer wins, the loser's chunks are gone
        from the FAISS file. The chunks survive in the per-document
        DB rows so a force-reindex rebuilds, but the index file is
        wrong until then. See AI review of #4200.

        The slow embedding + DB-chunk-insert path stays OUTSIDE this
        lock (handled by the caller). The lock only covers the fast
        FAISS reload→merge→save sequence (~10-100ms even for large
        indices), so cross-document parallelism is preserved up to
        the point where the in-memory states need reconciliation.

        Args:
            index_path: Path of the on-disk ``.faiss`` file.
            chunks_to_add: Chunks the caller wants to add.
            embedding_ids: Matching embedding/chunk IDs (same length).
            force_reindex: If True, delete any IDs from
                ``embedding_ids`` that already exist in the index
                (so the caller's metadata wins). If False, skip IDs
                that already exist (idempotent re-indexing).

        Returns:
            ``{"added": n, "skipped": m}`` for caller logging.
        """
        with _get_faiss_write_lock(self.username, str(index_path)):
            # Reload from disk to absorb other writers' saves. If the
            # file doesn't exist or fails verification, keep whatever
            # the caller already had in ``self.faiss_index`` (likely
            # a fresh in-memory index from load_or_create_faiss_index).
            if index_path.exists():
                verified, _reason = self.integrity_manager.verify_file(
                    index_path
                )
                if verified:
                    try:
                        # safe_load_faiss replaces load_local's dangerous
                        # pickle deserialization with a restricted unpickler
                        # (see faiss_safe_load) so a tampered .pkl cannot
                        # execute code on load.
                        self.faiss_index = safe_load_faiss(
                            str(index_path.parent),
                            self.embedding_manager.embeddings,
                            index_name=index_path.stem,
                            normalize_L2=True,
                        )
                    except Exception:
                        # Reload failed (torn write, etc.). Keep
                        # in-memory state — it's stale but valid;
                        # better than losing this write entirely.
                        logger.warning(
                            "Failed to reload FAISS for merge; "
                            "proceeding with in-memory state."
                        )

            # Force-reindex: remove old copies of IDs we're about to
            # re-add, so updated metadata replaces stale. Dedup via
            # set→list because ``embedding_ids`` can contain repeats
            # (same chunk hash appearing twice in the document) —
            # FAISS.delete with duplicate IDs raises.
            if force_reindex and hasattr(self.faiss_index, "docstore"):
                fresh_ids = set(self.faiss_index.docstore._dict.keys())
                old_chunk_ids = list(
                    {eid for eid in embedding_ids if eid in fresh_ids}
                )
                if old_chunk_ids:
                    logger.info(
                        f"Force re-index: removing {len(old_chunk_ids)} "
                        f"existing chunks from FAISS"
                    )
                    self.faiss_index.delete(old_chunk_ids)

            # Dedup against the freshly-loaded state.
            if not force_reindex and hasattr(self.faiss_index, "docstore"):
                fresh_ids = set(self.faiss_index.docstore._dict.keys())
            else:
                fresh_ids = None

            new_chunks, new_ids = self._deduplicate_chunks(
                chunks_to_add, embedding_ids, fresh_ids
            )

            if new_chunks:
                self.faiss_index.add_documents(new_chunks, ids=new_ids)

            self.faiss_index.save_local(
                str(index_path.parent), index_name=index_path.stem
            )
            self.integrity_manager.record_file(
                index_path,
                related_entity_type="rag_index",
                related_entity_id=self.rag_index_record.id,
            )

        return {
            "added": len(new_chunks),
            "skipped": len(chunks_to_add) - len(new_chunks),
            "added_ids": new_ids,
        }

    def load_or_create_faiss_index(self, collection_id: str) -> FAISS:
        """
        Load existing FAISS index or create new one.

        Args:
            collection_id: UUID of the collection

        Returns:
            FAISS vector store instance
        """
        rag_index = self._get_or_create_rag_index(collection_id)
        self.rag_index_record = rag_index

        index_path = Path(rag_index.index_path)

        # Hold the per-(username, index_path) write lock across the
        # entire verify → quarantine → load sequence. A narrower scope
        # leaves room for a concurrent save_local to race the
        # verification (verify sees bytes A, save overwrites with bytes
        # B, load_local reads bytes B which no longer match verified
        # checksum). See #4197. The fresh-build path below this block
        # is in-memory only and doesn't need the lock.
        if index_path.exists():
            load_lock = _get_faiss_write_lock(self.username, str(index_path))
            with load_lock:
                # Verify integrity before loading
                verified, reason = self.integrity_manager.verify_file(
                    index_path
                )
                if not verified:
                    logger.error(
                        f"Integrity verification failed for {index_path}: "
                        f"{reason}. Quarantining for recovery; creating "
                        f"new index."
                    )
                    self._quarantine_corrupt_index(index_path, reason)
                else:
                    # Probe the embedding model OUTSIDE the quarantine
                    # try-block: embed_query fails when the embedding
                    # provider is unreachable (e.g. Ollama down), which
                    # says nothing about the index files. Inside the try
                    # it would quarantine a healthy index and silently
                    # replace it with an empty one; here it propagates a
                    # clear provider error instead.
                    current_dim = len(
                        self.embedding_manager.embeddings.embed_query(
                            "dimension_check"
                        )
                    )
                    stored_dim = rag_index.embedding_dimension

                    try:
                        # Check for embedding dimension mismatch before loading
                        if stored_dim and current_dim != stored_dim:
                            logger.warning(
                                f"Embedding dimension mismatch detected! "
                                f"Index created with dim={stored_dim}, "
                                f"current model returns dim={current_dim}. "
                                f"Deleting old index and rebuilding."
                            )
                            # Delete old index files (legitimate deletion:
                            # the bytes are unreadable with the new model).
                            try:
                                index_path.unlink()
                                pkl_path = index_path.with_suffix(".pkl")
                                if pkl_path.exists():
                                    pkl_path.unlink()
                                logger.info(
                                    f"Deleted old FAISS index files at {index_path}"
                                )
                            except Exception:
                                logger.exception(
                                    "Failed to delete old index files"
                                )

                            # Update RAGIndex with new dimension and reset counts
                            with get_user_db_session(
                                self.username, self.db_password
                            ) as session:
                                idx = (
                                    session.query(RAGIndex)
                                    .filter_by(id=rag_index.id)
                                    .first()
                                )
                                if idx:
                                    idx.embedding_dimension = current_dim
                                    idx.chunk_count = 0
                                    idx.total_documents = 0
                                    session.commit()
                                    logger.info(
                                        f"Updated RAGIndex dimension to {current_dim}"
                                    )

                                # Clear rag_document_status for this index
                                session.query(RagDocumentStatus).filter_by(
                                    rag_index_id=rag_index.id
                                ).delete()
                                session.commit()
                                logger.info(
                                    "Cleared indexed status for documents in this "
                                    "collection"
                                )

                            # Update local reference for index creation below
                            rag_index.embedding_dimension = current_dim
                            # Fall through to create new index below
                        else:
                            # Dimensions match (or no stored dimension), load
                            # index via the restricted-unpickler loader (see
                            # faiss_safe_load) instead of load_local, so a
                            # tampered .pkl cannot execute code on load.
                            faiss_index = safe_load_faiss(
                                str(index_path.parent),
                                self.embedding_manager.embeddings,
                                index_name=index_path.stem,
                                normalize_L2=True,
                            )
                            logger.info(
                                f"Loaded existing FAISS index from {index_path}"
                            )
                            return faiss_index
                    except Exception:
                        # load_local raised (torn .pkl, malformed pickle,
                        # FAISS read failure). The .faiss bytes may still
                        # be recoverable, so quarantine before falling
                        # through to fresh build — don't just leave
                        # broken state on disk.
                        logger.warning(
                            "Failed to load FAISS index, quarantining "
                            "and creating new one"
                        )
                        if index_path.exists():
                            self._quarantine_corrupt_index(
                                index_path, "load_local_raised"
                            )

        # Create new FAISS index with configurable type and distance metric
        logger.info(
            f"Creating new FAISS index: type={self.index_type}, metric={self.distance_metric}, dimension={rag_index.embedding_dimension}"
        )

        # Create index based on type and distance metric
        if self.index_type == "hnsw":
            # HNSW: Fast approximate search, best for large collections
            # M=32 is a good default for connections per layer
            index = IndexHNSWFlat(rag_index.embedding_dimension, 32)
            logger.info("Created HNSW index with M=32 connections")
        elif self.index_type == "ivf":
            # IVF requires training, for now fall back to flat
            # TODO: Implement IVF with proper training
            logger.warning(
                "IVF index type not yet fully implemented, using Flat index"
            )
            if self.distance_metric in ("cosine", "dot_product"):
                index = IndexFlatIP(rag_index.embedding_dimension)
            else:
                index = IndexFlatL2(rag_index.embedding_dimension)
        else:  # "flat" or default
            # Flat index: Exact search
            if self.distance_metric in ("cosine", "dot_product"):
                # For cosine similarity, use inner product (IP) with normalized vectors
                index = IndexFlatIP(rag_index.embedding_dimension)
                logger.info(
                    "Created Flat index with Inner Product (for cosine similarity)"
                )
            else:  # l2
                index = IndexFlatL2(rag_index.embedding_dimension)
                logger.info("Created Flat index with L2 distance")

        faiss_index = FAISS(
            self.embedding_manager.embeddings,
            index=index,
            docstore=InMemoryDocstore(),  # Minimal - chunks in DB
            index_to_docstore_id={},
            normalize_L2=self.normalize_vectors,  # Use configurable normalization
        )
        logger.info(
            f"FAISS index created with normalization={self.normalize_vectors}"
        )
        return faiss_index

    def get_current_index_info(
        self, collection_id: Optional[str] = None
    ) -> Optional[Dict[str, Any]]:
        """
        Get information about the current RAG index for a collection.

        Args:
            collection_id: UUID of collection (defaults to Library if None)
        """
        with get_user_db_session(self.username, self.db_password) as session:
            # Get collection name in the format stored in RAGIndex (collection_<uuid>)
            if collection_id:
                collection = (
                    session.query(Collection)
                    .filter_by(id=collection_id)
                    .first()
                )
                collection_name = (
                    f"collection_{collection_id}" if collection else "unknown"
                )
            else:
                # Default to Library collection
                from ...database.library_init import get_default_library_id

                collection_id = get_default_library_id(
                    self.username, self.db_password
                )
                collection_name = f"collection_{collection_id}"

            rag_index = (
                session.query(RAGIndex)
                .filter_by(collection_name=collection_name, is_current=True)
                .first()
            )

            if not rag_index:
                # Debug: check all RAG indices for this collection
                all_indices = session.query(RAGIndex).all()
                logger.info(
                    f"No RAG index found for collection_name='{collection_name}'. All indices: {[(idx.collection_name, idx.is_current) for idx in all_indices]}"
                )
                return None

            # Calculate actual counts from rag_document_status table
            from ...database.models.library import RagDocumentStatus

            actual_chunk_count = (
                session.query(func.sum(RagDocumentStatus.chunk_count))
                .filter_by(collection_id=collection_id)
                .scalar()
                or 0
            )

            actual_doc_count = (
                session.query(RagDocumentStatus)
                .filter_by(collection_id=collection_id)
                .count()
            )

            return {
                "embedding_model": rag_index.embedding_model,
                "embedding_model_type": rag_index.embedding_model_type.value
                if rag_index.embedding_model_type
                else None,
                "embedding_dimension": rag_index.embedding_dimension,
                "chunk_size": rag_index.chunk_size,
                "chunk_overlap": rag_index.chunk_overlap,
                "chunk_count": actual_chunk_count,
                "total_documents": actual_doc_count,
                "created_at": rag_index.created_at.isoformat(),
                "last_updated_at": rag_index.last_updated_at.isoformat(),
            }

    def index_document(
        self, document_id: str, collection_id: str, force_reindex: bool = False
    ) -> Dict[str, Any]:
        """
        Index a single document into RAG for a specific collection.

        Args:
            document_id: UUID of the Document to index
            collection_id: UUID of the Collection to index for
            force_reindex: Whether to force reindexing even if already indexed

        Returns:
            Dict with status, chunk_count, and any errors
        """
        with get_user_db_session(self.username, self.db_password) as session:
            # Get the document
            document = session.query(Document).filter_by(id=document_id).first()

            if not document:
                return {"status": "error", "error": "Document not found"}

            # Get or create DocumentCollection entry
            doc_collection = ensure_in_collection(
                session, document_id, collection_id
            )

            # Check if already indexed for this collection
            if doc_collection.indexed and not force_reindex:
                return {
                    "status": "skipped",
                    "message": "Document already indexed for this collection",
                    "chunk_count": doc_collection.chunk_count,
                }

            # Validate text content
            if not document.text_content:
                return {
                    "status": "error",
                    "error": "Document has no text content",
                }

            try:
                # Create LangChain Document from text
                doc = LangchainDocument(
                    page_content=document.text_content,
                    metadata={
                        "source": document.original_url,
                        "document_id": document_id,  # Add document ID for source linking
                        "collection_id": collection_id,  # Add collection ID
                        "title": document.title
                        or document.filename
                        or "Untitled",
                        "document_title": document.title
                        or document.filename
                        or "Untitled",  # Add for compatibility
                        "authors": document.authors,
                        "published_date": str(document.published_date)
                        if document.published_date
                        else None,
                        "doi": document.doi,
                        "arxiv_id": document.arxiv_id,
                        "pmid": document.pmid,
                        "pmcid": document.pmcid,
                        "extraction_method": document.extraction_method,
                        "word_count": document.word_count,
                    },
                )

                # Split into chunks
                chunks = self.text_splitter.split_documents([doc])
                logger.info(
                    f"Split document {document_id} into {len(chunks)} chunks"
                )

                # Get collection name for chunk storage
                collection = (
                    session.query(Collection)
                    .filter_by(id=collection_id)
                    .first()
                )
                # Use collection_<uuid> format for internal storage
                collection_name = (
                    f"collection_{collection_id}" if collection else "unknown"
                )

                # Store chunks in database using embedding manager
                embedding_ids = self.embedding_manager._store_chunks_to_db(
                    chunks=chunks,
                    collection_name=collection_name,
                    source_type="document",
                    source_id=document_id,
                )

                # Load or create FAISS index (lazy; the merge step
                # below will reload from disk under the lock anyway).
                if self.faiss_index is None:
                    self.faiss_index = self.load_or_create_faiss_index(
                        collection_id
                    )

                # Read-modify-write the on-disk FAISS index under
                # the per-(user, index_path) lock. The lock spans
                # reload + dedup + add + save so concurrent indexers
                # of different documents into the same collection
                # don't lose each other's embeddings (see AI review
                # of #4200).
                index_path = Path(self.rag_index_record.index_path)
                unique_count = len(set(embedding_ids))
                batch_dups = len(chunks) - unique_count
                merge_stats = self._merge_and_persist_locked(
                    index_path,
                    chunks,
                    embedding_ids,
                    force_reindex=force_reindex,
                )
                if merge_stats["added"]:
                    if force_reindex:
                        logger.info(
                            f"Force re-index: added {merge_stats['added']} "
                            f"chunks with updated metadata to FAISS index"
                        )
                    else:
                        already_exist = unique_count - merge_stats["added"]
                        logger.info(
                            f"Added {merge_stats['added']} new embeddings to FAISS "
                            f"({already_exist} already exist, "
                            f"{batch_dups} batch duplicates removed)"
                        )
                else:
                    logger.info(
                        f"All {len(chunks)} chunks already exist in FAISS index, skipping"
                    )
                logger.info(
                    f"Saved FAISS index to {index_path} with integrity tracking"
                )

                from datetime import datetime, UTC

                # Check if document was already indexed (for stats update)
                existing_status = (
                    session.query(RagDocumentStatus)
                    .filter_by(
                        document_id=document_id, collection_id=collection_id
                    )
                    .first()
                )
                was_already_indexed = existing_status is not None

                # Mark document as indexed using rag_document_status table
                # Row existence = indexed, simple and clean
                timestamp = datetime.now(UTC)

                # Create or update RagDocumentStatus using ORM merge (atomic upsert)
                rag_status = RagDocumentStatus(
                    document_id=document_id,
                    collection_id=collection_id,
                    rag_index_id=self.rag_index_record.id,
                    chunk_count=len(chunks),
                    indexed_at=timestamp,
                )
                session.merge(rag_status)

                logger.info(
                    f"Marked document as indexed in rag_document_status: doc_id={document_id}, coll_id={collection_id}, chunks={len(chunks)}"
                )

                # Also update DocumentCollection table for backward compatibility
                session.query(DocumentCollection).filter_by(
                    document_id=document_id, collection_id=collection_id
                ).update(
                    {
                        "indexed": True,
                        "chunk_count": len(chunks),
                        "last_indexed_at": timestamp,
                    }
                )

                logger.info(
                    "Also updated DocumentCollection.indexed for backward compatibility"
                )

                # Update RAGIndex statistics (only if not already indexed)
                rag_index_obj = (
                    session.query(RAGIndex)
                    .filter_by(id=self.rag_index_record.id)
                    .first()
                )
                if rag_index_obj and not was_already_indexed:
                    rag_index_obj.chunk_count += len(chunks)
                    rag_index_obj.total_documents += 1
                    rag_index_obj.last_updated_at = datetime.now(UTC)
                    logger.info(
                        f"Updated RAGIndex stats: chunk_count +{len(chunks)}, total_documents +1"
                    )

                # Flush ORM changes to database before commit
                session.flush()
                logger.info(f"Flushed ORM changes for document {document_id}")

                # Commit the transaction. Durability is provided by
                # synchronous=NORMAL (sqlcipher_utils.py); SQLite
                # auto-checkpoints WAL at wal_autocheckpoint=250 frames.
                # An explicit PRAGMA wal_checkpoint(FULL) here used to
                # block other writers long enough to exhaust busy_timeout
                # under bulk-download concurrency (#4197).
                session.commit()

                logger.info(
                    f"Successfully indexed document {document_id} for collection {collection_id} "
                    f"with {len(chunks)} chunks"
                )

                return {
                    "status": "success",
                    "chunk_count": len(chunks),
                    "embedding_ids": embedding_ids,
                }

            except Exception as e:
                # The session is shared (thread-local) with the caller.
                # If session.flush() or session.commit() raised, the session
                # is in PendingRollbackError state until rolled back —
                # leaving subsequent operations to cascade. Roll back BEFORE
                # returning the error dict so the caller sees a clean
                # session. (Same pattern as the #3827 fix.)
                safe_rollback(session, "library_rag_service.index_document")
                logger.exception(
                    f"Error indexing document {document_id} for collection {collection_id}"
                )
                return {
                    "status": "error",
                    "error": f"Operation failed: {type(e).__name__}",
                }

    def index_all_documents(
        self,
        collection_id: str,
        force_reindex: bool = False,
        progress_callback=None,
    ) -> Dict[str, Any]:
        """
        Index all documents in a collection into RAG.

        Args:
            collection_id: UUID of the collection to index
            force_reindex: Whether to force reindexing already indexed documents
            progress_callback: Optional callback function called after each document with (current, total, doc_title, status)

        Returns:
            Dict with counts of successful, skipped, and failed documents
        """
        with get_user_db_session(self.username, self.db_password) as session:
            # Get all DocumentCollection entries for this collection
            query = session.query(DocumentCollection).filter_by(
                collection_id=collection_id
            )

            if not force_reindex:
                # Only index documents that haven't been indexed yet
                query = query.filter_by(indexed=False)

            doc_collections = query.all()

            if not doc_collections:
                return {
                    "status": "info",
                    "message": "No documents to index",
                    "successful": 0,
                    "skipped": 0,
                    "failed": 0,
                }

            results = {"successful": 0, "skipped": 0, "failed": 0, "errors": []}
            total = len(doc_collections)

            for idx, doc_collection in enumerate(doc_collections, 1):
                # Get the document for title info
                document = (
                    session.query(Document)
                    .filter_by(id=doc_collection.document_id)
                    .first()
                )
                title = document.title if document else "Unknown"

                result = self.index_document(
                    doc_collection.document_id, collection_id, force_reindex
                )

                if result["status"] == "success":
                    results["successful"] += 1
                elif result["status"] == "skipped":
                    results["skipped"] += 1
                else:
                    results["failed"] += 1
                    results["errors"].append(
                        {
                            "doc_id": doc_collection.document_id,
                            "title": title,
                            "error": result.get("error"),
                        }
                    )

                # Call progress callback if provided
                if progress_callback:
                    progress_callback(idx, total, title, result["status"])

            logger.info(
                f"Indexed collection {collection_id}: "
                f"{results['successful']} successful, "
                f"{results['skipped']} skipped, "
                f"{results['failed']} failed"
            )

            return results

    def remove_document_from_rag(
        self, document_id: str, collection_id: str
    ) -> Dict[str, Any]:
        """
        Remove a document's chunks from RAG for a specific collection.

        Args:
            document_id: UUID of the Document to remove
            collection_id: UUID of the Collection to remove from

        Returns:
            Dict with status and count of removed chunks
        """
        with get_user_db_session(self.username, self.db_password) as session:
            # Get the DocumentCollection entry
            doc_collection = (
                session.query(DocumentCollection)
                .filter_by(document_id=document_id, collection_id=collection_id)
                .first()
            )

            if not doc_collection:
                return {
                    "status": "error",
                    "error": "Document not found in collection",
                }

            try:
                # Get collection name in the format collection_<uuid>
                collection = (
                    session.query(Collection)
                    .filter_by(id=collection_id)
                    .first()
                )
                # Use collection_<uuid> format for internal storage
                collection_name = (
                    f"collection_{collection_id}" if collection else "unknown"
                )

                # Delete chunks from database
                deleted_count = self.embedding_manager._delete_chunks_from_db(
                    collection_name=collection_name,
                    source_id=document_id,
                )

                # Update DocumentCollection RAG status
                doc_collection.indexed = False
                doc_collection.chunk_count = 0
                doc_collection.last_indexed_at = None
                session.commit()

                logger.info(
                    f"Removed {deleted_count} chunks for document {document_id} from collection {collection_id}"
                )

                return {"status": "success", "deleted_count": deleted_count}

            except Exception as e:
                # session.commit() above can raise; without rollback the
                # shared thread-local session stays poisoned for the
                # caller's next operation (issue #3827 pattern).
                safe_rollback(
                    session, "library_rag_service.remove_document_from_rag"
                )
                logger.exception(
                    f"Error removing document {document_id} from collection {collection_id}"
                )
                return {
                    "status": "error",
                    "error": f"Operation failed: {type(e).__name__}",
                }

    def index_documents_batch(
        self,
        doc_info: List[tuple],
        collection_id: str,
        force_reindex: bool = False,
    ) -> Dict[str, Dict[str, Any]]:
        """
        Index multiple documents in a batch for a specific collection.

        Args:
            doc_info: List of (doc_id, title) tuples
            collection_id: UUID of the collection to index for
            force_reindex: Whether to force reindexing even if already indexed

        Returns:
            Dict mapping doc_id to individual result
        """
        results = {}
        doc_ids = [doc_id for doc_id, _ in doc_info]

        # Use single database session for querying
        with get_user_db_session(self.username, self.db_password) as session:
            # Pre-load all documents for this batch
            documents = (
                session.query(Document).filter(Document.id.in_(doc_ids)).all()
            )

            # Create lookup for quick access
            doc_lookup = {doc.id: doc for doc in documents}

            # Pre-load DocumentCollection entries
            doc_collections = (
                session.query(DocumentCollection)
                .filter(
                    DocumentCollection.document_id.in_(doc_ids),
                    DocumentCollection.collection_id == collection_id,
                )
                .all()
            )
            doc_collection_lookup = {
                dc.document_id: dc for dc in doc_collections
            }

            # Process each document in the batch
            for doc_id, title in doc_info:
                document = doc_lookup.get(doc_id)

                if not document:
                    results[doc_id] = {
                        "status": "error",
                        "error": "Document not found",
                    }
                    continue

                # Check if already indexed via DocumentCollection
                doc_collection = doc_collection_lookup.get(doc_id)
                if (
                    doc_collection
                    and doc_collection.indexed
                    and not force_reindex
                ):
                    results[doc_id] = {
                        "status": "skipped",
                        "message": "Document already indexed for this collection",
                        "chunk_count": doc_collection.chunk_count,
                    }
                    continue

                # Validate text content
                if not document.text_content:
                    results[doc_id] = {
                        "status": "error",
                        "error": "Document has no text content",
                    }
                    continue

                # Index the document
                try:
                    result = self.index_document(
                        doc_id, collection_id, force_reindex
                    )
                    results[doc_id] = result
                except Exception as e:
                    logger.exception(
                        f"Error indexing document {doc_id} in batch"
                    )
                    results[doc_id] = {
                        "status": "error",
                        "error": f"Indexing failed: {type(e).__name__}",
                    }

        return results

    def get_rag_stats(
        self, collection_id: Optional[str] = None
    ) -> Dict[str, Any]:
        """
        Get RAG statistics for a collection.

        Args:
            collection_id: UUID of the collection (defaults to Library)

        Returns:
            Dict with counts and metadata about indexed documents
        """
        with get_user_db_session(self.username, self.db_password) as session:
            # Get collection ID (default to Library)
            if not collection_id:
                from ...database.library_init import get_default_library_id

                collection_id = get_default_library_id(
                    self.username, self.db_password
                )

            # Count total documents in collection
            total_docs = (
                session.query(DocumentCollection)
                .filter_by(collection_id=collection_id)
                .count()
            )

            # Count indexed documents from rag_document_status table
            from ...database.models.library import RagDocumentStatus

            indexed_docs = (
                session.query(RagDocumentStatus)
                .filter_by(collection_id=collection_id)
                .count()
            )

            # Count total chunks from rag_document_status table
            total_chunks = (
                session.query(func.sum(RagDocumentStatus.chunk_count))
                .filter_by(collection_id=collection_id)
                .scalar()
                or 0
            )

            # Get collection name in the format stored in DocumentChunk (collection_<uuid>)
            collection = (
                session.query(Collection).filter_by(id=collection_id).first()
            )
            collection_name = (
                f"collection_{collection_id}" if collection else "library"
            )

            # Get embedding model info from chunks
            chunk_sample = (
                session.query(DocumentChunk)
                .filter_by(collection_name=collection_name)
                .first()
            )

            embedding_info = {}
            if chunk_sample:
                embedding_info = {
                    "model": chunk_sample.embedding_model,
                    "model_type": chunk_sample.embedding_model_type.value
                    if chunk_sample.embedding_model_type
                    else None,
                    "dimension": chunk_sample.embedding_dimension,
                }

            return {
                "total_documents": total_docs,
                "indexed_documents": indexed_docs,
                "unindexed_documents": total_docs - indexed_docs,
                "total_chunks": total_chunks,
                "embedding_info": embedding_info,
                "chunk_size": self.chunk_size,
                "chunk_overlap": self.chunk_overlap,
            }

    def index_local_file(self, file_path: str) -> Dict[str, Any]:
        """
        Index a local file from the filesystem into RAG.

        Args:
            file_path: Path to the file to index

        Returns:
            Dict with status, chunk_count, and any errors
        """
        from pathlib import Path
        import mimetypes

        file_path = Path(file_path)

        if not file_path.exists():
            return {"status": "error", "error": f"File not found: {file_path}"}

        if not file_path.is_file():
            return {"status": "error", "error": f"Not a file: {file_path}"}

        # Determine file type
        mime_type, _ = mimetypes.guess_type(str(file_path))

        # Read file content based on type
        try:
            if file_path.suffix.lower() in [".txt", ".md", ".markdown"]:
                # Text files
                with open(file_path, "r", encoding="utf-8") as f:
                    content = f.read()
            elif file_path.suffix.lower() in [".html", ".htm"]:
                # HTML files - strip tags
                from bs4 import BeautifulSoup

                with open(file_path, "r", encoding="utf-8") as f:
                    soup = BeautifulSoup(f.read(), "html.parser")
                    content = soup.get_text()
            elif file_path.suffix.lower() == ".pdf":
                # PDF files - extract text
                from pypdf import PdfReader

                content = ""
                with open(file_path, "rb") as f:
                    pdf_reader = PdfReader(f)
                    for page in pdf_reader.pages:
                        content += page.extract_text()
            else:
                return {
                    "status": "skipped",
                    "error": f"Unsupported file type: {file_path.suffix}",
                }

            if not content or len(content.strip()) < 10:
                return {
                    "status": "error",
                    "error": "File has no extractable text content",
                }

            # Create LangChain Document from text
            doc = LangchainDocument(
                page_content=content,
                metadata={
                    "source": str(file_path),
                    "source_id": f"local_{file_path.stem}_{hash(str(file_path))}",
                    "title": file_path.stem,
                    "document_title": file_path.stem,
                    "file_type": file_path.suffix.lower(),
                    "file_size": file_path.stat().st_size,
                    "source_type": "local_file",
                    "collection": "local_library",
                },
            )

            # Split into chunks
            chunks = self.text_splitter.split_documents([doc])
            logger.info(
                f"Split local file {file_path} into {len(chunks)} chunks"
            )

            # Store chunks in database (returns UUID-based IDs)
            embedding_ids = self.embedding_manager._store_chunks_to_db(
                chunks=chunks,
                collection_name="local_library",
                source_type="local_file",
                source_id=str(file_path),
            )

            # Load or create FAISS index using default library collection
            # (lazy; merge step below reloads under the lock anyway).
            if self.faiss_index is None:
                from ...database.library_init import get_default_library_id

                default_collection_id = get_default_library_id(
                    self.username, self.db_password
                )
                self.faiss_index = self.load_or_create_faiss_index(
                    default_collection_id
                )

            # Read-modify-write the on-disk FAISS under the lock.
            index_path = (
                Path(self.rag_index_record.index_path)
                if self.rag_index_record
                else None
            )
            if index_path:
                merge_stats = self._merge_and_persist_locked(
                    index_path, chunks, embedding_ids, force_reindex=False
                )
                logger.info(
                    f"Saved FAISS index to {index_path} with integrity tracking"
                )
            else:
                # No persistent index path — in-memory only path.
                # Preserve old behavior: dedup against in-memory state
                # and add directly without saving.
                if self.faiss_index is not None and hasattr(
                    self.faiss_index, "docstore"
                ):
                    existing_ids = set(self.faiss_index.docstore._dict.keys())
                else:
                    existing_ids = None
                new_chunks, new_ids = self._deduplicate_chunks(
                    chunks, embedding_ids, existing_ids
                )
                if new_chunks:
                    self.faiss_index.add_documents(new_chunks, ids=new_ids)
                merge_stats = {
                    "added": len(new_chunks),
                    "skipped": len(chunks) - len(new_chunks),
                    "added_ids": new_ids,
                }

            logger.info(
                f"Successfully indexed local file {file_path} with "
                f"{merge_stats['added']} new chunks "
                f"({merge_stats['skipped']} skipped)"
            )

            return {
                "status": "success",
                "chunk_count": merge_stats["added"],
                "embedding_ids": merge_stats["added_ids"],
            }

        except Exception as e:
            logger.exception(f"Error indexing local file {file_path}")
            return {
                "status": "error",
                "error": f"Operation failed: {type(e).__name__}",
            }

    def index_user_document(
        self, user_doc, collection_name: str, force_reindex: bool = False
    ) -> Dict[str, Any]:
        """
        Index a user-uploaded document into a specific collection.

        Args:
            user_doc: UserDocument object
            collection_name: Name of the collection (e.g., "collection_123")
            force_reindex: Whether to force reindexing

        Returns:
            Dict with status, chunk_count, and any errors
        """

        try:
            # Use the pre-extracted text content
            content = user_doc.text_content

            if not content or len(content.strip()) < 10:
                return {
                    "status": "error",
                    "error": "Document has no extractable text content",
                }

            # Create LangChain Document
            doc = LangchainDocument(
                page_content=content,
                metadata={
                    "source": f"user_upload_{user_doc.id}",
                    "source_id": user_doc.id,
                    "title": user_doc.filename,
                    "document_title": user_doc.filename,
                    "file_type": user_doc.file_type,
                    "file_size": user_doc.file_size,
                    "collection": collection_name,
                },
            )

            # Split into chunks
            chunks = self.text_splitter.split_documents([doc])
            logger.info(
                f"Split user document {user_doc.filename} into {len(chunks)} chunks"
            )

            # Store chunks in database
            embedding_ids = self.embedding_manager._store_chunks_to_db(
                chunks=chunks,
                collection_name=collection_name,
                source_type="user_document",
                source_id=user_doc.id,
            )

            # Load or create FAISS index for this collection (lazy;
            # merge step below reloads under the lock anyway).
            if self.faiss_index is None:
                # Extract collection_id from collection_name (format: "collection_<uuid>")
                collection_id = collection_name.removeprefix("collection_")
                self.faiss_index = self.load_or_create_faiss_index(
                    collection_id
                )

            unique_count = len(set(embedding_ids))
            batch_dups = len(chunks) - unique_count

            # Read-modify-write the on-disk FAISS under the lock so
            # concurrent uploads to the same collection don't lose
            # each other's chunks.
            index_path = (
                Path(self.rag_index_record.index_path)
                if self.rag_index_record
                else None
            )
            if index_path:
                merge_stats = self._merge_and_persist_locked(
                    index_path,
                    chunks,
                    embedding_ids,
                    force_reindex=force_reindex,
                )
            else:
                # No persistent index path — in-memory only path.
                # Preserve old behavior: handle force_reindex deletion
                # and dedup add against in-memory state without saving.
                if force_reindex and hasattr(self.faiss_index, "docstore"):
                    existing_ids = set(self.faiss_index.docstore._dict.keys())
                    old_chunk_ids = list(
                        {eid for eid in embedding_ids if eid in existing_ids}
                    )
                    if old_chunk_ids:
                        logger.info(
                            f"Force re-index: removing {len(old_chunk_ids)} "
                            f"existing chunks from FAISS"
                        )
                        self.faiss_index.delete(old_chunk_ids)
                if not force_reindex and hasattr(self.faiss_index, "docstore"):
                    existing_ids = set(self.faiss_index.docstore._dict.keys())
                else:
                    existing_ids = None
                new_chunks, new_ids = self._deduplicate_chunks(
                    chunks, embedding_ids, existing_ids
                )
                if new_chunks:
                    self.faiss_index.add_documents(new_chunks, ids=new_ids)
                merge_stats = {
                    "added": len(new_chunks),
                    "skipped": len(chunks) - len(new_chunks),
                    "added_ids": new_ids,
                }
            if merge_stats["added"]:
                if force_reindex:
                    logger.info(
                        f"Force re-index: added {merge_stats['added']} "
                        f"chunks with updated metadata to FAISS index"
                    )
                else:
                    already_exist = unique_count - merge_stats["added"]
                    logger.info(
                        f"Added {merge_stats['added']} new chunks to FAISS "
                        f"({already_exist} already exist, "
                        f"{batch_dups} batch duplicates removed)"
                    )
            else:
                logger.info(
                    f"All {len(chunks)} chunks already exist in FAISS index, skipping"
                )

            logger.info(
                f"Successfully indexed user document {user_doc.filename} with {len(chunks)} chunks"
            )

            return {
                "status": "success",
                "chunk_count": len(chunks),
                "embedding_ids": embedding_ids,
            }

        except Exception as e:
            logger.exception(
                f"Error indexing user document {user_doc.filename}"
            )
            return {
                "status": "error",
                "error": f"Operation failed: {type(e).__name__}",
            }

    def remove_collection_from_index(
        self, collection_name: str
    ) -> Dict[str, Any]:
        """
        Remove all documents from a collection from the FAISS index.

        Args:
            collection_name: Name of the collection (e.g., "collection_123")

        Returns:
            Dict with status and count of removed chunks
        """
        from ...database.models import DocumentChunk
        from ...database.session_context import get_user_db_session

        try:
            with get_user_db_session(
                self.username, self.db_password
            ) as session:
                # Get all chunk IDs for this collection. Select only the
                # id column — loading full DocumentChunk rows would pull
                # every chunk's text into memory just to build the id
                # list (#4560).
                chunks = (
                    session.query(DocumentChunk.id)
                    .filter_by(collection_name=collection_name)
                    .all()
                )

                if not chunks:
                    return {"status": "success", "deleted_count": 0}

                chunk_ids = [
                    f"{collection_name}_{chunk.id}" for chunk in chunks
                ]

                # Load FAISS index if not already loaded
                if self.faiss_index is None:
                    # Extract collection_id from collection_name (format: "collection_<uuid>")
                    collection_id = collection_name.removeprefix("collection_")
                    self.faiss_index = self.load_or_create_faiss_index(
                        collection_id
                    )

                # Remove from FAISS index. delete + save + record must all
                # be inside the same lock — otherwise a concurrent writer
                # could sandwich a save_local between our delete and our
                # save, leaving stale chunks back on disk (#4197).
                if hasattr(self.faiss_index, "delete"):
                    try:
                        index_path = (
                            Path(self.rag_index_record.index_path)
                            if self.rag_index_record
                            else None
                        )
                        if index_path:
                            with _get_faiss_write_lock(
                                self.username, str(index_path)
                            ):
                                self.faiss_index.delete(chunk_ids)
                                self.faiss_index.save_local(
                                    str(index_path.parent),
                                    index_name=index_path.stem,
                                )
                                self.integrity_manager.record_file(
                                    index_path,
                                    related_entity_type="rag_index",
                                    related_entity_id=self.rag_index_record.id,
                                )
                        else:
                            # No index path → in-memory-only delete
                            self.faiss_index.delete(chunk_ids)
                    except Exception:
                        logger.warning("Could not delete chunks from FAISS")

                logger.info(
                    f"Removed {len(chunk_ids)} chunks from collection {collection_name}"
                )

                return {"status": "success", "deleted_count": len(chunk_ids)}

        except Exception as e:
            logger.exception(
                f"Error removing collection {collection_name} from index"
            )
            return {
                "status": "error",
                "error": f"Operation failed: {type(e).__name__}",
            }
