velesdb-mobile 4.2.0

//! `VelesCollection` — UniFFI-exported collection operations for mobile.

use std::sync::Arc;

use velesdb_core::{
    Database as CoreDatabase, Filter, GatedRead, QueryOperationKind,
    VectorCollection as CoreCollection,
};

use crate::types::{
    IndividualSearchRequest, MobileAdvancedConfig, MobileCollectionDiagnostics,
    MobileCollectionStats, MobileIndexInfo, MobileQueryLimits, MobileStreamingConfig,
    SearchQuality, SearchResult, VelesError, VelesPoint,
};

// ============================================================================
// Collection
// ============================================================================

/// A collection of vectors with associated metadata.
///
/// `inner` is a *detached* core collection leaf with no observer reference, so
/// every governed read is routed back through the owning database's
/// control-plane gate (`db.gated_search` / `db.authorize_read`) rather than
/// hitting the leaf directly — restoring observer governance for the mobile
/// direct-search API (audit F-5.4, #1392). Reads with no observer registered
/// take a single `Option` check and then the same leaf call as before
/// (zero-overhead contract).
#[derive(uniffi::Object)]
pub struct VelesCollection {
    pub(crate) inner: CoreCollection,
    /// Shared handle to the owning database, carrying the read gate.
    pub(crate) db: Arc<CoreDatabase>,
    /// Collection name, used to address the gate.
    pub(crate) name: String,
}

/// Maps a batch of core search results into the UniFFI `SearchResult` shape
/// (id + score, payload dropped — the mobile direct-search API is id/score
/// oriented). Shared by every gated read leaf.
fn to_mobile_results(results: Vec<velesdb_core::SearchResult>) -> Vec<SearchResult> {
    results
        .into_iter()
        .map(|r| SearchResult {
            id: r.point.id,
            score: r.score,
            payload: None,
        })
        .collect()
}

/// AND-composes a caller filter with an observer scope filter. The result
/// matches only rows satisfying both, so composing a scope can only narrow.
pub(crate) fn and_scope(caller: Option<Filter>, scope: Option<Filter>) -> Option<Filter> {
    use velesdb_core::Condition;
    match (caller, scope) {
        (None, None) => None,
        (Some(c), None) => Some(c),
        (None, Some(s)) => Some(s),
        (Some(c), Some(s)) => Some(Filter::new(Condition::And {
            conditions: vec![c.condition, s.condition],
        })),
    }
}

/// Fails a non-`GatedRead` read closed when the observer returned a scope
/// filter the leaf cannot apply (id/score-only leaves with no metadata-filtered
/// twin). Refusing to run unscoped is the safe default — never widen past the
/// observer's narrowing.
pub(crate) fn deny_if_scoped(scope: Option<Filter>, context: &str) -> Result<(), VelesError> {
    if scope.is_some() {
        return Err(VelesError::database(format!(
            "{context} cannot honor the governance scope filter returned by the observer \
             (this entry point has no metadata-filtered leaf); refusing to run unscoped"
        )));
    }
    Ok(())
}

#[uniffi::export]
impl VelesCollection {
    /// Searches for the k nearest neighbors to the query vector.
    ///
    /// # Arguments
    ///
    /// * `vector` - Query vector
    /// * `limit` - Maximum number of results to return
    ///
    /// # Returns
    ///
    /// Vector of search results sorted by similarity.
    pub fn search(&self, vector: Vec<f32>, limit: u32) -> Result<Vec<SearchResult>, VelesError> {
        let results = self.db.gated_search(
            &self.name,
            None,
            None,
            GatedRead::Dense {
                query: &vector,
                k: usize::try_from(limit).unwrap_or(usize::MAX),
                ef: None,
                quality: None,
                filter: None,
            },
        )?;

        Ok(to_mobile_results(results))
    }

    /// Searches with a specific quality profile controlling recall/latency.
    ///
    /// # Arguments
    ///
    /// * `vector` - Query vector
    /// * `limit` - Maximum number of results to return
    /// * `quality` - Search quality profile (Fast, Balanced, Accurate, etc.)
    ///
    /// # Returns
    ///
    /// Vector of search results sorted by similarity.
    pub fn search_with_quality(
        &self,
        vector: Vec<f32>,
        limit: u32,
        quality: SearchQuality,
    ) -> Result<Vec<SearchResult>, VelesError> {
        let results = self.db.gated_search(
            &self.name,
            None,
            None,
            GatedRead::Dense {
                query: &vector,
                k: usize::try_from(limit).unwrap_or(usize::MAX),
                ef: None,
                quality: Some(quality.into()),
                filter: None,
            },
        )?;

        Ok(to_mobile_results(results))
    }

