mcp-nextcloud/nextcloud_mcp_server/vector/processor.py

"""Processor task for vector database synchronization.

Processes documents from stream: fetches content, generates embeddings, stores in Qdrant.
"""

import logging
import time
import uuid
from typing import Any, cast

import anyio
from anyio.abc import TaskStatus
from anyio.streams.memory import MemoryObjectReceiveStream
from httpx import HTTPStatusError
from qdrant_client.models import FieldCondition, Filter, MatchValue, PointStruct

from nextcloud_mcp_server.client import NextcloudClient
from nextcloud_mcp_server.config import get_settings
from nextcloud_mcp_server.document_processors import get_registry
from nextcloud_mcp_server.embedding import get_bm25_service, get_embedding_service
from nextcloud_mcp_server.observability.metrics import (
    record_qdrant_operation,
    record_vector_sync_processing,
    update_vector_sync_queue_size,
)
from nextcloud_mcp_server.observability.tracing import trace_operation
from nextcloud_mcp_server.search.pdf_highlighter import PDFHighlighter
from nextcloud_mcp_server.vector.document_chunker import DocumentChunker
from nextcloud_mcp_server.vector.html_processor import html_to_markdown
from nextcloud_mcp_server.vector.placeholder import delete_placeholder_point
from nextcloud_mcp_server.vector.qdrant_client import get_qdrant_client
from nextcloud_mcp_server.vector.scanner import DocumentTask

logger = logging.getLogger(__name__)


def assign_page_numbers(chunks, page_boundaries):
    """Assign page numbers to chunks based on page boundaries.

    Each chunk gets the page number where most of its content appears.
    For chunks spanning multiple pages, assigns the page containing the
    majority of the chunk's characters.

    Args:
        chunks: List of ChunkWithPosition objects
        page_boundaries: List of dicts with {page, start_offset, end_offset}

    Returns:
        None (modifies chunks in place)
    """
    if not page_boundaries:
        return

    for chunk in chunks:
        # Find which page(s) this chunk overlaps with
        max_overlap = 0
        assigned_page = None

        for boundary in page_boundaries:
            # Calculate overlap between chunk and page
            overlap_start = max(chunk.start_offset, boundary["start_offset"])
            overlap_end = min(chunk.end_offset, boundary["end_offset"])
            overlap = max(0, overlap_end - overlap_start)

            # Assign to page with maximum overlap
            if overlap > max_overlap:
                max_overlap = overlap
                assigned_page = boundary["page"]

        if assigned_page is not None:
            chunk.page_number = assigned_page


async def processor_task(
    worker_id: int,
    receive_stream: MemoryObjectReceiveStream[DocumentTask],
    shutdown_event: anyio.Event,
    nc_client: NextcloudClient,
    user_id: str,
    *,
    task_status: TaskStatus = anyio.TASK_STATUS_IGNORED,
):
    """
    Process documents from stream concurrently.

    Each processor task runs in a loop:
    1. Receive document from stream (with timeout)
    2. Fetch content from Nextcloud
    3. Tokenize and chunk text
    4. Generate embeddings (I/O bound - external API)
    5. Upload vectors to Qdrant

    Multiple processors run concurrently for I/O parallelism.

    Args:
        worker_id: Worker identifier for logging
        receive_stream: Stream to receive documents from
        shutdown_event: Event signaling shutdown
        nc_client: Authenticated Nextcloud client
        user_id: User being processed
        task_status: Status object for signaling task readiness
    """
    logger.info("Processor %s started", worker_id)

    # Signal that the task has started and is ready
    task_status.started()

    while not shutdown_event.is_set():
        try:
            # Get document with timeout (allows checking shutdown)
            with anyio.fail_after(1.0):
                doc_task = await receive_stream.receive()

            # Update queue size metric after receiving
            stream_stats = receive_stream.statistics()
            update_vector_sync_queue_size(stream_stats.current_buffer_used)

            # Process document
            await process_document(doc_task, nc_client)

            # Update queue size metric after processing
            stream_stats = receive_stream.statistics()
            update_vector_sync_queue_size(stream_stats.current_buffer_used)

        except TimeoutError:
            # No documents available, update metric to show empty queue
            stream_stats = receive_stream.statistics()
            update_vector_sync_queue_size(stream_stats.current_buffer_used)
            continue

        except anyio.EndOfStream:
            # Scanner finished and closed stream, exit gracefully
            logger.info("Processor %s: Scanner finished, exiting", worker_id)
            break

        except Exception as e:
            logger.error(
                "Processor %s error processing %s_%s: %s",
                worker_id,
                doc_task.doc_type,
                doc_task.doc_id,
                e,
                exc_info=True,
            )
            # Continue to next document (no task_done() needed with streams)

    logger.info("Processor %s stopped", worker_id)


