linkrs-query 0.1.0

//! Implementation of graph-related functions
//!
//! Provide functions for manipulating vertices and edges, including id, tags, labels, properties, type, src, dst, and rank.
//! Also includes graph traversal functions: neighbors, degree, shortest_path.

use crate::executor::expression::evaluation_context::graph_storage::GraphStorageRef;
use crate::executor::expression::ExpressionError;
use linkrs_core::types::VertexId;
use linkrs_core::value::list::List;
use linkrs_core::value::NullType;
use linkrs_core::Value;

/// Graph function enumeration
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum GraphFunction {
    Id,
    Tags,
    Labels,
    Properties,
    EdgeType,
    Src,
    Dst,
    Rank,
    StartNode,
    EndNode,
    Neighbors,
    Degree,
    OutEdges,
    InEdges,
    ShortestPath,
    Bfs,
    ConnectedComponents,
    VariableLengthPath,
    PageRank,
}

impl GraphFunction {
    /// Obtain the function name
    pub fn name(&self) -> &str {
        match self {
            Self::Id => "id",
            Self::Tags => "tags",
            Self::Labels => "labels",
            Self::Properties => "properties",
            Self::EdgeType => "type",
            Self::Src => "src",
            Self::Dst => "dst",
            Self::Rank => "rank",
            Self::StartNode => "startnode",
            Self::EndNode => "endnode",
            Self::Neighbors => "neighbors",
            Self::Degree => "degree",
            Self::OutEdges => "out_edges",
            Self::InEdges => "in_edges",
            Self::ShortestPath => "shortest_path",
            Self::Bfs => "bfs",
            Self::ConnectedComponents => "connected_components",
            Self::VariableLengthPath => "variable_length_path",
            Self::PageRank => "pagerank",
        }
    }

    /// Determine the number of parameters
    pub fn arity(&self) -> usize {
        match self {
            Self::Id => 1,
            Self::Tags => 1,
            Self::Labels => 1,
            Self::Properties => 1,
            Self::EdgeType => 1,
            Self::Src => 1,
            Self::Dst => 1,
            Self::Rank => 1,
            Self::StartNode => 1,
            Self::EndNode => 1,
            Self::Neighbors => 1,
            Self::Degree => 1,
            Self::OutEdges => 1,
            Self::InEdges => 1,
            Self::ShortestPath => 2,
            Self::Bfs => 2,
            Self::ConnectedComponents => 0,
            Self::VariableLengthPath => 3,
            Self::PageRank => 0,
        }
    }

    /// Is it a function with variable parameters?
    pub fn is_variadic(&self) -> bool {
        matches!(self, Self::VariableLengthPath)
    }

    /// Obtain the function description
    pub fn description(&self) -> &str {
        match self {
            Self::Id => "Get the ID of the vertex",
            Self::Tags => "Get all labels of a vertex",
            Self::Labels => "Get all labels (aliases) of a vertex",
            Self::Properties => "Get all attributes of a vertex or edge",
            Self::EdgeType => "Get the type of the edge",
            Self::Src => "Get the starting vertex ID of the edge",
            Self::Dst => "Get the target vertex ID of the edge",
            Self::Rank => "Get the edge's rank value",
            Self::StartNode => "Get the starting vertex of the edge",
            Self::EndNode => "Get the target vertex of the edge",
            Self::Neighbors => "Get all neighbor vertex IDs of a vertex",
            Self::Degree => "Get the degree (number of edges) of a vertex",
            Self::OutEdges => "Get all outgoing edges of a vertex",
            Self::InEdges => "Get all incoming edges of a vertex",
            Self::ShortestPath => "Find the shortest path between two vertices (returns path length)",
            Self::Bfs => "Breadth-first search traversal (start_vid, max_depth)",
            Self::ConnectedComponents => "Find all connected components in the graph",
            Self::VariableLengthPath => "Find all paths between two vertices within a depth range (start_vid, end_vid, max_depth) or (start_vid, end_vid, min_depth, max_depth)",
            Self::PageRank => "Calculate PageRank scores for all vertices in the graph",
        }
    }

