move reconnect to degensql

Author: ethereumdegen

Cargo.lock

@@ -1,6 +1,6 @@
 [package]
 name = "vibegraph"
-version = "0.3.11"
+version = "0.3.12"
 edition = "2021"
 default-run = "vibegraph"
 description = "Reads ethereum contract events from a lightweight RPC and caches them to a database"
@@ -30,7 +30,7 @@ thiserror = "1.0.49"
 include_dir = "0.7.3"
 inquire = "0.6.2"
 
-  degen-sql = "0.1.10" 
+  degen-sql = "0.1.12" 
  #degen-sql={path="../degen-sql"}
 
 rust_decimal = { version = "1.33.1", features = ["db-tokio-postgres"] }  

Cargo.toml

@@ -30,7 +30,7 @@ impl EventsModel {
     pub async fn insert_one(
     event: &ContractEvent  ,
 
-    psql_db: &Database,
+    psql_db: &mut Database,
 ) -> Result<i32, PostgresModelError> {
 
 
@@ -73,9 +73,8 @@ impl EventsModel {
 
 
         // Set a timeout (e.g., 5 seconds)
-        let insert_result = timeout(
-            Duration::from_secs(5), // Set timeout duration
-            psql_db.query_one(
+        let insert_result =  
+            psql_db.query_one_with_reconnect(
                 "
                 INSERT INTO events 
                 (
@@ -107,79 +106,25 @@ impl EventsModel {
                     &log_index,
                     &transaction_index
                 ],
-            ),
+                3
+            
         ).await;
 
-        match insert_result {
-            Ok(Ok(row)) => Ok(row.get(0)), // Successfully inserted and retrieved ID
-            Ok(Err(e)) => {
-                eprintln!("Database error: {:?}", e);
-                Err(PostgresModelError::Postgres(e))
-            }
-            Err(_) => {
-                eprintln!("Database timeout occurred.");
-                Err(PostgresModelError::Timeout) // You may need to define a Timeout variant in PostgresModelError
-            }
-        }
-        /*
-        
-        let insert_result = psql_db.query_one(
-            "
-            INSERT INTO events 
-            (
-            contract_address,
-            name,
-            signature,
-            args,
-            data,
-            chain_id,
-            transaction_hash,
-            block_number,
-            block_hash,
-            log_index,
-            transaction_index            
-            ) 
-            VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11)
-            RETURNING id;
-            ",
-            &[
-                &contract_address, 
-                &name,
-                &signature,
-                &args,
-                &data,
-                &chain_id,
-                &transaction_hash,
-                &block_number,
-                &block_hash,
-                &log_index  ,
-                &transaction_index                 
-                ],
-        ).await;
-    
-        match insert_result {
-            Ok(row) => Ok(row.get(0)),  // Successfully inserted new row and got its ID.
-            Err(e) => {
-                eprintln!("Database error: Event {:?}", e);
-                
-                Err(  PostgresModelError::Postgres(e) )
-                
-                
-            }
-        }*/
+          insert_result.map(|r| r.get(0)) 
+         
     }
 
 
 pub async fn find_most_recent_event(
     contract_address: Address,
-    psql_db: &Database,
+    psql_db: &mut Database,
 ) -> Result< ContractEvent, PostgresModelError> {
 
 
     let parsed_contract_address = to_checksum(&contract_address, None).to_string();
 
 
-    let row = psql_db.query_one(
+    let row = psql_db.query_one_with_reconnect(
         "
         SELECT 
             contract_address,
@@ -200,6 +145,7 @@ pub async fn find_most_recent_event(
         LIMIT 1;
         ",
         &[&parsed_contract_address],
+        3
     ).await;
 
     match row {
@@ -237,7 +183,7 @@ pub async fn find_most_recent_event(
         }, 
         Err(e) => {
             eprintln!("Database error: Recent Event {:?}", e);
-            Err(PostgresModelError::Postgres(e))
+            Err( e) 
         }
     }
 }

src/db/postgres/models/events_model.rs

@@ -194,7 +194,7 @@ pub fn try_identify_event_for_log(
 
 pub async fn find_most_recent_event_blocknumber( 
     contract_address: Address,
-    psql_db: &Database
+    psql_db: &mut Database
 ) -> Option<U64> {
 
 

src/event.rs

@@ -309,44 +309,43 @@ async fn collect_events(
 
 
 
-    let mut encountered_insertion_timeout = false;
 
     for event_log in event_logs {
 
           info!("decoded event log {:?}", event_log);
 
          // let psql_db = &app_state.database;
 
-          let psql_db = app_state.database.lock().await; // Lock database correctly
+          let mut psql_db = app_state.database.lock().await; // Lock database correctly
 
-          let inserted = EventsModel::insert_one(&event_log, &*psql_db ).await;
+          let inserted = EventsModel::insert_one(&event_log, &mut *psql_db ).await;
 
           info!("inserted {:?}", inserted);
 
-          if inserted.is_err_and( |e| e == PostgresModelError::Timeout ) {
-            encountered_insertion_timeout = true ;
-          }
+        //  if inserted.is_err_and( |e| e == PostgresModelError::Timeout ) {
+         //   encountered_insertion_timeout = true ;
+         // }
 
 
     }
 
-    if !encountered_insertion_timeout {
+   // if !encountered_insertion_timeout {
           //progress the current indexing block
           app_state.indexing_state.current_indexing_block = end_block + 1; 
-    }else {
+  /*  }else {
         warn!("Encountered a timeout with the database- retrying");
 
         // Unlock database first before reconnecting
            // drop(psql_db);
 
             // Reconnect and update `app_state.database`
             let mut db_lock = app_state.database.lock().await;
-            if let Err(e) = db_lock.reconnect(   app_config.db_conn_url.clone() ).await {
+            if let Err(e) = db_lock.reconnect(     ).await {
                 eprintln!("Database reconnection failed: {:?}", e);
             } else {
                 info!("Database reconnected successfully.");
             }
-    }
+    }*/
     // there there was an error, we are going to cycle this period again .
 
 
@@ -390,8 +389,8 @@ async fn initialize(
 
 
      let most_recent_event_blocknumber = {
-        let psql_db = app_state.database.lock().await;
-        find_most_recent_event_blocknumber(contract_address, &psql_db).await
+        let mut psql_db = app_state.database.lock().await;
+        find_most_recent_event_blocknumber(contract_address, &mut psql_db).await
     }; // `psql_db` is dropped here to avoid deadlock later