feat: Neo4j slow query logger

Cargo.toml

@@ -32,3 +32,5 @@ thiserror = "2.0.17"
 tokio = { version = "1.49.0", features = ["full"] }
 tokio-shared-rt = "0.1"
 tracing = "0.1.44"
+tracing-test = "0.2"
+futures = "0.3"

examples/api/api-config.toml

@@ -18,4 +18,10 @@ redis = "redis://127.0.0.1:6379"
 uri = "bolt://localhost:7687"
 # Not needed in the Community Edition
 #user = "neo4j"
-password = "12345678"
+password = "12345678"
+# Queries taking longer than this (ms) will be logged as warnings
+slow_query_logging_threshold_ms = 100
+# Enable or disable slow query logging
+#slow_query_logging_enabled = true
+# Include the full cypher (with interpolated params) in slow-query warnings
+#slow_query_logging_include_cypher = false

examples/watcher/watcher-config.toml

@@ -31,3 +31,9 @@ uri = "bolt://localhost:7687"
 # Not needed in the Community Edition
 #user = "neo4j"
 password = "12345678"
+# Queries taking longer than this (ms) will be logged as warnings
+slow_query_logging_threshold_ms = 100
+# Enable or disable slow query logging
+#slow_query_logging_enabled = true
+# Include the full cypher (with interpolated params) in slow-query warnings
+#slow_query_logging_include_cypher = false

nexus-common/Cargo.toml

@@ -11,6 +11,7 @@ license = "MIT"
 async-trait = { workspace = true }
 dirs = "6.0.0"
 chrono = { workspace = true }
+futures = { workspace = true }
 neo4rs = { workspace = true }
 opentelemetry = { workspace = true }
 opentelemetry-appender-tracing = "0.31.1"
@@ -33,3 +34,4 @@ utoipa = "5.4.0"
 [dev-dependencies]
 tempfile = { workspace = true }
 tokio-shared-rt = { workspace = true }
+tracing-test = { workspace = true }

nexus-common/default.config.toml

@@ -51,3 +51,9 @@ uri = "bolt://localhost:7687"
 # Not needed in the Community Edition the profile username, just the password
 #user = "neo4j"
 password = "12345678"
+# Queries taking longer than this (ms) will be logged as warnings
+slow_query_logging_threshold_ms = 100
+# Enable or disable slow query logging
+slow_query_logging_enabled = true
+# Include the Cypher query text in slow query log entries
+#slow_query_logging_include_cypher = false

nexus-common/src/db/config/mod.rs

@@ -2,7 +2,7 @@ use serde::{Deserialize, Serialize};
 use std::fmt::Debug;
 
 mod neo4j;
-pub use neo4j::Neo4JConfig;
+pub use neo4j::{Neo4JConfig, DEFAULT_SLOW_QUERY_THRESHOLD_MS};
 
 pub const REDIS_URI: &str = "redis://localhost:6379";
 

nexus-common/src/db/config/neo4j.rs

@@ -3,26 +3,55 @@ use serde::{Deserialize, Serialize};
 pub const NEO4J_URI: &str = "bolt://localhost:7687";
 pub const NEO4J_USER: &str = "neo4j";
 pub const NEO4J_PASS: &str = "12345678";
+pub const DEFAULT_SLOW_QUERY_THRESHOLD_MS: u64 = 100;
 
 // Create temporal struct to wrap database config
 #[derive(Debug, Clone, Serialize, Deserialize, Eq, PartialEq)]
 pub struct Neo4JConfig {
     pub uri: String,
+
     #[serde(default = "default_neo4j_user")]
     pub user: String,
+
     pub password: String,
+
+    /// Queries exceeding this threshold (in milliseconds) are logged as warnings.
+    /// Only used when `slow_query_logging_enabled` is true.
+    #[serde(default = "default_slow_query_logging_threshold_ms")]
+    pub slow_query_logging_threshold_ms: u64,
+
+    /// Enable slow-query logging. Defaults to true.
+    /// Set to false for CLI/admin commands where tracing overhead is unnecessary.
+    #[serde(default = "default_slow_query_logging_enabled")]
+    pub slow_query_logging_enabled: bool,
+
+    /// Include the full cypher (with interpolated params) in slow-query warnings.
+    /// Useful for debugging but verbose. Defaults to false.
+    #[serde(default)]
+    pub slow_query_logging_include_cypher: bool,
 }
 
 fn default_neo4j_user() -> String {
     String::from("neo4j")
 }
 
+fn default_slow_query_logging_threshold_ms() -> u64 {
+    DEFAULT_SLOW_QUERY_THRESHOLD_MS
+}
+
+fn default_slow_query_logging_enabled() -> bool {
+    true
+}
+
 impl Default for Neo4JConfig {
     fn default() -> Self {
         Self {
             uri: String::from(NEO4J_URI),
             user: String::from(NEO4J_USER),
             password: String::from(NEO4J_PASS),
+            slow_query_logging_threshold_ms: DEFAULT_SLOW_QUERY_THRESHOLD_MS,
+            slow_query_logging_enabled: true,
+            slow_query_logging_include_cypher: false,
         }
     }
 }