    /// Inserts or updates a single point.
    ///
    /// # Arguments
    ///
    /// * `point` - The point to upsert
    pub fn upsert(&self, point: VelesPoint) -> Result<(), VelesError> {
        let core_point = parse_point(point)?;
        self.inner.upsert(vec![core_point])?;
        Ok(())
    }

    /// Inserts or updates multiple points in batch.
    ///
    /// # Arguments
    ///
    /// * `points` - Points to upsert
    pub fn upsert_batch(&self, points: Vec<VelesPoint>) -> Result<(), VelesError> {
        let core_points: Result<Vec<velesdb_core::Point>, VelesError> =
            points.into_iter().map(parse_point).collect();

        self.inner.upsert(core_points?)?;
        Ok(())
    }

    /// Deletes a point by ID.
    pub fn delete(&self, id: u64) -> Result<(), VelesError> {
        self.inner.delete(&[id])?;
        Ok(())
    }

    /// Returns the number of points in the collection.
    #[allow(clippy::cast_possible_truncation)]
    pub fn count(&self) -> u64 {
        self.inner.config().point_count as u64
    }

    /// Returns the vector dimension.
    #[allow(clippy::cast_possible_truncation)]
    pub fn dimension(&self) -> u32 {
        self.inner.config().dimension as u32
    }

    /// Gets points by their IDs.
    ///
    /// # Arguments
    ///
    /// * `ids` - List of point IDs to retrieve
    ///
    /// # Returns
    ///
    /// Vector of points found. Missing IDs are silently skipped.
    pub fn get(&self, ids: Vec<u64>) -> Vec<VelesPoint> {
        self.inner
            .get(&ids)
            .into_iter()
            .flatten()
            .map(|p| VelesPoint {
                id: p.id,
                vector: p.vector,
                payload: p.payload.map(|v| v.to_string()),
            })
            .collect()
    }

    /// Gets a single point by ID.
    ///
    /// # Arguments
    ///
    /// * `id` - Point ID to retrieve
    ///
    /// # Returns
    ///
    /// The point if found, None otherwise.
    pub fn get_by_id(&self, id: u64) -> Option<VelesPoint> {
        self.inner
            .get(&[id])
            .into_iter()
            .flatten()
            .next()
            .map(|p| VelesPoint {
                id: p.id,
                vector: p.vector,
                payload: p.payload.map(|v| v.to_string()),
            })
    }

    /// Checks if this is a metadata-only collection.
    pub fn is_metadata_only(&self) -> bool {
        self.inner.config().metadata_only
    }

    /// Performs full-text search using BM25.
    ///
    /// # Arguments
    ///
    /// * `query` - Text query to search for
    /// * `limit` - Maximum number of results to return
    ///
    /// # Returns
    ///
    /// Vector of search results sorted by BM25 score.
    pub fn text_search(&self, query: String, limit: u32) -> Result<Vec<SearchResult>, VelesError> {
        let results = self.db.gated_search(
            &self.name,
            None,
            None,
            GatedRead::Text {
                query: &query,
                k: usize::try_from(limit).unwrap_or(usize::MAX),
                filter: None,
            },
        )?;

        Ok(to_mobile_results(results))
    }

    /// Performs hybrid search combining vector similarity and BM25 text search.
    ///
    /// # Arguments
    ///
    /// * `vector` - Query vector for similarity search
    /// * `text_query` - Text query for BM25 search
    /// * `limit` - Maximum number of results
    /// * `vector_weight` - Weight for vector similarity (0.0-1.0)
    ///
    /// # Returns
    ///
    /// Vector of search results sorted by fused score.
    pub fn hybrid_search(
        &self,
        vector: Vec<f32>,
        text_query: String,
        limit: u32,
        vector_weight: f32,
    ) -> Result<Vec<SearchResult>, VelesError> {
        let results = self.db.gated_search(
            &self.name,
            None,
            None,
            GatedRead::Hybrid {
                vector: &vector,
                text: &text_query,
                k: usize::try_from(limit).unwrap_or(usize::MAX),
                alpha: Some(vector_weight),
                filter: None,
            },
        )?;

        Ok(to_mobile_results(results))
    }

