class SimpleVecDBVectorStore(VectorStore):
    """LangChain-compatible wrapper for SimpleVecDB."""

    def __init__(
        self,
        db_path: str = ":memory:",
        embedding: Embeddings | None = None,
        collection_name: str = "default",
        **kwargs: Any,
    ):
        self.embedding = embedding  # LangChain expects this
        self._db = VectorDB(path=db_path, **kwargs)
        self._collection = self._db.collection(collection_name)

    @property
    def embeddings(self) -> Embeddings | None:
        """LangChain ``VectorStore.embeddings`` contract.

        The base class exposes the embedding model via this property and
        framework code (``as_retriever()``, tag generation, etc.) reads it.
        Storing the field as ``self.embedding`` (singular) without overriding
        this property would silently return ``None`` to those callers.
        """
        return self.embedding

    @classmethod
    def from_texts(
        cls,
        texts: list[str],
        embedding: Embeddings,
        metadatas: list[dict] | None = None,
        db_path: str = ":memory:",
        collection_name: str = "default",
        **kwargs: Any,
    ) -> "SimpleVecDBVectorStore":
        """
        Initialize from texts (embeds them automatically).

        Args:
            texts: List of texts to add.
            embedding: LangChain Embeddings model.
            metadatas: Optional list of metadata dicts.
            db_path: Path to SQLite database.
            collection_name: Name of the collection to use.
            **kwargs: Additional arguments for VectorDB.

        Returns:
            Initialized SimpleVecDBVectorStore.
        """
        store = cls(
            embedding=embedding,
            db_path=db_path,
            collection_name=collection_name,
            **kwargs,
        )
        store.add_texts(texts, metadatas)
        return store

    def add_texts(
        self,
        texts: Iterable[str],
        metadatas: list[dict] | None = None,
        **kwargs: Any,
    ) -> list[str]:
        """
        Add texts (embed if no pre-computed). Returns IDs as str.

        Args:
            texts: Iterable of texts to add.
            metadatas: Optional list of metadata dicts.
            **kwargs: Additional arguments (e.g., ids).

        Returns:
            List of document IDs.
        """
        texts_list = list(texts)
        embeddings = None
        if self.embedding:
            embeddings = self.embedding.embed_documents(texts_list)
        ids = self._collection.add_texts(
            texts=texts_list,
            metadatas=metadatas,
            embeddings=embeddings,
            ids=kwargs.get("ids"),
        )
        return [str(id_) for id_ in ids]

    def similarity_search(
        self,
        query: str,
        k: int = 4,
        **kwargs: Any,
    ) -> list[LangChainDocument]:
        """
        Search by text query (auto-embeds).

        Args:
            query: Text query string.
            k: Number of results to return.
            **kwargs: Additional arguments (e.g., filter).

        Returns:
            List of LangChain Documents.
        """
        if self.embedding:
            query_vec = self.embedding.embed_query(query)
        else:
            raise ValueError("Embedding model required for text queries")
        results = self._collection.similarity_search(
            query=query_vec,
            k=k,
            filter=kwargs.get("filter"),
        )
        return [
            LangChainDocument(page_content=doc.page_content, metadata=doc.metadata)
            for doc, _ in results
        ]

    def similarity_search_with_score(
        self,
        query: str,
        k: int = 4,
        **kwargs: Any,
    ) -> list[tuple[LangChainDocument, float]]:
        """
        Return with scores (distances).

        Args:
            query: Text query string.
            k: Number of results to return.
            **kwargs: Additional arguments (e.g., filter).

        Returns:
            List of (Document, score) tuples.
        """
        if self.embedding:
            query_vec = self.embedding.embed_query(query)
        else:
            raise ValueError("Embedding model required")
        results = self._collection.similarity_search(
            query=query_vec,
            k=k,
            filter=kwargs.get("filter"),
        )
        return [
            (
                LangChainDocument(page_content=doc.page_content, metadata=doc.metadata),
                score,
            )
            for doc, score in results
        ]

    def delete(self, ids: list[str] | None = None, **kwargs: Any) -> None:
        """
        Delete documents by ID.

        Args:
            ids: List of document IDs to delete.
            **kwargs: Unused.
        """
        if ids:
            int_ids = [int(id_) for id_ in ids]
            self._collection.delete_by_ids(int_ids)

    def max_marginal_relevance_search(
        self,
        query: str,
        k: int = 4,
        fetch_k: int = constants.DEFAULT_FETCH_K,
        lambda_mult: float = 0.5,
        **kwargs: Any,
    ) -> list[LangChainDocument]:
        """
        Max marginal relevance search.

        Args:
            query: Text query string.
            k: Number of results to return.
            fetch_k: Number of candidates to fetch.
            lambda_mult: Diversity trade-off (unused in core currently).
            **kwargs: Additional arguments (e.g., filter).

        Returns:
            List of LangChain Documents.
        """
        if self.embedding:
            query_vec = self.embedding.embed_query(query)
        else:
            raise ValueError("Embedding model required for text queries")
        results = self._collection.max_marginal_relevance_search(
            query=query_vec,
            k=k,
            fetch_k=fetch_k,
            filter=kwargs.get("filter"),
        )
        return [
            LangChainDocument(page_content=doc.page_content, metadata=doc.metadata)
            for doc in results
        ]

    def keyword_search(
        self,
        query: str,
        k: int = 4,
        **kwargs: Any,
    ) -> list[LangChainDocument]:
        """Return BM25-ranked documents without requiring embeddings."""

        results = self._collection.keyword_search(
            query, k=k, filter=kwargs.get("filter")
        )
        return [
            LangChainDocument(page_content=doc.page_content, metadata=doc.metadata)
            for doc, _ in results
        ]

    def hybrid_search(
        self,
        query: str,
        k: int = 4,
        **kwargs: Any,
    ) -> list[LangChainDocument]:
        """Blend BM25 + vector rankings using Reciprocal Rank Fusion."""

        query_vec = None
        if self.embedding and hasattr(self.embedding, "embed_query"):
            query_vec = self.embedding.embed_query(query)

        results = self._collection.hybrid_search(
            query,
            k=k,
            filter=kwargs.get("filter"),
            query_vector=query_vec,
            vector_k=kwargs.get("vector_k"),
            keyword_k=kwargs.get("keyword_k"),
            rrf_k=kwargs.get("rrf_k", constants.DEFAULT_RRF_K),
        )
        return [
            LangChainDocument(page_content=doc.page_content, metadata=doc.metadata)
            for doc, _ in results
        ]

    # The sync implementations below do blocking I/O and embedding inference.
    # Calling them directly from an async context blocks the event loop, so
    # offload via asyncio.to_thread. This is still not "true" async (the
    # underlying SQLite/usearch calls remain blocking), but it stops a
    # single request from starving every other coroutine in the loop.
    async def aadd_texts(self, *args, **kwargs):
        import asyncio

        return await asyncio.to_thread(self.add_texts, *args, **kwargs)

    async def asimilarity_search(self, *args, **kwargs):
        import asyncio

        return await asyncio.to_thread(self.similarity_search, *args, **kwargs)

    async def amax_marginal_relevance_search(
        self,
        *args,
        **kwargs,
    ) -> list[LangChainDocument]:
        import asyncio

        return await asyncio.to_thread(
            self.max_marginal_relevance_search, *args, **kwargs
        )

