velesdb_mobile/
collection.rs

1//! `VelesCollection` — UniFFI-exported collection operations for mobile.
2
3use std::sync::Arc;
4
5use velesdb_core::{
6    Database as CoreDatabase, Filter, GatedRead, QueryOperationKind,
7    VectorCollection as CoreCollection,
8};
9
10use crate::types::{
11    IndividualSearchRequest, MobileAdvancedConfig, MobileCollectionDiagnostics,
12    MobileCollectionStats, MobileIndexInfo, MobileQueryLimits, MobileStreamingConfig,
13    SearchQuality, SearchResult, VelesError, VelesPoint,
14};
15
16// ============================================================================
17// Collection
18// ============================================================================
19
20/// A collection of vectors with associated metadata.
21///
22/// `inner` is a *detached* core collection leaf with no observer reference, so
23/// every governed read is routed back through the owning database's
24/// control-plane gate (`db.gated_search` / `db.authorize_read`) rather than
25/// hitting the leaf directly — restoring observer governance for the mobile
26/// direct-search API (audit F-5.4, #1392). Reads with no observer registered
27/// take a single `Option` check and then the same leaf call as before
28/// (zero-overhead contract).
29#[derive(uniffi::Object)]
30pub struct VelesCollection {
31    pub(crate) inner: CoreCollection,
32    /// Shared handle to the owning database, carrying the read gate.
33    pub(crate) db: Arc<CoreDatabase>,
34    /// Collection name, used to address the gate.
35    pub(crate) name: String,
36}
37
38/// Maps a batch of core search results into the UniFFI `SearchResult` shape
39/// (id + score, payload dropped — the mobile direct-search API is id/score
40/// oriented). Shared by every gated read leaf.
41fn to_mobile_results(results: Vec<velesdb_core::SearchResult>) -> Vec<SearchResult> {
42    results
43        .into_iter()
44        .map(|r| SearchResult {
45            id: r.point.id,
46            score: r.score,
47            payload: None,
48        })
49        .collect()
50}
51
52/// AND-composes a caller filter with an observer scope filter. The result
53/// matches only rows satisfying both, so composing a scope can only narrow.
54pub(crate) fn and_scope(caller: Option<Filter>, scope: Option<Filter>) -> Option<Filter> {
55    use velesdb_core::Condition;
56    match (caller, scope) {
57        (None, None) => None,
58        (Some(c), None) => Some(c),
59        (None, Some(s)) => Some(s),
60        (Some(c), Some(s)) => Some(Filter::new(Condition::And {
61            conditions: vec![c.condition, s.condition],
62        })),
63    }
64}
65
66/// Fails a non-`GatedRead` read closed when the observer returned a scope
67/// filter the leaf cannot apply (id/score-only leaves with no metadata-filtered
68/// twin). Refusing to run unscoped is the safe default — never widen past the
69/// observer's narrowing.
70pub(crate) fn deny_if_scoped(scope: Option<Filter>, context: &str) -> Result<(), VelesError> {
71    if scope.is_some() {
72        return Err(VelesError::database(format!(
73            "{context} cannot honor the governance scope filter returned by the observer \
74             (this entry point has no metadata-filtered leaf); refusing to run unscoped"
75        )));
76    }
77    Ok(())
78}
79
80#[uniffi::export]
81impl VelesCollection {
82    /// Searches for the k nearest neighbors to the query vector.
83    ///
84    /// # Arguments
85    ///
86    /// * `vector` - Query vector
87    /// * `limit` - Maximum number of results to return
88    ///
89    /// # Returns
90    ///
91    /// Vector of search results sorted by similarity.
92    pub fn search(&self, vector: Vec<f32>, limit: u32) -> Result<Vec<SearchResult>, VelesError> {
93        let results = self.db.gated_search(
94            &self.name,
95            None,
96            None,
97            GatedRead::Dense {
98                query: &vector,
99                k: usize::try_from(limit).unwrap_or(usize::MAX),
100                ef: None,
101                quality: None,
102                filter: None,
103            },
104        )?;
105
106        Ok(to_mobile_results(results))
107    }
108
109    /// Searches with a specific quality profile controlling recall/latency.
110    ///
111    /// # Arguments
112    ///
113    /// * `vector` - Query vector
114    /// * `limit` - Maximum number of results to return
115    /// * `quality` - Search quality profile (Fast, Balanced, Accurate, etc.)
116    ///
117    /// # Returns
118    ///
119    /// Vector of search results sorted by similarity.
120    pub fn search_with_quality(
121        &self,
122        vector: Vec<f32>,
123        limit: u32,
124        quality: SearchQuality,
125    ) -> Result<Vec<SearchResult>, VelesError> {
126        let results = self.db.gated_search(
127            &self.name,
128            None,
129            None,
130            GatedRead::Dense {
131                query: &vector,
132                k: usize::try_from(limit).unwrap_or(usize::MAX),
133                ef: None,
134                quality: Some(quality.into()),
135                filter: None,
136            },
137        )?;
138
139        Ok(to_mobile_results(results))
140    }
141
142    /// Inserts or updates a single point.
143    ///
144    /// # Arguments
145    ///
146    /// * `point` - The point to upsert
147    pub fn upsert(&self, point: VelesPoint) -> Result<(), VelesError> {
148        let core_point = parse_point(point)?;
149        self.inner.upsert(vec![core_point])?;
150        Ok(())
151    }
152
153    /// Inserts or updates multiple points in batch.
154    ///
155    /// # Arguments
156    ///
157    /// * `points` - Points to upsert
158    pub fn upsert_batch(&self, points: Vec<VelesPoint>) -> Result<(), VelesError> {
159        let core_points: Result<Vec<velesdb_core::Point>, VelesError> =
160            points.into_iter().map(parse_point).collect();
161
162        self.inner.upsert(core_points?)?;
163        Ok(())
164    }
165
166    /// Deletes a point by ID.
167    pub fn delete(&self, id: u64) -> Result<(), VelesError> {
168        self.inner.delete(&[id])?;
169        Ok(())
170    }
171
172    /// Returns the number of points in the collection.
173    #[allow(clippy::cast_possible_truncation)]
174    pub fn count(&self) -> u64 {
175        self.inner.config().point_count as u64
176    }
177
178    /// Returns the vector dimension.
179    #[allow(clippy::cast_possible_truncation)]
180    pub fn dimension(&self) -> u32 {
181        self.inner.config().dimension as u32
182    }
183
184    /// Gets points by their IDs.
185    ///
186    /// # Arguments
187    ///
188    /// * `ids` - List of point IDs to retrieve
189    ///
190    /// # Returns
191    ///
192    /// Vector of points found. Missing IDs are silently skipped.
193    pub fn get(&self, ids: Vec<u64>) -> Vec<VelesPoint> {
194        self.inner
195            .get(&ids)
196            .into_iter()
197            .flatten()
198            .map(|p| VelesPoint {
199                id: p.id,
200                vector: p.vector,
201                payload: p.payload.map(|v| v.to_string()),
202            })
203            .collect()
204    }
205
206    /// Gets a single point by ID.
207    ///
208    /// # Arguments
209    ///
210    /// * `id` - Point ID to retrieve
211    ///
212    /// # Returns
213    ///
214    /// The point if found, None otherwise.
215    pub fn get_by_id(&self, id: u64) -> Option<VelesPoint> {
216        self.inner
217            .get(&[id])
218            .into_iter()
219            .flatten()
220            .next()
221            .map(|p| VelesPoint {
222                id: p.id,
223                vector: p.vector,
224                payload: p.payload.map(|v| v.to_string()),
225            })
226    }
227
228    /// Checks if this is a metadata-only collection.
229    pub fn is_metadata_only(&self) -> bool {
230        self.inner.config().metadata_only
231    }
232
233    /// Performs full-text search using BM25.
234    ///
235    /// # Arguments
236    ///
237    /// * `query` - Text query to search for
238    /// * `limit` - Maximum number of results to return
239    ///
240    /// # Returns
241    ///
242    /// Vector of search results sorted by BM25 score.
243    pub fn text_search(&self, query: String, limit: u32) -> Result<Vec<SearchResult>, VelesError> {
244        let results = self.db.gated_search(
245            &self.name,
246            None,
247            None,
248            GatedRead::Text {
249                query: &query,
250                k: usize::try_from(limit).unwrap_or(usize::MAX),
251                filter: None,
252            },
253        )?;
254
255        Ok(to_mobile_results(results))
256    }
257
258    /// Performs hybrid search combining vector similarity and BM25 text search.
259    ///
260    /// # Arguments
261    ///
262    /// * `vector` - Query vector for similarity search
263    /// * `text_query` - Text query for BM25 search
264    /// * `limit` - Maximum number of results
265    /// * `vector_weight` - Weight for vector similarity (0.0-1.0)
266    ///
267    /// # Returns
268    ///
269    /// Vector of search results sorted by fused score.
270    pub fn hybrid_search(
271        &self,
272        vector: Vec<f32>,
273        text_query: String,
274        limit: u32,
275        vector_weight: f32,
276    ) -> Result<Vec<SearchResult>, VelesError> {
277        let results = self.db.gated_search(
278            &self.name,
279            None,
280            None,
281            GatedRead::Hybrid {
282                vector: &vector,
283                text: &text_query,
284                k: usize::try_from(limit).unwrap_or(usize::MAX),
285                alpha: Some(vector_weight),
286                filter: None,
287            },
288        )?;
289
290        Ok(to_mobile_results(results))
291    }
292
293    /// Searches with metadata filtering.
294    ///
295    /// # Arguments
296    ///
297    /// * `vector` - Query vector
298    /// * `limit` - Maximum number of results
299    /// * `filter_json` - JSON filter string (e.g., `{"condition": {"type": "eq", "field": "category", "value": "tech"}}`)
300    ///
301    /// # Returns
302    ///
303    /// Vector of search results matching the filter.
304    pub fn search_with_filter(
305        &self,
306        vector: Vec<f32>,
307        limit: u32,
308        filter_json: String,
309    ) -> Result<Vec<SearchResult>, VelesError> {
310        // Parse filter JSON
311        let filter: Filter = serde_json::from_str(&filter_json)
312            .map_err(|e| VelesError::database(format!("Invalid filter JSON: {e}")))?;
313
314        let results = self.db.gated_search(
315            &self.name,
316            None,
317            None,
318            GatedRead::Dense {
319                query: &vector,
320                k: usize::try_from(limit).unwrap_or(usize::MAX),
321                ef: None,
322                quality: None,
323                filter: Some(&filter),
324            },
325        )?;
326
327        Ok(to_mobile_results(results))
328    }
329
330    /// Performs batch search for multiple query vectors in parallel.
331    ///
332    /// # Arguments
333    ///
334    /// * `searches` - List of search requests
335    ///
336    /// # Returns
337    ///
338    /// List of result lists (one per query vector).
339    pub fn batch_search(
340        &self,
341        searches: Vec<IndividualSearchRequest>,
342    ) -> Result<Vec<Vec<SearchResult>>, VelesError> {
343        let query_refs: Vec<&[f32]> = searches.iter().map(|s| s.vector.as_slice()).collect();
344
345        let filters: Result<Vec<Option<Filter>>, VelesError> = searches
346            .iter()
347            .map(|s| {
348                s.filter
349                    .as_ref()
350                    .map(|f_json| {
351                        serde_json::from_str(f_json).map_err(|e| {
352                            VelesError::database(format!("Invalid filter JSON in batch: {e}"))
353                        })
354                    })
355                    .transpose()
356            })
357            .collect();
358
359        // Authorize the batch as a whole through the read gate, then AND any
360        // observer scope filter into every per-query filter so the filtered
361        // leaf enforces the narrowing (Deny propagates as an error).
362        let scope =
363            self.db
364                .authorize_read(&self.name, QueryOperationKind::VectorSearch, None, None)?;
365        let filters: Vec<Option<Filter>> = filters?
366            .into_iter()
367            .map(|f| and_scope(f, scope.clone()))
368            .collect();
369        let max_top_k = searches.iter().map(|s| s.top_k).max().unwrap_or(10);
370
371        let all_results = self.inner.search_batch_with_filters(
372            &query_refs,
373            usize::try_from(max_top_k).unwrap_or(usize::MAX),
374            &filters,
375        )?;
376
377        Ok(all_results
378            .into_iter()
379            .zip(searches)
380            .map(
381                |(results, s): (Vec<velesdb_core::SearchResult>, IndividualSearchRequest)| {
382                    results
383                        .into_iter()
384                        .take(usize::try_from(s.top_k).unwrap_or(usize::MAX))
385                        .map(|r| SearchResult {
386                            id: r.point.id,
387                            score: r.score,
388                            payload: None,
389                        })
390                        .collect()
391                },
392            )
393            .collect())
394    }
395
396    /// Performs text search with metadata filtering.
397    ///
398    /// # Arguments
399    ///
400    /// * `query` - Text query
401    /// * `limit` - Maximum number of results
402    /// * `filter_json` - JSON filter string
403    pub fn text_search_with_filter(
404        &self,
405        query: String,
406        limit: u32,
407        filter_json: String,
408    ) -> Result<Vec<SearchResult>, VelesError> {
409        let filter: Filter = serde_json::from_str(&filter_json)
410            .map_err(|e| VelesError::database(format!("Invalid filter JSON: {e}")))?;
411
412        let results = self.db.gated_search(
413            &self.name,
414            None,
415            None,
416            GatedRead::Text {
417                query: &query,
418                k: usize::try_from(limit).unwrap_or(usize::MAX),
419                filter: Some(&filter),
420            },
421        )?;
422
423        Ok(to_mobile_results(results))
424    }
425
426    /// Performs hybrid search with metadata filtering.
427    ///
428    /// # Arguments
429    ///
430    /// * `vector` - Query vector
431    /// * `text_query` - Text query
432    /// * `limit` - Maximum number of results
433    /// * `vector_weight` - Weight for vector similarity (0.0-1.0)
434    /// * `filter_json` - JSON filter string
435    pub fn hybrid_search_with_filter(
436        &self,
437        vector: Vec<f32>,
438        text_query: String,
439        limit: u32,
440        vector_weight: f32,
441        filter_json: String,
442    ) -> Result<Vec<SearchResult>, VelesError> {
443        let filter: Filter = serde_json::from_str(&filter_json)
444            .map_err(|e| VelesError::database(format!("Invalid filter JSON: {e}")))?;
445
446        let results = self.db.gated_search(
447            &self.name,
448            None,
449            None,
450            GatedRead::Hybrid {
451                vector: &vector,
452                text: &text_query,
453                k: usize::try_from(limit).unwrap_or(usize::MAX),
454                alpha: Some(vector_weight),
455                filter: Some(&filter),
456            },
457        )?;
458
459        Ok(to_mobile_results(results))
460    }
461
462    /// Executes a VelesQL query.
463    ///
464    /// # Arguments
465    ///
466    /// * `query_str` - VelesQL query string
467    /// * `params_json` - Optional JSON object with query parameters
468    ///
469    /// # Returns
470    ///
471    /// Vector of search results.
472    ///
473    /// # Example
474    ///
475    /// ```swift
476    /// let results = try collection.query(
477    ///     "SELECT * FROM vectors WHERE category = 'tech' LIMIT 10",
478    ///     nil
479    /// )
480    /// ```
481    pub fn query(
482        &self,
483        query_str: String,
484        params_json: Option<String>,
485    ) -> Result<Vec<SearchResult>, VelesError> {
486        // Parse the VelesQL query
487        let parsed = velesdb_core::velesql::Parser::parse(&query_str)
488            .map_err(|e| VelesError::database(format!("VelesQL parse error: {}", e.message)))?;
489
490        // Parse params from JSON if provided
491        let params: std::collections::HashMap<String, serde_json::Value> = params_json
492            .map(|json| serde_json::from_str(&json))
493            .transpose()
494            .map_err(|e| VelesError::database(format!("Invalid params JSON: {e}")))?
495            .unwrap_or_default();
496
497        // Execute through the owning database (not the detached collection
498        // leaf) so the VelesQL read path passes the control-plane observer gate
499        // — the detached `VectorCollection::execute_query` has no observer
500        // reference and would bypass governance (audit F-5.4, #1392).
501        let results = self
502            .db
503            .execute_query(&parsed, &params)
504            .map_err(|e| VelesError::database(format!("Query execution failed: {e}")))?;
505
506        Ok(results
507            .into_iter()
508            .map(|r| SearchResult {
509                id: r.point.id,
510                score: r.score,
511                payload: r.point.payload.as_ref().map(|p| p.to_string()),
512            })
513            .collect())
514    }
515
516    // multi_query_search and multi_query_search_with_filter are in collection_sparse.rs
517
518    /// Enables streaming ingestion on this collection.
519    ///
520    /// Must be called before [`stream_insert`](Self::stream_insert); otherwise
521    /// stream inserts fail with "not configured". Calling it again replaces the
522    /// existing ingester.
523    ///
524    /// # Arguments
525    ///
526    /// * `config` - Optional [`MobileStreamingConfig`]. `None` uses the engine
527    ///   defaults (`buffer_size=10000`, `batch_size=128`, `flush_interval_ms=50`).
528    pub fn enable_streaming(
529        &self,
530        config: Option<MobileStreamingConfig>,
531    ) -> Result<(), VelesError> {
532        let core_config = config.map_or_else(velesdb_core::StreamingConfig::default, |c| {
533            velesdb_core::StreamingConfig::new(
534                usize::try_from(c.buffer_size).unwrap_or(usize::MAX),
535                usize::try_from(c.batch_size).unwrap_or(usize::MAX),
536                c.flush_interval_ms,
537            )
538        });
539        // Spawning the drain task needs an ambient runtime; enter the shared
540        // streaming runtime so the task is scheduled on it and survives this call.
541        let rt = crate::streaming_runtime::stream_runtime()?;
542        let _guard = rt.enter();
543        self.inner.enable_streaming(core_config);
544        Ok(())
545    }
546
547    /// Queues a batch of points for streaming ingestion.
548    ///
549    /// Requires [`enable_streaming`](Self::enable_streaming) to have been called
550    /// first. Returns the number of points successfully queued.
551    ///
552    /// # Arguments
553    ///
554    /// * `points` - Points to queue for ingestion
555    pub fn stream_insert(&self, points: Vec<VelesPoint>) -> Result<u64, VelesError> {
556        let core_points: Result<Vec<velesdb_core::Point>, VelesError> =
557            points.into_iter().map(parse_point).collect();
558
559        let queued = self.inner.stream_insert_batch(core_points?).map_err(|e| {
560            VelesError::database(format!(
561                "Stream insert failed (buffer full or not configured): {e}"
562            ))
563        })?;
564        Ok(u64::try_from(queued).unwrap_or(u64::MAX))
565    }
566
567    /// Flushes collection data to durable storage.
568    pub fn flush(&self) -> Result<(), VelesError> {
569        self.inner.flush()?;
570        Ok(())
571    }
572
573    /// Compacts on-disk storage, reclaiming space left by deleted vectors.
574    ///
575    /// Returns the number of bytes reclaimed.
576    pub fn compact_storage(&self) -> Result<u64, VelesError> {
577        Ok(u64::try_from(self.inner.compact_storage()?).unwrap_or(u64::MAX))
578    }
579
580    /// Returns the current query guardrail limits for this collection.
581    pub fn guard_rails(&self) -> MobileQueryLimits {
582        self.inner.guard_rails().limits().into()
583    }
584
585    /// Applies post-creation overrides to advanced configuration fields
586    /// (`pq_rescore_oversampling`, `deferred_indexing`,
587    /// `async_index_builder`) and persists the updated config. Each `Some`
588    /// field is applied; each `None` field is left unchanged.
589    pub fn apply_advanced_config(&self, config: MobileAdvancedConfig) -> Result<(), VelesError> {
590        self.inner.apply_advanced_config(
591            config.pq_rescore_oversampling.map(Some),
592            config.deferred_indexing.map(|c| Some(c.into())),
593            config.async_index_builder.map(|c| Some(c.into())),
594        )?;
595        Ok(())
596    }
597
598    /// Returns all point IDs currently present in the collection.
599    pub fn all_ids(&self) -> Vec<u64> {
600        self.inner.all_ids()
601    }
602
603    /// Creates a secondary metadata index for a payload field.
604    pub fn create_index(&self, field_name: String) -> Result<(), VelesError> {
605        self.inner.create_index(&field_name)?;
606        Ok(())
607    }
608
609    /// Checks whether a secondary metadata index exists for a field.
610    pub fn has_secondary_index(&self, field_name: String) -> bool {
611        self.inner.has_secondary_index(&field_name)
612    }
613
614    /// Creates a graph/property index for equality lookups.
615    pub fn create_property_index(&self, label: String, property: String) -> Result<(), VelesError> {
616        self.inner.create_property_index(&label, &property)?;
617        Ok(())
618    }
619
620    /// Creates a graph/range index for range queries.
621    pub fn create_range_index(&self, label: String, property: String) -> Result<(), VelesError> {
622        self.inner.create_range_index(&label, &property)?;
623        Ok(())
624    }
625
626    /// Checks if a property index exists.
627    pub fn has_property_index(&self, label: String, property: String) -> bool {
628        self.inner.has_property_index(&label, &property)
629    }
630
631    /// Checks if a range index exists.
632    pub fn has_range_index(&self, label: String, property: String) -> bool {
633        self.inner.has_range_index(&label, &property)
634    }
635
636    /// Lists all index definitions on this collection.
637    pub fn list_indexes(&self) -> Vec<MobileIndexInfo> {
638        self.inner
639            .list_indexes()
640            .into_iter()
641            .map(MobileIndexInfo::from)
642            .collect()
643    }
644
645    /// Drops an index and returns true when something was removed.
646    pub fn drop_index(&self, label: String, property: String) -> Result<bool, VelesError> {
647        Ok(self.inner.drop_index(&label, &property)?)
648    }
649
650    /// Returns total memory usage used by indexes.
651    pub fn indexes_memory_usage(&self) -> u64 {
652        u64::try_from(self.inner.indexes_memory_usage()).unwrap_or(u64::MAX)
653    }
654
655    /// Runs ANALYZE and returns fresh statistics for this collection.
656    pub fn analyze(&self) -> Result<MobileCollectionStats, VelesError> {
657        Ok(self.inner.analyze()?.into())
658    }
659
660    /// Returns the latest known collection statistics snapshot.
661    pub fn get_stats(&self) -> MobileCollectionStats {
662        self.inner.get_stats().into()
663    }
664
665    /// Returns a health/readiness diagnostics snapshot for this collection.
666    pub fn diagnostics(&self) -> MobileCollectionDiagnostics {
667        self.inner.diagnostics().into()
668    }
669}
670
671/// Converts a [`VelesPoint`] into a core point, parsing the optional JSON payload.
672fn parse_point(p: VelesPoint) -> Result<velesdb_core::Point, VelesError> {
673    let payload = p
674        .payload
675        .map(|s| serde_json::from_str(&s))
676        .transpose()
677        .map_err(|e| VelesError::database(format!("Invalid JSON payload: {e}")))?;
678    Ok(velesdb_core::Point::new(p.id, p.vector, payload))
679}
680
681// Sparse vector operations are in collection_sparse.rs
velesdb_mobile/collection.rs

velesdb_mobile/
collection.rs
