velesdb_mobile/

graph.rs

//! Graph bindings for VelesDB Mobile (UniFFI).
//!
//! Provides UniFFI bindings for graph operations on iOS and Android.

use std::collections::HashMap;
use std::sync::Arc;

use parking_lot::RwLock;

/// A graph node for knowledge graph construction.
#[derive(Debug, Clone, uniffi::Record, serde::Serialize, serde::Deserialize)]
pub struct MobileGraphNode {
    /// Unique identifier.
    pub id: u64,
    /// Node type/label.
    pub label: String,
    /// JSON properties as string.
    pub properties_json: Option<String>,
    /// Optional vector embedding.
    pub vector: Option<Vec<f32>>,
}

/// A graph edge representing a relationship.
#[derive(Debug, Clone, uniffi::Record, serde::Serialize, serde::Deserialize)]
pub struct MobileGraphEdge {
    /// Unique identifier.
    pub id: u64,
    /// Source node ID.
    pub source: u64,
    /// Target node ID.
    pub target: u64,
    /// Relationship type.
    pub label: String,
    /// JSON properties as string.
    pub properties_json: Option<String>,
}

/// Traversal result from BFS/DFS.
///
/// FFI projection of [`velesdb_core::TraversalResult`]. The mobile field
/// `node_id` corresponds to core's `target_id` (the node reached); `path`
/// and `depth` mirror core's fields one-for-one. See the `From` impls below
/// for the canonical mapping — they make any future core field drift a
/// compile error rather than silent divergence.
#[derive(Debug, Clone, uniffi::Record)]
pub struct TraversalResult {
    /// Target node ID reached (core: `target_id`).
    pub node_id: u64,
    /// Edge IDs along the path from the source to this node (core: `path`).
    pub path: Vec<u64>,
    /// Depth from source (number of hops).
    pub depth: u32,
}

/// Serializes a core property map to the mobile `properties_json` shape.
///
/// Returns `None` for an empty map (no properties) and a JSON object string
/// otherwise. A serialization failure also yields `None` so the projection is
/// total (FFI conversions cannot return a `Result`).
fn properties_to_json(
    properties: &std::collections::HashMap<String, serde_json::Value>,
) -> Option<String> {
    if properties.is_empty() {
        return None;
    }
    serde_json::to_string(properties).ok()
}

impl From<velesdb_core::GraphNode> for MobileGraphNode {
    fn from(node: velesdb_core::GraphNode) -> Self {
        Self {
            id: node.id(),
            label: node.label().to_string(),
            properties_json: properties_to_json(node.properties()),
            vector: node.vector().cloned(),
        }
    }
}

impl From<velesdb_core::GraphEdge> for MobileGraphEdge {
    fn from(edge: velesdb_core::GraphEdge) -> Self {
        Self {
            id: edge.id(),
            source: edge.source(),
            target: edge.target(),
            label: edge.label().to_string(),
            properties_json: properties_to_json(edge.properties()),
        }
    }
}

impl From<velesdb_core::TraversalResult> for TraversalResult {
    fn from(result: velesdb_core::TraversalResult) -> Self {
        Self {
            node_id: result.target_id,
            path: result.path,
            depth: result.depth,
        }
    }
}

/// In-memory graph store for mobile knowledge graphs.
///
/// Nodes and edges are held in RAM for fast in-session traversal and are not
/// written automatically. Call [`save`](Self::save) to persist a snapshot to
/// disk and [`load`](Self::load) to restore it across app restarts.
#[derive(uniffi::Object)]
pub struct MobileGraphStore {
    nodes: RwLock<HashMap<u64, MobileGraphNode>>,
    edges: RwLock<HashMap<u64, MobileGraphEdge>>,
    outgoing: RwLock<HashMap<u64, Vec<u64>>>,
    incoming: RwLock<HashMap<u64, Vec<u64>>>,
}