    /// Searches with metadata filtering.
    ///
    /// # Arguments
    ///
    /// * `vector` - Query vector
    /// * `limit` - Maximum number of results
    /// * `filter_json` - JSON filter string (e.g., `{"condition": {"type": "eq", "field": "category", "value": "tech"}}`)
    ///
    /// # Returns
    ///
    /// Vector of search results matching the filter.
    pub fn search_with_filter(
        &self,
        vector: Vec<f32>,
        limit: u32,
        filter_json: String,
    ) -> Result<Vec<SearchResult>, VelesError> {
        // Parse filter JSON
        let filter: Filter = serde_json::from_str(&filter_json)
            .map_err(|e| VelesError::database(format!("Invalid filter JSON: {e}")))?;

        let results = self.db.gated_search(
            &self.name,
            None,
            None,
            GatedRead::Dense {
                query: &vector,
                k: usize::try_from(limit).unwrap_or(usize::MAX),
                ef: None,
                quality: None,
                filter: Some(&filter),
            },
        )?;

        Ok(to_mobile_results(results))
    }

    /// Performs batch search for multiple query vectors in parallel.
    ///
    /// # Arguments
    ///
    /// * `searches` - List of search requests
    ///
    /// # Returns
    ///
    /// List of result lists (one per query vector).
    pub fn batch_search(
        &self,
        searches: Vec<IndividualSearchRequest>,
    ) -> Result<Vec<Vec<SearchResult>>, VelesError> {
        let query_refs: Vec<&[f32]> = searches.iter().map(|s| s.vector.as_slice()).collect();

        let filters: Result<Vec<Option<Filter>>, VelesError> = searches
            .iter()
            .map(|s| {
                s.filter
                    .as_ref()
                    .map(|f_json| {
                        serde_json::from_str(f_json).map_err(|e| {
                            VelesError::database(format!("Invalid filter JSON in batch: {e}"))
                        })
                    })
                    .transpose()
            })
            .collect();

        // Authorize the batch as a whole through the read gate, then AND any
        // observer scope filter into every per-query filter so the filtered
        // leaf enforces the narrowing (Deny propagates as an error).
        let scope =
            self.db
                .authorize_read(&self.name, QueryOperationKind::VectorSearch, None, None)?;
        let filters: Vec<Option<Filter>> = filters?
            .into_iter()
            .map(|f| and_scope(f, scope.clone()))
            .collect();
        let max_top_k = searches.iter().map(|s| s.top_k).max().unwrap_or(10);

        let all_results = self.inner.search_batch_with_filters(
            &query_refs,
            usize::try_from(max_top_k).unwrap_or(usize::MAX),
            &filters,
        )?;

        Ok(all_results
            .into_iter()
            .zip(searches)
            .map(
                |(results, s): (Vec<velesdb_core::SearchResult>, IndividualSearchRequest)| {
                    results
                        .into_iter()
                        .take(usize::try_from(s.top_k).unwrap_or(usize::MAX))
                        .map(|r| SearchResult {
                            id: r.point.id,
                            score: r.score,
                            payload: None,
                        })
                        .collect()
                },
            )
            .collect())
    }

    /// Performs text search with metadata filtering.
    ///
    /// # Arguments
    ///
    /// * `query` - Text query
    /// * `limit` - Maximum number of results
    /// * `filter_json` - JSON filter string
    pub fn text_search_with_filter(
        &self,
        query: String,
        limit: u32,
        filter_json: String,
    ) -> Result<Vec<SearchResult>, VelesError> {
        let filter: Filter = serde_json::from_str(&filter_json)
            .map_err(|e| VelesError::database(format!("Invalid filter JSON: {e}")))?;

        let results = self.db.gated_search(
            &self.name,
            None,
            None,
            GatedRead::Text {
                query: &query,
                k: usize::try_from(limit).unwrap_or(usize::MAX),
                filter: Some(&filter),
            },
        )?;

        Ok(to_mobile_results(results))
    }

