"""
Content Type Handlers for Unified Embeddings

Pluggable system for different content sources (store agents, blocks, docs).
Each handler knows how to fetch and process its content type for embedding.
"""

from __future__ import annotations

import asyncio
import functools
import itertools
import logging
from abc import ABC, abstractmethod
from dataclasses import dataclass
from pathlib import Path
from typing import TYPE_CHECKING, Any, get_args, get_origin

from prisma.enums import ContentType

from backend.blocks import get_blocks
from backend.blocks.llm import LlmModel
from backend.data.db import query_raw_with_schema
from backend.util.text import split_camelcase

if TYPE_CHECKING:
    from backend.blocks._base import AnyBlockSchema

logger = logging.getLogger(__name__)


def _contains_type(annotation: Any, target: type) -> bool:
    """Check if an annotation is or contains the target type (handles Optional/Union/Annotated)."""
    if annotation is target:
        return True
    origin = get_origin(annotation)
    if origin is None:
        return False
    return any(_contains_type(arg, target) for arg in get_args(annotation))


@dataclass
class ContentItem:
    """Represents a piece of content to be embedded."""

    content_id: str  # Unique identifier (DB ID or file path)
    content_type: ContentType
    searchable_text: str  # Combined text for embedding
    metadata: dict[str, Any]  # Content-specific metadata
    user_id: str | None = None  # For user-scoped content


class ContentHandler(ABC):
    """Base handler for fetching and processing content for embeddings."""

    @property
    @abstractmethod
    def content_type(self) -> ContentType:
        """The ContentType this handler manages."""
        pass

    @abstractmethod
    async def get_missing_items(self, batch_size: int) -> list[ContentItem]:
        """
        Fetch items that don't have embeddings yet.

        Args:
            batch_size: Maximum number of items to return

        Returns:
            List of ContentItem objects ready for embedding
        """
        pass

    @abstractmethod
    async def get_stats(self) -> dict[str, int]:
        """
        Get statistics about embedding coverage.

        Returns:
            Dict with keys: total, with_embeddings, without_embeddings
        """
        pass

    @abstractmethod
    async def get_valid_content_ids(self) -> set[str]:
        """Return the set of ``contentId`` values that should currently
        have an embedding for this content type.

        Used by :func:`cleanup_orphaned_embeddings` to compute the
        difference against ``UnifiedContentEmbedding`` and delete rows
        whose source content has been removed / hidden / untitled. The
        definition of "valid" must match the corresponding handler's
        :meth:`get_missing_items` filter so the embedder isn't fighting
        the orphan-cleaner.
        """
        pass