class SimpleVecDBLlamaStore(BasePydanticVectorStore):
    """LlamaIndex-compatible wrapper for SimpleVecDB."""

    stores_text: bool = True
    is_embedding_query: bool = True

    def __init__(
        self,
        db_path: str = ":memory:",
        collection_name: str = "default",
        **kwargs: Any,
    ):
        # Pass stores_text as a literal value, not self.stores_text
        super().__init__(stores_text=True)
        self._db = VectorDB(path=db_path, **kwargs)
        self._collection = self._db.collection(collection_name)
        # Map internal DB IDs to node IDs
        self._id_map: dict[int, str] = {}
        # Detect a v2.5 (or earlier) collection where the LlamaIndex node_id
        # was not persisted into metadata. delete(ref_doc_id) silently fails
        # against such rows because the metadata-fallback query finds
        # nothing. Emit a one-shot DeprecationWarning so the operator knows
        # to call migrate_node_id_metadata() before relying on delete().
        try:
            self._warn_if_legacy_collection()
        except Exception:
            _logger.debug(
                "Legacy-collection probe failed", exc_info=True
            )

    def _warn_if_legacy_collection(self) -> None:
        """Probe one row; warn if it lacks the v2.6 node_id metadata stamp."""
        sample = self._collection.get_documents(limit=1)
        if not sample:
            return
        _doc_id, _text, metadata = sample[0]
        if "_simplevecdb_node_id" in (metadata or {}):
            return
        warnings.warn(
            "SimpleVecDBLlamaStore: the underlying collection contains "
            "documents created before v2.6.0 that do not carry a "
            "'_simplevecdb_node_id' metadata stamp. delete(node_id) will "
            "silently no-op against those rows until you call "
            "store.migrate_node_id_metadata(). NOTE: legacy rows were "
            "stamped with the integer DB row id (not the original "
            "LlamaIndex node_id, which was never persisted), so "
            "delete(original_node_id) cannot be recovered for them; "
            "re-index to obtain stable node ids.",
            DeprecationWarning,
            stacklevel=3,
        )

    @property
    def client(self) -> Any:
        """Return the underlying client (our VectorDB)."""
        return self._db

    def migrate_node_id_metadata(self) -> int:
        """Backfill ``_simplevecdb_node_id`` for documents inserted before 2.6.0.

        Pre-2.6.0 versions did not persist the LlamaIndex node_id into
        document metadata, so ``delete(ref_doc_id)`` could not find the
        right row after a process restart. This helper walks every
        document in the underlying collection and stamps the internal DB
        id as the node_id for any row that lacks ``_simplevecdb_node_id``
        metadata. Idempotent — already-stamped rows are skipped, so a
        retry after a partial run converges.

        Limitation
        ----------
        Pre-2.6.0 rows never had their original LlamaIndex node_id
        persisted, so the backfill stamps ``str(doc_id)`` (the integer DB
        rowid) as the node_id. After migration:

        - ``delete(str(doc_id))`` works against migrated rows.
        - ``delete(<original-llama-uuid>)`` still cannot find these rows;
          the original ids were never written to disk and cannot be
          recovered. Re-indexing the upstream documents is the only way
          to restore stable node ids that match the LlamaIndex side.

        Atomicity: the underlying ``update_metadata`` issues one
        transaction per chunk of 500 rows. A process kill between chunks
        leaves the migration half-done; calling this method again will
        finish it (the per-row idempotency guard skips already-stamped
        rows).

        Returns:
            Number of documents updated.
        """
        docs = self._collection.get_documents()
        updates: list[tuple[int, dict[str, Any]]] = []
        for doc_id, _text, metadata in docs:
            if not metadata.get("_simplevecdb_node_id"):
                merged = dict(metadata or {})
                merged["_simplevecdb_node_id"] = str(doc_id)
                updates.append((int(doc_id), merged))
                self._id_map[int(doc_id)] = str(doc_id)
        if not updates:
            return 0
        return self._collection.update_metadata(updates)

    @property
    def store_text(self) -> bool:
        """Whether the store keeps text content."""
        return self.stores_text

    def add(self, nodes: Sequence[BaseNode], **kwargs: Any) -> list[str]:
        """
        Add nodes with embeddings.

        The node_id is persisted into the document's metadata under
        ``_simplevecdb_node_id`` so it survives process restarts. The
        in-memory ``_id_map`` is also populated as a fast cache for the
        current session.
        """
        texts = [node.get_content() for node in nodes]

        # Stamp the node_id into metadata so delete() can recover the mapping
        # after a restart. When LlamaIndex did not assign a node_id, generate
        # a UUID up front so the metadata stamp is committed in the same
        # transaction as the row insert — there is no window where the row
        # exists without ``_simplevecdb_node_id`` in its metadata.
        node_ids_resolved: list[str] = []
        metadatas: list[dict[str, Any]] = []
        for node in nodes:
            node_id = node.node_id or str(uuid.uuid4())
            md = dict(node.metadata or {})
            md["_simplevecdb_node_id"] = node_id
            node_ids_resolved.append(node_id)
            metadatas.append(md)

        embeddings = None
        if nodes and nodes[0].embedding is not None:
            emb_list = []
            all_have_embeddings = True
            for node in nodes:
                if node.embedding is None:
                    all_have_embeddings = False
                    break
                emb_list.append(node.embedding)
            if all_have_embeddings:
                embeddings = emb_list

        internal_ids = self._collection.add_texts(texts, metadatas, embeddings)

        for internal_id, node_id in zip(internal_ids, node_ids_resolved):
            self._id_map[internal_id] = node_id

        return node_ids_resolved

    def delete(self, ref_doc_id: str, **delete_kwargs: Any) -> None:
        """
        Delete by ref_doc_id (node ID).

        First consults the in-memory ``_id_map`` for the current session;
        on a miss (typically after a restart) falls back to a metadata
        query against ``_simplevecdb_node_id`` so deletion is reliable
        across process boundaries.
        """
        internal_id: int | None = None
        for int_id, node_id in self._id_map.items():
            if node_id == ref_doc_id:
                internal_id = int_id
                break

        if internal_id is None:
            # Fall back to metadata lookup — mapping was not in this
            # process's _id_map, so we have to find it on disk. We catch
            # only TypeError (older catalog signatures lack
            # ``filter_dict=``) and NotImplementedError (filter mode not
            # supported); other exceptions — including database errors,
            # locked-DB, schema mismatches — propagate so the caller
            # cannot mistake a real failure for a successful no-op.
            try:
                docs = self._collection.get_documents(
                    filter_dict={"_simplevecdb_node_id": ref_doc_id}, limit=1
                )
            except (TypeError, NotImplementedError) as exc:
                _logger.debug(
                    "filter_dict-based delete fallback unavailable: %s",
                    exc,
                )
                docs = []
            except sqlite3.DatabaseError:
                # Database-layer error: surface it instead of swallowing.
                # A locked or corrupted DB previously turned into a
                # silent no-op, hiding data-loss bugs.
                raise
            if docs:
                internal_id = int(docs[0][0])

        if internal_id is not None:
            self._collection.delete_by_ids([internal_id])
            self._id_map.pop(internal_id, None)

    def delete_nodes(
        self,
        node_ids: list[str] | None = None,
        filters: MetadataFilters | None = None,
        **delete_kwargs: Any,
    ) -> None:
        """
        Delete nodes from vector store.

        Args:
            node_ids: List of node IDs to delete.
            filters: Metadata filters. Currently unsupported — passing a
                non-None ``filters`` raises ``NotImplementedError`` rather
                than silently ignoring it (which would let callers think
                the deletion happened).
            **delete_kwargs: Unused.
        """
        if filters is not None:
            raise NotImplementedError(
                "delete_nodes(filters=...) is not yet supported by simplevecdb. "
                "Resolve the filter to node_ids first via the underlying "
                "VectorCollection.find_ids_by_filter() or query()."
            )
        if node_ids:
            for node_id in node_ids:
                self.delete(node_id)

    def _filters_to_dict(
        self, filters: MetadataFilters | None
    ) -> dict[str, Any] | None:
        if filters is None:
            return None
        result: dict[str, Any] = {}
        if hasattr(filters, "filters"):
            for filter_item in filters.filters:  # type: ignore[attr-defined]
                if hasattr(filter_item, "key") and hasattr(filter_item, "value"):
                    key = getattr(filter_item, "key")
                    value = getattr(filter_item, "value")
                    result[key] = value
        return result or None

    def _build_query_result(
        self,
        docs_with_scores: list[tuple["Document", float]],
        score_transform,
    ) -> VectorStoreQueryResult:
        nodes: list[TextNode] = []
        similarities: list[float] = []
        ids: list[str] = []

        for tiny_doc, score in docs_with_scores:
            # Prefer the persisted node_id over an unstable Python hash().
            # Python's hash() is randomized per process (PYTHONHASHSEED) and
            # can collide; ``_simplevecdb_node_id`` is stamped into metadata
            # at insert time, survives restarts, and uniquely identifies the
            # node.
            metadata = tiny_doc.metadata or {}
            node_id = metadata.get("_simplevecdb_node_id") or str(
                abs(hash(tiny_doc.page_content))
            )
            node = TextNode(
                text=tiny_doc.page_content,
                metadata=tiny_doc.metadata or {},
                id_=node_id,
                relationships={},
            )
            nodes.append(node)
            similarities.append(score_transform(score))
            ids.append(node_id)

        return VectorStoreQueryResult(nodes=nodes, similarities=similarities, ids=ids)

    def query(self, query: VectorStoreQuery, **kwargs: Any) -> VectorStoreQueryResult:
        """Support dense, keyword, or hybrid lookups based on the requested mode."""

        filter_dict = self._filters_to_dict(query.filters)
        mode = getattr(query, "mode", VectorStoreQueryMode.DEFAULT)
        mode_value = getattr(mode, "value", mode)
        normalized_mode = str(mode_value).lower() if mode_value else "default"

        keyword_modes = {
            VectorStoreQueryMode.SPARSE.value,
            VectorStoreQueryMode.TEXT_SEARCH.value,
        }
        hybrid_modes = {
            VectorStoreQueryMode.HYBRID.value,
            VectorStoreQueryMode.SEMANTIC_HYBRID.value,
        }

        if normalized_mode in keyword_modes:
            if not query.query_str:
                raise ValueError("Keyword search requires query_str")
            results = self._collection.keyword_search(
                query.query_str,
                k=query.similarity_top_k,
                filter=filter_dict,
            )
            return self._build_query_result(results, lambda score: 1.0 / (1.0 + score))

        if normalized_mode in hybrid_modes:
            if not query.query_str:
                raise ValueError("Hybrid search requires query_str")
            results = self._collection.hybrid_search(
                query.query_str,
                k=query.similarity_top_k,
                filter=filter_dict,
                query_vector=query.query_embedding,
            )
            return self._build_query_result(results, lambda score: float(score))

        # Fallback to dense/vector search
        query_emb = query.query_embedding
        if query_emb is None:
            if query.query_str:
                query_input: str | list[float] = query.query_str
            else:
                raise ValueError("Either query_embedding or query_str must be provided")
        else:
            query_input = query_emb

        results = self._collection.similarity_search(
            query=query_input,
            k=query.similarity_top_k,
            filter=filter_dict,
        )
        return self._build_query_result(results, lambda distance: 1 - distance)