    /// Performs hybrid search with metadata filtering.
    ///
    /// # Arguments
    ///
    /// * `vector` - Query vector
    /// * `text_query` - Text query
    /// * `limit` - Maximum number of results
    /// * `vector_weight` - Weight for vector similarity (0.0-1.0)
    /// * `filter_json` - JSON filter string
    pub fn hybrid_search_with_filter(
        &self,
        vector: Vec<f32>,
        text_query: String,
        limit: u32,
        vector_weight: f32,
        filter_json: String,
    ) -> Result<Vec<SearchResult>, VelesError> {
        let filter: Filter = serde_json::from_str(&filter_json)
            .map_err(|e| VelesError::database(format!("Invalid filter JSON: {e}")))?;

        let results = self.db.gated_search(
            &self.name,
            None,
            None,
            GatedRead::Hybrid {
                vector: &vector,
                text: &text_query,
                k: usize::try_from(limit).unwrap_or(usize::MAX),
                alpha: Some(vector_weight),
                filter: Some(&filter),
            },
        )?;

        Ok(to_mobile_results(results))
    }

    /// Executes a VelesQL query.
    ///
    /// # Arguments
    ///
    /// * `query_str` - VelesQL query string
    /// * `params_json` - Optional JSON object with query parameters
    ///
    /// # Returns
    ///
    /// Vector of search results.
    ///
    /// # Example
    ///
    /// ```swift
    /// let results = try collection.query(
    ///     "SELECT * FROM vectors WHERE category = 'tech' LIMIT 10",
    ///     nil
    /// )
    /// ```
    pub fn query(
        &self,
        query_str: String,
        params_json: Option<String>,
    ) -> Result<Vec<SearchResult>, VelesError> {
        // Parse the VelesQL query
        let parsed = velesdb_core::velesql::Parser::parse(&query_str)
            .map_err(|e| VelesError::database(format!("VelesQL parse error: {}", e.message)))?;

        // Parse params from JSON if provided
        let params: std::collections::HashMap<String, serde_json::Value> = params_json
            .map(|json| serde_json::from_str(&json))
            .transpose()
            .map_err(|e| VelesError::database(format!("Invalid params JSON: {e}")))?
            .unwrap_or_default();

        // Execute through the owning database (not the detached collection
        // leaf) so the VelesQL read path passes the control-plane observer gate
        // — the detached `VectorCollection::execute_query` has no observer
        // reference and would bypass governance (audit F-5.4, #1392).
        let results = self
            .db
            .execute_query(&parsed, &params)
            .map_err(|e| VelesError::database(format!("Query execution failed: {e}")))?;

        Ok(results
            .into_iter()
            .map(|r| SearchResult {
                id: r.point.id,
                score: r.score,
                payload: r.point.payload.as_ref().map(|p| p.to_string()),
            })
            .collect())
    }

    // multi_query_search and multi_query_search_with_filter are in collection_sparse.rs

    /// Enables streaming ingestion on this collection.
    ///
    /// Must be called before [`stream_insert`](Self::stream_insert); otherwise
    /// stream inserts fail with "not configured". Calling it again replaces the
    /// existing ingester.
    ///
    /// # Arguments
    ///
    /// * `config` - Optional [`MobileStreamingConfig`]. `None` uses the engine
    ///   defaults (`buffer_size=10000`, `batch_size=128`, `flush_interval_ms=50`).
    pub fn enable_streaming(
        &self,
        config: Option<MobileStreamingConfig>,
    ) -> Result<(), VelesError> {
        let core_config = config.map_or_else(velesdb_core::StreamingConfig::default, |c| {
            velesdb_core::StreamingConfig::new(
                usize::try_from(c.buffer_size).unwrap_or(usize::MAX),
                usize::try_from(c.batch_size).unwrap_or(usize::MAX),
                c.flush_interval_ms,
            )
        });
        // Spawning the drain task needs an ambient runtime; enter the shared
        // streaming runtime so the task is scheduled on it and survives this call.
        let rt = crate::streaming_runtime::stream_runtime()?;
        let _guard = rt.enter();
        self.inner.enable_streaming(core_config);
        Ok(())
    }

    /// Queues a batch of points for streaming ingestion.
    ///
    /// Requires [`enable_streaming`](Self::enable_streaming) to have been called
    /// first. Returns the number of points successfully queued.
    ///
    /// # Arguments
    ///
    /// * `points` - Points to queue for ingestion
    pub fn stream_insert(&self, points: Vec<VelesPoint>) -> Result<u64, VelesError> {
        let core_points: Result<Vec<velesdb_core::Point>, VelesError> =
            points.into_iter().map(parse_point).collect();

        let queued = self.inner.stream_insert_batch(core_points?).map_err(|e| {
            VelesError::database(format!(
                "Stream insert failed (buffer full or not configured): {e}"
            ))
        })?;
        Ok(u64::try_from(queued).unwrap_or(u64::MAX))
    }