class StoreAgentHandler(ContentHandler):
    """Handler for marketplace store agent listings."""

    @property
    def content_type(self) -> ContentType:
        return ContentType.STORE_AGENT

    async def get_missing_items(self, batch_size: int) -> list[ContentItem]:
        """Fetch approved store listings without embeddings."""
        from backend.api.features.search.embeddings import build_searchable_text

        missing = await query_raw_with_schema(
            """
            SELECT
                slv.id,
                slv.name,
                slv.description,
                slv."subHeading",
                slv.categories,
                sl.slug,
                p.username AS creator
            FROM {schema_prefix}"StoreListingVersion" slv
            JOIN {schema_prefix}"StoreListing" sl
                ON slv."storeListingId" = sl.id
            JOIN {schema_prefix}"Profile" p
                ON p."userId" = sl."owningUserId"
            LEFT JOIN {schema_prefix}"UnifiedContentEmbedding" uce
                ON slv.id = uce."contentId" AND uce."contentType" = 'STORE_AGENT'::{schema_prefix}"ContentType"
            WHERE slv."submissionStatus" = 'APPROVED'
            AND slv."isDeleted" = false
            AND sl."isDeleted" = false
            AND uce."contentId" IS NULL
            LIMIT $1
            """,
            batch_size,
        )

        return [
            ContentItem(
                content_id=row["id"],
                content_type=ContentType.STORE_AGENT,
                searchable_text=build_searchable_text(
                    name=row["name"],
                    description=row["description"],
                    sub_heading=row["subHeading"],
                    categories=row["categories"] or [],
                ),
                # ``creator`` (Profile.username) + ``slug`` (StoreListing.slug)
                # build the marketplace URL — the FE search-result click
                # handler in ChatSidebar.tsx routes to
                # ``/marketplace/agent/{creator}/{slug}`` and silently no-ops
                # without them.
                metadata={
                    "name": row["name"],
                    "categories": row["categories"] or [],
                    "creator": row["creator"],
                    "slug": row["slug"],
                },
                user_id=None,  # Store agents are public
            )
            for row in missing
        ]

    async def get_stats(self) -> dict[str, int]:
        """Get statistics about store agent embedding coverage."""
        # Count approved versions
        approved_result = await query_raw_with_schema(
            """
            SELECT COUNT(*) as count
            FROM {schema_prefix}"StoreListingVersion"
            WHERE "submissionStatus" = 'APPROVED'
            AND "isDeleted" = false
            """
        )
        total_approved = approved_result[0]["count"] if approved_result else 0

        # Count versions with embeddings
        embedded_result = await query_raw_with_schema(
            """
            SELECT COUNT(*) as count
            FROM {schema_prefix}"StoreListingVersion" slv
            JOIN {schema_prefix}"UnifiedContentEmbedding" uce ON slv.id = uce."contentId" AND uce."contentType" = 'STORE_AGENT'::{schema_prefix}"ContentType"
            WHERE slv."submissionStatus" = 'APPROVED'
            AND slv."isDeleted" = false
            """
        )
        with_embeddings = embedded_result[0]["count"] if embedded_result else 0

        return {
            "total": total_approved,
            "with_embeddings": with_embeddings,
            "without_embeddings": total_approved - with_embeddings,
        }

    async def get_valid_content_ids(self) -> set[str]:
        # Approved StoreListingVersion rows from non-deleted listings
        # mirror ``get_missing_items``: rejected / soft-deleted listings
        # should have their embeddings cleaned up.
        rows = await query_raw_with_schema(
            """
            SELECT slv.id
            FROM {schema_prefix}"StoreListingVersion" slv
            JOIN {schema_prefix}"StoreListing" sl ON slv."storeListingId" = sl.id
            WHERE slv."submissionStatus" = 'APPROVED'
              AND slv."isDeleted" = false
              AND sl."isDeleted" = false
            """,
        )
        return {row["id"] for row in rows}


@functools.lru_cache(maxsize=1)
def _get_enabled_blocks() -> dict[str, AnyBlockSchema]:
    """Return ``{block_id: block_instance}`` for all enabled, instantiable blocks.

    Disabled blocks and blocks that fail to instantiate are silently skipped
    (with a warning log), so callers never need their own try/except loop.

    Results are cached for the process lifetime via ``lru_cache`` because
    blocks are registered at import time and never change while running.
    """
    enabled: dict[str, AnyBlockSchema] = {}
    for block_id, block_cls in get_blocks().items():
        try:
            instance = block_cls()
        except Exception as e:
            logger.warning(f"Skipping block {block_id}: init failed: {e}")
            continue
        if not instance.disabled:
            enabled[block_id] = instance
    return enabled