async def process_document(doc_task: DocumentTask, nc_client: NextcloudClient):
    """
    Process a single document: fetch, tokenize, embed, store in Qdrant.

    Implements retry logic with exponential backoff for transient failures.

    Args:
        doc_task: Document task to process
        nc_client: Authenticated Nextcloud client
    """
    start_time = time.time()

    logger.debug(
        "Processing %s_%s for %s (%s)",
        doc_task.doc_type,
        doc_task.doc_id,
        doc_task.user_id,
        doc_task.operation,
    )

    with trace_operation(
        "vector_sync.process_document",
        attributes={
            "vector_sync.operation": "process",
            "vector_sync.user_id": doc_task.user_id,
            "vector_sync.doc_id": doc_task.doc_id,
            "vector_sync.doc_type": doc_task.doc_type,
            "vector_sync.doc_operation": doc_task.operation,
        },
    ):
        try:
            qdrant_client = await get_qdrant_client()
            settings = get_settings()

            # Handle deletion
            if doc_task.operation == "delete":
                await qdrant_client.delete(
                    collection_name=settings.get_collection_name(),
                    points_selector=Filter(
                        must=[
                            FieldCondition(
                                key="user_id",
                                match=MatchValue(value=doc_task.user_id),
                            ),
                            FieldCondition(
                                key="doc_id",
                                match=MatchValue(value=doc_task.doc_id),
                            ),
                            FieldCondition(
                                key="doc_type",
                                match=MatchValue(value=doc_task.doc_type),
                            ),
                        ]
                    ),
                )
                logger.info(
                    "Deleted %s_%s for %s",
                    doc_task.doc_type,
                    doc_task.doc_id,
                    doc_task.user_id,
                )

                # Record successful deletion metrics
                duration = time.time() - start_time
                record_qdrant_operation("delete", "success")
                record_vector_sync_processing(duration, "success")
                return

            # Handle indexing with retry
            max_retries = 3
            retry_delay = 1.0

            for attempt in range(max_retries):
                try:
                    await _index_document(doc_task, nc_client, qdrant_client)

                    # Record successful processing metrics
                    duration = time.time() - start_time
                    record_qdrant_operation("upsert", "success")
                    record_vector_sync_processing(duration, "success")
                    return  # Success

                except (HTTPStatusError, Exception) as e:
                    if attempt < max_retries - 1:
                        logger.warning(
                            "Retry %s/%s for %s_%s: %s",
                            attempt + 1,
                            max_retries,
                            doc_task.doc_type,
                            doc_task.doc_id,
                            e,
                        )
                        await anyio.sleep(retry_delay)
                        retry_delay *= 2  # Exponential backoff
                    else:
                        logger.error(
                            "Failed to index %s_%s after %s retries: %s",
                            doc_task.doc_type,
                            doc_task.doc_id,
                            max_retries,
                            e,
                        )
                        # Record failed processing metrics
                        duration = time.time() - start_time
                        record_qdrant_operation("upsert", "error")
                        record_vector_sync_processing(duration, "error")
                        raise

        except Exception:
            # Catch any other unexpected errors
            duration = time.time() - start_time
            record_vector_sync_processing(duration, "error")
            raise