    pub fn execute(&self, args: &[Value]) -> Result<Value, ExpressionError> {
        if !self.is_variadic() && args.len() != self.arity() {
            return Err(ExpressionError::invalid_arity(
                self.name(),
                self.arity(),
                args.len(),
            ));
        }
        match self {
            Self::Id => execute_id(args),
            Self::Tags => execute_tags(args),
            Self::Labels => execute_labels(args),
            Self::Properties => execute_properties(args),
            Self::EdgeType => execute_edge_type(args),
            Self::Src => execute_src(args),
            Self::Dst => execute_dst(args),
            Self::Rank => execute_rank(args),
            Self::StartNode => execute_startnode(args),
            Self::EndNode => execute_endnode(args),
            Self::Neighbors => execute_neighbors(args),
            Self::Degree => execute_degree(args),
            Self::OutEdges => execute_out_edges(args),
            Self::InEdges => execute_in_edges(args),
            Self::ShortestPath => execute_shortest_path(args),
            Self::Bfs => execute_bfs(args),
            Self::ConnectedComponents => execute_connected_components(args),
            Self::VariableLengthPath => execute_variable_length_path(args),
            Self::PageRank => execute_pagerank(args),
        }
    }

    /// Execute with graph storage access.
    /// Falls back to execute() if no results can be obtained from storage.
    pub fn execute_with_storage(
        &self,
        args: &[Value],
        storage: &GraphStorageRef,
    ) -> Result<Value, ExpressionError> {
        if !self.is_variadic() && args.len() != self.arity() {
            return Err(ExpressionError::invalid_arity(
                self.name(),
                self.arity(),
                args.len(),
            ));
        }
        match self {
            Self::StartNode => execute_startnode_with_storage(args, storage),
            Self::EndNode => execute_endnode_with_storage(args, storage),
            Self::Neighbors => execute_neighbors_with_storage(args, storage),
            Self::Degree => execute_degree_with_storage(args, storage),
            Self::OutEdges => execute_out_edges_with_storage(args, storage),
            Self::InEdges => execute_in_edges_with_storage(args, storage),
            Self::ShortestPath => execute_shortest_path_with_storage(args, storage),
            Self::Bfs => execute_bfs_with_storage(args, storage),
            Self::ConnectedComponents => execute_connected_components_with_storage(args, storage),
            Self::VariableLengthPath => execute_variable_length_path_with_storage(args, storage),
            Self::PageRank => execute_pagerank_with_storage(args, storage),
            _ => self.execute(args),
        }
    }
}

fn execute_id(args: &[Value]) -> Result<Value, ExpressionError> {
    match &args[0] {
        Value::Vertex(v) => Ok(Value::from(v.vid)),
        Value::Null(_) => Ok(Value::Null(NullType::Null)),
        _ => Err(ExpressionError::type_error(
            "The id function requires a vertex type",
        )),
    }
}

fn execute_tags(args: &[Value]) -> Result<Value, ExpressionError> {
    match &args[0] {
        Value::Vertex(v) => {
            let tags: Vec<Value> = vec![Value::string(v.tag.name.clone())];
            Ok(Value::list(List { values: tags }))
        }
        Value::Null(_) => Ok(Value::Null(NullType::Null)),
        _ => Err(ExpressionError::type_error("tags requires vertex type")),
    }
}

fn execute_labels(args: &[Value]) -> Result<Value, ExpressionError> {
    execute_tags(args)
}

fn execute_properties(args: &[Value]) -> Result<Value, ExpressionError> {
    match &args[0] {
        Value::Vertex(v) => {
            let mut props = std::collections::HashMap::new();
            props.extend(
                v.properties()
                    .iter()
                    .map(|(k, v)| (Value::string(k.clone()), v.clone())),
            );
            Ok(Value::map(props))
        }
        Value::Edge(e) => Ok(Value::string_map(e.props.clone())),
        Value::EdgeHeader(_) => Err(ExpressionError::type_error(
            "properties requires a materialized edge: the edge identity reference carries no properties",
        )),
        Value::Map(m) => Ok(Value::map((**m).clone())),
        Value::Null(_) => Ok(Value::Null(NullType::Null)),
        _ => Err(ExpressionError::type_error(
            "properties requires vertex, edge or map type",
        )),
    }
}

fn execute_edge_type(args: &[Value]) -> Result<Value, ExpressionError> {
    match &args[0] {
        Value::Edge(e) => Ok(Value::string(e.edge_type.clone())),
        Value::EdgeHeader(h) => Ok(Value::string(h.edge_type.clone())),
        Value::Null(_) => Ok(Value::Null(NullType::Null)),
        _ => Err(ExpressionError::type_error(
            "The type function requires an edge type",
        )),
    }
}

