"""
Library Management Service

Handles querying and managing the downloaded document library:
- Search and filter documents
- Get statistics and analytics
- Manage collections and favorites
- Handle file operations
"""

from pathlib import Path
from typing import Dict, List, Optional
from urllib.parse import urlparse

from loguru import logger
from sqlalchemy import or_, func, case
from sqlalchemy.orm import aliased, defer

from ...constants import FILE_PATH_SENTINELS
from ...database.models.download_tracker import DownloadTracker
from ...database.models.library import (
    Collection,
    Document,
    DocumentBlob,
    DocumentCollection,
    DocumentStatus,
)
from ...database.models.metrics import ResearchRating
from ...database.models.research import ResearchHistory, ResearchResource
from ...database.session_context import get_user_db_session
from ...security import PathValidator
from ...config.paths import get_library_directory
from ..utils import (
    get_absolute_path_from_settings,
    get_url_hash,
    open_file_location,
)

# Filter dropdowns (library page + ``/api/research-list``) load research
# sessions into a client-side ``<select>``. Bound the query so a very large
# history cannot pull every row into memory / the DOM on each page load
# (#4560). The query already projects to three small columns, so this is a
# safety cap rather than a crash fix; realistic histories are far smaller, and
# comprehensive large-scale filtering is a server-side-search follow-up.
_DROPDOWN_RESEARCH_LIMIT = 1000

# ``get_unique_domains`` scans one URL row per downloaded document to build the
# domain filter. Stream it in batches (``yield_per``) rather than materializing
# every row at once, so a very large library cannot exhaust memory (#4560). The
# distinct-domain set it accumulates is small regardless of library size.
_DOMAIN_SCAN_BATCH_SIZE = 1000


