linkrs-api 0.1.0

//! Vector Index API – Core Layer
//!
//! Provides transport layer independent vector index management and search operations.

use crate::api_core::error::{CoreError, CoreResult};
use linkrs_sync::backend::VectorBackend;
use linkrs_sync::vector_sync::{SearchOptions, VectorIndexLocation, VectorSyncCoordinator};
use simvec::{
    types::{validate_distance_metric, IndexMetadata, PointId},
    CollectionConfig, DistanceMetric, VectorPoint,
};
use std::sync::Arc;

/// Parameters for paginated vector point scanning.
pub struct ScrollQuery<'a> {
    pub space_id: u64,
    pub tag_name: &'a str,
    pub field_name: &'a str,
    pub limit: usize,
    pub offset: Option<&'a str>,
    pub with_payload: Option<bool>,
    pub with_vector: Option<bool>,
}

/// Metrics every backend accepts at index-creation time; anything else is
/// rejected up front instead of failing deep inside one engine.
fn validate_metric(distance: DistanceMetric) -> CoreResult<()> {
    validate_distance_metric(distance).map_err(CoreError::VectorError)
}

/// Vector search result
#[derive(Debug, Clone)]
pub struct VectorSearchResult {
    pub id: PointId,
    pub score: f32,
    pub vector: Option<Vec<f32>>,
    pub payload: Option<std::collections::HashMap<String, serde_json::Value>>,
}

/// Write mode for vector point mutations.
///
/// All writes are transactional through the durable outbox (`WAL + SQLite`)
/// so they participate in the graph transaction's commit/abort and benefit
/// from `read-your-writes` consistency.
#[derive(Debug, Clone)]
pub enum VectorWriteMode {
    /// Transactional write through the durable outbox.
    Transactional {
        txn_id: linkrs_core::types::TransactionId,
        space_id: u64,
        tag: String,
        field: String,
    },
}

/// Vector Index API – Core Layer
pub struct VectorApi {
    backend: VectorBackend,
    coordinator: Arc<VectorSyncCoordinator>,
    sync_manager: Arc<linkrs_sync::SyncManager>,
}

impl VectorApi {
    /// Create a VectorApi with the full assembly the vector subsystem needs.
    ///
    /// The coordinator and sync manager are required, not optional: every
    /// assembly owning a vector backend also owns the sync manager hosting
    /// the vector coordinator, and a partially wired instance would serve
    /// degraded reads or empty index listings instead of failing loudly.
    pub fn new(
        backend: VectorBackend,
        coordinator: Arc<VectorSyncCoordinator>,
        sync_manager: Arc<linkrs_sync::SyncManager>,
    ) -> Self {
        Self {
            backend,
            coordinator,
            sync_manager,
        }
    }

    /// Get the vector backend
    pub fn backend(&self) -> &VectorBackend {
        &self.backend
    }

    /// Get the sync coordinator
    pub fn coordinator(&self) -> &Arc<VectorSyncCoordinator> {
        &self.coordinator
    }

    /// Create a vector index
    ///
    /// Vector indexes are vertex-only: point IDs, payloads, and rebuild
    /// sources are all modeled per vertex, and edge mutations never produce
    /// vector points. `tag_name` must be a vertex tag; edge type names are
    /// rejected by the query and HTTP creation paths before reaching here.
    pub async fn create_index(
        &self,
        space_id: u64,
        tag_name: &str,
        field_name: &str,
        vector_size: usize,
        distance: DistanceMetric,
    ) -> CoreResult<String> {
        self.create_index_with_config(
            space_id,
            tag_name,
            field_name,
            CollectionConfig::new(vector_size, distance),
        )
        .await
    }