fn execute_src(args: &[Value]) -> Result<Value, ExpressionError> {
    match &args[0] {
        Value::Edge(e) => Ok(Value::from(e.src)),
        Value::EdgeHeader(h) => Ok(Value::from(h.src)),
        Value::Null(_) => Ok(Value::Null(NullType::Null)),
        _ => Err(ExpressionError::type_error(
            "The src function requires the edge type",
        )),
    }
}

fn execute_dst(args: &[Value]) -> Result<Value, ExpressionError> {
    match &args[0] {
        Value::Edge(e) => Ok(Value::from(e.dst)),
        Value::EdgeHeader(h) => Ok(Value::from(h.dst)),
        Value::Null(_) => Ok(Value::Null(NullType::Null)),
        _ => Err(ExpressionError::type_error(
            "The dst function requires an edge type",
        )),
    }
}

fn execute_rank(args: &[Value]) -> Result<Value, ExpressionError> {
    match &args[0] {
        Value::Edge(e) => Ok(Value::BigInt(e.ranking)),
        Value::EdgeHeader(h) => Ok(Value::BigInt(h.ranking)),
        Value::Null(_) => Ok(Value::Null(NullType::Null)),
        _ => Err(ExpressionError::type_error(
            "The rank function requires an edge type",
        )),
    }
}

fn execute_startnode(args: &[Value]) -> Result<Value, ExpressionError> {
    match &args[0] {
        Value::Edge(_) | Value::EdgeHeader(_) => Err(ExpressionError::function_error(
            "startnode() requires graph storage access to resolve the endpoint label; use within a query context",
        )),
        Value::Null(_) => Ok(Value::Null(NullType::Null)),
        _ => Err(ExpressionError::type_error(
            "The startnode function requires the edge type",
        )),
    }
}

fn execute_endnode(args: &[Value]) -> Result<Value, ExpressionError> {
    match &args[0] {
        Value::Edge(_) | Value::EdgeHeader(_) => Err(ExpressionError::function_error(
            "endnode() requires graph storage access to resolve the endpoint label; use within a query context",
        )),
        Value::Null(_) => Ok(Value::Null(NullType::Null)),
        _ => Err(ExpressionError::type_error(
            "The endnode function requires an edge type",
        )),
    }
}

/// Resolve one edge endpoint to a tagged vertex through the edge type
/// schema. Returns Null when the endpoint vertex no longer exists; errors
/// when the edge type or its endpoint labels are unknown.
fn endpoint_vertex_with_storage(
    edge_type: &str,
    vid: VertexId,
    endpoint_src: bool,
    storage: &GraphStorageRef,
) -> Result<Value, ExpressionError> {
    let reader = storage.storage.read();
    let info = reader
        .get_edge_type(&storage.space, edge_type)
        .map_err(|e| ExpressionError::function_error(format!("Storage error: {}", e)))?
        .ok_or_else(|| {
            ExpressionError::function_error(format!("Edge type '{edge_type}' not found"))
        })?;
    let tag = if endpoint_src {
        &info.src_tag_name
    } else {
        &info.dst_tag_name
    };
    if tag.is_empty() {
        return Err(ExpressionError::function_error(format!(
            "Edge type '{edge_type}' declares no endpoint label"
        )));
    }
    let vertex = reader
        .get_vertex(&storage.space, tag, &vid)
        .map_err(|e| ExpressionError::function_error(format!("Storage error: {}", e)))?;
    match vertex {
        Some(vertex) => Ok(Value::Vertex(Box::new(vertex))),
        None => Ok(Value::Null(NullType::Null)),
    }
}

fn execute_startnode_with_storage(
    args: &[Value],
    storage: &GraphStorageRef,
) -> Result<Value, ExpressionError> {
    match &args[0] {
        Value::Edge(e) => endpoint_vertex_with_storage(&e.edge_type, e.src, true, storage),
        Value::EdgeHeader(h) => endpoint_vertex_with_storage(&h.edge_type, h.src, true, storage),
        Value::Null(_) => Ok(Value::Null(NullType::Null)),
        _ => Err(ExpressionError::type_error(
            "The startnode function requires the edge type",
        )),
    }
}

