zeph_memory/

qdrant_ops.rs

// SPDX-FileCopyrightText: 2026 Andrei G <bug-ops>
// SPDX-License-Identifier: MIT OR Apache-2.0

//! Low-level Qdrant operations shared across crates.
//!
//! [`QdrantOps`] is the single point of contact with the `qdrant-client` crate.
//! All higher-level stores ([`crate::embedding_store::EmbeddingStore`],
//! [`crate::embedding_registry::EmbeddingRegistry`]) route through this type.

use std::collections::HashMap;
use std::future::Future;
use std::pin::Pin;
use std::sync::atomic::AtomicBool;
use std::time::Duration;

use crate::vector_store::BoxFuture;
use qdrant_client::Qdrant;
use qdrant_client::qdrant::vector_output::Vector as VectorVariant;
use qdrant_client::qdrant::{
    CreateCollectionBuilder, DeletePointsBuilder, Distance, Filter, GetPointsBuilder, PointId,
    PointStruct, PointsIdsList, QueryPointsBuilder, ScoredPoint, ScrollPointsBuilder,
    UpsertPointsBuilder, VectorParamsBuilder, value::Kind,
};

type QdrantResult<T> = Result<T, Box<qdrant_client::QdrantError>>;

/// Default per-call timeout applied to every Qdrant gRPC operation (#5484).
///
/// Qdrant calls normally complete in well under a second; 10s bounds the await against a
/// hung server or a stalled network path without misfiring on ordinary load spikes.
/// Override via [`QdrantOps::with_timeout`].
const DEFAULT_TIMEOUT: Duration = Duration::from_secs(10);

/// Thin wrapper over [`Qdrant`] client encapsulating common collection operations.
#[derive(Clone)]
pub struct QdrantOps {
    client: Qdrant,
    timeout: Duration,
}

impl std::fmt::Debug for QdrantOps {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("QdrantOps")
            .field("timeout", &self.timeout)
            .finish_non_exhaustive()
    }
}

impl QdrantOps {
    /// Create a new `QdrantOps` connected to the given URL with optional API key.
    ///
    /// When `api_key` is `Some(k)` and `k` is non-empty, the key is attached to the gRPC
    /// client builder. Empty strings are treated as `None` so callers can pass empty config
    /// values without accidentally disabling auth.
    ///
    /// The key is consumed by `qdrant-client` during `.build()` and is never stored on
    /// `QdrantOps` itself.
    ///
    /// # Errors
    ///
    /// Returns an error if the Qdrant client cannot be created (e.g. malformed URL).
    pub fn new(url: &str, api_key: Option<&str>) -> QdrantResult<Self> {
        let mut builder = Qdrant::from_url(url);
        if let Some(key) = api_key.filter(|k| !k.trim().is_empty()) {
            builder = builder.api_key(key.trim());
        }
        let client = builder.build().map_err(Box::new)?;
        Ok(Self {
            client,
            timeout: DEFAULT_TIMEOUT,
        })
    }

    /// Override the per-call timeout applied to every Qdrant gRPC operation.
    ///
    /// Defaults to 10 seconds. A slow or hung Qdrant server would otherwise block the
    /// calling async task indefinitely (#5484). The production bootstrap path drives this
    /// from `MemoryConfig::qdrant_timeout_secs`.
    ///
    /// # Examples
    ///
    /// ```
    /// use std::time::Duration;
    /// use zeph_memory::QdrantOps;
    ///
    /// let ops = QdrantOps::new("http://localhost:6334", None)
    ///     .unwrap()
    ///     .with_timeout(Duration::from_secs(3));
    /// ```
    #[must_use]
    pub fn with_timeout(mut self, timeout: Duration) -> Self {
        self.timeout = timeout;
        self
    }

    /// Access the underlying Qdrant client for advanced operations.
    ///
    /// # Warning
    ///
    /// Calls made directly through the returned client bypass the [`Self::with_timeout`]
    /// guard (#5484) — they are not wrapped by `Self::timed`. Prefer the inherent
    /// `QdrantOps` methods, which are all timeout-guarded, unless the client exposes an
    /// operation this type does not wrap.
    #[must_use]
    pub fn client(&self) -> &Qdrant {
        &self.client
    }