class BlockHandler(ContentHandler):
    """Handler for block definitions (Python classes)."""

    @property
    def content_type(self) -> ContentType:
        return ContentType.BLOCK

    async def get_missing_items(self, batch_size: int) -> list[ContentItem]:
        """Fetch blocks without embeddings."""
        # to_thread keeps the first (heavy) call off the event loop.  On
        # subsequent calls the lru_cache makes this a dict lookup, so the
        # thread-pool overhead is negligible compared to the DB queries below.
        enabled = await asyncio.to_thread(_get_enabled_blocks)
        if not enabled:
            return []

        block_ids = list(enabled.keys())

        # Query for existing embeddings
        placeholders = ",".join([f"${i+1}" for i in range(len(block_ids))])
        existing_result = await query_raw_with_schema(
            f"""
            SELECT "contentId"
            FROM {{schema_prefix}}"UnifiedContentEmbedding"
            WHERE "contentType" = 'BLOCK'::{{schema_prefix}}"ContentType"
            AND "contentId" = ANY(ARRAY[{placeholders}])
            """,
            *block_ids,
        )

        existing_ids = {row["contentId"] for row in existing_result}

        # Convert to ContentItem — disabled filtering already done by
        # _get_enabled_blocks so batch_size won't be exhausted by disabled blocks.
        missing = ((bid, b) for bid, b in enabled.items() if bid not in existing_ids)
        items = []
        for block_id, block in itertools.islice(missing, batch_size):
            try:
                # Build searchable text from block metadata
                if not block.name:
                    logger.warning(
                        f"Block {block_id} has no name — using block_id as fallback"
                    )
                display_name = split_camelcase(block.name) if block.name else ""
                parts = []
                if display_name:
                    parts.append(display_name)
                if block.description:
                    parts.append(block.description)
                if block.categories:
                    parts.append(" ".join(str(cat.value) for cat in block.categories))

                # Add input schema field descriptions
                parts += [
                    f"{field_name}: {field_info.description}"
                    for field_name, field_info in block.input_schema.model_fields.items()
                    if field_info.description
                ]

                searchable_text = " ".join(parts)

                categories_list = (
                    [cat.value for cat in block.categories] if block.categories else []
                )

                # Extract provider names from credentials fields
                credentials_info = block.input_schema.get_credentials_fields_info()
                is_integration = len(credentials_info) > 0
                provider_names = [
                    provider.value.lower()
                    for info in credentials_info.values()
                    for provider in info.provider
                ]

                # Check if block has LlmModel field in input schema
                has_llm_model_field = any(
                    _contains_type(field.annotation, LlmModel)
                    for field in block.input_schema.model_fields.values()
                )

                items.append(
                    ContentItem(
                        content_id=block_id,
                        content_type=ContentType.BLOCK,
                        searchable_text=searchable_text,
                        metadata={
                            "name": display_name or block.name or block_id,
                            "categories": categories_list,
                            "providers": provider_names,
                            "has_llm_model_field": has_llm_model_field,
                            "is_integration": is_integration,
                        },
                        user_id=None,
                    )
                )
            except Exception as e:
                logger.warning(f"Failed to process block {block_id}: {e}")
                continue

        return items

    async def get_stats(self) -> dict[str, int]:
        """Get statistics about block embedding coverage."""
        enabled = await asyncio.to_thread(_get_enabled_blocks)
        total_blocks = len(enabled)

        if total_blocks == 0:
            return {"total": 0, "with_embeddings": 0, "without_embeddings": 0}

        block_ids = list(enabled.keys())
        placeholders = ",".join([f"${i+1}" for i in range(len(block_ids))])

        embedded_result = await query_raw_with_schema(
            f"""
            SELECT COUNT(*) as count
            FROM {{schema_prefix}}"UnifiedContentEmbedding"
            WHERE "contentType" = 'BLOCK'::{{schema_prefix}}"ContentType"
            AND "contentId" = ANY(ARRAY[{placeholders}])
            """,
            *block_ids,
        )

        with_embeddings = embedded_result[0]["count"] if embedded_result else 0

        return {
            "total": total_blocks,
            "with_embeddings": with_embeddings,
            "without_embeddings": total_blocks - with_embeddings,
        }

    async def get_valid_content_ids(self) -> set[str]:
        # The enabled-block registry is the source of truth — a block
        # that's been removed from the registry or disabled is no
        # longer a valid embedding target.
        enabled = await asyncio.to_thread(_get_enabled_blocks)
        return set(enabled.keys())


@dataclass
class MarkdownSection:
    """Represents a section of a markdown document."""

    title: str  # Section heading text
    content: str  # Section content (including the heading line)
    level: int  # Heading level (1 for #, 2 for ##, etc.)
    index: int  # Section index within the document