/// On-disk snapshot of a [`MobileGraphStore`] (JSON). The outgoing/incoming
/// adjacency is rebuilt from `edges` on load, so it is not stored.
#[derive(serde::Serialize, serde::Deserialize)]
struct GraphSnapshot {
    nodes: Vec<MobileGraphNode>,
    edges: Vec<MobileGraphEdge>,
}

#[uniffi::export]
impl MobileGraphStore {
    /// Creates a new empty graph store.
    #[uniffi::constructor]
    pub fn new() -> Arc<Self> {
        Arc::new(Self {
            nodes: RwLock::new(HashMap::new()),
            edges: RwLock::new(HashMap::new()),
            outgoing: RwLock::new(HashMap::new()),
            incoming: RwLock::new(HashMap::new()),
        })
    }

    /// Persists the current nodes and edges to `path` as JSON so the graph
    /// survives an app restart (the store is otherwise in-memory only).
    ///
    /// # Lock Order
    ///
    /// Acquires and RELEASES each read guard in turn — `edges` first, then
    /// `nodes` — so no two of this store's locks are ever held at once. The
    /// earlier form built the snapshot in one struct literal, whose temporary
    /// guards both lived to the end of the statement, taking `nodes` then
    /// `edges` — the exact reverse of the `edges → outgoing → incoming → nodes`
    /// order every mutator uses (`add_edge`, `remove_node`, `clear`). A
    /// concurrent `save`/`remove_node` from two UniFFI-called threads was a
    /// textbook ABBA deadlock; holding at most one lock here cannot deadlock
    /// against any acquisition order.
    pub fn save(&self, path: String) -> Result<(), crate::VelesError> {
        let edges: Vec<MobileGraphEdge> = self.edges.read().values().cloned().collect();
        let nodes: Vec<MobileGraphNode> = self.nodes.read().values().cloned().collect();
        let snapshot = GraphSnapshot { nodes, edges };
        let bytes = serde_json::to_vec(&snapshot)
            .map_err(|e| crate::VelesError::database(format!("Graph serialize failed: {e}")))?;
        std::fs::write(&path, bytes)
            .map_err(|e| crate::VelesError::database(format!("Graph save to '{path}' failed: {e}")))
    }

    /// Loads a graph previously written by [`save`](Self::save) from `path`,
    /// rebuilding the adjacency from the stored edges.
    #[uniffi::constructor]
    pub fn load(path: String) -> Result<Arc<Self>, crate::VelesError> {
        let bytes = std::fs::read(&path).map_err(|e| {
            crate::VelesError::database(format!("Graph load from '{path}' failed: {e}"))
        })?;
        let snapshot: GraphSnapshot = serde_json::from_slice(&bytes)
            .map_err(|e| crate::VelesError::database(format!("Graph deserialize failed: {e}")))?;
        let store = Self::new();
        for node in snapshot.nodes {
            store.add_node(node);
        }
        for edge in snapshot.edges {
            store.add_edge(edge)?;
        }
        Ok(store)
    }

    /// Adds a node to the graph.
    pub fn add_node(&self, node: MobileGraphNode) {
        let mut nodes = self.nodes.write();
        nodes.insert(node.id, node);
    }

    /// Adds an edge to the graph.
    ///
    /// # Lock Order
    ///
    /// Acquires locks in consistent order: edges → outgoing → incoming
    /// WITHOUT dropping between operations to ensure atomicity.
    /// This prevents race conditions with concurrent remove_node() calls.
    pub fn add_edge(&self, edge: MobileGraphEdge) -> Result<(), crate::VelesError> {
        // CRITICAL FIX: Acquire all locks BEFORE any mutation
        // and hold them until the operation is complete.
        // Lock order: edges → outgoing → incoming (consistent with remove_node)
        let mut edges = self.edges.write();
        let mut outgoing = self.outgoing.write();
        let mut incoming = self.incoming.write();

        if edges.contains_key(&edge.id) {
            return Err(crate::VelesError::database(format!(
                "Edge with ID {} already exists",
                edge.id
            )));
        }

        let source = edge.source;
        let target = edge.target;
        let id = edge.id;

        // All mutations happen while holding all locks
        edges.insert(id, edge);
        outgoing.entry(source).or_default().push(id);
        incoming.entry(target).or_default().push(id);

        // Locks are released here (all at once) when guards go out of scope
        Ok(())
    }