fn execute_endnode_with_storage(
    args: &[Value],
    storage: &GraphStorageRef,
) -> Result<Value, ExpressionError> {
    match &args[0] {
        Value::Edge(e) => endpoint_vertex_with_storage(&e.edge_type, e.dst, false, storage),
        Value::EdgeHeader(h) => endpoint_vertex_with_storage(&h.edge_type, h.dst, false, storage),
        Value::Null(_) => Ok(Value::Null(NullType::Null)),
        _ => Err(ExpressionError::type_error(
            "The endnode function requires an edge type",
        )),
    }
}

fn extract_vertex_id(value: &Value) -> Result<VertexId, ExpressionError> {
    match value {
        Value::Vertex(v) => Ok(v.vid),
        Value::BigInt(id) => {
            VertexId::try_from_int64(*id).map_err(|e| ExpressionError::type_error(e.to_string()))
        }
        Value::Int(id) => VertexId::try_from_int64(*id as i64)
            .map_err(|e| ExpressionError::type_error(e.to_string())),
        Value::Null(_) => Err(ExpressionError::type_error(
            "Expected a vertex or vertex ID, got null",
        )),
        _ => Err(ExpressionError::type_error(
            "Expected a vertex or vertex ID",
        )),
    }
}

fn execute_neighbors(args: &[Value]) -> Result<Value, ExpressionError> {
    match &args[0] {
        Value::Vertex(v) => Ok(Value::from(v.vid)),
        Value::Null(_) => Ok(Value::Null(NullType::Null)),
        _ => Err(ExpressionError::type_error(
            "The neighbors function requires a vertex type",
        )),
    }
}

fn execute_degree(args: &[Value]) -> Result<Value, ExpressionError> {
    match &args[0] {
        Value::Vertex(v) => {
            let degree = v
                .properties()
                .keys()
                .filter(|k| k.starts_with("neighbor_"))
                .count();
            Ok(Value::BigInt(degree as i64))
        }
        Value::Null(_) => Ok(Value::Null(NullType::Null)),
        _ => Err(ExpressionError::type_error(
            "The degree function requires a vertex type",
        )),
    }
}

fn execute_out_edges(args: &[Value]) -> Result<Value, ExpressionError> {
    match &args[0] {
        Value::Vertex(v) => {
            let edges: Vec<Value> = v
                .properties()
                .iter()
                .filter(|(k, _)| k.starts_with("out_"))
                .map(|(_, v)| v.clone())
                .collect();
            Ok(Value::list(List { values: edges }))
        }
        Value::Null(_) => Ok(Value::Null(NullType::Null)),
        _ => Err(ExpressionError::type_error(
            "The out_edges function requires a vertex type",
        )),
    }
}

fn execute_in_edges(args: &[Value]) -> Result<Value, ExpressionError> {
    match &args[0] {
        Value::Vertex(v) => {
            let edges: Vec<Value> = v
                .properties()
                .iter()
                .filter(|(k, _)| k.starts_with("in_"))
                .map(|(_, v)| v.clone())
                .collect();
            Ok(Value::list(List { values: edges }))
        }
        Value::Null(_) => Ok(Value::Null(NullType::Null)),
        _ => Err(ExpressionError::type_error(
            "The in_edges function requires a vertex type",
        )),
    }
}

fn execute_shortest_path(args: &[Value]) -> Result<Value, ExpressionError> {
    let start_vid = match &args[0] {
        Value::BigInt(id) => *id,
        Value::Int(id) => *id as i64,
        _ => {
            return Err(ExpressionError::type_error(
                "shortest_path requires integer vertex IDs",
            ))
        }
    };
    let end_vid = match &args[1] {
        Value::BigInt(id) => *id,
        Value::Int(id) => *id as i64,
        _ => {
            return Err(ExpressionError::type_error(
                "shortest_path requires integer vertex IDs",
            ))
        }
    };
    if start_vid < 0 || end_vid < 0 {
        return Err(ExpressionError::type_error(
            "shortest_path requires non-negative integer vertex IDs",
        ));
    }
    if start_vid == end_vid {
        return Ok(Value::BigInt(0));
    }
    Ok(Value::BigInt(-1))
}

fn execute_bfs(_args: &[Value]) -> Result<Value, ExpressionError> {
    Err(ExpressionError::function_error(
        "bfs() requires graph storage access; use within a query context".to_string(),
    ))
}

fn execute_connected_components(_args: &[Value]) -> Result<Value, ExpressionError> {
    Err(ExpressionError::function_error(
        "connected_components() requires graph storage access; use within a query context"
            .to_string(),
    ))
}