    /// Create a vector index with full collection config (quantization/hnsw)
    ///
    /// Vertex-only, like [`VectorApi::create_index`]: `tag_name` must be a
    /// vertex tag, never an edge type name.
    pub async fn create_index_with_config(
        &self,
        space_id: u64,
        tag_name: &str,
        field_name: &str,
        config: CollectionConfig,
    ) -> CoreResult<String> {
        validate_metric(config.distance)?;
        if let Some(qc) = &config.quantization_config {
            qc.validate(config.vector_size)
                .map_err(|e| CoreError::VectorError(e.to_string()))?;
        }

        if config.quantization_config.is_some() || config.hnsw_config.is_some() {
            return self
                .coordinator
                .create_index_with_config(space_id, tag_name, field_name, config)
                .await
                .map_err(|e| CoreError::VectorError(e.to_string()));
        }
        self.coordinator
            .create_vector_index(
                space_id,
                tag_name,
                field_name,
                config.vector_size,
                config.distance,
            )
            .await
            .map_err(|e| CoreError::VectorError(e.to_string()))
    }

    /// Drop a vector index
    pub async fn drop_index(
        &self,
        space_id: u64,
        tag_name: &str,
        field_name: &str,
    ) -> CoreResult<()> {
        self.coordinator
            .drop_vector_index(space_id, tag_name, field_name)
            .await
            .map_err(|e| CoreError::VectorError(e.to_string()))
    }

    /// Get vector index info (from logical index metadata)
    pub fn get_index_info(
        &self,
        space_id: u64,
        tag_name: &str,
        field_name: &str,
    ) -> CoreResult<Option<IndexMetadata>> {
        Ok(self.coordinator.index_info(space_id, tag_name, field_name))
    }

    /// List all vector indexes
    pub fn list_indexes(&self) -> Vec<String> {
        self.coordinator
            .list_indexes()
            .into_iter()
            .map(|w| w.collection_name)
            .collect()
    }

    /// List vector indexes whose last rebuild failed.
    ///
    /// These indexes still serve live data but need operator attention: retry
    /// [`VectorApi::rebuild_collection`].
    pub fn inconsistent_indexes(&self) -> Vec<linkrs_core::index_events::RebuildProgress> {
        self.coordinator.inconsistent_vector_indexes()
    }

    /// Insert a vector point with explicit write mode.
    pub async fn insert_vector_with_mode(
        &self,
        _space_id: u64,
        _tag_name: &str,
        _field_name: &str,
        point: VectorPoint,
        mode: VectorWriteMode,
    ) -> CoreResult<()> {
        let VectorWriteMode::Transactional {
            txn_id,
            space_id: txn_space,
            tag,
            field,
        } = mode;
        let manager = &self.sync_manager;
        let vector = point.vector.clone();
        let payload = point.payload.clone().unwrap_or_default();
        let mut properties: Vec<(Arc<str>, linkrs_core::Value)> = Vec::new();
        for (k, v) in payload {
            if let Ok(val) = serde_json::from_value::<linkrs_core::Value>(v) {
                properties.push((Arc::from(k.as_str()), val));
            }
        }
        properties.push((
            Arc::from(field.as_str()),
            linkrs_core::Value::vector(vector),
        ));
        let vertex_id = linkrs_core::Value::string(point.id.to_string());
        manager
            .on_vertex_change_with_txn(
                txn_id,
                txn_space,
                &tag,
                &vertex_id,
                &properties,
                linkrs_sync::types::ChangeType::Insert,
            )
            .map_err(|e| CoreError::VectorError(e.to_string()))?;
        Ok(())
    }

    /// Insert vector points in batch with explicit write mode.
    pub async fn insert_vector_batch_with_mode(
        &self,
        space_id: u64,
        _tag_name: &str,
        _field_name: &str,
        points: Vec<VectorPoint>,
        mode: VectorWriteMode,
    ) -> CoreResult<()> {
        let VectorWriteMode::Transactional {
            txn_id,
            space_id: txn_space,
            tag,
            field,
        } = mode;
        let manager = &self.sync_manager;
        for point in points {
            let vector = point.vector.clone();
            let payload = point.payload.clone().unwrap_or_default();
            let mut properties: Vec<(Arc<str>, linkrs_core::Value)> = Vec::new();
            for (k, v) in payload {
                if let Ok(val) = serde_json::from_value::<linkrs_core::Value>(v) {
                    properties.push((Arc::from(k.as_str()), val));
                }
            }
            properties.push((
                Arc::from(field.as_str()),
                linkrs_core::Value::vector(vector),
            ));
            let vertex_id = linkrs_core::Value::string(point.id.to_string());
            manager
                .on_vertex_change_with_txn(
                    txn_id,
                    txn_space,
                    &tag,
                    &vertex_id,
                    &properties,
                    linkrs_sync::types::ChangeType::Insert,
                )
                .map_err(|e| CoreError::VectorError(e.to_string()))?;
        }
        let _ = space_id;
        Ok(())
    }