class DocumentationHandler(ContentHandler):
    """Handler for documentation files (.md/.mdx).

    Chunks documents by markdown headings to create multiple embeddings per file.
    Each section (## heading) becomes a separate embedding for better retrieval.
    """

    @property
    def content_type(self) -> ContentType:
        return ContentType.DOCUMENTATION

    def _get_docs_root(self) -> Path:
        """Get the documentation root directory."""
        # content_handlers.py is at: backend/backend/api/features/store/content_handlers.py
        # Need to go up to project root then into docs/
        # In container: /app/autogpt_platform/backend/backend/api/features/store -> /app/docs
        # In development: /repo/autogpt_platform/backend/backend/api/features/store -> /repo/docs
        this_file = Path(
            __file__
        )  # .../backend/backend/api/features/store/content_handlers.py
        project_root = (
            this_file.parent.parent.parent.parent.parent.parent.parent
        )  # -> /app or /repo
        docs_root = project_root / "docs"
        return docs_root

    def _extract_doc_title(self, file_path: Path) -> str:
        """Extract the document title from a markdown file."""
        try:
            content = file_path.read_text(encoding="utf-8")
            lines = content.split("\n")

            # Try to extract title from first # heading
            for line in lines:
                if line.startswith("# "):
                    return line[2:].strip()

            # If no title found, use filename
            return file_path.stem.replace("-", " ").replace("_", " ").title()
        except Exception as e:
            logger.warning(f"Failed to read title from {file_path}: {e}")
            return file_path.stem.replace("-", " ").replace("_", " ").title()

    def _chunk_markdown_by_headings(
        self, file_path: Path, min_heading_level: int = 2
    ) -> list[MarkdownSection]:
        """
        Split a markdown file into sections based on headings.

        Args:
            file_path: Path to the markdown file
            min_heading_level: Minimum heading level to split on (default: 2 for ##)

        Returns:
            List of MarkdownSection objects, one per section.
            If no headings found, returns a single section with all content.
        """
        try:
            content = file_path.read_text(encoding="utf-8")
        except Exception as e:
            logger.warning(f"Failed to read {file_path}: {e}")
            return []

        lines = content.split("\n")
        sections: list[MarkdownSection] = []
        current_section_lines: list[str] = []
        current_title = ""
        current_level = 0
        section_index = 0
        doc_title = ""

        for line in lines:
            # Check if line is a heading
            if line.startswith("#"):
                # Count heading level
                level = 0
                for char in line:
                    if char == "#":
                        level += 1
                    else:
                        break

                heading_text = line[level:].strip()

                # Track document title (level 1 heading)
                if level == 1 and not doc_title:
                    doc_title = heading_text
                    # Don't create a section for just the title - add it to first section
                    current_section_lines.append(line)
                    continue

                # Check if this heading should start a new section
                if level >= min_heading_level:
                    # Save previous section if it has content
                    if current_section_lines:
                        section_content = "\n".join(current_section_lines).strip()
                        if section_content:
                            # Use doc title for first section if no specific title
                            title = current_title if current_title else doc_title
                            if not title:
                                title = file_path.stem.replace("-", " ").replace(
                                    "_", " "
                                )
                            sections.append(
                                MarkdownSection(
                                    title=title,
                                    content=section_content,
                                    level=current_level if current_level else 1,
                                    index=section_index,
                                )
                            )
                            section_index += 1

                    # Start new section
                    current_section_lines = [line]
                    current_title = heading_text
                    current_level = level
                else:
                    # Lower level heading (e.g., # when splitting on ##)
                    current_section_lines.append(line)
            else:
                current_section_lines.append(line)

        # Don't forget the last section
        if current_section_lines:
            section_content = "\n".join(current_section_lines).strip()
            if section_content:
                title = current_title if current_title else doc_title
                if not title:
                    title = file_path.stem.replace("-", " ").replace("_", " ")
                sections.append(
                    MarkdownSection(
                        title=title,
                        content=section_content,
                        level=current_level if current_level else 1,
                        index=section_index,
                    )
                )

        # If no sections were created (no headings found), create one section with all content
        if not sections and content.strip():
            title = (
                doc_title
                if doc_title
                else file_path.stem.replace("-", " ").replace("_", " ")
            )
            sections.append(
                MarkdownSection(
                    title=title,
                    content=content.strip(),
                    level=1,
                    index=0,
                )
            )

        return sections

    def _make_section_content_id(self, doc_path: str, section_index: int) -> str:
        """Create a unique content ID for a document section.

        Format: doc_path::section_index
        Example: 'platform/getting-started.md::0'
        """
        return f"{doc_path}::{section_index}"

    def _parse_section_content_id(self, content_id: str) -> tuple[str, int]:
        """Parse a section content ID back into doc_path and section_index.

        Returns: (doc_path, section_index)
        """
        if "::" in content_id:
            parts = content_id.rsplit("::", 1)
            return parts[0], int(parts[1])
        # Legacy format (whole document)
        return content_id, 0

    async def get_missing_items(self, batch_size: int) -> list[ContentItem]:
        """Fetch documentation sections without embeddings.

        Chunks each document by markdown headings and creates embeddings for each section.
        Content IDs use the format: 'path/to/doc.md::section_index'
        """
        docs_root = self._get_docs_root()

        if not docs_root.exists():
            logger.warning(f"Documentation root not found: {docs_root}")
            return []

        # Find all .md and .mdx files
        all_docs = list(docs_root.rglob("*.md")) + list(docs_root.rglob("*.mdx"))

        if not all_docs:
            return []

        # Build list of all sections from all documents
        all_sections: list[tuple[str, Path, MarkdownSection]] = []
        for doc_file in all_docs:
            doc_path = str(doc_file.relative_to(docs_root))
            sections = self._chunk_markdown_by_headings(doc_file)
            for section in sections:
                all_sections.append((doc_path, doc_file, section))

        if not all_sections:
            return []

        # Generate content IDs for all sections
        section_content_ids = [
            self._make_section_content_id(doc_path, section.index)
            for doc_path, _, section in all_sections
        ]

        # Check which ones have embeddings
        placeholders = ",".join([f"${i+1}" for i in range(len(section_content_ids))])
        existing_result = await query_raw_with_schema(
            f"""
            SELECT "contentId"
            FROM {{schema_prefix}}"UnifiedContentEmbedding"
            WHERE "contentType" = 'DOCUMENTATION'::{{schema_prefix}}"ContentType"
            AND "contentId" = ANY(ARRAY[{placeholders}])
            """,
            *section_content_ids,
        )

        existing_ids = {row["contentId"] for row in existing_result}

        # Filter to missing sections
        missing_sections = [
            (doc_path, doc_file, section, content_id)
            for (doc_path, doc_file, section), content_id in zip(
                all_sections, section_content_ids
            )
            if content_id not in existing_ids
        ]

        # Convert to ContentItem (up to batch_size)
        items = []
        for doc_path, doc_file, section, content_id in missing_sections[:batch_size]:
            try:
                # Get document title for context
                doc_title = self._extract_doc_title(doc_file)

                # Build searchable text with context
                # Include doc title and section title for better search relevance
                searchable_text = f"{doc_title} - {section.title}\n\n{section.content}"

                items.append(
                    ContentItem(
                        content_id=content_id,
                        content_type=ContentType.DOCUMENTATION,
                        searchable_text=searchable_text,
                        metadata={
                            "doc_title": doc_title,
                            "section_title": section.title,
                            "section_index": section.index,
                            "heading_level": section.level,
                            "path": doc_path,
                        },
                        user_id=None,  # Documentation is public
                    )
                )
            except Exception as e:
                logger.warning(f"Failed to process section {content_id}: {e}")
                continue

        return items

    def _get_all_section_content_ids(self, docs_root: Path) -> set[str]:
        """Get all current section content IDs from the docs directory.

        Used for stats and cleanup to know what sections should exist.
        """
        all_docs = list(docs_root.rglob("*.md")) + list(docs_root.rglob("*.mdx"))
        content_ids = set()

        for doc_file in all_docs:
            doc_path = str(doc_file.relative_to(docs_root))
            sections = self._chunk_markdown_by_headings(doc_file)
            for section in sections:
                content_ids.add(self._make_section_content_id(doc_path, section.index))

        return content_ids

    async def get_stats(self) -> dict[str, int]:
        """Get statistics about documentation embedding coverage.

        Counts sections (not documents) since each section gets its own embedding.
        """
        docs_root = self._get_docs_root()

        if not docs_root.exists():
            return {"total": 0, "with_embeddings": 0, "without_embeddings": 0}

        # Get all section content IDs
        all_section_ids = self._get_all_section_content_ids(docs_root)
        total_sections = len(all_section_ids)

        if total_sections == 0:
            return {"total": 0, "with_embeddings": 0, "without_embeddings": 0}

        # Count embeddings in database for DOCUMENTATION type
        embedded_result = await query_raw_with_schema(
            """
            SELECT COUNT(*) as count
            FROM {schema_prefix}"UnifiedContentEmbedding"
            WHERE "contentType" = 'DOCUMENTATION'::{schema_prefix}"ContentType"
            """
        )

        with_embeddings = embedded_result[0]["count"] if embedded_result else 0

        return {
            "total": total_sections,
            "with_embeddings": with_embeddings,
            "without_embeddings": total_sections - with_embeddings,
        }

    async def get_valid_content_ids(self) -> set[str]:
        # Section IDs are derived from on-disk markdown — a deleted file
        # or removed section drops from this set and its embedding gets
        # cleaned up.
        docs_root = self._get_docs_root()
        if not docs_root.exists():
            return set()
        # Filesystem walk is sync; keep it off the event loop.
        return await asyncio.to_thread(self._get_all_section_content_ids, docs_root)