    /// Run a Qdrant gRPC future under the configured [`Self::with_timeout`] bound.
    ///
    /// Converts an elapsed timeout into `QdrantError::Io` so every existing call site
    /// (which already returns `Result<T, Box<QdrantError>>`) needs no signature change.
    ///
    /// Takes a boxed future (rather than `impl Future`) so the timeout wrapper does not
    /// inline the callee's gRPC future type into every caller's generated state machine —
    /// with a generic parameter, that inlining compounds at each layer of async call
    /// nesting and trips `clippy::large_futures` across the whole call chain.
    async fn timed<T>(
        &self,
        fut: Pin<Box<dyn Future<Output = Result<T, qdrant_client::QdrantError>> + Send + '_>>,
    ) -> QdrantResult<T> {
        match tokio::time::timeout(self.timeout, fut).await {
            Ok(result) => result.map_err(Box::new),
            Err(_) => Err(Box::new(qdrant_client::QdrantError::Io(
                std::io::Error::new(
                    std::io::ErrorKind::TimedOut,
                    format!("Qdrant gRPC call exceeded {:?}", self.timeout),
                ),
            ))),
        }
    }

    /// Ensure a collection exists with cosine distance vectors.
    ///
    /// If the collection already exists but has a different vector dimension than `vector_size`,
    /// the collection is deleted and recreated. All existing data in the collection is lost.
    /// Callers on interactive/destructive-sensitive paths should pre-check via
    /// [`Self::collection_exists`] + [`Self::get_collection_vector_size`] and gate the call
    /// behind explicit confirmation instead of relying on this method's silent recreate (see
    /// `zeph knowledge ingest`'s notes-sink resource setup for the reference pattern, #5444).
    ///
    /// # Errors
    ///
    /// Returns an error if Qdrant cannot be reached or collection creation fails.
    #[tracing::instrument(name = "memory.qdrant.ensure_collection", skip_all, err)]
    pub async fn ensure_collection(&self, collection: &str, vector_size: u64) -> QdrantResult<()> {
        if self
            .timed(Box::pin(self.client.collection_exists(collection)))
            .await?
        {
            let existing_size = self.get_collection_vector_size(collection).await?;
            if existing_size == Some(vector_size) {
                return Ok(());
            }
            tracing::warn!(
                collection,
                existing = ?existing_size,
                required = vector_size,
                "vector dimension mismatch — recreating collection (existing data will be lost)"
            );
            self.timed(Box::pin(self.client.delete_collection(collection)))
                .await?;
        }
        self.timed(Box::pin(
            self.client.create_collection(
                CreateCollectionBuilder::new(collection)
                    .vectors_config(VectorParamsBuilder::new(vector_size, Distance::Cosine)),
            ),
        ))
        .await?;
        Ok(())
    }