    /// Gets a node by ID.
    pub fn get_node(&self, id: u64) -> Option<MobileGraphNode> {
        let nodes = self.nodes.read();
        nodes.get(&id).cloned()
    }

    /// Gets an edge by ID.
    pub fn get_edge(&self, id: u64) -> Option<MobileGraphEdge> {
        let edges = self.edges.read();
        edges.get(&id).cloned()
    }

    /// Returns the number of nodes.
    pub fn node_count(&self) -> u64 {
        let nodes = self.nodes.read();
        nodes.len() as u64
    }

    /// Returns the number of edges.
    pub fn edge_count(&self) -> u64 {
        let edges = self.edges.read();
        edges.len() as u64
    }

    /// Gets outgoing edges from a node.
    ///
    /// # Lock Order
    ///
    /// Acquires locks in consistent order: edges → outgoing
    /// to prevent ABBA deadlock with write operations.
    pub fn get_outgoing(&self, node_id: u64) -> Vec<MobileGraphEdge> {
        self.get_edges_from_index(node_id, &self.outgoing)
    }

    /// Gets incoming edges to a node.
    ///
    /// # Lock Order
    ///
    /// Acquires locks in consistent order: edges → incoming
    /// to prevent ABBA deadlock with write operations.
    pub fn get_incoming(&self, node_id: u64) -> Vec<MobileGraphEdge> {
        self.get_edges_from_index(node_id, &self.incoming)
    }

    /// Gets outgoing edges filtered by label.
    pub fn get_outgoing_by_label(&self, node_id: u64, label: String) -> Vec<MobileGraphEdge> {
        self.get_outgoing(node_id)
            .into_iter()
            .filter(|e| e.label == label)
            .collect()
    }

    /// Gets neighbors reachable from a node (1-hop).
    pub fn get_neighbors(&self, node_id: u64) -> Vec<u64> {
        self.get_outgoing(node_id)
            .into_iter()
            .map(|e| e.target)
            .collect()
    }

    /// Performs BFS traversal from a source node.
    ///
    /// # Arguments
    ///
    /// * `source_id` - Starting node ID
    /// * `max_depth` - Maximum traversal depth
    /// * `limit` - Maximum number of results
    pub fn bfs_traverse(&self, source_id: u64, max_depth: u32, limit: u32) -> Vec<TraversalResult> {
        self.bfs_traverse_parallel(vec![source_id], max_depth, limit)
    }

    /// Performs multi-source BFS traversal with deduplication.
    ///
    /// Starts BFS from multiple source nodes simultaneously and deduplicates
    /// results by target node ID (first-seen wins).
    ///
    /// # Arguments
    ///
    /// * `source_ids` - Starting node IDs
    /// * `max_depth` - Maximum traversal depth
    /// * `limit` - Maximum number of results
    pub fn bfs_traverse_parallel(
        &self,
        source_ids: Vec<u64>,
        max_depth: u32,
        limit: u32,
    ) -> Vec<TraversalResult> {
        use std::collections::{HashSet, VecDeque};

        let mut results: Vec<TraversalResult> = Vec::new();
        let mut visited: HashSet<u64> = HashSet::new();
        let mut queue: VecDeque<(u64, u32, Vec<u64>)> = VecDeque::new();

        for &source_id in &source_ids {
            if visited.insert(source_id) {
                queue.push_back((source_id, 0, Vec::new()));
            }
        }

        while let Some((node_id, depth, path)) = queue.pop_front() {
            if results.len() >= limit as usize {
                break;
            }

            if depth > 0 {
                results.push(TraversalResult {
                    node_id,
                    path: path.clone(),
                    depth,
                });
            }

            self.enqueue_neighbors(node_id, depth, max_depth, &path, &mut visited, &mut queue);
        }

        results
    }

