velesdb_server/handlers/graph/

handlers.rs

//! Graph HTTP handlers for VelesDB REST API.
//!
//! All graph operations are routed through `AppState.db.get_graph_collection()`.
//! No separate GraphService state — graph data persists via GraphCollection/GraphEngine.
//!
//! Extended handlers (parity endpoints) live in [`super::handlers_extended`].

use std::sync::Arc;

use axum::{
    extract::{Path, Query, State},
    http::StatusCode,
    response::IntoResponse,
    Json,
};
use velesdb_core::collection::graph::{GraphEdge, TraversalConfig};
use velesdb_core::observer::QueryOperationKind;

use crate::handlers::helpers::{auto_core_error_response, run_blocking, run_blocking_typed};
use crate::types::ErrorResponse;
use crate::AppState;

use super::types::{
    AddEdgeRequest, AddEdgesBatchRequest, AddEdgesBatchResponse, DegreeResponse, EdgeQueryParams,
    EdgeResponse, EdgesResponse, TraversalStats, TraverseRequest, TraverseResponse,
};

/// Shared graph preamble: record metric and resolve collection.
///
/// Mirrors [`super::super::search::search_preamble`] for graph handlers.
/// `GraphCollection` does not expose guard rails, so only the metrics
/// recording and collection resolution steps are performed.
#[allow(clippy::result_large_err)]
pub(super) fn graph_preamble(
    state: &AppState,
    name: &str,
) -> Result<velesdb_core::GraphCollection, (StatusCode, Json<ErrorResponse>)> {
    state.onboarding_metrics.record_graph_request();
    get_graph_collection_or_404(state, name)
}

/// Resolves a `GraphCollection` by name.
///
/// # Returns
///
/// * `Ok(collection)` if a graph collection with this name exists.
/// * `Err(404 Not Found)` if no collection with this name exists.
/// * `Err(409 Conflict)` if a collection exists with this name but is not
///   a graph collection (type mismatch with vector or metadata collection).
///
/// # Contract
///
/// This function previously auto-created a schemaless graph collection
/// on first use. That behaviour is retired (F-05): a missing graph
/// collection now yields a 404 response instead of being created
/// silently. Callers must issue `POST /collections` with
/// `collection_type = "graph"` before targeting graph endpoints.
pub(super) fn get_graph_collection_or_404(
    state: &AppState,
    name: &str,
) -> Result<velesdb_core::GraphCollection, (StatusCode, Json<ErrorResponse>)> {
    if let Some(c) = state.db.get_graph_collection(name) {
        return Ok(c);
    }

    // Check if a non-graph collection exists with this name (type mismatch → 409).
    if state.db.get_vector_collection(name).is_some()
        || state.db.get_metadata_collection(name).is_some()
    {
        return Err((
            StatusCode::CONFLICT,
            Json(ErrorResponse {
                error: format!(
                    "Collection '{name}' exists but is not a graph collection. \
                     Use /collections/{name}/graph only on graph-typed collections.",
                ),
                code: None,
            }),
        ));
    }

    // PR #586 Devin fix: propagate `VELES-002 CollectionNotFound` so
    // typed-error clients surface `CollectionNotFoundError` instead of
    // a status-derived `'NOT_FOUND'` string. The "create it first"
    // hint stays in the message for human operators.
    let err = velesdb_core::Error::CollectionNotFound(name.to_string());
    Err((
        StatusCode::NOT_FOUND,
        Json(ErrorResponse {
            error: format!(
                "{err}. Create it first with \
                 POST /collections and collection_type = \"graph\".",
            ),
            code: Some(err.code().to_string()),
        }),
    ))
}

/// Shared graph preamble for **read** operations: resolves the collection via
/// [`graph_preamble`], then routes the read through the control-plane gate
/// (CORE-2). Graph reads have no metadata-filter channel to narrow, so a
/// denied or scope-narrowed decision refuses the request (fail closed)
/// rather than running it unfiltered.
///
/// Mirrors the gate `MATCH` (`handlers::match_query`) and embedding
/// [`super::handlers_extended::graph_search`] already apply — centralized
/// here so every plain REST graph read is governed the same way instead of
/// each handler wiring the check individually.
#[allow(clippy::result_large_err)]
pub(super) fn graph_read_preamble(
    state: &AppState,
    name: &str,
    operation: QueryOperationKind,
) -> Result<velesdb_core::GraphCollection, (StatusCode, Json<ErrorResponse>)> {
    let coll = graph_preamble(state, name)?;
    match state.db.authorize_read(name, operation, None, None) {
        Ok(None) => Ok(coll),
        Ok(Some(_)) | Err(_) => Err((
            StatusCode::FORBIDDEN,
            Json(ErrorResponse {
                error: "Read denied by governance policy".to_string(),
                code: None,
            }),
        )),
    }
}