    /// Returns the configured vector size of an existing collection, or `None` if it cannot be
    /// determined (e.g. named-vector collections).
    ///
    /// # Precondition
    ///
    /// The caller must already know the collection exists (e.g. via a prior
    /// [`Self::collection_exists`] check that returned `true`). This method does **not** treat
    /// "collection missing" and "dimension unreadable" as the same case: calling it on a
    /// non-existent collection surfaces the underlying `collection_info` gRPC error (`Err`), it
    /// does **not** return `Ok(None)`. Every existing call site (this type's own
    /// [`Self::ensure_collection`], [`crate::EmbeddingRegistry`]) checks `collection_exists`
    /// first for this reason.
    ///
    /// Used by [`crate::EmbeddingRegistry`] to validate query vector dimensions before issuing
    /// gRPC searches. Qdrant gRPC silently returns near-zero cosine scores when the query vector
    /// dimension does not match the stored collection dimension (unlike REST which returns 400).
    /// Also used by callers that need to detect an [`Self::ensure_collection`] dimension mismatch
    /// *before* invoking it, so a destructive recreate can be gated behind confirmation.
    ///
    /// # Errors
    ///
    /// Returns an error on hard Qdrant communication failures, including calling this on a
    /// collection that does not exist (see Precondition above).
    #[tracing::instrument(
        name = "memory.qdrant.get_collection_vector_size",
        skip_all,
        err,
        level = "debug"
    )]
    pub async fn get_collection_vector_size(&self, collection: &str) -> QdrantResult<Option<u64>> {
        let info = self
            .timed(Box::pin(self.client.collection_info(collection)))
            .await?;
        let size = info
            .result
            .and_then(|r| r.config)
            .and_then(|cfg| cfg.params)
            .and_then(|params| params.vectors_config)
            .and_then(|vc| vc.config)
            .and_then(|cfg| match cfg {
                qdrant_client::qdrant::vectors_config::Config::Params(vp) => Some(vp.size),
                // Named-vector collections are not supported here; treat as unknown.
                qdrant_client::qdrant::vectors_config::Config::ParamsMap(_) => None,
            });
        Ok(size)
    }

    /// Check whether a collection exists.
    ///
    /// # Errors
    ///
    /// Returns an error if Qdrant cannot be reached.
    #[tracing::instrument(name = "memory.qdrant.collection_exists", skip_all, err)]
    pub async fn collection_exists(&self, collection: &str) -> QdrantResult<bool> {
        self.timed(Box::pin(self.client.collection_exists(collection)))
            .await
    }

    /// Delete a collection.
    ///
    /// # Errors
    ///
    /// Returns an error if the collection cannot be deleted.
    #[tracing::instrument(name = "memory.qdrant.delete_collection", skip_all, err)]
    pub async fn delete_collection(&self, collection: &str) -> QdrantResult<()> {
        self.timed(Box::pin(self.client.delete_collection(collection)))
            .await?;
        Ok(())
    }

    /// Upsert points into a collection.
    ///
    /// # Errors
    ///
    /// Returns an error if the upsert fails.
    #[tracing::instrument(name = "memory.qdrant.upsert", skip_all, err)]
    pub async fn upsert(&self, collection: &str, points: Vec<PointStruct>) -> QdrantResult<()> {
        self.timed(Box::pin(self.client.upsert_points(
            UpsertPointsBuilder::new(collection, points).wait(true),
        )))
        .await?;
        Ok(())
    }

    /// Search for similar vectors, returning scored points with payloads.
    ///
    /// Uses the Qdrant Query API (`client.query`) which handles query vector normalization
    /// server-side for Cosine collections, producing correct similarity scores.
    ///
    /// # Errors
    ///
    /// Returns an error if the search fails.
    #[tracing::instrument(name = "memory.qdrant.search", skip_all, err)]
    pub async fn search(
        &self,
        collection: &str,
        vector: Vec<f32>,
        limit: u64,
        filter: Option<Filter>,
    ) -> QdrantResult<Vec<ScoredPoint>> {
        let mut builder = QueryPointsBuilder::new(collection)
            .query(vector)
            .limit(limit)
            .with_payload(true);
        if let Some(f) = filter {
            builder = builder.filter(f);
        }
        let results = self.timed(Box::pin(self.client.query(builder))).await?;
        Ok(results.result)
    }

    /// Delete points by their IDs.
    ///
    /// # Errors
    ///
    /// Returns an error if the deletion fails.
    #[tracing::instrument(name = "memory.qdrant.delete_by_ids", skip_all, err)]
    pub async fn delete_by_ids(&self, collection: &str, ids: Vec<PointId>) -> QdrantResult<()> {
        if ids.is_empty() {
            return Ok(());
        }
        self.timed(Box::pin(
            self.client.delete_points(
                DeletePointsBuilder::new(collection)
                    .points(PointsIdsList { ids })
                    .wait(true),
            ),
        ))
        .await?;
        Ok(())
    }

    /// Scroll all points in a collection, extracting string payload fields.
    ///
    /// Returns a map of `key_field` value -> { `field_name` -> `field_value` }.
    ///
    /// # Errors
    ///
    /// Returns an error if the scroll operation fails.
    #[tracing::instrument(name = "memory.qdrant.scroll_all", skip_all, err)]
    pub async fn scroll_all(
        &self,
        collection: &str,
        key_field: &str,
    ) -> QdrantResult<HashMap<String, HashMap<String, String>>> {
        let mut result = HashMap::new();
        let mut offset: Option<PointId> = None;

        loop {
            let mut builder = ScrollPointsBuilder::new(collection)
                .with_payload(true)
                .with_vectors(false)
                .limit(100);

            if let Some(ref off) = offset {
                builder = builder.offset(off.clone());
            }

            let response = self.timed(Box::pin(self.client.scroll(builder))).await?;

            for point in &response.result {
                let Some(key_val) = point.payload.get(key_field) else {
                    continue;
                };
                let Some(Kind::StringValue(key)) = &key_val.kind else {
                    continue;
                };

                let mut fields = HashMap::new();
                for (k, val) in &point.payload {
                    if let Some(Kind::StringValue(s)) = &val.kind {
                        fields.insert(k.clone(), s.clone());
                    }
                }
                result.insert(key.clone(), fields);
            }

            match response.next_page_offset {
                Some(next) => offset = Some(next),
                None => break,
            }
        }

        Ok(result)
    }

    /// Scroll all points in a collection, returning `(point_id, string_payload_fields)` pairs.
    ///
    /// Only points whose payload contains `key_field` as a `StringValue` are included.
    /// The Qdrant point ID is preserved as the first tuple element.
    ///
    /// # Errors
    ///
    /// Returns an error if the scroll operation fails.
    #[tracing::instrument(name = "memory.qdrant.scroll_all_with_point_ids", skip_all, err)]
    pub async fn scroll_all_with_point_ids(
        &self,
        collection: &str,
        key_field: &str,
    ) -> QdrantResult<Vec<(String, HashMap<String, String>)>> {
        let mut result = Vec::new();
        let mut offset: Option<PointId> = None;

        loop {
            let mut builder = ScrollPointsBuilder::new(collection)
                .with_payload(true)
                .with_vectors(false)
                .limit(100);

            if let Some(ref off) = offset {
                builder = builder.offset(off.clone());
            }

            let response = self.timed(Box::pin(self.client.scroll(builder))).await?;

            for point in &response.result {
                let Some(key_val) = point.payload.get(key_field) else {
                    continue;
                };
                let Some(Kind::StringValue(_)) = &key_val.kind else {
                    continue;
                };
                let Some(point_id_str) = point_id_to_string(point.id.clone()) else {
                    continue;
                };

                let mut fields = HashMap::new();
                for (k, val) in &point.payload {
                    if let Some(Kind::StringValue(s)) = &val.kind {
                        fields.insert(k.clone(), s.clone());
                    }
                }
                result.push((point_id_str, fields));
            }

            match response.next_page_offset {
                Some(next) => offset = Some(next),
                None => break,
            }
        }

        Ok(result)
    }

    /// Create a collection with scalar INT8 quantization if it does not exist,
    /// then create keyword indexes for the given fields.
    ///
    /// If the collection already exists but has a different vector dimension than `vector_size`,
    /// the collection is deleted and recreated. All existing data in the collection is lost.
    ///
    /// # Errors
    ///
    /// Returns an error if any Qdrant operation fails.
    #[tracing::instrument(
        name = "memory.qdrant.ensure_collection_with_quantization",
        skip_all,
        err
    )]
    pub async fn ensure_collection_with_quantization(
        &self,
        collection: &str,
        vector_size: u64,
        keyword_fields: &[&str],
    ) -> Result<(), crate::VectorStoreError> {
        use qdrant_client::qdrant::{
            CreateFieldIndexCollectionBuilder, FieldType, ScalarQuantizationBuilder,
        };
        if self
            .timed(Box::pin(self.client.collection_exists(collection)))
            .await
            .map_err(|e| crate::VectorStoreError::Collection(e.to_string()))?
        {
            let existing_size = self
                .get_collection_vector_size(collection)
                .await
                .map_err(|e| crate::VectorStoreError::Collection(e.to_string()))?;
            if existing_size == Some(vector_size) {
                return Ok(());
            }
            tracing::warn!(
                collection,
                existing = ?existing_size,
                required = vector_size,
                "vector dimension mismatch — recreating collection (existing data will be lost)"
            );
            self.timed(Box::pin(self.client.delete_collection(collection)))
                .await
                .map_err(|e| crate::VectorStoreError::Collection(e.to_string()))?;
        }
        self.timed(Box::pin(
            self.client.create_collection(
                CreateCollectionBuilder::new(collection)
                    .vectors_config(VectorParamsBuilder::new(vector_size, Distance::Cosine))
                    .quantization_config(ScalarQuantizationBuilder::default()),
            ),
        ))
        .await
        .map_err(|e| crate::VectorStoreError::Collection(e.to_string()))?;

        for field in keyword_fields {
            self.timed(Box::pin(self.client.create_field_index(
                CreateFieldIndexCollectionBuilder::new(collection, *field, FieldType::Keyword),
            )))
            .await
            .map_err(|e| crate::VectorStoreError::Collection(e.to_string()))?;
        }
        Ok(())
    }

    /// Convert a JSON value to a Qdrant payload map.
    ///
    /// # Errors
    ///
    /// Returns a JSON error if deserialization fails.
    pub fn json_to_payload(
        value: serde_json::Value,
    ) -> Result<HashMap<String, qdrant_client::qdrant::Value>, serde_json::Error> {
        serde_json::from_value(value)
    }
}