    /// Removes a node and all connected edges.
    ///
    /// # Lock Order
    ///
    /// Acquires locks in consistent order: edges → outgoing → incoming → nodes
    /// to prevent deadlock with concurrent add_edge() calls.
    pub fn remove_node(&self, node_id: u64) {
        // CRITICAL: Acquire locks in consistent order (edges → outgoing → incoming → nodes)
        // to prevent deadlock with add_edge() which uses (edges → outgoing → incoming)
        let mut edges = self.edges.write();
        let mut outgoing = self.outgoing.write();
        let mut incoming = self.incoming.write();
        let mut nodes = self.nodes.write();

        nodes.remove(&node_id);

        let outgoing_ids: Vec<u64> = outgoing.remove(&node_id).unwrap_or_default();
        for edge_id in outgoing_ids {
            if let Some(edge) = edges.remove(&edge_id) {
                if let Some(ids) = incoming.get_mut(&edge.target) {
                    ids.retain(|&id| id != edge_id);
                }
            }
        }

        let incoming_ids: Vec<u64> = incoming.remove(&node_id).unwrap_or_default();
        for edge_id in incoming_ids {
            if let Some(edge) = edges.remove(&edge_id) {
                if let Some(ids) = outgoing.get_mut(&edge.source) {
                    ids.retain(|&id| id != edge_id);
                }
            }
        }
    }

    /// Removes an edge by ID.
    ///
    /// # Lock Order
    ///
    /// Acquires locks in consistent order: edges → outgoing → incoming
    /// WITHOUT dropping between operations to ensure atomicity.
    pub fn remove_edge(&self, edge_id: u64) {
        // CRITICAL FIX: Acquire all locks BEFORE any mutation
        let mut edges = self.edges.write();
        let mut outgoing = self.outgoing.write();
        let mut incoming = self.incoming.write();

        if let Some(edge) = edges.remove(&edge_id) {
            if let Some(ids) = outgoing.get_mut(&edge.source) {
                ids.retain(|&id| id != edge_id);
            }
            if let Some(ids) = incoming.get_mut(&edge.target) {
                ids.retain(|&id| id != edge_id);
            }
        }
        // All locks released here
    }

    /// Clears all nodes and edges.
    ///
    /// # Lock Order
    ///
    /// Acquires locks in consistent order: edges → outgoing → incoming → nodes
    pub fn clear(&self) {
        // Consistent lock order: edges → outgoing → incoming → nodes
        let mut edges = self.edges.write();
        let mut outgoing = self.outgoing.write();
        let mut incoming = self.incoming.write();
        let mut nodes = self.nodes.write();

        edges.clear();
        outgoing.clear();
        incoming.clear();
        nodes.clear();
    }

    /// Performs DFS traversal from a source node.
    ///
    /// # Arguments
    ///
    /// * `source_id` - Starting node ID
    /// * `max_depth` - Maximum traversal depth
    /// * `limit` - Maximum number of results
    pub fn dfs_traverse(&self, source_id: u64, max_depth: u32, limit: u32) -> Vec<TraversalResult> {
        use std::collections::HashSet;

        let mut results: Vec<TraversalResult> = Vec::new();
        let mut visited: HashSet<u64> = HashSet::new();
        let mut stack: Vec<(u64, u32, Vec<u64>)> = vec![(source_id, 0, Vec::new())];

        while let Some((node_id, depth, path)) = stack.pop() {
            if results.len() >= limit as usize {
                break;
            }

            if visited.contains(&node_id) {
                continue;
            }
            visited.insert(node_id);

            if depth > 0 {
                results.push(TraversalResult {
                    node_id,
                    path: path.clone(),
                    depth,
                });
            }

            if depth < max_depth {
                let neighbors: Vec<_> = self
                    .get_outgoing(node_id)
                    .into_iter()
                    .filter(|e| !visited.contains(&e.target))
                    .collect();

                for edge in neighbors.into_iter().rev() {
                    let mut next_path = path.clone();
                    next_path.push(edge.id);
                    stack.push((edge.target, depth + 1, next_path));
                }
            }
        }

        results
    }