/// Get edges from a collection's graph filtered by label.
#[utoipa::path(
    get,
    path = "/collections/{name}/graph/edges",
    params(("name" = String, Path, description = "Collection name"), EdgeQueryParams),
    responses(
        (status = 200, description = "Edges retrieved successfully", body = EdgesResponse),
        (status = 400, description = "Missing required 'label' query parameter", body = ErrorResponse),
        (status = 404, description = "Collection not found", body = ErrorResponse),
        (status = 500, description = "Internal server error", body = ErrorResponse)
    ),
    tag = "graph"
)]
pub async fn get_edges(
    Path(name): Path<String>,
    Query(params): Query<EdgeQueryParams>,
    State(state): State<Arc<AppState>>,
) -> Result<Json<EdgesResponse>, (StatusCode, Json<ErrorResponse>)> {
    let label = params.label.ok_or_else(|| {
        (
            StatusCode::BAD_REQUEST,
            Json(ErrorResponse {
                error: "Query parameter 'label' is required. Listing all edges requires pagination (not yet implemented).".to_string(),
                code: None,
            }),
        )
    })?;

    let coll = graph_read_preamble(&state, &name, QueryOperationKind::GraphTraversal)?;

    // Edge listing takes graph store locks — run it on the blocking pool.
    let raw_edges = run_blocking_typed(move || coll.get_edges(Some(&label))).await?;

    let edges: Vec<EdgeResponse> = raw_edges
        .into_iter()
        .map(|e| EdgeResponse {
            id: e.id(),
            source: e.source(),
            target: e.target(),
            label: e.label().to_string(),
            properties: serde_json::to_value(e.properties()).unwrap_or_default(),
        })
        .collect();

    let count = edges.len();
    Ok(Json(EdgesResponse { edges, count }))
}

/// The write-result match [`add_edge`] and [`add_edges_batch`] share: the
/// blocking-pool outcome routed through `auto_core_error_response` (so e.g.
/// `EdgeExists` surfaces as 409 + VELES-019, never a 500 string), success
/// shaped by the caller. One copy, so the two handlers cannot drift on the
/// error route.
fn created_or_core_error<T>(
    outcome: Result<Result<T, velesdb_core::Error>, axum::response::Response>,
    success: impl FnOnce(T) -> axum::response::Response,
) -> axum::response::Response {
    match outcome {
        Ok(Ok(value)) => success(value),
        Ok(Err(e)) => auto_core_error_response(&e),
        Err(resp) => resp,
    }
}

/// Add an edge to a collection's graph.
#[utoipa::path(
    post,
    path = "/collections/{name}/graph/edges",
    request_body = AddEdgeRequest,
    responses(
        (status = 201, description = "Edge added successfully"),
        (status = 400, description = "Invalid request", body = ErrorResponse),
        (status = 404, description = "Collection not found, or source/target node has no stored payload (VELES-022 NodeNotFound)", body = ErrorResponse),
        (status = 500, description = "Internal server error", body = ErrorResponse),
        (status = 422, response = crate::types::MalformedBody)
    ),
    tag = "graph"
)]
pub async fn add_edge(
    Path(name): Path<String>,
    State(state): State<Arc<AppState>>,
    Json(request): Json<AddEdgeRequest>,
) -> axum::response::Response {
    let edge = match build_edge(request) {
        Ok(e) => e,
        Err(resp) => return resp.into_response(),
    };

    let coll = match graph_preamble(&state, &name) {
        Ok(c) => c,
        Err(resp) => return resp.into_response(),
    };

    // Edge insertion takes write locks and persists — run it on the blocking
    // pool. Route the core error through `auto_core_error_response` so e.g.
    // `EdgeExists` surfaces as 409 + VELES-019 instead of a generic 500 string.
    created_or_core_error(run_blocking(move || coll.add_edge(edge)).await, |()| {
        StatusCode::CREATED.into_response()
    })
}

/// Converts an [`AddEdgeRequest`] into a core [`GraphEdge`], validating the
/// properties shape and edge fields. Shared by [`add_edge`] and
/// [`add_edges_batch`].
#[allow(clippy::result_large_err)]
fn build_edge(request: AddEdgeRequest) -> Result<GraphEdge, (StatusCode, Json<ErrorResponse>)> {
    let properties: std::collections::HashMap<String, serde_json::Value> = match request.properties
    {
        serde_json::Value::Object(map) => map.into_iter().collect(),
        serde_json::Value::Null => std::collections::HashMap::new(),
        _ => {
            return Err((
                StatusCode::BAD_REQUEST,
                Json(ErrorResponse {
                    error: "Properties must be an object or null".to_string(),
                    code: None,
                }),
            ));
        }
    };

    let edge = GraphEdge::new(request.id, request.source, request.target, &request.label)
        .map_err(|e| {
            (
                StatusCode::BAD_REQUEST,
                Json(ErrorResponse {
                    error: format!("Invalid edge: {e}"),
                    code: None,
                }),
            )
        })?
        .with_properties(properties);
    Ok(edge)
}