// --- Storage-backed implementations ---

fn execute_neighbors_with_storage(
    args: &[Value],
    storage: &GraphStorageRef,
) -> Result<Value, ExpressionError> {
    let vid = extract_vertex_id(&args[0])?;
    let neighbors = storage
        .get_neighbors(&vid)
        .map_err(ExpressionError::function_error)?;
    let neighbor_ids: Vec<Value> = neighbors
        .into_iter()
        .map(|(nid, _)| Value::from(nid))
        .collect();
    Ok(Value::list(List {
        values: neighbor_ids,
    }))
}

fn execute_degree_with_storage(
    args: &[Value],
    storage: &GraphStorageRef,
) -> Result<Value, ExpressionError> {
    let vid = extract_vertex_id(&args[0])?;
    let neighbors = storage
        .get_neighbors(&vid)
        .map_err(ExpressionError::function_error)?;
    Ok(Value::BigInt(neighbors.len() as i64))
}

fn execute_out_edges_with_storage(
    args: &[Value],
    storage: &GraphStorageRef,
) -> Result<Value, ExpressionError> {
    use linkrs_core::types::EdgeDirection;
    let vid = extract_vertex_id(&args[0])?;
    let reader = storage.storage.read();
    let edges = reader
        .get_node_edges(&storage.space, &vid, EdgeDirection::Out, &[])
        .map_err(|e| ExpressionError::function_error(format!("Storage error: {}", e)))?;
    drop(reader);
    let edge_values: Vec<Value> = edges
        .into_iter()
        .map(|e| Value::Edge(Box::new(e)))
        .collect();
    Ok(Value::list(List {
        values: edge_values,
    }))
}

fn execute_in_edges_with_storage(
    args: &[Value],
    storage: &GraphStorageRef,
) -> Result<Value, ExpressionError> {
    use linkrs_core::types::EdgeDirection;
    let vid = extract_vertex_id(&args[0])?;
    let reader = storage.storage.read();
    let edges = reader
        .get_node_edges(&storage.space, &vid, EdgeDirection::In, &[])
        .map_err(|e| ExpressionError::function_error(format!("Storage error: {}", e)))?;
    drop(reader);
    let edge_values: Vec<Value> = edges
        .into_iter()
        .map(|e| Value::Edge(Box::new(e)))
        .collect();
    Ok(Value::list(List {
        values: edge_values,
    }))
}

fn execute_shortest_path_with_storage(
    args: &[Value],
    storage: &GraphStorageRef,
) -> Result<Value, ExpressionError> {
    let start_vid = extract_vertex_id(&args[0])?;
    let end_vid = extract_vertex_id(&args[1])?;

    if start_vid == end_vid {
        return Ok(Value::BigInt(0));
    }

    // Simple BFS to find shortest path length
    use std::collections::{HashMap, VecDeque};
    let mut visited: HashMap<VertexId, i64> = HashMap::new();
    let mut queue: VecDeque<VertexId> = VecDeque::new();

    visited.insert(start_vid, 0);
    queue.push_back(start_vid);

    while let Some(current) = queue.pop_front() {
        let distance = visited[&current];

        let neighbors = storage
            .get_neighbors(&current)
            .map_err(ExpressionError::function_error)?;

        for (neighbor_id, _) in neighbors {
            if neighbor_id == end_vid {
                return Ok(Value::BigInt(distance + 1));
            }
            if let std::collections::hash_map::Entry::Vacant(e) = visited.entry(neighbor_id) {
                e.insert(distance + 1);
                queue.push_back(neighbor_id);
            }
        }
    }

    Ok(Value::BigInt(-1))
}

fn execute_bfs_with_storage(
    args: &[Value],
    storage: &GraphStorageRef,
) -> Result<Value, ExpressionError> {
    let start_vid = extract_vertex_id(&args[0])?;
    let max_depth = match &args[1] {
        Value::BigInt(d) => *d,
        Value::Int(d) => *d as i64,
        _ => {
            return Err(ExpressionError::type_error(
                "bfs max_depth must be an integer",
            ))
        }
    };

    use std::collections::{HashSet, VecDeque};
    let mut visited: HashSet<VertexId> = HashSet::new();
    let mut queue: VecDeque<(VertexId, i64)> = VecDeque::new();
    let mut result: Vec<Value> = Vec::new();

    visited.insert(start_vid);
    queue.push_back((start_vid, 0));
    result.push(Value::from(start_vid));

    while let Some((current, depth)) = queue.pop_front() {
        if depth >= max_depth {
            continue;
        }

        let neighbors = storage
            .get_neighbors(&current)
            .map_err(ExpressionError::function_error)?;

        for (neighbor_id, _) in neighbors {
            if visited.insert(neighbor_id) {
                queue.push_back((neighbor_id, depth + 1));
                result.push(Value::from(neighbor_id));
            }
        }
    }

    Ok(Value::list(List { values: result }))
}