    /// Checks if a node exists.
    pub fn has_node(&self, id: u64) -> bool {
        let nodes = self.nodes.read();
        nodes.contains_key(&id)
    }

    /// Checks if an edge exists.
    pub fn has_edge(&self, id: u64) -> bool {
        let edges = self.edges.read();
        edges.contains_key(&id)
    }

    /// Gets the out-degree (number of outgoing edges) of a node.
    #[allow(clippy::cast_possible_truncation)]
    pub fn out_degree(&self, node_id: u64) -> u32 {
        let outgoing = self.outgoing.read();
        // Safe: graph degree unlikely to exceed u32::MAX (4 billion edges from one node)
        outgoing.get(&node_id).map_or(0, |v| v.len() as u32)
    }

    /// Gets the in-degree (number of incoming edges) of a node.
    #[allow(clippy::cast_possible_truncation)]
    pub fn in_degree(&self, node_id: u64) -> u32 {
        let incoming = self.incoming.read();
        // Safe: graph degree unlikely to exceed u32::MAX (4 billion edges to one node)
        incoming.get(&node_id).map_or(0, |v| v.len() as u32)
    }

    /// Gets all nodes with a specific label.
    pub fn get_nodes_by_label(&self, label: String) -> Vec<MobileGraphNode> {
        let nodes = self.nodes.read();
        nodes
            .values()
            .filter(|n| n.label == label)
            .cloned()
            .collect()
    }

    /// Gets all edges with a specific label.
    pub fn get_edges_by_label(&self, label: String) -> Vec<MobileGraphEdge> {
        let edges = self.edges.read();
        edges
            .values()
            .filter(|e| e.label == label)
            .cloned()
            .collect()
    }
}

/// Internal helpers (not exposed via UniFFI).
impl MobileGraphStore {
    /// Resolves edge IDs from an adjacency index to full edge objects.
    ///
    /// # Lock Order
    ///
    /// Acquires `edges` read-lock first, then the `index` read-lock, matching
    /// the write-side lock order (edges -> outgoing -> incoming).
    fn get_edges_from_index(
        &self,
        node_id: u64,
        index: &RwLock<HashMap<u64, Vec<u64>>>,
    ) -> Vec<MobileGraphEdge> {
        let edges = self.edges.read();
        let idx = index.read();
        idx.get(&node_id)
            .map(|ids| ids.iter().filter_map(|id| edges.get(id).cloned()).collect())
            .unwrap_or_default()
    }

    /// Enqueues unvisited outgoing neighbors of `node_id` for further traversal.
    ///
    /// Each enqueued entry carries the edge-ID path taken to reach the neighbor
    /// (`path` so far plus the traversed edge), mirroring core's
    /// `TraversalResult::path`. No-op when `depth` has already reached
    /// `max_depth`.
    fn enqueue_neighbors(
        &self,
        node_id: u64,
        depth: u32,
        max_depth: u32,
        path: &[u64],
        visited: &mut std::collections::HashSet<u64>,
        queue: &mut std::collections::VecDeque<(u64, u32, Vec<u64>)>,
    ) {
        if depth >= max_depth {
            return;
        }
        for edge in self.get_outgoing(node_id) {
            if visited.insert(edge.target) {
                let mut next_path = path.to_vec();
                next_path.push(edge.id);
                queue.push_back((edge.target, depth + 1, next_path));
            }
        }
    }
}

impl Default for MobileGraphStore {
    fn default() -> Self {
        Self {
            nodes: RwLock::new(HashMap::new()),
            edges: RwLock::new(HashMap::new()),
            outgoing: RwLock::new(HashMap::new()),
            incoming: RwLock::new(HashMap::new()),
        }
    }
}

#[cfg(test)]
#[path = "graph_tests.rs"]
mod tests;