/// Add multiple edges to a collection's graph in one batched operation.
#[utoipa::path(
    post,
    path = "/collections/{name}/graph/edges/batch",
    request_body = AddEdgesBatchRequest,
    responses(
        (status = 201, description = "Edges added successfully", body = AddEdgesBatchResponse),
        (status = 400, description = "Invalid request", body = ErrorResponse),
        (status = 404, description = "Collection not found, or a source/target node has no stored payload (VELES-022 NodeNotFound) — the whole batch is rejected", body = ErrorResponse),
        (status = 500, description = "Internal server error", body = ErrorResponse),
        (status = 422, response = crate::types::MalformedBody)
    ),
    tag = "graph"
)]
pub async fn add_edges_batch(
    Path(name): Path<String>,
    State(state): State<Arc<AppState>>,
    Json(request): Json<AddEdgesBatchRequest>,
) -> axum::response::Response {
    let edges = match request
        .edges
        .into_iter()
        .map(build_edge)
        .collect::<Result<Vec<_>, _>>()
    {
        Ok(edges) => edges,
        Err(resp) => return resp.into_response(),
    };

    let coll = match graph_preamble(&state, &name) {
        Ok(c) => c,
        Err(resp) => return resp.into_response(),
    };

    // Batch edge insertion takes write locks and persists — run it on the
    // blocking pool. Route the core error through `auto_core_error_response`
    // so e.g. `EdgeExists` surfaces as 409 + VELES-019 instead of a 500 string.
    created_or_core_error(
        run_blocking(move || coll.add_edges_batch(edges)).await,
        |added| (StatusCode::CREATED, Json(AddEdgesBatchResponse { added })).into_response(),
    )
}

/// Traverse the graph using BFS or DFS from a source node.
#[utoipa::path(
    post,
    path = "/collections/{name}/graph/traverse",
    request_body = TraverseRequest,
    responses(
        (status = 200, description = "Traversal completed successfully", body = TraverseResponse),
        (status = 400, description = "Invalid request", body = ErrorResponse),
        (status = 404, description = "Collection not found", body = ErrorResponse),
        (status = 500, description = "Internal server error", body = ErrorResponse),
        (status = 422, response = crate::types::MalformedBody)
    ),
    tag = "graph"
)]
pub async fn traverse_graph(
    Path(name): Path<String>,
    State(state): State<Arc<AppState>>,
    Json(request): Json<TraverseRequest>,
) -> Result<Json<TraverseResponse>, (StatusCode, Json<ErrorResponse>)> {
    let coll = graph_read_preamble(&state, &name, QueryOperationKind::GraphTraversal)?;

    let use_bfs = match request.strategy.to_lowercase().as_str() {
        "bfs" => true,
        "dfs" => false,
        _ => {
            return Err((
                StatusCode::BAD_REQUEST,
                Json(ErrorResponse {
                    error: format!(
                        "Invalid strategy '{}'. Use 'bfs' or 'dfs'.",
                        request.strategy
                    ),
                    code: None,
                }),
            ));
        }
    };

    let limit = request.limit;
    let source = request.source;
    let config = TraversalConfig::with_range(1, request.max_depth)
        .with_limit(limit)
        .with_rel_types(request.rel_types);

    // Traversal is synchronous, lock-taking core code — run it on the
    // blocking pool so the async workers stay responsive.
    let raw_results = run_blocking_typed(move || {
        if use_bfs {
            coll.traverse_bfs(source, &config)
        } else {
            coll.traverse_dfs(source, &config)
        }
    })
    .await?;

    let results: Vec<super::types::TraversalResultItem> = raw_results
        .into_iter()
        .map(|r| super::types::TraversalResultItem {
            target_id: r.target_id,
            depth: r.depth,
            path: r.path,
        })
        .collect();

    let depth_reached = results.iter().map(|r| r.depth).max().unwrap_or(0);
    let visited = results.len();
    let has_more = visited >= limit;

    Ok(Json(TraverseResponse {
        results,
        has_more,
        stats: TraversalStats {
            visited,
            depth_reached,
        },
    }))
}

/// Get the degree (in and out) of a specific node.
#[utoipa::path(
    get,
    path = "/collections/{name}/graph/nodes/{node_id}/degree",
    params(
        ("name" = String, Path, description = "Collection name"),
        ("node_id" = String, Path, description = "Node ID (u64 as a string; precision-safe above 2^53-1)", pattern = "^[0-9]+$")
    ),
    responses(
        (status = 200, description = "Degree retrieved successfully", body = DegreeResponse),
        (status = 404, description = "Collection not found", body = ErrorResponse),
        (status = 500, description = "Internal server error", body = ErrorResponse)
    ),
    tag = "graph"
)]
pub async fn get_node_degree(
    Path((name, node_id)): Path<(String, u64)>,
    State(state): State<Arc<AppState>>,
) -> Result<Json<DegreeResponse>, (StatusCode, Json<ErrorResponse>)> {
    let coll = graph_read_preamble(&state, &name, QueryOperationKind::GraphTraversal)?;
    // Degree lookup takes edge-store shard locks — run it on the blocking pool.
    let (in_degree, out_degree) = run_blocking_typed(move || coll.node_degree(node_id)).await?;
    Ok(Json(DegreeResponse {
        in_degree,
        out_degree,
    }))
}