class LibraryAgentHandler(ContentHandler):
    """Handler for user-scoped library agents.

    Library agents are private (one row per user) so embeddings carry a
    non-null userId. The unique key on (contentType, contentId, userId)
    keeps the same library_agent_id from colliding across users that may
    have forked the same graph.
    """

    @property
    def content_type(self) -> ContentType:
        return ContentType.LIBRARY_AGENT

    async def get_missing_items(self, batch_size: int) -> list[ContentItem]:
        missing = await query_raw_with_schema(
            """
            SELECT
                la.id,
                la."userId",
                g.name,
                g.description,
                g.instructions
            FROM {schema_prefix}"LibraryAgent" la
            JOIN {schema_prefix}"AgentGraph" g
              ON g.id = la."agentGraphId" AND g.version = la."agentGraphVersion"
            LEFT JOIN {schema_prefix}"UnifiedContentEmbedding" uce
              ON uce."contentId" = la.id
              AND uce."contentType" = 'LIBRARY_AGENT'::{schema_prefix}"ContentType"
              AND uce."userId" = la."userId"
            WHERE la."isDeleted" = false
              AND la."isHidden" = false
              AND uce."contentId" IS NULL
            LIMIT $1
            """,
            batch_size,
        )

        items: list[ContentItem] = []
        for row in missing:
            name = row["name"] or ""
            description = row["description"] or ""
            instructions = row["instructions"] or ""
            searchable_text = " ".join(
                part for part in (name, description, instructions) if part
            )
            items.append(
                ContentItem(
                    content_id=row["id"],
                    content_type=ContentType.LIBRARY_AGENT,
                    searchable_text=searchable_text,
                    metadata={"name": name},
                    user_id=row["userId"],
                )
            )
        return items

    async def get_stats(self) -> dict[str, int]:
        total_result = await query_raw_with_schema(
            """
            SELECT COUNT(*) as count
            FROM {schema_prefix}"LibraryAgent"
            WHERE "isDeleted" = false AND "isHidden" = false
            """
        )
        total = total_result[0]["count"] if total_result else 0

        with_result = await query_raw_with_schema(
            """
            SELECT COUNT(*) as count
            FROM {schema_prefix}"LibraryAgent" la
            JOIN {schema_prefix}"UnifiedContentEmbedding" uce
              ON uce."contentId" = la.id
              AND uce."contentType" = 'LIBRARY_AGENT'::{schema_prefix}"ContentType"
              AND uce."userId" = la."userId"
            WHERE la."isDeleted" = false AND la."isHidden" = false
            """
        )
        with_embeddings = with_result[0]["count"] if with_result else 0

        return {
            "total": total,
            "with_embeddings": with_embeddings,
            "without_embeddings": total - with_embeddings,
        }

    async def get_valid_content_ids(self) -> set[str]:
        # Mirrors ``get_missing_items``: only non-deleted, non-hidden
        # LibraryAgent rows are considered valid embedding targets.
        rows = await query_raw_with_schema(
            """
            SELECT id
            FROM {schema_prefix}"LibraryAgent"
            WHERE "isDeleted" = false AND "isHidden" = false
            """,
        )
        return {row["id"] for row in rows}