async def _index_document(
    doc_task: DocumentTask, nc_client: NextcloudClient, qdrant_client
):
    """
    Index a single document (called by process_document with retry).

    Args:
        doc_task: Document task to index
        nc_client: Authenticated Nextcloud client
        qdrant_client: Qdrant client instance
    """
    settings = get_settings()

    # Fetch document content
    with trace_operation(
        "vector_sync.fetch_content",
        attributes={
            "vector_sync.doc_type": doc_task.doc_type,
            "vector_sync.doc_id": doc_task.doc_id,
        },
    ):
        if doc_task.doc_type == "note":
            document = await nc_client.notes.get_note(int(doc_task.doc_id))
            content = f"{document['title']}\n\n{document['content']}"
            title = document["title"]
            etag = document.get("etag", "")
            file_metadata = {}  # No file-specific metadata for notes
            file_path = None  # Notes don't have file paths
            content_bytes = None  # Notes don't have binary content
            content_type = None
        elif doc_task.doc_type == "news_item":
            item = await nc_client.news.get_item(int(doc_task.doc_id))
            # Convert HTML body to Markdown for better embedding
            body_markdown = html_to_markdown(item.get("body", ""))
            # Build content: title + URL + body
            item_title = item.get("title", "")
            item_url = item.get("url", "")
            feed_title = item.get("feedTitle", "")

            # Structure content for embedding
            content_parts = [item_title]
            if feed_title:
                content_parts.append(f"Source: {feed_title}")
            if item_url:
                content_parts.append(f"URL: {item_url}")
            content_parts.append("")  # Blank line
            content_parts.append(body_markdown)
            content = "\n".join(content_parts)

            title = item_title
            etag = item.get("guidHash", "")
            # Store news-specific metadata for later use in payload
            file_metadata = {
                "feed_id": item.get("feedId"),
                "feed_title": feed_title,
                "author": item.get("author"),
                "pub_date": item.get("pubDate"),
                "starred": item.get("starred", False),
                "unread": item.get("unread", True),
                "url": item_url,
                "guid_hash": item.get("guidHash"),
                "enclosure_link": item.get("enclosureLink"),
                "enclosure_mime": item.get("enclosureMime"),
            }
            file_path = None
            content_bytes = None
            content_type = None
        elif doc_task.doc_type == "deck_card":
            # Fetch card from Deck API
            # Use metadata from scanner if available (O(1) lookup)
            # Otherwise fall back to iteration (legacy data)
            card = None
            board = None
            stack = None

            if (
                doc_task.metadata
                and "board_id" in doc_task.metadata
                and "stack_id" in doc_task.metadata
            ):
                # Fast path: Direct lookup with known board_id/stack_id
                board_id = doc_task.metadata["board_id"]
                stack_id = doc_task.metadata["stack_id"]
                try:
                    card = await nc_client.deck.get_card(
                        board_id=int(board_id),
                        stack_id=int(stack_id),
                        card_id=int(doc_task.doc_id),
                    )
                    # Fetch board and stack info for metadata
                    boards = await nc_client.deck.get_boards()
                    for b in boards:
                        if b.id == int(board_id):
                            board = b
                            stacks = await nc_client.deck.get_stacks(b.id)
                            for s in stacks:
                                if s.id == int(stack_id):
                                    stack = s
                                    break
                            break
                except Exception as e:
                    logger.warning(
                        "Failed to fetch card with metadata (board_id=%s, stack_id=%s, card_id=%s): %s, falling back to iteration",
                        board_id,
                        stack_id,
                        doc_task.doc_id,
                        e,
                    )

            # Fallback: Iterate through all boards/stacks (for legacy data or if fast path failed)
            if card is None:
                boards = await nc_client.deck.get_boards()
                card_found = False

                for b in boards:
                    if card_found:
                        break
                    # Skip deleted boards (soft delete: deletedAt > 0)
                    if b.deletedAt > 0:
                        continue
                    stacks = await nc_client.deck.get_stacks(b.id)
                    for s in stacks:
                        if card_found:
                            break
                        if s.cards:
                            for c in s.cards:
                                if c.id == int(doc_task.doc_id):
                                    card = c
                                    board = b
                                    stack = s
                                    card_found = True
                                    break

                if not card_found:
                    raise ValueError(
                        f"Deck card {doc_task.doc_id} not found in any board/stack"
                    )

            # Type narrowing: card, board, stack are all set if we reach here
            assert card is not None
            assert board is not None
            assert stack is not None

            # Build content from card title and description
            content_parts = [card.title]
            if card.description:
                content_parts.append(card.description)
            content = "\n\n".join(content_parts)
            title = card.title

            # Store deck-specific metadata
            file_metadata = {
                "board_id": board.id,
                "board_title": board.title,
                "stack_id": stack.id,
                "stack_title": stack.title,
                "card_type": card.type,
                "duedate": (card.duedate.isoformat() if card.duedate else None),
                "archived": card.archived,
                "owner": (
                    card.owner.uid if hasattr(card.owner, "uid") else str(card.owner)
                ),
            }
            etag = card.etag or ""
            file_path = None
            content_bytes = None
            content_type = None
        elif doc_task.doc_type == "file":
            # For files, doc_id is now the numeric file ID, file_path comes from DocumentTask
            if not doc_task.file_path:
                raise ValueError(
                    f"File path required for file indexing but not provided (file_id={doc_task.doc_id})"
                )
            file_path = doc_task.file_path

            # Read file content via WebDAV
            content_bytes, content_type = await nc_client.webdav.read_file(file_path)
        else:
            raise ValueError(f"Unsupported doc_type: {doc_task.doc_type}")

    # Process file content (text extraction)
    if doc_task.doc_type == "file":
        # Type narrowing: content_bytes and content_type are set for files
        assert content_bytes is not None
        assert content_type is not None
        assert file_path is not None

        with trace_operation(
            "vector_sync.document_process",
            attributes={
                "vector_sync.content_type": content_type,
                "vector_sync.file_size": len(content_bytes),
            },
        ):
            # Use document processor registry to extract text
            registry = get_registry()

            try:
                result = await registry.process(
                    content=content_bytes,
                    content_type=content_type,
                    filename=file_path,
                )
                content = result.text
                file_metadata = result.metadata
                title = file_metadata.get("title") or file_path.split("/")[-1]
                etag = ""  # WebDAV read_file doesn't return etag

                # Diagnostic: Log page boundary information if available
                if "page_boundaries" in file_metadata:
                    page_boundaries = file_metadata["page_boundaries"]
                    logger.info(
                        "Page boundaries for %s: %s pages, text length: %s",
                        file_path,
                        len(page_boundaries),
                        len(content),
                    )
                    # Log first 3 page boundaries for debugging
                    for boundary in page_boundaries[:3]:
                        logger.debug(
                            "  Page %s: offsets [%s:%s]",
                            boundary["page"],
                            boundary["start_offset"],
                            boundary["end_offset"],
                        )
                    # Verify last boundary matches text length
                    if page_boundaries:
                        last_boundary = page_boundaries[-1]
                        if last_boundary["end_offset"] != len(content):
                            logger.warning(
                                "Text length mismatch: content=%s, last_boundary_end=%s",
                                len(content),
                                last_boundary["end_offset"],
                            )
                else:
                    logger.debug("No page_boundaries in metadata for %s", file_path)
            except Exception as e:
                logger.error("Failed to process file %s: %s", file_path, e)
                raise

    # Tokenize and chunk (using configured chunk size and overlap)
    with trace_operation(
        "vector_sync.chunk_text",
        attributes={
            "vector_sync.input_chars": len(content),
            "vector_sync.chunk_size": settings.document_chunk_size,
            "vector_sync.overlap": settings.document_chunk_overlap,
        },
    ):
        chunker = DocumentChunker(
            chunk_size=settings.document_chunk_size,
            overlap=settings.document_chunk_overlap,
        )
        chunks = await chunker.chunk_text(content)

    # Assign page numbers to chunks if page boundaries are available (PDFs)
    page_boundaries = file_metadata.get("page_boundaries")
    if doc_task.doc_type == "file" and page_boundaries is not None:
        # Type narrowing: page_boundaries is guaranteed to be list[dict] here
        page_boundaries_list = cast(list[dict[str, Any]], page_boundaries)
        with trace_operation(
            "vector_sync.assign_page_numbers",
            attributes={
                "vector_sync.chunk_count": len(chunks),
                "vector_sync.page_count": len(page_boundaries_list),
            },
        ):
            assign_page_numbers(chunks, page_boundaries_list)

            # Diagnostic: Verify page number assignment
            assigned_count = sum(1 for c in chunks if c.page_number is not None)
            logger.info(
                "Assigned page numbers to %s/%s chunks for %s",
                assigned_count,
                len(chunks),
                file_path,
            )

            # Log first 3 chunks to see their page assignments
            for i, chunk in enumerate(chunks[:3]):
                logger.debug(
                    "  Chunk %s: page=%s, offsets=[%s:%s]",
                    i,
                    chunk.page_number,
                    chunk.start_offset,
                    chunk.end_offset,
                )

            # Warning if NO page numbers were assigned
            if assigned_count == 0:
                logger.warning(
                    "NO page numbers assigned! Text length: %s, Chunks: %s, Chunk offset range: [%s:%s], Page boundaries: %s pages, First boundary: %s",
                    len(content),
                    len(chunks),
                    chunks[0].start_offset,
                    chunks[-1].end_offset,
                    len(page_boundaries_list),
                    page_boundaries_list[0] if page_boundaries_list else "None",
                )

    # Extract chunk texts for embedding
    chunk_texts = [chunk.text for chunk in chunks]

    # Initialize results containers
    dense_embeddings: list = []
    sparse_embeddings: list = []
    # chunk_index -> list[(x0, y0, x1, y1)] of normalized rectangles
    # in [0, 1] relative to page width/height. The page is taken from
    # `chunk.page_number` (offset-based) and stored as `page_number`
    # in the Qdrant payload, so we don't carry an `actual_page_num` here.
    chunk_bboxes: dict[int, list[tuple[float, float, float, float]]] = {}

    # Determine if we need PDF highlighting
    is_pdf = doc_task.doc_type == "file" and content_type == "application/pdf"

    # Define async tasks for parallel execution
    async def generate_dense_embeddings():
        """Generate dense embeddings (I/O bound - external API call)."""
        nonlocal dense_embeddings
        with trace_operation(
            "vector_sync.embed_dense",
            attributes={
                "vector_sync.chunk_count": len(chunk_texts),
                "vector_sync.total_chars": sum(len(t) for t in chunk_texts),
            },
        ):
            embedding_service = get_embedding_service()
            dense_embeddings = await embedding_service.embed_batch(chunk_texts)

    async def generate_sparse_embeddings():
        """Generate sparse embeddings (BM25 for keyword matching)."""
        nonlocal sparse_embeddings
        with trace_operation(
            "vector_sync.embed_sparse",
            attributes={
                "vector_sync.chunk_count": len(chunk_texts),
            },
        ):
            bm25_service = await get_bm25_service()
            sparse_embeddings = await bm25_service.encode_batch(chunk_texts)

    async def generate_highlights():
        """Compute chunk bounding boxes for PDF chunks (CPU-bound, no rendering)."""
        nonlocal chunk_bboxes
        if not is_pdf:
            return

        # Type narrowing: content_bytes is set for PDF files
        assert content_bytes is not None

        with trace_operation(
            "vector_sync.compute_chunk_bboxes",
            attributes={
                "vector_sync.chunk_count": len(chunks),
                "vector_sync.pdf_size": len(content_bytes),
            },
        ):
            chunk_data: list[tuple[int, int, int, int | None, str]] = [
                (i, chunk.start_offset, chunk.end_offset, chunk.page_number, chunk.text)
                for i, chunk in enumerate(chunks)
                if chunk.page_number is not None
            ]

            page_boundaries = file_metadata.get("page_boundaries")
            if not page_boundaries:
                logger.warning(
                    "No page boundaries available, skipping bbox computation"
                )
                return

            page_boundaries_list = cast(list[dict[str, Any]], page_boundaries)

            logger.info("Computing chunk bboxes for %s PDF chunks", len(chunk_data))

            batch_results = await anyio.to_thread.run_sync(  # type: ignore[attr-defined]
                lambda: PDFHighlighter.compute_chunk_bboxes_batch(
                    pdf_bytes=content_bytes,
                    chunks=chunk_data,
                    page_boundaries=page_boundaries_list,
                    full_text=content,
                )
            )

            for chunk_index, (bboxes, _) in batch_results.items():
                chunk_bboxes[chunk_index] = bboxes

            logger.info(
                "Computed bboxes for %s/%s chunks", len(chunk_bboxes), len(chunks)
            )

    # Run all embedding/highlighting operations in parallel
    # - Dense embeddings: I/O bound (API call)
    # - Sparse embeddings: CPU bound (local BM25)
    # - Highlighting: CPU bound (PyMuPDF rendering, runs in thread pool)
    with trace_operation(
        "vector_sync.parallel_processing",
        attributes={
            "vector_sync.is_pdf": is_pdf,
            "vector_sync.chunk_count": len(chunks),
        },
    ):
        async with anyio.create_task_group() as tg:
            tg.start_soon(generate_dense_embeddings)
            tg.start_soon(generate_sparse_embeddings)
            tg.start_soon(generate_highlights)

    # Prepare Qdrant points
    indexed_at = int(time.time())
    points = []

    # Surface deck card data quality issues at indexing time rather than
    # only at verification time (where _verify_deck_cards falls through to
    # legacy-data pass-through when board_id/stack_id are missing). This is
    # logged once per document — not per chunk — to avoid log spam.
    if doc_task.doc_type == "deck_card":
        missing_deck_fields = [
            field for field in ("board_id", "stack_id") if not file_metadata.get(field)
        ]
        if missing_deck_fields:
            logger.warning(
                "Indexing deck_card %s for user %s with missing metadata: %s; "
                "verification will fall back to legacy-data pass-through",
                doc_task.doc_id,
                doc_task.user_id,
                missing_deck_fields,
            )

    for i, (chunk, dense_emb, sparse_emb) in enumerate(
        zip(chunks, dense_embeddings, sparse_embeddings)
    ):
        # Generate deterministic UUID for point ID
        # Using uuid5 with DNS namespace and combining doc info
        point_name = f"{doc_task.doc_type}:{doc_task.doc_id}:chunk:{i}"
        point_id = str(uuid.uuid5(uuid.NAMESPACE_DNS, point_name))

        points.append(
            PointStruct(
                id=point_id,
                vector={
                    "dense": dense_emb,
                    "sparse": sparse_emb,
                },
                payload={
                    "user_id": doc_task.user_id,
                    "doc_id": doc_task.doc_id,
                    "doc_type": doc_task.doc_type,
                    "is_placeholder": False,  # Real indexed document (not placeholder)
                    "title": title,
                    "excerpt": chunk.text,  # Full chunk text (up to chunk_size, default 2048 chars)
                    "indexed_at": indexed_at,
                    "modified_at": doc_task.modified_at,
                    "etag": etag,
                    "chunk_index": i,
                    "total_chunks": len(chunks),
                    "chunk_start_offset": chunk.start_offset,
                    "chunk_end_offset": chunk.end_offset,
                    "metadata_version": 2,  # v2 includes position metadata
                    # File-specific metadata (PDF, etc.)
                    **(
                        {
                            "file_path": file_path,  # Store file path for retrieval
                            "mime_type": content_type,  # From WebDAV response
                            "file_size": file_metadata.get("file_size"),
                            "page_number": chunk.page_number,
                            "page_count": file_metadata.get("page_count"),
                            "author": file_metadata.get("author"),
                            "creation_date": file_metadata.get("creation_date"),
                            "has_images": file_metadata.get("has_images", False),
                            "image_count": file_metadata.get("image_count", 0),
                        }
                        if doc_task.doc_type == "file"
                        else {}
                    ),
                    # News item-specific metadata
                    **(
                        {
                            "feed_id": file_metadata.get("feed_id"),
                            "feed_title": file_metadata.get("feed_title"),
                            "author": file_metadata.get("author"),
                            "pub_date": file_metadata.get("pub_date"),
                            "starred": file_metadata.get("starred"),
                            "unread": file_metadata.get("unread"),
                            "url": file_metadata.get("url"),
                            "guid_hash": file_metadata.get("guid_hash"),
                            "enclosure_link": file_metadata.get("enclosure_link"),
                            "enclosure_mime": file_metadata.get("enclosure_mime"),
                        }
                        if doc_task.doc_type == "news_item"
                        else {}
                    ),
                    # Deck card-specific metadata
                    **(
                        {
                            "board_id": file_metadata.get("board_id"),
                            "board_title": file_metadata.get("board_title"),
                            "stack_id": file_metadata.get("stack_id"),
                            "stack_title": file_metadata.get("stack_title"),
                            "card_type": file_metadata.get("card_type"),
                            "duedate": file_metadata.get("duedate"),
                            "owner": file_metadata.get("owner"),
                        }
                        if doc_task.doc_type == "deck_card"
                        else {}
                    ),
                    # Chunk bbox (PDF only) — normalized rectangles in [0,1]
                    # relative to page width/height. Replaces the legacy
                    # `highlighted_page_image` (Deck #76). The page number
                    # comes from `page_number` (set above for PDF chunks).
                    **({"chunk_bbox": chunk_bboxes[i]} if i in chunk_bboxes else {}),
                },
            )
        )

    # Delete placeholder before writing real vectors
    # This prevents duplicates and cleans up the placeholder state
    try:
        await delete_placeholder_point(
            doc_id=doc_task.doc_id,
            doc_type=doc_task.doc_type,
            user_id=doc_task.user_id,
        )
    except Exception as e:
        # Log but don't fail indexing if placeholder deletion fails
        logger.warning(
            "Failed to delete placeholder for %s_%s: %s",
            doc_task.doc_type,
            doc_task.doc_id,
            e,
        )

    # Upsert to Qdrant in batches. Now that we no longer embed PNG payloads,
    # per-point payloads are small (chunk text + small metadata), so we can
    # safely use a larger batch size.
    BATCH_SIZE = 100
    with trace_operation(
        "vector_sync.qdrant_upsert",
        attributes={
            "vector_sync.point_count": len(points),
            "vector_sync.collection": settings.get_collection_name(),
            "vector_sync.bboxes_count": len(chunk_bboxes),
            "vector_sync.batch_size": BATCH_SIZE,
        },
    ):
        for batch_start in range(0, len(points), BATCH_SIZE):
            batch_end = min(batch_start + BATCH_SIZE, len(points))
            batch = points[batch_start:batch_end]
            await qdrant_client.upsert(
                collection_name=settings.get_collection_name(),
                points=batch,
                wait=True,
            )
            if batch_end < len(points):
                logger.debug(
                    "Upserted batch %s/%s",
                    batch_start // BATCH_SIZE + 1,
                    (len(points) + BATCH_SIZE - 1) // BATCH_SIZE,
                )

    logger.info(
        "Indexed %s_%s for %s (%s chunks)",
        doc_task.doc_type,
        doc_task.doc_id,
        doc_task.user_id,
        len(chunks),
    )