nexus-common/src/db/connectors/neo4j.rs

@@ -1,26 +1,23 @@
-use neo4rs::{query, Graph};
+use crate::db::graph::Query;
 use std::fmt;
-use std::sync::OnceLock;
+use std::sync::{Arc, OnceLock};
+use std::time::Duration;
 use tracing::{debug, info};
 
 use crate::db::graph::error::{GraphError, GraphResult};
+use crate::db::graph::{Graph, GraphOps, TracedGraph};
 use crate::db::setup::setup_graph;
 use crate::db::Neo4JConfig;
 use crate::types::DynError;
 
 pub struct Neo4jConnector {
-    pub graph: Graph,
+    graph: Arc<dyn GraphOps>,
 }
 
 impl Neo4jConnector {
     /// Initialize and register the global Neo4j connector and verify connectivity
     pub async fn init(neo4j_config: &Neo4JConfig) -> Result<(), DynError> {
-        let neo4j_connector = Neo4jConnector::new_connection(
-            &neo4j_config.uri,
-            &neo4j_config.user,
-            &neo4j_config.password,
-        )
-        .await?;
+        let neo4j_connector = Neo4jConnector::new_connection(neo4j_config).await?;
 
         neo4j_connector.ping(&neo4j_config.uri).await?;
 
@@ -35,17 +32,28 @@ impl Neo4jConnector {
     }
 
     /// Create and return a new connector after defining a database connection
-    async fn new_connection(uri: &str, user: &str, password: &str) -> GraphResult<Self> {
-        let graph = Graph::new(uri, user, password).await?;
-        let neo4j_connector = Neo4jConnector { graph };
-        info!("Created Neo4j connector");
+    async fn new_connection(config: &Neo4JConfig) -> GraphResult<Self> {
+        let neo4j_graph = neo4rs::Graph::new(&config.uri, &config.user, &config.password).await?;
+        let graph = Graph::new(neo4j_graph);
+
+        let graph: Arc<dyn GraphOps> = if config.slow_query_logging_enabled {
+            let threshold = Duration::from_millis(config.slow_query_logging_threshold_ms);
+            Arc::new(
+                TracedGraph::new(graph)
+                    .with_slow_query_threshold(threshold)
+                    .with_log_cypher(config.slow_query_logging_include_cypher),
+            )
+        } else {
+            Arc::new(graph)
+        };
 
-        Ok(neo4j_connector)
+        info!("Created Neo4j connector");
+        Ok(Neo4jConnector { graph })
     }
 
     /// Perform a health-check PING over the Bolt protocol to the Neo4j server
     async fn ping(&self, neo4j_uri: &str) -> Result<(), DynError> {
-        if let Err(neo4j_err) = self.graph.execute(query("RETURN 1")).await {
+        if let Err(neo4j_err) = self.graph.run(Query::new("ping", "RETURN 1")).await {
             return Err(format!("Failed to PING to Neo4j at {neo4j_uri}, {neo4j_err}").into());
         }
 
@@ -57,13 +65,13 @@ impl Neo4jConnector {
 impl fmt::Debug for Neo4jConnector {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         f.debug_struct("Neo4jConnector")
-            .field("graph", &"Graph instance")
+            .field("graph", &"GraphOps instance")
             .finish()
     }
 }
 
 /// Helper to retrieve a Neo4j graph connection.
-pub fn get_neo4j_graph() -> GraphResult<Graph> {
+pub fn get_neo4j_graph() -> GraphResult<Arc<dyn GraphOps>> {
     NEO4J_CONNECTOR
         .get()
         .ok_or(GraphError::ConnectionNotInitialized)

nexus-common/src/db/graph/exec.rs

@@ -1,5 +1,7 @@
+use super::query::Query;
 use crate::db::{get_neo4j_graph, graph::error::GraphResult};
-use neo4rs::{Query, Row};
+use futures::TryStreamExt;
+use neo4rs::Row;
 use serde::de::DeserializeOwned;
 
 /// Represents the outcome of a mutation-like query in the graph database.
@@ -37,7 +39,7 @@ pub async fn execute_graph_operation(query: Query) -> GraphResult<OperationOutco
 pub async fn exec_single_row(query: Query) -> GraphResult<()> {
     let graph = get_neo4j_graph()?;
     let mut result = graph.execute(query).await?;
-    result.next().await?;
+    result.try_next().await?;
     Ok(())
 }
 
@@ -46,20 +48,15 @@ pub async fn fetch_row_from_graph(query: Query) -> GraphResult<Option<Row>> {
 
     let mut result = graph.execute(query).await?;
 
-    result.next().await.map_err(Into::into)
+    result.try_next().await.map_err(Into::into)
 }
 
 pub async fn fetch_all_rows_from_graph(query: Query) -> GraphResult<Vec<Row>> {
     let graph = get_neo4j_graph()?;
 
-    let mut result = graph.execute(query).await?;
-    let mut rows = Vec::new();
-
-    while let Some(row) = result.next().await? {
-        rows.push(row);
-    }
+    let result = graph.execute(query).await?;
 
-    Ok(rows)
+    result.try_collect().await.map_err(Into::into)
 }
 
 /// Fetch the value of type T mapped to a specific key from the first row of a graph query's result

nexus-common/src/db/graph/mod.rs

@@ -1,6 +1,11 @@
 pub mod error;
 pub mod exec;
 pub mod queries;
+mod query;
 pub mod setup;
+mod traced;
 
 pub use error::{GraphError, GraphResult};
+pub use query::Query;
+pub use traced::GraphOps;
+pub(crate) use traced::{Graph, TracedGraph};

nexus-common/src/db/graph/queries/del.rs

@@ -1,10 +1,11 @@
-use neo4rs::{query, Query};
+use crate::db::graph::Query;
 
 /// Deletes a user node and all its relationships
 /// # Arguments
 /// * `user_id` - The unique identifier of the user to be deleted
 pub fn delete_user(user_id: &str) -> Query {
-    query(
+    Query::new(
+        "delete_user",
         "MATCH (u:User {id: $id})
          DETACH DELETE u;",
     )
@@ -16,7 +17,8 @@ pub fn delete_user(user_id: &str) -> Query {
 /// * `author_id` - The unique identifier of the user who authored the post.
 /// * `post_id` - The unique identifier of the post to be deleted.
 pub fn delete_post(author_id: &str, post_id: &str) -> Query {
-    query(
+    Query::new(
+        "delete_post",
         "MATCH (u:User {id: $author_id})-[:AUTHORED]->(p:Post {id: $post_id})
          DETACH DELETE p;",
     )
@@ -29,11 +31,12 @@ pub fn delete_post(author_id: &str, post_id: &str) -> Query {
 /// * `follower_id` - The unique identifier of the user who is following another user.
 /// * `followee_id` - The unique identifier of the user being followed
 pub fn delete_follow(follower_id: &str, followee_id: &str) -> Query {
-    query(
+    Query::new(
+        "delete_follow",
         "// Important that MATCH to check if both users are in the graph
         MATCH (follower:User {id: $follower_id}), (followee:User {id: $followee_id})
         // Check if follow already exist
-        OPTIONAL MATCH (follower)-[existing:FOLLOWS]->(followee) 
+        OPTIONAL MATCH (follower)-[existing:FOLLOWS]->(followee)
         DELETE existing
         // Returns true if the relationship does not exist as 'flag'
         RETURN existing IS NULL AS flag;",
@@ -47,7 +50,8 @@ pub fn delete_follow(follower_id: &str, followee_id: &str) -> Query {
 /// * `user_id` - The unique identifier of the user who created the bookmark.
 /// * `bookmark_id` - The unique identifier of the bookmark relationship to be deleted.
 pub fn delete_bookmark(user_id: &str, bookmark_id: &str) -> Query {
-    query(
+    Query::new(
+        "delete_bookmark",
         "MATCH (u:User {id: $user_id})-[b:BOOKMARKED {id: $bookmark_id}]->(post:Post)<-[:AUTHORED]-(author:User)
          WITH post.id as post_id, author.id as author_id, b
          DELETE b
@@ -62,7 +66,8 @@ pub fn delete_bookmark(user_id: &str, bookmark_id: &str) -> Query {
 /// * `user_id` - The unique identifier of the user who created the tag.
 /// * `tag_id` - The unique identifier of the `TAGGED` relationship to be deleted.
 pub fn delete_tag(user_id: &str, tag_id: &str) -> Query {
-    query(
+    Query::new(
+        "delete_tag",
         "MATCH (user:User {id: $user_id})-[tag:TAGGED {id: $tag_id}]->(target)
          OPTIONAL MATCH (target)<-[:AUTHORED]-(author:User)
          WITH CASE WHEN target:User THEN target.id ELSE null END AS user_id,
@@ -82,7 +87,8 @@ pub fn delete_tag(user_id: &str, tag_id: &str) -> Query {
 /// * `owner_id` - The unique identifier of the user who owns the file
 /// * `file_id` - The unique identifier of the file to be deleted
 pub fn delete_file(owner_id: &str, file_id: &str) -> Query {
-    query(
+    Query::new(
+        "delete_file",
         "MATCH (f:File {id: $id, owner_id: $owner_id})
          DETACH DELETE f;",
     )