def build_workspace_file_text(name: str, path: str) -> str:
    """Build the embedding-searchable text for a workspace file.

    Shared between :class:`WorkspaceFileHandler` (batch backfill) and
    the per-write indexer in ``workspace/embeddings.py`` so both encode
    the file under the same ``searchableText`` representation. If they
    drift, the "skip if unchanged" guard in the indexer keeps re-
    embedding the same row forever.
    """
    # Strip the leading slash and extension from the path stem so the
    # token "report" from "/documents/report.pdf" reinforces a rename
    # to "Final.pdf" without polluting the text with directory noise.
    from pathlib import PurePosixPath

    stem = PurePosixPath(path).stem if path else ""
    parts = [name or "", stem]
    # Dedup while preserving order: a freshly-created file usually has
    # ``name == stem`` and we don't want the same token weighted twice.
    seen: set[str] = set()
    out: list[str] = []
    for p in parts:
        p = p.strip()
        if p and p.lower() not in seen:
            seen.add(p.lower())
            out.append(p)
    return " ".join(out)


class WorkspaceFileHandler(ContentHandler):
    """Handler for user-scoped workspace files.

    Files are private per workspace owner. ``searchable_text`` is the
    user-visible ``name`` (plus the path stem when it adds extra signal —
    e.g. files renamed to a generic name still match against their path)
    so search results surface meaningful matches without us having to
    embed file contents.
    """

    @property
    def content_type(self) -> ContentType:
        return ContentType.WORKSPACE_FILE

    async def get_missing_items(self, batch_size: int) -> list[ContentItem]:
        missing = await query_raw_with_schema(
            """
            SELECT
                uwf.id,
                uwf.name,
                uwf.path,
                uwf."mimeType",
                uw."userId"
            FROM {schema_prefix}"UserWorkspaceFile" uwf
            JOIN {schema_prefix}"UserWorkspace" uw ON uw.id = uwf."workspaceId"
            LEFT JOIN {schema_prefix}"UnifiedContentEmbedding" uce
              ON uce."contentId" = uwf.id
              AND uce."contentType" = 'WORKSPACE_FILE'::{schema_prefix}"ContentType"
              AND uce."userId" = uw."userId"
            WHERE uwf."isDeleted" = false
              AND uce."contentId" IS NULL
            LIMIT $1
            """,
            batch_size,
        )

        items: list[ContentItem] = []
        for row in missing:
            text = build_workspace_file_text(row["name"] or "", row["path"] or "")
            if not text:
                continue
            items.append(
                ContentItem(
                    content_id=row["id"],
                    content_type=ContentType.WORKSPACE_FILE,
                    searchable_text=text,
                    metadata={
                        "name": row["name"] or "",
                        "path": row["path"] or "",
                        "mime_type": row["mimeType"] or "",
                    },
                    user_id=row["userId"],
                )
            )
        return items

    async def get_stats(self) -> dict[str, int]:
        total_result = await query_raw_with_schema(
            """
            SELECT COUNT(*) as count
            FROM {schema_prefix}"UserWorkspaceFile"
            WHERE "isDeleted" = false
            """
        )
        total = total_result[0]["count"] if total_result else 0

        with_result = await query_raw_with_schema(
            """
            SELECT COUNT(*) as count
            FROM {schema_prefix}"UserWorkspaceFile" uwf
            JOIN {schema_prefix}"UserWorkspace" uw ON uw.id = uwf."workspaceId"
            JOIN {schema_prefix}"UnifiedContentEmbedding" uce
              ON uce."contentId" = uwf.id
              AND uce."contentType" = 'WORKSPACE_FILE'::{schema_prefix}"ContentType"
              AND uce."userId" = uw."userId"
            WHERE uwf."isDeleted" = false
            """
        )
        with_embeddings = with_result[0]["count"] if with_result else 0

        return {
            "total": total,
            "with_embeddings": with_embeddings,
            "without_embeddings": total - with_embeddings,
        }

    async def get_valid_content_ids(self) -> set[str]:
        # Mirrors ``get_missing_items``: a soft-deleted file's embedding
        # is orphaned and should be cleaned up.
        rows = await query_raw_with_schema(
            """
            SELECT id
            FROM {schema_prefix}"UserWorkspaceFile"
            WHERE "isDeleted" = false
            """,
        )
        return {row["id"] for row in rows}