    /// Delete a vector point with explicit write mode.
    pub async fn delete_vector_with_mode(
        &self,
        space_id: u64,
        tag_name: &str,
        field_name: &str,
        point_id: &str,
        mode: VectorWriteMode,
    ) -> CoreResult<()> {
        let VectorWriteMode::Transactional {
            txn_id,
            space_id: txn_space,
            tag,
            field,
        } = mode;
        let manager = &self.sync_manager;
        let vertex_id = linkrs_core::Value::string(point_id);
        manager
            .on_vertex_change_with_txn(
                txn_id,
                txn_space,
                &tag,
                &vertex_id,
                &[],
                linkrs_sync::types::ChangeType::Delete,
            )
            .map_err(|e| CoreError::VectorError(e.to_string()))?;
        let _ = (space_id, tag_name, field_name, field);
        Ok(())
    }

    /// Delete vector points in batch with explicit write mode.
    pub async fn delete_vector_batch_with_mode(
        &self,
        space_id: u64,
        tag_name: &str,
        field_name: &str,
        point_ids: Vec<&str>,
        mode: VectorWriteMode,
    ) -> CoreResult<()> {
        let VectorWriteMode::Transactional {
            txn_id,
            space_id: txn_space,
            tag,
            field,
        } = mode;
        let manager = &self.sync_manager;
        for point_id in point_ids {
            let vertex_id = linkrs_core::Value::string(point_id);
            manager
                .on_vertex_change_with_txn(
                    txn_id,
                    txn_space,
                    &tag,
                    &vertex_id,
                    &[],
                    linkrs_sync::types::ChangeType::Delete,
                )
                .map_err(|e| CoreError::VectorError(e.to_string()))?;
        }
        let _ = (space_id, tag_name, field_name, field);
        Ok(())
    }

    /// Rebuild a vector index from primary storage without dropping it.
    ///
    /// Backfill and catch-up replay converge a temp collection while the
    /// live slice keeps serving; a fenced publish then swaps the temp in.
    /// Requires a configured `SyncManager` (durable outbox + vector
    /// coordinator). Returns the number of applied point operations.
    pub async fn rebuild_collection(
        &self,
        space_id: u64,
        tag_name: &str,
        field_name: &str,
        source: &mut dyn linkrs_sync::VectorDocSource,
        options: linkrs_sync::VectorRebuildOptions,
    ) -> CoreResult<u64> {
        let manager = &self.sync_manager;
        manager
            .rebuild_vector_index(space_id, tag_name, field_name, source, options)
            .await
            .map_err(|e| CoreError::VectorError(e.to_string()))
    }

    /// Clear a vector index: drop all indexed points without backfill.
    ///
    /// This is an explicit destructive operation, not a rebuild. The caller
    /// must pass `force = true` to confirm the data loss; the call is audit
    /// logged. To repair a damaged index without data loss, use
    /// [`VectorApi::rebuild_collection`] (or drop and recreate the index).
    pub async fn clear_collection(
        &self,
        space_id: u64,
        tag_name: &str,
        field_name: &str,
        force: bool,
    ) -> CoreResult<()> {
        if !force {
            return Err(CoreError::VectorError(
                "Refusing to clear vector index without explicit confirmation: retry with force = true"
                    .to_string(),
            ));
        }
        log::warn!(
            "Clearing vector index explicitly confirmed: space {} tag {} field {}",
            space_id,
            tag_name,
            field_name
        );
        self.coordinator
            .index_manager()
            .purge_index_data(space_id, tag_name, field_name)
            .await
            .map_err(|e| CoreError::VectorError(e.to_string()))
    }