fn execute_connected_components_with_storage(
    _args: &[Value],
    storage: &GraphStorageRef,
) -> Result<Value, ExpressionError> {
    // Get all vertices from storage
    let reader = storage.storage.read();
    let all_vertices = reader
        .scan_vertices(&storage.space)
        .map_err(|e| ExpressionError::function_error(format!("Storage error: {}", e)))?;
    drop(reader);

    use std::collections::{HashSet, VecDeque};

    let all_vids: HashSet<VertexId> = all_vertices.iter().map(|v| v.vid).collect();
    let mut visited: HashSet<VertexId> = HashSet::new();
    let mut components: Vec<Vec<Value>> = Vec::new();

    for start_vid in &all_vids {
        if visited.contains(start_vid) {
            continue;
        }

        let mut component: Vec<Value> = Vec::new();
        let mut queue: VecDeque<VertexId> = VecDeque::new();
        visited.insert(*start_vid);
        queue.push_back(*start_vid);
        component.push(Value::from(*start_vid));

        while let Some(current) = queue.pop_front() {
            let neighbors = match storage.get_neighbors(&current) {
                Ok(n) => n,
                Err(_) => continue,
            };
            for (neighbor_id, _) in neighbors {
                if visited.insert(neighbor_id) {
                    queue.push_back(neighbor_id);
                    component.push(Value::from(neighbor_id));
                }
            }
        }

        components.push(component);
    }

    let component_lists: Vec<Value> = components
        .into_iter()
        .map(|comp| Value::list(List { values: comp }))
        .collect();

    Ok(Value::list(List {
        values: component_lists,
    }))
}

fn execute_variable_length_path(_args: &[Value]) -> Result<Value, ExpressionError> {
    Err(ExpressionError::type_error(
        "variable_length_path() requires graph storage access; use within a query context",
    ))
}

fn execute_pagerank(_args: &[Value]) -> Result<Value, ExpressionError> {
    Err(ExpressionError::function_error(
        "pagerank() requires graph storage access; use within a query context".to_string(),
    ))
}

fn execute_variable_length_path_with_storage(
    args: &[Value],
    storage: &GraphStorageRef,
) -> Result<Value, ExpressionError> {
    let (start_vid, end_vid, min_depth, max_depth) = match args.len() {
        3 => {
            let sv = extract_vertex_id(&args[0])?;
            let ev = extract_vertex_id(&args[1])?;
            let md = match &args[2] {
                Value::BigInt(d) => *d,
                Value::Int(d) => *d as i64,
                _ => {
                    return Err(ExpressionError::type_error(
                        "variable_length_path max_depth must be an integer",
                    ))
                }
            };
            (sv, ev, 1i64, md)
        }
        4 => {
            let sv = extract_vertex_id(&args[0])?;
            let ev = extract_vertex_id(&args[1])?;
            let mind = match &args[2] {
                Value::BigInt(d) => *d,
                Value::Int(d) => *d as i64,
                _ => {
                    return Err(ExpressionError::type_error(
                        "variable_length_path min_depth must be an integer",
                    ))
                }
            };
            let maxd = match &args[3] {
                Value::BigInt(d) => *d,
                Value::Int(d) => *d as i64,
                _ => {
                    return Err(ExpressionError::type_error(
                        "variable_length_path max_depth must be an integer",
                    ))
                }
            };
            (sv, ev, mind, maxd)
        }
        _ => {
            return Err(ExpressionError::type_error(
                "variable_length_path takes 3 or 4 arguments (start_vid, end_vid, max_depth) or (start_vid, end_vid, min_depth, max_depth)",
            ))
        }
    };

    if min_depth < 1 || max_depth < min_depth {
        return Err(ExpressionError::type_error(
            "variable_length_path: min_depth must be >= 1 and max_depth must be >= min_depth",
        ));
    }

    if start_vid == end_vid && min_depth <= 0 {
        return Ok(Value::list(List {
            values: vec![Value::list(List {
                values: vec![Value::from(start_vid)],
            })],
        }));
    }

    // DFS with depth limit to find all paths from start_vid to end_vid
    // within the depth range [min_depth, max_depth]
    use std::collections::VecDeque;
    // Each element: (current_vertex, path_so_far, depth)
    // path_so_far is the list of vertex IDs visited (excluding current)
    let mut results: Vec<Vec<Value>> = Vec::new();
    let mut queue: VecDeque<(VertexId, Vec<Value>, i64)> = VecDeque::new();

    queue.push_back((start_vid, vec![Value::from(start_vid)], 0));

    while let Some((current, path, depth)) = queue.pop_front() {
        if depth >= max_depth {
            continue;
        }

        let neighbors = match storage.get_neighbors(&current) {
            Ok(n) => n,
            Err(_) => continue,
        };

        for (neighbor_id, _) in neighbors {
            // Avoid cycles - skip if neighbor already in path
            let neighbor_val = Value::from(neighbor_id);
            if path.contains(&neighbor_val) {
                continue;
            }

            let new_depth = depth + 1;
            let mut new_path = path.clone();
            new_path.push(neighbor_val.clone());

            if neighbor_id == end_vid && new_depth >= min_depth {
                results.push(new_path.clone());
            }

            // Limit total results to avoid excessive memory usage
            if results.len() >= 1000 {
                break;
            }

            queue.push_back((neighbor_id, new_path, new_depth));
        }

        if results.len() >= 1000 {
            break;
        }
    }

    let path_values: Vec<Value> = results
        .into_iter()
        .map(|path| Value::list(List { values: path }))
        .collect();

    Ok(Value::list(List {
        values: path_values,
    }))
}

