Fix Neo4j connection when running in Docker

This change fixes the connection string to Neo4j when running
inside a Docker container. It also delays app startup until a
healthy Neo4j graph client can be constructed and execute a query
against the database. Lastly, it fixes a boolean bug when checking
that a system shouldn't exist before fetching it from ESI.

Cargo.toml

@@ -3,8 +3,6 @@ name = "eve-graph"
 version = "0.1.0"
 edition = "2021"
 
-# See more keys and their definitions at https://doc.rust-lang.org/cargo/reference/manifest.html
-
 [dependencies]
 reqwest = { version = "0.11", default-features = false, features = ["blocking", "json", "serde_json", "rustls-tls"] }
 serde = { version = "1", features = ["derive"] }
@@ -16,4 +14,5 @@ uuid = "1.6.0"
 thiserror = "1.0.50"
 warp = "0.3.6"
 chrono = "0.4.31"
-openssl = { version = "0.10", features = ["vendored"] }
+openssl = { version = "0.10", features = ["vendored"] }
+time = "0.3.30"

src/database.rs

@@ -1,25 +1,75 @@
 use std::sync::Arc;
+use std::time::Duration;
 
+use neo4rs::Error::ConnectionError;
 use neo4rs::{query, Error, Error::DeserializationError, Graph, Row};
 use serde::{Deserialize, Serialize};
 
 use crate::eve_scout::EveScoutSignature;
 