    /// Search vectors with options
    pub async fn search_with_options(
        &self,
        options: SearchOptions,
    ) -> CoreResult<Vec<VectorSearchResult>> {
        self.coordinator
            .search_with_options(options)
            .await
            .map(|results| {
                results
                    .into_iter()
                    .map(|r| VectorSearchResult {
                        id: r.id,
                        score: r.score,
                        vector: r.vector.map(|v| v.to_vec()),
                        payload: r.payload.map(|p| p.into_iter().collect()),
                    })
                    .collect()
            })
            .map_err(|e| CoreError::VectorError(e.to_string()))
    }

    /// Get a vector point by ID
    pub async fn get_vector(
        &self,
        space_id: u64,
        tag_name: &str,
        field_name: &str,
        point_id: &str,
    ) -> CoreResult<Option<VectorPoint>> {
        let collection_name =
            VectorIndexLocation::new(space_id, tag_name, field_name).to_collection_name();
        self.backend
            .get_vector(&collection_name, point_id)
            .await
            .map_err(|e| CoreError::VectorError(e.to_string()))
    }

    /// Get vector index count
    pub async fn count(&self, space_id: u64, tag_name: &str, field_name: &str) -> CoreResult<u64> {
        let collection_name =
            VectorIndexLocation::new(space_id, tag_name, field_name).to_collection_name();
        self.backend
            .count(&collection_name)
            .await
            .map_err(|e| CoreError::VectorError(e.to_string()))
    }

    /// Replace the entire payload for the given points.
    pub async fn set_payload(
        &self,
        space_id: u64,
        tag_name: &str,
        field_name: &str,
        point_ids: Vec<&str>,
        payload: simvec::types::Payload,
    ) -> CoreResult<()> {
        let collection_name =
            VectorIndexLocation::new(space_id, tag_name, field_name).to_collection_name();
        self.backend
            .set_payload(&collection_name, point_ids, payload)
            .await
            .map_err(|e| CoreError::VectorError(e.to_string()))
    }

    /// Merge the given fields into the payload of the given points.
    /// Only the supplied keys are updated; other existing keys are preserved.
    pub async fn set_payload_fields(
        &self,
        space_id: u64,
        tag_name: &str,
        field_name: &str,
        point_ids: Vec<&str>,
        fields: simvec::types::Payload,
    ) -> CoreResult<()> {
        let collection_name =
            VectorIndexLocation::new(space_id, tag_name, field_name).to_collection_name();
        self.backend
            .set_payload_fields(&collection_name, point_ids, fields)
            .await
            .map_err(|e| CoreError::VectorError(e.to_string()))
    }

    /// Remove specific keys from the payload of the given points.
    pub async fn delete_payload(
        &self,
        space_id: u64,
        tag_name: &str,
        field_name: &str,
        point_ids: Vec<&str>,
        keys: Vec<&str>,
    ) -> CoreResult<()> {
        let collection_name =
            VectorIndexLocation::new(space_id, tag_name, field_name).to_collection_name();
        self.backend
            .delete_payload(&collection_name, point_ids, keys)
            .await
            .map_err(|e| CoreError::VectorError(e.to_string()))
    }

    /// Paginated scan over points in a collection.
    pub async fn scroll(
        &self,
        query: ScrollQuery<'_>,
    ) -> CoreResult<(Vec<VectorPoint>, Option<String>)> {
        let collection_name =
            VectorIndexLocation::new(query.space_id, query.tag_name, query.field_name)
                .to_collection_name();
        self.backend
            .scroll(
                &collection_name,
                query.limit,
                query.offset,
                query.with_payload,
                query.with_vector,
            )
            .await
            .map_err(|e| CoreError::VectorError(e.to_string()))
    }
}