fn execute_pagerank_with_storage(
    _args: &[Value],
    storage: &GraphStorageRef,
) -> Result<Value, ExpressionError> {
    let reader = storage.storage.read();
    let all_vertices = reader
        .scan_vertices(&storage.space)
        .map_err(|e| ExpressionError::function_error(format!("Storage error: {}", e)))?;
    drop(reader);

    let vertex_count = all_vertices.len();
    if vertex_count == 0 {
        return Ok(Value::map(std::collections::HashMap::new()));
    }

    use std::collections::HashMap;

    let vertex_ids: Vec<VertexId> = all_vertices.iter().map(|v| v.vid).collect();
    let mut vid_to_index: HashMap<VertexId, usize> = HashMap::new();
    for (i, vid) in vertex_ids.iter().enumerate() {
        vid_to_index.insert(*vid, i);
    }

    let damping_factor: f64 = 0.85;
    let max_iterations: usize = 100;
    let tolerance: f64 = 1e-6;

    let mut scores: Vec<f64> = vec![1.0 / vertex_count as f64; vertex_count];
    let mut out_degrees: Vec<usize> = vec![0; vertex_count];

    let mut adjacency: Vec<Vec<usize>> = vec![Vec::new(); vertex_count];

    for vid in &vertex_ids {
        let reader = storage.storage.read();
        let edges = reader
            .get_node_edges(
                &storage.space,
                vid,
                linkrs_core::types::EdgeDirection::Out,
                &[],
            )
            .map_err(|e| ExpressionError::function_error(format!("Storage error: {}", e)))?;
        drop(reader);

        let src_index = vid_to_index[vid];
        out_degrees[src_index] = edges.len();

        for edge in edges {
            if let Some(&dst_index) = vid_to_index.get(&edge.dst) {
                adjacency[src_index].push(dst_index);
            }
        }
    }

    for _ in 0..max_iterations {
        let mut new_scores: Vec<f64> = vec![0.0; vertex_count];
        let mut dangling_sum: f64 = 0.0;

        for i in 0..vertex_count {
            if out_degrees[i] == 0 {
                dangling_sum += scores[i];
            }
        }

        let base = (1.0 - damping_factor) / vertex_count as f64
            + damping_factor * dangling_sum / vertex_count as f64;
        for score in new_scores.iter_mut() {
            *score = base;
        }

        for i in 0..vertex_count {
            if out_degrees[i] > 0 {
                let share = scores[i] / out_degrees[i] as f64;
                for &j in &adjacency[i] {
                    new_scores[j] += damping_factor * share;
                }
            }
        }

        let mut diff: f64 = 0.0;
        for i in 0..vertex_count {
            diff += (new_scores[i] - scores[i]).abs();
        }

        scores = new_scores;

        if diff < tolerance {
            break;
        }
    }

    let mut result: HashMap<String, Value> = HashMap::new();
    for (i, vid) in vertex_ids.iter().enumerate() {
        let vid_str = vid.to_string();
        result.insert(vid_str, Value::Double(scores[i]));
    }

    Ok(Value::string_map(result))
}