class ChatSessionHandler(ContentHandler):
    """Handler for user-scoped chat sessions.

    Only sessions with a non-empty ``title`` are embedded — untitled
    sessions are noise for search and CoPilot auto-fills the title within
    the first turn anyway.
    """

    @property
    def content_type(self) -> ContentType:
        return ContentType.CHAT_SESSION

    async def get_missing_items(self, batch_size: int) -> list[ContentItem]:
        missing = await query_raw_with_schema(
            """
            SELECT
                cs.id,
                cs."userId",
                cs.title
            FROM {schema_prefix}"ChatSession" cs
            LEFT JOIN {schema_prefix}"UnifiedContentEmbedding" uce
              ON uce."contentId" = cs.id
              AND uce."contentType" = 'CHAT_SESSION'::{schema_prefix}"ContentType"
              AND uce."userId" = cs."userId"
            WHERE cs.title IS NOT NULL
              AND btrim(cs.title) <> ''
              AND uce."contentId" IS NULL
            LIMIT $1
            """,
            batch_size,
        )

        items: list[ContentItem] = []
        for row in missing:
            title = (row["title"] or "").strip()
            if not title:
                continue
            items.append(
                ContentItem(
                    content_id=row["id"],
                    content_type=ContentType.CHAT_SESSION,
                    searchable_text=title,
                    metadata={"title": title},
                    user_id=row["userId"],
                )
            )
        return items

    async def get_stats(self) -> dict[str, int]:
        total_result = await query_raw_with_schema(
            """
            SELECT COUNT(*) as count
            FROM {schema_prefix}"ChatSession"
            WHERE title IS NOT NULL AND btrim(title) <> ''
            """
        )
        total = total_result[0]["count"] if total_result else 0

        with_result = await query_raw_with_schema(
            """
            SELECT COUNT(*) as count
            FROM {schema_prefix}"ChatSession" cs
            JOIN {schema_prefix}"UnifiedContentEmbedding" uce
              ON uce."contentId" = cs.id
              AND uce."contentType" = 'CHAT_SESSION'::{schema_prefix}"ContentType"
              AND uce."userId" = cs."userId"
            WHERE cs.title IS NOT NULL AND btrim(cs.title) <> ''
            """
        )
        with_embeddings = with_result[0]["count"] if with_result else 0

        return {
            "total": total,
            "with_embeddings": with_embeddings,
            "without_embeddings": total - with_embeddings,
        }

    async def get_valid_content_ids(self) -> set[str]:
        # Mirrors ``get_missing_items``: sessions without a non-empty
        # title aren't embedded; a session that was deleted or had its
        # title cleared should have its embedding cleaned up.
        rows = await query_raw_with_schema(
            """
            SELECT id
            FROM {schema_prefix}"ChatSession"
            WHERE title IS NOT NULL AND btrim(title) <> ''
            """,
        )
        return {row["id"] for row in rows}


# Content handler registry
CONTENT_HANDLERS: dict[ContentType, ContentHandler] = {
    ContentType.STORE_AGENT: StoreAgentHandler(),
    ContentType.BLOCK: BlockHandler(),
    ContentType.DOCUMENTATION: DocumentationHandler(),
    ContentType.LIBRARY_AGENT: LibraryAgentHandler(),
    ContentType.WORKSPACE_FILE: WorkspaceFileHandler(),
    ContentType.CHAT_SESSION: ChatSessionHandler(),
}