impl crate::vector_store::VectorStore for QdrantOps {
    fn search_clamp_diagnostics(&self) -> (&'static str, &'static AtomicBool) {
        static CLAMP_WARNED: AtomicBool = AtomicBool::new(false);
        ("QdrantOps::search", &CLAMP_WARNED)
    }

    fn ensure_collection(
        &self,
        collection: &str,
        vector_size: u64,
    ) -> BoxFuture<'_, Result<(), crate::VectorStoreError>> {
        let collection = collection.to_owned();
        Box::pin(async move {
            self.ensure_collection(&collection, vector_size)
                .await
                .map_err(|e| crate::VectorStoreError::Collection(e.to_string()))
        })
    }

    fn collection_exists(
        &self,
        collection: &str,
    ) -> BoxFuture<'_, Result<bool, crate::VectorStoreError>> {
        let collection = collection.to_owned();
        Box::pin(async move {
            self.collection_exists(&collection)
                .await
                .map_err(|e| crate::VectorStoreError::Collection(e.to_string()))
        })
    }

    fn delete_collection(
        &self,
        collection: &str,
    ) -> BoxFuture<'_, Result<(), crate::VectorStoreError>> {
        let collection = collection.to_owned();
        Box::pin(async move {
            self.delete_collection(&collection)
                .await
                .map_err(|e| crate::VectorStoreError::Collection(e.to_string()))
        })
    }

    fn upsert(
        &self,
        collection: &str,
        points: Vec<crate::VectorPoint>,
    ) -> BoxFuture<'_, Result<(), crate::VectorStoreError>> {
        let collection = collection.to_owned();
        Box::pin(async move {
            let qdrant_points: Vec<PointStruct> = points
                .into_iter()
                .map(|p| {
                    let payload: HashMap<String, qdrant_client::qdrant::Value> =
                        serde_json::from_value(serde_json::Value::Object(
                            p.payload.into_iter().collect(),
                        ))
                        .unwrap_or_default();
                    PointStruct::new(p.id, p.vector, payload)
                })
                .collect();
            self.upsert(&collection, qdrant_points)
                .await
                .map_err(|e| crate::VectorStoreError::Upsert(e.to_string()))
        })
    }

    fn search_clamped(
        &self,
        collection: &str,
        vector: Vec<f32>,
        limit: u64,
        filter: Option<crate::VectorFilter>,
    ) -> BoxFuture<'_, Result<Vec<crate::ScoredVectorPoint>, crate::VectorStoreError>> {
        let collection = collection.to_owned();
        Box::pin(async move {
            let qdrant_filter = filter.map(vector_filter_to_qdrant);
            let results = self
                .search(&collection, vector, limit, qdrant_filter)
                .await
                .map_err(|e| crate::VectorStoreError::Search(e.to_string()))?;
            Ok(results.into_iter().map(scored_point_to_vector).collect())
        })
    }

    fn delete_by_ids(
        &self,
        collection: &str,
        ids: Vec<String>,
    ) -> BoxFuture<'_, Result<(), crate::VectorStoreError>> {
        let collection = collection.to_owned();
        Box::pin(async move {
            let point_ids: Vec<PointId> = ids.into_iter().map(PointId::from).collect();
            self.delete_by_ids(&collection, point_ids)
                .await
                .map_err(|e| crate::VectorStoreError::Delete(e.to_string()))
        })
    }

    fn scroll_all(
        &self,
        collection: &str,
        key_field: &str,
    ) -> BoxFuture<'_, Result<HashMap<String, HashMap<String, String>>, crate::VectorStoreError>>
    {
        let collection = collection.to_owned();
        let key_field = key_field.to_owned();
        Box::pin(async move {
            self.scroll_all(&collection, &key_field)
                .await
                .map_err(|e| crate::VectorStoreError::Scroll(e.to_string()))
        })
    }

    fn scroll_all_with_point_ids(
        &self,
        collection: &str,
        key_field: &str,
    ) -> BoxFuture<'_, Result<crate::vector_store::ScrollWithIdsResult, crate::VectorStoreError>>
    {
        let collection = collection.to_owned();
        let key_field = key_field.to_owned();
        Box::pin(async move {
            self.scroll_all_with_point_ids(&collection, &key_field)
                .await
                .map_err(|e| crate::VectorStoreError::Scroll(e.to_string()))
        })
    }

    fn health_check(&self) -> BoxFuture<'_, Result<bool, crate::VectorStoreError>> {
        use tracing::Instrument as _;
        Box::pin(
            async move {
                match self.timed(Box::pin(self.client.health_check())).await {
                    Ok(_) => Ok(true),
                    Err(e) => {
                        tracing::warn!(err = %e, "health_check failed");
                        Err(crate::VectorStoreError::Collection(e.to_string()))
                    }
                }
            }
            .instrument(tracing::debug_span!("memory.qdrant.health_check")),
        )
    }

    fn create_keyword_indexes(
        &self,
        collection: &str,
        fields: &[&str],
    ) -> BoxFuture<'_, Result<(), crate::VectorStoreError>> {
        use qdrant_client::qdrant::{CreateFieldIndexCollectionBuilder, FieldType};
        use tracing::Instrument as _;
        let collection = collection.to_owned();
        let fields: Vec<String> = fields.iter().map(|f| (*f).to_owned()).collect();
        Box::pin(
            async move {
                for field in &fields {
                    self.timed(Box::pin(self.client.create_field_index(
                        CreateFieldIndexCollectionBuilder::new(
                            &collection,
                            field.as_str(),
                            FieldType::Keyword,
                        ),
                    )))
                    .await
                    .map_err(|e| crate::VectorStoreError::Collection(e.to_string()))?;
                }
                Ok(())
            }
            .instrument(tracing::debug_span!("memory.qdrant.create_keyword_indexes")),
        )
    }

    fn get_points(
        &self,
        collection: &str,
        ids: Vec<String>,
    ) -> BoxFuture<'_, Result<Vec<crate::VectorPoint>, crate::VectorStoreError>> {
        use tracing::Instrument as _;
        let collection = collection.to_owned();
        Box::pin(
            async move {
                if ids.is_empty() {
                    return Ok(Vec::new());
                }
                let point_ids: Vec<PointId> = ids.into_iter().map(PointId::from).collect();
                let response = self
                    .timed(Box::pin(
                        self.client.get_points(
                            GetPointsBuilder::new(&collection, point_ids)
                                .with_vectors(true)
                                .with_payload(true),
                        ),
                    ))
                    .await
                    .map_err(|e| {
                        tracing::error!(err = %e, "get_points failed");
                        crate::VectorStoreError::Search(e.to_string())
                    })?;

                let mut result = Vec::with_capacity(response.result.len());
                for point in response.result {
                    let Some(id_str) = point_id_to_string(point.id) else {
                        continue;
                    };
                    // Use VectorsOutput::get_vector() to extract the default dense vector.
                    let vector = match point.vectors.and_then(|v| v.get_vector()) {
                        Some(VectorVariant::Dense(dv)) => dv.data,
                        _ => continue,
                    };
                    let payload: HashMap<String, serde_json::Value> = point
                        .payload
                        .into_iter()
                        .filter_map(|(k, v)| {
                            let json = qdrant_value_to_json(v.kind?)?;
                            Some((k, json))
                        })
                        .collect();
                    result.push(crate::VectorPoint {
                        id: id_str,
                        vector,
                        payload,
                    });
                }
                Ok(result)
            }
            .instrument(tracing::debug_span!("memory.qdrant.get_points")),
        )
    }
}