-pub async fn get_graph_client() -> Arc<Graph> {
-    let uri = "bolt://localhost:7687";
+pub async fn get_graph_client_with_retry(max_retries: usize) -> Result<Arc<Graph>, Error> {
+    let neo4j_container_name = "neo4j";
+    let uri = format!("bolt://{}:7687", neo4j_container_name);
     let user = "neo4j";
     let pass = "neo4jneo4j";
-    Arc::new(Graph::new(uri, user, pass).await.unwrap())
+
+    for attempt in 1..=max_retries {
+        println!("Trying to build a Neo4j graph client, attempt {}", attempt);
+        match Graph::new(uri.clone(), user, pass).await {
+            Ok(graph) => {
+                let graph = Arc::new(graph);
+                match check_neo4j_health(graph.clone()).await {
+                    Ok(_) => {
+                        println!("Successfully built a healthy Neo4j graph client");
+                        return Ok(graph);
+                    }
+                    Err(err) => {
+                        println!("Neo4j isn't ready yet: {}", err);
+                    }
+                };
+            }
+            Err(err) => println!("Failed to build a Neo4j graph client: {}", err),
+        };
+
+        let seconds = 5;
+        println!(
+            "Waiting {} seconds before trying to build another Neo4j graph client",
+            seconds
+        );
+        tokio::time::sleep(Duration::from_secs(seconds)).await;
+    }
+
+    println!(
+        "Failed to get graph client after max of {} attempts",
+        max_retries
+    );
+    Err(ConnectionError)
+}
+
+async fn check_neo4j_health(graph: Arc<Graph>) -> Result<(), Error> {
+    let test_query = "MATCH (n) RETURN n LIMIT 1";
+    let mut result = match graph.execute(query(test_query)).await {
+        Ok(row_stream) => row_stream,
+        Err(err) => {
+            return Err(ConnectionError);
+        }
+    };
+
+    // Any result means neo4j is responding
+    match result.next().await? {
+        None => Ok(()),
+        Some(_) => Ok(()),
+    }
 }
 
 pub async fn system_id_exists(graph: Arc<Graph>, system_id: i64) -> Result<bool, Error> {
     let system_exists = "MATCH (s:System {system_id: $system_id}) RETURN COUNT(s) as count LIMIT 1";
     println!("Querying database for system_id {}", system_id);
     let mut result = graph
         .execute(query(system_exists).param("system_id", system_id))
-        .await?;
+        .await
+        .expect("Graph client failed to execute the system exists query");
 
-    println!("Matching query result");
     match result.next().await? {
         Some(row) => Ok(row_count_is_positive(row)),
         None => Ok(false),

src/main.rs

@@ -26,7 +26,7 @@ mod eve_scout;
 async fn main() {
     println!("Starting eve-graph");
     let client = Client::new();
-    let graph = get_graph_client().await;
+    let graph = get_graph_client_with_retry(5).await.unwrap();
 
     let systems_refresh = warp::path!("systems" / "refresh");
     let systems_risk = warp::path!("systems" / "risk");
@@ -220,7 +220,7 @@ async fn pull_system_if_missing(
 
     match system_id_exists(graph.clone(), system_id).await {
         Ok(exists) => {
-            if exists {
+            if !exists {
                 println!(
                     "System {} does not already exist in the database",
                     system_id
@@ -332,7 +332,7 @@ async fn error_if_any_member_has_error<T: 'static>(
             return Some(Err(e));
         }
     }
-    None
+    Some(Ok(()))
 }
 
 async fn pull_system_kills(client: Client, graph: Arc<Graph>) -> Result<i32, ReplicationError> {
@@ -425,15 +425,15 @@ async fn pull_stargate(
 mod tests {
     use reqwest::Client;
 
-    use crate::database::{get_graph_client, save_system, System};
+    use crate::database::{get_graph_client_with_retry, save_system, System};
     use crate::esi::get_system_details;
     use crate::{pull_all_stargates, pull_system_jumps, pull_system_kills, pull_system_stargates};
 
     #[tokio::test]
     #[ignore]
     async fn can_save_system_to_database() {
         let client = Client::new();
-        let graph = get_graph_client().await;
+        let graph = get_graph_client_with_retry(1).await.unwrap();
         let system_id = 30000201;
         let system_response = get_system_details(&client, system_id).await.unwrap();
 
@@ -448,7 +448,8 @@ mod tests {
     #[tokio::test]
     #[ignore]
     async fn should_pull_all_stargates() {
-        match pull_all_stargates(Client::new(), get_graph_client().await).await {
+        match pull_all_stargates(Client::new(), get_graph_client_with_retry(1).await.unwrap()).await
+        {
             Ok(_) => {}
             Err(_) => {
                 println!("failed to pull all stargates")
@@ -460,7 +461,7 @@ mod tests {
     #[ignore]
     async fn should_pull_system_stargates() {
         let client = Client::new();
-        let graph = get_graph_client().await;
+        let graph = get_graph_client_with_retry(1).await.unwrap();
         let system_id = 30000193;
 
         match pull_system_stargates(client.clone(), graph.clone(), system_id).await {
@@ -475,7 +476,7 @@ mod tests {
     #[ignore]
     async fn should_pull_system_jumps() {
         let client = Client::new();
-        let graph = get_graph_client().await;
+        let graph = get_graph_client_with_retry(1).await.unwrap();
 
         let total_jumps = pull_system_jumps(client, graph).await.unwrap();
 
@@ -486,7 +487,7 @@ mod tests {
     #[ignore]
     async fn should_pull_system_kills() {
         let client = Client::new();
-        let graph = get_graph_client().await;
+        let graph = get_graph_client_with_retry(1).await.unwrap();
 
         let total_kills = pull_system_kills(client, graph).await.unwrap();
 

tests/database_tests.rs

@@ -1,13 +1,13 @@
 use eve_graph::database::{
     build_jump_risk_graph, build_system_jump_graph, drop_jump_risk_graph, drop_system_jump_graph,
-    get_all_system_ids, get_graph_client, get_system,
+    get_all_system_ids, get_graph_client_with_retry, get_system,
 };
 
 #[tokio::test]
 #[ignore]
 async fn should_get_system() {
     let system_id = 30000276;
-    let system = get_system(get_graph_client().await, system_id)
+    let system = get_system(get_graph_client_with_retry(1).await.unwrap(), system_id)
         .await
         .unwrap()
         .unwrap();
@@ -17,39 +17,39 @@ async fn should_get_system() {
 #[tokio::test]
 #[ignore]
 async fn should_get_all_system_ids() {
-    let system_ids = get_all_system_ids(get_graph_client().await).await;
+    let system_ids = get_all_system_ids(get_graph_client_with_retry(1).await.unwrap()).await;
 
     assert_eq!(system_ids.unwrap().len(), 8436)
 }
 
 #[tokio::test]
 #[ignore]
 async fn should_drop_system_jump_graph() {
-    let graph = get_graph_client().await;
+    let graph = get_graph_client_with_retry(1).await.unwrap();
     let dropped_graph_name = drop_system_jump_graph(&graph).await.unwrap();
     assert_eq!(dropped_graph_name, "system-map")
 }
 
 #[tokio::test]
 #[ignore]
 async fn should_create_system_jump_graph() {
-    let graph = get_graph_client().await;
+    let graph = get_graph_client_with_retry(1).await.unwrap();
     let new_graph_name = build_system_jump_graph(graph).await.unwrap();
     assert_eq!(new_graph_name, "system-map")
 }
 
 #[tokio::test]
 #[ignore]
 async fn should_drop_jump_risk_graph() {
-    let graph = get_graph_client().await;
+    let graph = get_graph_client_with_retry(1).await.unwrap();
     let dropped_graph_name = drop_jump_risk_graph(graph).await.unwrap();
     assert_eq!(dropped_graph_name, "jump-risk")
 }
 
 #[tokio::test]
 #[ignore]
 async fn should_create_jump_risk_graph() {
-    let graph = get_graph_client().await;
+    let graph = get_graph_client_with_retry(1).await.unwrap();
     let new_graph_name = build_jump_risk_graph(graph).await.unwrap();
     assert_eq!(new_graph_name, "jump-risk")
 }