#[cfg(test)]
mod tests {
    use super::*;
    use linkrs_core::types::VertexId;
    use linkrs_core::vertex_edge_path::{Edge, Tag, Vertex};
    use std::collections::HashMap;

    fn create_test_vertex() -> Vertex {
        let mut props = HashMap::from([
            ("name".into(), Value::string("Alice")),
            ("age".into(), Value::Int(30)),
        ]);
        props.insert("dept".into(), Value::string("Engineering"));
        let tag = Tag::new("person".into(), props);
        Vertex::new(VertexId::try_from_int64(1).expect("valid vertex id"), tag)
    }

    fn create_test_edge() -> Edge {
        Edge::new(
            VertexId::try_from_int64(1).expect("valid vertex id"),
            VertexId::try_from_int64(2).expect("valid vertex id"),
            "knows".to_string(),
            0,
            HashMap::from([("since".into(), Value::Int(2020))]),
        )
    }

    #[test]
    fn test_id_function() {
        let vertex = create_test_vertex();
        let result = GraphFunction::Id
            .execute(&[Value::Vertex(Box::new(vertex))])
            .expect("The id function execution should succeed");
        assert_eq!(result, Value::BigInt(1));
    }

    #[test]
    fn test_tags_function() {
        let vertex = create_test_vertex();
        let result = GraphFunction::Tags
            .execute(&[Value::Vertex(Box::new(vertex))])
            .expect("The tags function should be executed successfully");
        if let Value::List(tags) = result {
            assert_eq!(tags.values.len(), 1);
        } else {
            panic!("The `tags` function should return a list.");
        }
    }

    #[test]
    fn test_properties_vertex() {
        let vertex = create_test_vertex();
        let result = GraphFunction::Properties
            .execute(&[Value::Vertex(Box::new(vertex))])
            .expect("The properties function should execute successfully");
        if let Value::Map(props) = result {
            assert!(props.contains_key(&Value::string("name")));
            assert!(props.contains_key(&Value::string("age")));
            assert!(props.contains_key(&Value::string("dept")));
        } else {
            panic!("The `properties` function should return a map.");
        }
    }

    #[test]
    fn test_type_function() {
        let edge = create_test_edge();
        let result = GraphFunction::EdgeType
            .execute(&[Value::Edge(Box::new(edge))])
            .expect("The type function execution should succeed");
        assert_eq!(result, Value::string("knows"));
    }

    #[test]
    fn test_src_function() {
        let edge = create_test_edge();
        let result = GraphFunction::Src
            .execute(&[Value::Edge(Box::new(edge))])
            .expect("The src function execution should succeed");
        assert_eq!(result, Value::BigInt(1));
    }

    #[test]
    fn test_dst_function() {
        let edge = create_test_edge();
        let result = GraphFunction::Dst
            .execute(&[Value::Edge(Box::new(edge))])
            .expect("The dst function execution should succeed");
        assert_eq!(result, Value::BigInt(2));
    }

    #[test]
    fn test_rank_function() {
        let edge = create_test_edge();
        let result = GraphFunction::Rank
            .execute(&[Value::Edge(Box::new(edge))])
            .expect("The execution of the rank function should succeed");
        assert_eq!(result, Value::BigInt(0));
    }

    #[test]
    fn test_null_handling() {
        let null_value = Value::Null(NullType::Null);

        assert_eq!(
            GraphFunction::Id
                .execute(std::slice::from_ref(&null_value))
                .expect("The id function should handle NULL"),
            Value::Null(NullType::Null)
        );
        assert_eq!(
            GraphFunction::Tags
                .execute(std::slice::from_ref(&null_value))
                .expect("The tags function should handle NULL"),
            Value::Null(NullType::Null)
        );
        assert_eq!(
            GraphFunction::Properties
                .execute(std::slice::from_ref(&null_value))
                .expect("The properties function should handle NULL."),
            Value::Null(NullType::Null)
        );
    }
}