fn vector_filter_to_qdrant(filter: crate::VectorFilter) -> Filter {
    let must: Vec<_> = filter
        .must
        .into_iter()
        .map(field_condition_to_qdrant)
        .collect();
    let must_not: Vec<_> = filter
        .must_not
        .into_iter()
        .map(field_condition_to_qdrant)
        .collect();

    let mut f = Filter::default();
    if !must.is_empty() {
        f.must = must;
    }
    if !must_not.is_empty() {
        f.must_not = must_not;
    }
    f
}

fn field_condition_to_qdrant(cond: crate::FieldCondition) -> qdrant_client::qdrant::Condition {
    match cond.value {
        crate::FieldValue::Integer(v) => qdrant_client::qdrant::Condition::matches(cond.field, v),
        crate::FieldValue::Text(v) => qdrant_client::qdrant::Condition::matches(cond.field, v),
    }
}

/// Convert a Qdrant [`qdrant_client::qdrant::PointId`] to its string representation.
///
/// Returns `None` when the id variant is unrecognised.
fn point_id_to_string(pid: Option<qdrant_client::qdrant::PointId>) -> Option<String> {
    match pid?.point_id_options? {
        qdrant_client::qdrant::point_id::PointIdOptions::Uuid(u) => Some(u),
        qdrant_client::qdrant::point_id::PointIdOptions::Num(n) => Some(n.to_string()),
    }
}