class LibraryService:
    """Service for managing and querying the document library."""

    def __init__(self, username: str):
        """Initialize library service for a user."""
        self.username = username

    def _has_blob_in_db(self, session, document_id: str) -> bool:
        """Check if a PDF blob exists in the database for a document."""
        return (
            session.query(DocumentBlob.document_id)
            .filter_by(document_id=document_id)
            .first()
            is not None
        )

    def _get_safe_absolute_path(self, file_path: str) -> Optional[str]:
        """
        Get the absolute path for a file, safely handling invalid paths.

        Args:
            file_path: Relative file path from library root

        Returns:
            Absolute path as string, or None if path is invalid/unsafe
        """
        if not file_path or file_path in FILE_PATH_SENTINELS:
            return None
        abs_path = get_absolute_path_from_settings(file_path)
        return str(abs_path) if abs_path else None

    def _is_arxiv_url(self, url: str) -> bool:
        """Check if URL is from arXiv domain."""
        try:
            hostname = urlparse(url).hostname
            return bool(
                hostname
                and (hostname == "arxiv.org" or hostname.endswith(".arxiv.org"))
            )
        except Exception:
            return False

    def _is_pubmed_url(self, url: str) -> bool:
        """Check if URL is from PubMed or NCBI domains."""
        try:
            parsed = urlparse(url)
            hostname = parsed.hostname
            if not hostname:
                return False

            # Check for pubmed.ncbi.nlm.nih.gov
            if hostname == "pubmed.ncbi.nlm.nih.gov":
                return True

            # Check for ncbi.nlm.nih.gov with PMC path
            if hostname == "ncbi.nlm.nih.gov" and "/pmc" in parsed.path:
                return True

            # Check for pubmed in subdomain
            if "pubmed" in hostname:
                return True

            return False
        except Exception:
            return False

    def _apply_date_filter(self, query, model_class, date_filter: str):
        """Apply date range filter based on processed_at timestamp."""
        from datetime import datetime, timedelta, timezone

        now = datetime.now(timezone.utc)
        if date_filter == "today":
            cutoff = now.replace(hour=0, minute=0, second=0, microsecond=0)
        elif date_filter == "week":
            cutoff = now - timedelta(days=7)
        elif date_filter == "month":
            cutoff = now - timedelta(days=30)
        else:
            return query
        return query.filter(model_class.processed_at >= cutoff)

    @staticmethod
    def _escape_like(value: str) -> str:
        """Escape SQL LIKE wildcards (%, _) and the escape char itself.

        Without this, a value like ``my_journal`` would have ``_`` interpreted
        as a single-character wildcard, and ``%`` would match anything.
        Used together with ``escape="\\\\"`` on the .like()/.ilike() call.
        """
        return (
            value.replace("\\", "\\\\").replace("%", "\\%").replace("_", "\\_")
        )

    def _apply_domain_filter(self, query, model_class, domain: str):
        """Apply domain filter to query for Document.

        The dropdown is fully data-driven (populated from get_unique_domains),
        so the filter is a generic substring match against original_url.
        """
        pattern = f"%{self._escape_like(domain)}%"
        return query.filter(model_class.original_url.like(pattern, escape="\\"))

    def _apply_search_filter(self, query, model_class, search_query: str):
        """Apply search filter to query for Document."""
        search_pattern = f"%{self._escape_like(search_query)}%"
        return query.filter(
            or_(
                model_class.title.ilike(search_pattern, escape="\\"),
                model_class.authors.ilike(search_pattern, escape="\\"),
                model_class.doi.ilike(search_pattern, escape="\\"),
                ResearchResource.title.ilike(search_pattern, escape="\\"),
            )
        )

    def get_library_stats(self) -> Dict:
        """Get overall library statistics."""
        with get_user_db_session(self.username) as session:
            # Get document counts
            total_docs = session.query(Document).count()
            total_pdfs = (
                session.query(Document).filter_by(file_type="pdf").count()
            )

            # Get size stats
            size_result = session.query(
                func.sum(Document.file_size),
                func.avg(Document.file_size),
            ).first()

            total_size = size_result[0] or 0
            avg_size = size_result[1] or 0

            # Get research stats
            research_count = session.query(
                func.count(func.distinct(Document.research_id))
            ).scalar()

            # Get domain stats - count unique domains from URLs
            # Extract domain from original_url using SQL functions
            from sqlalchemy import case, func as sql_func

            # Count unique domains by extracting them from URLs
            domain_subquery = session.query(
                sql_func.distinct(
                    case(
                        (
                            Document.original_url.like("%arxiv.org%"),
                            "arxiv.org",
                        ),
                        (
                            Document.original_url.like("%pubmed%"),
                            "pubmed",
                        ),
                        (
                            Document.original_url.like("%ncbi.nlm.nih.gov%"),
                            "pubmed",
                        ),
                        else_="other",
                    )
                )
            ).subquery()

            domain_count = (
                session.query(sql_func.count())
                .select_from(domain_subquery)
                .scalar()
            )

            # Get download tracker stats
            pending_downloads = (
                session.query(DownloadTracker)
                .filter_by(is_downloaded=False)
                .count()
            )

            return {
                "total_documents": total_docs,
                "total_pdfs": total_pdfs,
                "total_size_bytes": total_size,
                "total_size_mb": total_size / (1024 * 1024)
                if total_size
                else 0,
                "average_size_mb": avg_size / (1024 * 1024) if avg_size else 0,
                "research_sessions": research_count,
                "unique_domains": domain_count,
                "pending_downloads": pending_downloads,
                "storage_path": self._get_storage_path(),
            }

    def count_documents(
        self,
        research_id: Optional[str] = None,
        domain: Optional[str] = None,
        collection_id: Optional[str] = None,
        date_filter: Optional[str] = None,
    ) -> int:
        """Count documents matching the given filters (for pagination)."""
        with get_user_db_session(self.username) as session:
            from ...database.library_init import get_default_library_id

            if not collection_id:
                collection_id = get_default_library_id(self.username)

            q = (
                session.query(func.count(Document.id))
                .join(
                    DocumentCollection,
                    Document.id == DocumentCollection.document_id,
                )
                .filter(DocumentCollection.collection_id == collection_id)
                .filter(Document.status == "completed")
            )

            if research_id:
                q = q.filter(Document.research_id == research_id)
            if domain:
                q = self._apply_domain_filter(q, Document, domain)
            if date_filter:
                q = self._apply_date_filter(q, Document, date_filter)

            return q.scalar() or 0

    def get_documents(
        self,
        research_id: Optional[str] = None,
        domain: Optional[str] = None,
        file_type: Optional[str] = None,
        favorites_only: bool = False,
        search_query: Optional[str] = None,
        collection_id: Optional[str] = None,
        date_filter: Optional[str] = None,
        limit: int = 100,
        offset: int = 0,
    ) -> List[Dict]:
        """
        Get documents with filtering options.

        Returns enriched document information with research details.
        """
        with get_user_db_session(self.username) as session:
            # Get default Library collection ID if not specified
            from ...database.library_init import get_default_library_id

            if not collection_id:
                collection_id = get_default_library_id(self.username)

            logger.info(
                f"[LibraryService] Getting documents for collection_id: {collection_id}, research_id: {research_id}, domain: {domain}"
            )

            all_documents = []

            # Step 1: subquery to get paginated document IDs.
            # Pagination and sorting happen here on Document alone,
            # avoiding non-determinism from outer-joining related tables.
            doc_subq = (
                session.query(Document.id)
                .join(
                    DocumentCollection,
                    Document.id == DocumentCollection.document_id,
                )
                .filter(DocumentCollection.collection_id == collection_id)
            )

            # Apply filters
            if research_id:
                doc_subq = doc_subq.filter(Document.research_id == research_id)

            if domain:
                doc_subq = self._apply_domain_filter(doc_subq, Document, domain)

            if date_filter:
                doc_subq = self._apply_date_filter(
                    doc_subq, Document, date_filter
                )

            if file_type:
                doc_subq = doc_subq.filter(Document.file_type == file_type)

            if favorites_only:
                doc_subq = doc_subq.filter(Document.favorite.is_(True))

            if search_query:
                # _apply_search_filter references ResearchResource.title,
                # so we must outerjoin it here. DISTINCT prevents fan-out
                # from the outerjoin duplicating Document IDs.
                doc_subq = doc_subq.outerjoin(
                    ResearchResource,
                    (Document.resource_id == ResearchResource.id)
                    | (ResearchResource.document_id == Document.id),
                ).distinct()
                doc_subq = self._apply_search_filter(
                    doc_subq, Document, search_query
                )

            # Filter to only completed documents
            doc_subq = doc_subq.filter(Document.status == "completed")

            # Sort at SQL level (SQLite-safe NULL handling)
            doc_subq = doc_subq.order_by(
                case((Document.processed_at.isnot(None), 0), else_=1),
                Document.processed_at.desc(),
            )

            # Apply SQL-level pagination
            doc_subq = doc_subq.offset(offset).limit(limit)
            doc_id_subq = doc_subq.subquery()

            # Step 2: join the paginated document IDs with related tables.
            # Use two separate outer joins for ResearchResource to avoid
            # the OR-condition join that can fan out to multiple rows:
            #   - ResourceByFK: matched via Document.resource_id (primary FK)
            #   - ResourceByDoc: matched via ResearchResource.document_id
            # We prefer ResourceByFK; fall back to ResourceByDoc in Python.
            ResourceByFK = aliased(ResearchResource)
            ResourceByDoc = aliased(ResearchResource)

            query = (
                session.query(
                    Document,
                    ResourceByFK,
                    ResourceByDoc,
                    ResearchHistory,
                    DocumentCollection,
                )
                .join(doc_id_subq, Document.id == doc_id_subq.c.id)
                .join(
                    DocumentCollection,
                    Document.id == DocumentCollection.document_id,
                )
                .outerjoin(
                    ResourceByFK,
                    Document.resource_id == ResourceByFK.id,
                )
                .outerjoin(
                    ResourceByDoc,
                    ResourceByDoc.document_id == Document.id,
                )
                .outerjoin(
                    ResearchHistory,
                    Document.research_id == ResearchHistory.id,
                )
                .filter(DocumentCollection.collection_id == collection_id)
                # Re-apply sort so final results are ordered
                .order_by(
                    case((Document.processed_at.isnot(None), 0), else_=1),
                    Document.processed_at.desc(),
                )
            )

            # Execute query
            results = query.all()
            logger.info(
                f"[LibraryService] Found {len(results)} documents in collection {collection_id}"
            )

            # Batch-check blob existence to avoid N+1 queries
            doc_ids = [row[0].id for row in results]
            blob_ids = set()
            if doc_ids:
                blob_ids = {
                    r[0]
                    for r in session.query(DocumentBlob.document_id)
                    .filter(DocumentBlob.document_id.in_(doc_ids))
                    .all()
                }

            # Process results — deduplicate by doc.id since the ResourceByDoc
            # outer join can fan out when multiple ResearchResource rows
            # point to the same document via document_id.
            seen_doc_ids = set()
            for doc, res_by_fk, res_by_doc, research, doc_collection in results:
                if doc.id in seen_doc_ids:
                    continue
                seen_doc_ids.add(doc.id)
                # Prefer the resource matched via Document.resource_id FK;
                # fall back to the one matched via ResearchResource.document_id.
                resource = res_by_fk or res_by_doc
                # Determine availability flags - use Document.file_path directly
                file_absolute_path = None
                if doc.file_path and doc.file_path not in FILE_PATH_SENTINELS:
                    abs_path = get_absolute_path_from_settings(doc.file_path)
                    if abs_path:
                        file_absolute_path = str(abs_path)

                # Check if PDF is available (filesystem OR database)
                has_pdf = bool(file_absolute_path)
                if not has_pdf and doc.storage_mode == "database":
                    has_pdf = doc.id in blob_ids
                has_text_db = bool(doc.text_content)  # Text now in Document

                # Use DocumentCollection from query results
                has_rag_indexed = (
                    doc_collection.indexed if doc_collection else False
                )
                rag_chunk_count = (
                    doc_collection.chunk_count if doc_collection else 0
                )

                all_documents.append(
                    {
                        "id": doc.id,
                        "resource_id": doc.resource_id,
                        "research_id": doc.research_id,
                        # Document info
                        "document_title": doc.title
                        or (resource.title if resource else doc.filename),
                        "authors": doc.authors,
                        "published_date": doc.published_date,
                        "doi": doc.doi,
                        "arxiv_id": doc.arxiv_id,
                        "pmid": doc.pmid,
                        # File info
                        "file_path": doc.file_path,
                        "file_absolute_path": file_absolute_path,
                        "file_name": Path(doc.file_path).name
                        if doc.file_path
                        and doc.file_path not in FILE_PATH_SENTINELS
                        else doc.filename,
                        "file_size": doc.file_size,
                        "file_type": doc.file_type,
                        # URLs
                        "original_url": doc.original_url,
                        "domain": self._extract_domain(doc.original_url)
                        if doc.original_url
                        else "User Upload",
                        # Status
                        "download_status": doc.status or "completed",
                        "downloaded_at": doc.processed_at.isoformat()
                        if doc.processed_at
                        else (
                            doc.uploaded_at.isoformat()
                            if hasattr(doc, "uploaded_at") and doc.uploaded_at
                            else None
                        ),
                        "favorite": doc.favorite
                        if hasattr(doc, "favorite")
                        else False,
                        "tags": doc.tags if hasattr(doc, "tags") else [],
                        # Research info (None for user uploads)
                        "research_title": research.title or research.query[:80]
                        if research
                        else "User Upload",
                        "research_query": research.query if research else None,
                        "research_mode": research.mode if research else None,
                        "research_date": research.created_at
                        if research
                        else None,
                        # Classification flags
                        "is_arxiv": self._is_arxiv_url(doc.original_url)
                        if doc.original_url
                        else False,
                        "is_pubmed": self._is_pubmed_url(doc.original_url)
                        if doc.original_url
                        else False,
                        "is_pdf": doc.file_type == "pdf",
                        # Availability flags
                        "has_pdf": has_pdf,
                        "has_text_db": has_text_db,
                        "has_rag_indexed": has_rag_indexed,
                        "rag_chunk_count": rag_chunk_count,
                    }
                )

            # Sorting and pagination are now handled at SQL level
            return all_documents

    def get_all_collections(self) -> List[Dict]:
        """Get all collections with document and indexed document counts."""
        with get_user_db_session(self.username) as session:
            # Query collections with document counts and indexed counts
            results = (
                session.query(
                    Collection,
                    func.count(DocumentCollection.document_id).label(
                        "document_count"
                    ),
                    func.count(
                        case(
                            (
                                DocumentCollection.indexed == True,  # noqa: E712
                                DocumentCollection.document_id,
                            ),
                            else_=None,
                        )
                    ).label("indexed_document_count"),
                )
                .outerjoin(
                    DocumentCollection,
                    Collection.id == DocumentCollection.collection_id,
                )
                .group_by(Collection.id)
                .order_by(Collection.is_default.desc(), Collection.name)
                .all()
            )

            logger.info(f"[LibraryService] Found {len(results)} collections")

            collections = []
            for collection, doc_count, indexed_count in results:
                logger.debug(
                    f"[LibraryService] Collection: {collection.name} (ID: {collection.id}), documents: {doc_count}, indexed: {indexed_count}"
                )
                collections.append(
                    {
                        "id": collection.id,
                        "name": collection.name,
                        "description": collection.description,
                        "is_default": collection.is_default,
                        "document_count": doc_count or 0,
                        "indexed_document_count": indexed_count or 0,
                    }
                )

            return collections

    def get_research_list_for_dropdown(self) -> List[Dict]:
        """Get minimal research session info for filter dropdowns.

        Returns only id, title, and query — no joins or aggregates — for at
        most the ``_DROPDOWN_RESEARCH_LIMIT`` most-recent sessions, so a very
        large history cannot load every row into memory / the DOM (#4560).
        """
        with get_user_db_session(self.username) as session:
            results = (
                session.query(
                    ResearchHistory.id,
                    ResearchHistory.title,
                    ResearchHistory.query,
                )
                .order_by(ResearchHistory.created_at.desc())
                .limit(_DROPDOWN_RESEARCH_LIMIT)
                .all()
            )
            return [
                {"id": r.id, "title": r.title, "query": r.query}
                for r in results
            ]

    def get_research_list_with_stats(
        self,
        limit: int = 0,
        offset: int = 0,
    ) -> List[Dict]:
        """Get research sessions with download statistics.

        Args:
            limit: Maximum number of results (0 = no limit, for backwards compat).
            offset: Number of rows to skip. Only applied when limit > 0.
        """
        with get_user_db_session(self.username) as session:
            # Query research sessions with resource counts
            query = (
                session.query(
                    ResearchHistory,
                    func.count(func.distinct(ResearchResource.id)).label(
                        "total_resources"
                    ),
                    func.count(
                        func.distinct(
                            case(
                                (Document.status == "completed", Document.id),
                                else_=None,
                            )
                        )
                    ).label("downloaded_count"),
                    func.count(
                        func.distinct(
                            case(
                                (
                                    ResearchResource.url.like("%.pdf")
                                    | ResearchResource.url.like("%arxiv.org%")
                                    | ResearchResource.url.like(
                                        "%ncbi.nlm.nih.gov/pmc%"
                                    ),
                                    ResearchResource.id,
                                ),
                                else_=None,
                            )
                        )
                    ).label("downloadable_count"),
                )
                .outerjoin(
                    ResearchResource,
                    ResearchHistory.id == ResearchResource.research_id,
                )
                .outerjoin(
                    Document,
                    (ResearchResource.id == Document.resource_id)
                    | (ResearchResource.document_id == Document.id),
                )
                .group_by(ResearchHistory.id)
                .order_by(ResearchHistory.created_at.desc())
            )

            # Apply SQL-level pagination when limit is set
            if limit > 0:
                query = query.offset(offset).limit(limit)

            results = query.all()

            # Preload all ratings to avoid N+1 queries
            research_ids = [r[0].id for r in results]
            all_ratings = (
                session.query(ResearchRating)
                .filter(ResearchRating.research_id.in_(research_ids))
                .all()
                if research_ids
                else []
            )
            ratings_by_research = {r.research_id: r for r in all_ratings}

            # Batch domain queries to avoid N+1 (same pattern as ratings)
            domain_case = case(
                (
                    ResearchResource.url.like("%arxiv.org%"),
                    "arxiv.org",
                ),
                (ResearchResource.url.like("%pubmed%"), "pubmed"),
                (
                    ResearchResource.url.like("%ncbi.nlm.nih.gov%"),
                    "pubmed",
                ),
                else_="other",
            )
            all_domains = (
                session.query(
                    ResearchResource.research_id,
                    domain_case.label("domain"),
                    func.count().label("count"),
                )
                .filter(ResearchResource.research_id.in_(research_ids))
                .group_by(ResearchResource.research_id, domain_case)
                .all()
                if research_ids
                else []
            )
            domains_by_research: Dict[str, list] = {}
            for rid, domain, count in all_domains:
                domains_by_research.setdefault(rid, []).append((domain, count))

            research_list = []
            for (
                research,
                total_resources,
                downloaded_count,
                downloadable_count,
            ) in results:
                # Get rating from preloaded dict
                rating = ratings_by_research.get(research.id)

                # Get domain breakdown from preloaded dict
                domains = domains_by_research.get(research.id, [])

                research_list.append(
                    {
                        "id": research.id,
                        "title": research.title,
                        "query": research.query,
                        "mode": research.mode,
                        "status": research.status,
                        "created_at": research.created_at,
                        "duration_seconds": research.duration_seconds,
                        "total_resources": total_resources or 0,
                        "downloaded_count": downloaded_count or 0,
                        "downloadable_count": downloadable_count or 0,
                        "rating": rating.rating if rating else None,
                        "top_domains": [(d, c) for d, c in domains if d],
                    }
                )

            return research_list

    def get_download_manager_summary_stats(self) -> Dict:
        """Get aggregate download stats across ALL research sessions.

        This is a lightweight query that only returns totals — used for
        the download-manager header so stats remain accurate regardless
        of which page the user is viewing.
        """
        with get_user_db_session(self.username) as session:
            row = (
                session.query(
                    func.count(func.distinct(ResearchHistory.id)).label(
                        "total_researches"
                    ),
                    func.count(func.distinct(ResearchResource.id)).label(
                        "total_resources"
                    ),
                    func.count(
                        func.distinct(
                            case(
                                (Document.status == "completed", Document.id),
                                else_=None,
                            )
                        )
                    ).label("downloaded_count"),
                    func.count(
                        func.distinct(
                            case(
                                (
                                    ResearchResource.url.like("%.pdf")
                                    | ResearchResource.url.like("%arxiv.org%")
                                    | ResearchResource.url.like(
                                        "%ncbi.nlm.nih.gov/pmc%"
                                    ),
                                    ResearchResource.id,
                                ),
                                else_=None,
                            )
                        )
                    ).label("downloadable_count"),
                )
                .select_from(ResearchHistory)
                .outerjoin(
                    ResearchResource,
                    ResearchHistory.id == ResearchResource.research_id,
                )
                .outerjoin(
                    Document,
                    (ResearchResource.id == Document.resource_id)
                    | (ResearchResource.document_id == Document.id),
                )
                .one()
            )

            total_researches = row.total_researches or 0
            total_resources = row.total_resources or 0
            downloaded = row.downloaded_count or 0
            downloadable = row.downloadable_count or 0

            return {
                "total_researches": total_researches,
                "total_resources": total_resources,
                "already_downloaded": downloaded,
                "available_to_download": max(downloadable - downloaded, 0),
            }

    def get_pdf_previews_batch(
        self, research_ids: List, limit_per_research: int = 10
    ) -> Dict[str, Dict]:
        """Batch-fetch PDF documents and domain breakdowns for multiple research sessions.

        Returns a dict keyed by research_id with:
            - "pdf_sources": list of document dicts (capped at limit_per_research)
            - "domains": dict of domain -> {total, pdfs, downloaded}
        """
        if not research_ids:
            return {}

        with get_user_db_session(self.username) as session:
            results = (
                session.query(Document, ResearchResource)
                .outerjoin(
                    ResearchResource,
                    (Document.resource_id == ResearchResource.id)
                    | (ResearchResource.document_id == Document.id),
                )
                .filter(
                    Document.research_id.in_(research_ids),
                    Document.file_type == "pdf",
                )
                .order_by(Document.processed_at.desc())
                .limit(limit_per_research * len(research_ids))
                .all()
            )

            previews: Dict[str, Dict] = {}
            seen_doc_ids: set = set()
            for doc, resource in results:
                if doc.id in seen_doc_ids:
                    continue
                seen_doc_ids.add(doc.id)

                rid = doc.research_id
                if rid not in previews:
                    previews[rid] = {"pdf_sources": [], "domains": {}}

                entry = previews[rid]

                # Domain breakdown (within the SQL LIMIT budget)
                domain = "unknown"
                if resource and resource.url:
                    try:
                        domain = urlparse(resource.url).netloc or "unknown"
                    except Exception:
                        logger.debug("Failed to parse resource URL for domain")
                elif doc.original_url:
                    try:
                        domain = urlparse(doc.original_url).netloc or "unknown"
                    except Exception:
                        logger.debug("Failed to parse document URL for domain")

                if domain not in entry["domains"]:
                    entry["domains"][domain] = {
                        "total": 0,
                        "pdfs": 0,
                        "downloaded": 0,
                    }
                entry["domains"][domain]["total"] += 1
                if doc.file_type == "pdf":
                    entry["domains"][domain]["pdfs"] += 1
                if doc.status == "completed":
                    entry["domains"][domain]["downloaded"] += 1

                # PDF sources preview (capped)
                if len(entry["pdf_sources"]) < limit_per_research:
                    title = "Untitled"
                    if resource and resource.title:
                        title = resource.title
                    elif doc.filename:
                        title = doc.filename

                    entry["pdf_sources"].append(
                        {
                            "document_title": title,
                            "domain": domain,
                            "file_type": doc.file_type,
                            "download_status": doc.status or "unknown",
                        }
                    )

            return previews

    def get_document_by_id(self, doc_id: str) -> Optional[Dict]:
        """
        Get a specific document by its ID.

        Returns document information with file path.
        """
        with get_user_db_session(self.username) as session:
            # Find document - use outer joins to support both research downloads and user uploads
            result = (
                session.query(Document, ResearchResource, ResearchHistory)
                .outerjoin(
                    ResearchResource,
                    (Document.resource_id == ResearchResource.id)
                    | (ResearchResource.document_id == Document.id),
                )
                .outerjoin(
                    ResearchHistory,
                    Document.research_id == ResearchHistory.id,
                )
                .filter(Document.id == doc_id)
                .first()
            )

            if result:
                # Found document
                doc, resource, research = result

                # Get RAG indexing status across all collections
                doc_collections = (
                    session.query(DocumentCollection, Collection)
                    .join(Collection)
                    .filter(DocumentCollection.document_id == doc_id)
                    .all()
                )

                # Check if indexed in any collection
                has_rag_indexed = any(
                    dc.indexed for dc, coll in doc_collections
                )
                total_chunks = sum(
                    dc.chunk_count for dc, coll in doc_collections if dc.indexed
                )

                # Build collections list
                collections_list = [
                    {
                        "id": coll.id,
                        "name": coll.name,
                        "indexed": dc.indexed,
                        "chunk_count": dc.chunk_count,
                    }
                    for dc, coll in doc_collections
                ]

                # Calculate word count from text content
                word_count = (
                    len(doc.text_content.split()) if doc.text_content else 0
                )

                # Check if PDF is available (database OR filesystem)
                has_pdf = bool(
                    doc.file_path and doc.file_path not in FILE_PATH_SENTINELS
                )
                if not has_pdf and doc.storage_mode == "database":
                    has_pdf = self._has_blob_in_db(session, doc.id)

                return {
                    "id": doc.id,
                    "resource_id": doc.resource_id,
                    "research_id": doc.research_id,
                    "document_title": doc.title
                    or (resource.title if resource else doc.filename),
                    "original_url": doc.original_url
                    or (resource.url if resource else None),
                    "file_path": doc.file_path,
                    "file_absolute_path": self._get_safe_absolute_path(
                        doc.file_path
                    ),
                    "file_name": Path(doc.file_path).name
                    if doc.file_path
                    and doc.file_path not in FILE_PATH_SENTINELS
                    else doc.filename,
                    "file_size": doc.file_size,
                    "file_type": doc.file_type,
                    "mime_type": doc.mime_type,
                    "domain": self._extract_domain(resource.url)
                    if resource
                    else "User Upload",
                    "download_status": doc.status,
                    "downloaded_at": doc.processed_at.isoformat()
                    if doc.processed_at
                    and hasattr(doc.processed_at, "isoformat")
                    else str(doc.processed_at)
                    if doc.processed_at
                    else (
                        doc.uploaded_at.isoformat()
                        if hasattr(doc, "uploaded_at") and doc.uploaded_at
                        else None
                    ),
                    "favorite": doc.favorite
                    if hasattr(doc, "favorite")
                    else False,
                    "tags": doc.tags if hasattr(doc, "tags") else [],
                    "research_title": research.query[:100]
                    if research
                    else "User Upload",
                    "research_created_at": research.created_at
                    if research and isinstance(research.created_at, str)
                    else research.created_at.isoformat()
                    if research and research.created_at
                    else None,
                    # Document fields
                    "is_pdf": doc.file_type == "pdf",
                    "has_pdf": has_pdf,
                    "has_text_db": bool(doc.text_content),
                    "has_rag_indexed": has_rag_indexed,
                    "rag_chunk_count": total_chunks,
                    "word_count": word_count,
                    "collections": collections_list,
                }

            # Not found
            return None

    def toggle_favorite(self, document_id: str) -> bool:
        """Toggle favorite status of a document."""
        with get_user_db_session(self.username) as session:
            doc = session.query(Document).get(document_id)
            if doc:
                doc.favorite = not doc.favorite
                session.commit()
                return doc.favorite
            return False

    def delete_document(self, document_id: str) -> bool:
        """Delete a document from library (file and database entry)."""
        with get_user_db_session(self.username) as session:
            doc = session.query(Document).get(document_id)
            if not doc:
                return False

            # Get file path from tracker (only if document has original_url)
            tracker = None
            if doc.original_url:
                tracker = (
                    session.query(DownloadTracker)
                    .filter_by(url_hash=self._get_url_hash(doc.original_url))
                    .first()
                )

            # Delete physical file
            if tracker and tracker.file_path:
                try:
                    file_path = get_absolute_path_from_settings(
                        tracker.file_path
                    )
                    if file_path and file_path.is_file():
                        file_path.unlink()
                        logger.info(f"Deleted file: {file_path}")
                except Exception:
                    logger.exception("Failed to delete file")

            # Update tracker
            if tracker:
                tracker.is_downloaded = False
                tracker.file_path = None

            # Delete document and all related records
            from ..deletion.utils.cascade_helper import CascadeHelper

            CascadeHelper.delete_document_completely(session, document_id)
            session.commit()

            return True

    def open_file_location(self, document_id: str) -> bool:
        """Open the folder containing the document."""
        with get_user_db_session(self.username) as session:
            doc = session.query(Document).get(document_id)
            if not doc:
                return False

            tracker = None
            if doc.original_url:
                tracker = (
                    session.query(DownloadTracker)
                    .filter_by(url_hash=self._get_url_hash(doc.original_url))
                    .first()
                )

            if tracker and tracker.file_path:
                # Validate path is within library root to prevent traversal attacks
                library_root = get_absolute_path_from_settings("")
                if not library_root:
                    logger.warning("Could not determine library root")
                    return False
                try:
                    validated_path = PathValidator.validate_safe_path(
                        tracker.file_path, library_root, allow_absolute=False
                    )
                    if validated_path and validated_path.is_file():
                        return open_file_location(str(validated_path))
                except ValueError:
                    logger.warning("Path validation failed")
                    return False

            return False

    def get_unique_domains(self) -> List[str]:
        """Get sorted list of unique netlocs from all document URLs.

        Streams the URL column in batches (``yield_per``) and accumulates
        netlocs into a set, so a very large library is never fully
        materialized in memory at once (#4560). The query already projects to
        the single ``original_url`` column.
        """
        with get_user_db_session(self.username) as session:
            netlocs = set()
            rows = (
                session.query(Document.original_url)
                .filter(Document.original_url.isnot(None))
                .yield_per(_DOMAIN_SCAN_BATCH_SIZE)
            )
            for (url,) in rows:
                domain = self._extract_domain(url)
                if domain:
                    netlocs.add(domain)
            return sorted(netlocs)

    def _extract_domain(self, url: str) -> str:
        """Extract domain from URL."""
        from urllib.parse import urlparse

        try:
            return urlparse(url).netloc
        except (ValueError, AttributeError):
            return ""

    def _get_url_hash(self, url: str) -> str:
        """Generate hash for URL."""
        import re

        # Normalize URL
        url = re.sub(r"^https?://", "", url)
        url = re.sub(r"^www\.", "", url)
        url = url.rstrip("/")

        return get_url_hash(url)

    def _get_storage_path(self) -> str:
        """Get library storage path from settings (respects LDR_DATA_DIR)."""
        from ...utilities.db_utils import get_settings_manager

        settings = get_settings_manager()
        return str(
            Path(
                settings.get_setting(
                    "research_library.storage_path",
                    str(get_library_directory()),
                )
            )
            .expanduser()
            .resolve()
        )

    def sync_library_with_filesystem(self) -> Dict:
        """
        Sync library database with filesystem.
        Check which PDF files exist and update database accordingly.

        Returns:
            Statistics about the sync operation
        """
        with get_user_db_session(self.username) as session:
            # Sync only research downloads — uploads have no original_url
            # and no DownloadTracker, so they don't belong in this routine
            # (and the destructive `else` branch below would otherwise
            # silently delete every uploaded document).
            # Don't load the large text_content body: this sync loop only
            # reads original_url/id/title (and may delete the row), so loading
            # every completed document's full text would needlessly exhaust
            # memory on a large library (#4560).
            documents = (
                session.query(Document)
                .filter_by(status=DocumentStatus.COMPLETED)
                .filter(Document.original_url.isnot(None))
                .options(defer(Document.text_content))
                .all()
            )

            stats = {
                "total_documents": len(documents),
                "files_found": 0,
                "files_missing": 0,
                "trackers_updated": 0,
                "missing_files": [],
            }

            # Sync documents with filesystem
            for doc in documents:
                # Get download tracker
                tracker = (
                    session.query(DownloadTracker)
                    .filter_by(url_hash=self._get_url_hash(doc.original_url))
                    .first()
                )

                if tracker and tracker.file_path:
                    # Check if file exists
                    file_path = get_absolute_path_from_settings(
                        tracker.file_path
                    )
                    if file_path and file_path.is_file():
                        stats["files_found"] += 1
                    else:
                        # File missing or path invalid - mark for re-download
                        stats["files_missing"] += 1
                        stats["missing_files"].append(
                            {
                                "id": doc.id,
                                "title": doc.title,
                                "path": str(file_path)
                                if file_path
                                else "invalid",
                                "url": doc.original_url,
                            }
                        )

                        # Reset tracker
                        tracker.is_downloaded = False
                        tracker.file_path = None

                        # Delete the document entry so it can be re-queued
                        from ..deletion.utils.cascade_helper import (
                            CascadeHelper,
                        )

                        CascadeHelper.delete_document_completely(
                            session, doc.id
                        )
                        stats["trackers_updated"] += 1
                else:
                    # No tracker or path - delete the document entry
                    stats["files_missing"] += 1
                    from ..deletion.utils.cascade_helper import CascadeHelper

                    CascadeHelper.delete_document_completely(session, doc.id)

            session.commit()
            logger.info(
                f"Library sync completed: {stats['files_found']} found, {stats['files_missing']} missing"
            )

            return stats

    def mark_for_redownload(self, document_ids: List[str]) -> int:
        """
        Mark specific documents for re-download.

        Args:
            document_ids: List of document IDs to mark for re-download

        Returns:
            Number of documents marked
        """
        with get_user_db_session(self.username) as session:
            count = 0
            for doc_id in document_ids:
                doc = session.query(Document).get(doc_id)
                if doc:
                    if not doc.original_url:
                        # Uploads have no source URL and no DownloadTracker,
                        # so re-download is not meaningful for them.
                        logger.warning(
                            f"Skipping mark-for-redownload on {doc_id[:8]}…: "
                            "document has no original_url (likely user upload)"
                        )
                        continue
                    # Get tracker and reset it
                    tracker = (
                        session.query(DownloadTracker)
                        .filter_by(
                            url_hash=self._get_url_hash(doc.original_url)
                        )
                        .first()
                    )

                    if tracker:
                        tracker.is_downloaded = False
                        tracker.file_path = None

                    # Mark document as pending
                    doc.status = DocumentStatus.PENDING
                    count += 1

            session.commit()
            logger.info(f"Marked {count} documents for re-download")
            return count