    /// Flushes collection data to durable storage.
    pub fn flush(&self) -> Result<(), VelesError> {
        self.inner.flush()?;
        Ok(())
    }

    /// Compacts on-disk storage, reclaiming space left by deleted vectors.
    ///
    /// Returns the number of bytes reclaimed.
    pub fn compact_storage(&self) -> Result<u64, VelesError> {
        Ok(u64::try_from(self.inner.compact_storage()?).unwrap_or(u64::MAX))
    }

    /// Returns the current query guardrail limits for this collection.
    pub fn guard_rails(&self) -> MobileQueryLimits {
        self.inner.guard_rails().limits().into()
    }

    /// Applies post-creation overrides to advanced configuration fields
    /// (`pq_rescore_oversampling`, `deferred_indexing`,
    /// `async_index_builder`) and persists the updated config. Each `Some`
    /// field is applied; each `None` field is left unchanged.
    pub fn apply_advanced_config(&self, config: MobileAdvancedConfig) -> Result<(), VelesError> {
        self.inner.apply_advanced_config(
            config.pq_rescore_oversampling.map(Some),
            config.deferred_indexing.map(|c| Some(c.into())),
            config.async_index_builder.map(|c| Some(c.into())),
        )?;
        Ok(())
    }

    /// Returns all point IDs currently present in the collection.
    pub fn all_ids(&self) -> Vec<u64> {
        self.inner.all_ids()
    }

    /// Creates a secondary metadata index for a payload field.
    pub fn create_index(&self, field_name: String) -> Result<(), VelesError> {
        self.inner.create_index(&field_name)?;
        Ok(())
    }

    /// Checks whether a secondary metadata index exists for a field.
    pub fn has_secondary_index(&self, field_name: String) -> bool {
        self.inner.has_secondary_index(&field_name)
    }

    /// Creates a graph/property index for equality lookups.
    pub fn create_property_index(&self, label: String, property: String) -> Result<(), VelesError> {
        self.inner.create_property_index(&label, &property)?;
        Ok(())
    }

    /// Creates a graph/range index for range queries.
    pub fn create_range_index(&self, label: String, property: String) -> Result<(), VelesError> {
        self.inner.create_range_index(&label, &property)?;
        Ok(())
    }

    /// Checks if a property index exists.
    pub fn has_property_index(&self, label: String, property: String) -> bool {
        self.inner.has_property_index(&label, &property)
    }

    /// Checks if a range index exists.
    pub fn has_range_index(&self, label: String, property: String) -> bool {
        self.inner.has_range_index(&label, &property)
    }

    /// Lists all index definitions on this collection.
    pub fn list_indexes(&self) -> Vec<MobileIndexInfo> {
        self.inner
            .list_indexes()
            .into_iter()
            .map(MobileIndexInfo::from)
            .collect()
    }

    /// Drops an index and returns true when something was removed.
    pub fn drop_index(&self, label: String, property: String) -> Result<bool, VelesError> {
        Ok(self.inner.drop_index(&label, &property)?)
    }

    /// Returns total memory usage used by indexes.
    pub fn indexes_memory_usage(&self) -> u64 {
        u64::try_from(self.inner.indexes_memory_usage()).unwrap_or(u64::MAX)
    }

    /// Runs ANALYZE and returns fresh statistics for this collection.
    pub fn analyze(&self) -> Result<MobileCollectionStats, VelesError> {
        Ok(self.inner.analyze()?.into())
    }

    /// Returns the latest known collection statistics snapshot.
    pub fn get_stats(&self) -> MobileCollectionStats {
        self.inner.get_stats().into()
    }

    /// Returns a health/readiness diagnostics snapshot for this collection.
    pub fn diagnostics(&self) -> MobileCollectionDiagnostics {
        self.inner.diagnostics().into()
    }
}

/// Converts a [`VelesPoint`] into a core point, parsing the optional JSON payload.
fn parse_point(p: VelesPoint) -> Result<velesdb_core::Point, VelesError> {
    let payload = p
        .payload
        .map(|s| serde_json::from_str(&s))
        .transpose()
        .map_err(|e| VelesError::database(format!("Invalid JSON payload: {e}")))?;
    Ok(velesdb_core::Point::new(p.id, p.vector, payload))
}

// Sparse vector operations are in collection_sparse.rs