/// Convert a Qdrant [`Kind`] to a `serde_json::Value`.
///
/// Returns `None` for unsupported kinds (structs, lists, nulls).
fn qdrant_value_to_json(kind: Kind) -> Option<serde_json::Value> {
    match kind {
        Kind::StringValue(s) => Some(serde_json::Value::String(s)),
        Kind::IntegerValue(i) => Some(serde_json::Value::Number(i.into())),
        Kind::DoubleValue(d) => serde_json::Number::from_f64(d).map(serde_json::Value::Number),
        Kind::BoolValue(b) => Some(serde_json::Value::Bool(b)),
        _ => None,
    }
}

fn scored_point_to_vector(point: ScoredPoint) -> crate::ScoredVectorPoint {
    let payload: HashMap<String, serde_json::Value> = point
        .payload
        .into_iter()
        .filter_map(|(k, v)| Some((k, qdrant_value_to_json(v.kind?)?)))
        .collect();

    let id = point_id_to_string(point.id).unwrap_or_default();

    crate::ScoredVectorPoint {
        id,
        score: point.score,
        payload,
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn new_valid_url() {
        let ops = QdrantOps::new("http://localhost:6334", None);
        assert!(ops.is_ok());
    }

    #[test]
    fn new_invalid_url() {
        let ops = QdrantOps::new("not a valid url", None);
        assert!(ops.is_err());
    }

    /// Empty `api_key` must be silently treated as `None` — the whitespace guard in
    /// `QdrantOps::new` must not panic and the client must be built successfully.
    #[test]
    fn new_empty_api_key_is_treated_as_none() {
        let result = QdrantOps::new("http://127.0.0.1:9999", Some(""));
        assert!(result.is_ok(), "empty key must not cause a build error");
    }

    /// Whitespace-only keys must be dropped the same way as empty keys.
    #[test]
    fn new_whitespace_api_key_is_treated_as_none() {
        let result = QdrantOps::new("http://127.0.0.1:9999", Some("   "));
        assert!(
            result.is_ok(),
            "whitespace-only key must not cause a build error"
        );
    }

    /// A non-empty, non-whitespace key must be accepted without errors.
    #[test]
    fn new_with_api_key_constructs_successfully() {
        let result = QdrantOps::new("http://127.0.0.1:9999", Some("valid-key"));
        assert!(result.is_ok(), "valid key must not cause a build error");
    }

    /// `new` must apply [`DEFAULT_TIMEOUT`] and [`QdrantOps::with_timeout`] must override it
    /// (#5484). The override is observable via the `Debug` impl since `timeout` is private.
    #[test]
    fn with_timeout_overrides_default() {
        let ops = QdrantOps::new("http://localhost:6334", None).unwrap();
        assert!(format!("{ops:?}").contains("10s"), "default must be 10s");

        let ops = ops.with_timeout(Duration::from_secs(2));
        assert!(
            format!("{ops:?}").contains("2s"),
            "with_timeout must override the default"
        );
    }

    /// `timed` must return an error instead of hanging forever when the wrapped future never
    /// resolves — the core guarantee of #5484.
    #[tokio::test]
    async fn timed_returns_error_instead_of_hanging() {
        let ops = QdrantOps::new("http://localhost:6334", None)
            .unwrap()
            .with_timeout(Duration::from_millis(10));

        let never_resolves: Pin<
            Box<dyn Future<Output = Result<(), qdrant_client::QdrantError>> + Send>,
        > = Box::pin(std::future::pending());

        let result = ops.timed(never_resolves).await;
        assert!(result.is_err(), "must time out instead of hanging forever");
    }

    #[test]
    fn debug_format() {
        let ops = QdrantOps::new("http://localhost:6334", None).unwrap();
        let dbg = format!("{ops:?}");
        assert!(dbg.contains("QdrantOps"));
    }

    #[test]
    fn json_to_payload_valid() {
        let value = serde_json::json!({"key": "value", "num": 42});
        let result = QdrantOps::json_to_payload(value);
        assert!(result.is_ok());
    }

    #[test]
    fn json_to_payload_empty() {
        let result = QdrantOps::json_to_payload(serde_json::json!({}));
        assert!(result.is_ok());
        assert!(result.unwrap().is_empty());
    }

    #[test]
    fn delete_by_ids_empty_is_ok_sync() {
        // Constructing QdrantOps with a valid URL succeeds even without a live server.
        // delete_by_ids with empty list short-circuits before any network call.
        // We validate the early-return logic via the async test below.
        let ops = QdrantOps::new("http://localhost:6334", None);
        assert!(ops.is_ok());
    }

    /// Requires a live Qdrant instance at localhost:6334.
    #[tokio::test]
    #[ignore = "requires a live Qdrant instance at localhost:6334"]
    async fn ensure_collection_with_quantization_idempotent() {
        let ops = QdrantOps::new("http://localhost:6334", None).unwrap();
        let collection = "test_quant_idempotent";

        // Clean up from any prior run
        let _ = ops.delete_collection(collection).await;

        // First call — creates collection
        ops.ensure_collection_with_quantization(collection, 128, &["language", "file_path"])
            .await
            .unwrap();

        assert!(ops.collection_exists(collection).await.unwrap());

        // Second call — idempotent, must not error
        ops.ensure_collection_with_quantization(collection, 128, &["language", "file_path"])
            .await
            .unwrap();

        // Cleanup
        ops.delete_collection(collection).await.unwrap();
    }

    /// Requires a live Qdrant instance at localhost:6334.
    #[tokio::test]
    #[ignore = "requires a live Qdrant instance at localhost:6334"]
    async fn delete_by_ids_empty_no_network_call() {
        let ops = QdrantOps::new("http://localhost:6334", None).unwrap();
        // Empty ID list must short-circuit and return Ok without hitting Qdrant.
        let result = ops.delete_by_ids("nonexistent_collection", vec![]).await;
        assert!(result.is_ok());
    }

    /// Requires a live Qdrant instance at localhost:6334.
    #[tokio::test]
    #[ignore = "requires a live Qdrant instance at localhost:6334"]
    async fn ensure_collection_idempotent_same_size() {
        let ops = QdrantOps::new("http://localhost:6334", None).unwrap();
        let collection = "test_ensure_idempotent";

        let _ = ops.delete_collection(collection).await;

        ops.ensure_collection(collection, 128).await.unwrap();
        assert!(ops.collection_exists(collection).await.unwrap());

        // Second call with same size must be a no-op.
        ops.ensure_collection(collection, 128).await.unwrap();
        assert!(ops.collection_exists(collection).await.unwrap());

        ops.delete_collection(collection).await.unwrap();
    }

    /// Requires a live Qdrant instance at localhost:6334.
    ///
    /// Verifies that `ensure_collection` detects a vector dimension mismatch and
    /// recreates the collection instead of silently reusing the wrong-dimension one.
    #[tokio::test]
    #[ignore = "requires a live Qdrant instance at localhost:6334"]
    async fn ensure_collection_recreates_on_dimension_mismatch() {
        let ops = QdrantOps::new("http://localhost:6334", None).unwrap();
        let collection = "test_dim_mismatch";

        let _ = ops.delete_collection(collection).await;

        // Create with 128 dims.
        ops.ensure_collection(collection, 128).await.unwrap();
        assert_eq!(
            ops.get_collection_vector_size(collection).await.unwrap(),
            Some(128)
        );

        // Call again with a different size — must recreate.
        ops.ensure_collection(collection, 256).await.unwrap();
        assert_eq!(
            ops.get_collection_vector_size(collection).await.unwrap(),
            Some(256),
            "collection must have been recreated with the new dimension"
        );

        ops.delete_collection(collection).await.unwrap();
    }

    /// Requires a live Qdrant instance at localhost:6334.
    ///
    /// Verifies that `ensure_collection_with_quantization` also detects dimension mismatch.
    #[tokio::test]
    #[ignore = "requires a live Qdrant instance at localhost:6334"]
    async fn ensure_collection_with_quantization_recreates_on_dimension_mismatch() {
        let ops = QdrantOps::new("http://localhost:6334", None).unwrap();
        let collection = "test_quant_dim_mismatch";

        let _ = ops.delete_collection(collection).await;

        ops.ensure_collection_with_quantization(collection, 128, &["language"])
            .await
            .unwrap();
        assert_eq!(
            ops.get_collection_vector_size(collection).await.unwrap(),
            Some(128)
        );

        // Call again with a different size — must recreate.
        ops.ensure_collection_with_quantization(collection, 384, &["language"])
            .await
            .unwrap();
        assert_eq!(
            ops.get_collection_vector_size(collection).await.unwrap(),
            Some(384),
            "collection must have been recreated with the new dimension"
        );

        ops.delete_collection(collection).await.unwrap();
    }

    /// Issue #6616: the `VectorStore::search` trait impl on `QdrantOps` must clamp an
    /// oversized `limit` itself, not rely on `EmbeddingStore`/`EmbeddingRegistry` wrapper
    /// methods (issue #6553) to clamp before forwarding. `QdrantOps` has no test seam
    /// (concrete gRPC client), so — matching the pattern in `embedding_registry.rs`'s
    /// `search_raw_oversized_limit_does_not_panic` — this connects to an unreachable
    /// endpoint and asserts the one-shot `tracing::warn!` fired before the network call was
    /// attempted, rather than the clamped result count.
    #[tokio::test]
    #[tracing_test::traced_test]
    async fn vector_store_search_clamps_oversized_limit() {
        use crate::vector_store::VectorStore;

        let ops = QdrantOps::new("http://127.0.0.1:1", None).unwrap(); // unreachable — forces network error
        let _ = VectorStore::search(&ops, "col", vec![1.0, 0.0], u64::MAX, None).await;
        assert!(
            logs_contain("requested search limit exceeds MAX_SEARCH_LIMIT"),
            "expected the one-shot clamp warning to fire for an oversized limit"
        );
    }
}