Merge pull request #41 from chaindexing/re-introduce-chain-reorg-doma…

…in-in-chaindexing-processes

Re introduce chain reorg domain in chaindexing processes

chaindexing/src/event_handlers.rs

@@ -1,15 +1,14 @@
 use std::{sync::Arc, time::Duration};
 
 mod handle_events;
-mod handled_events;
+mod maybe_handle_chain_reorg;
 
 use tokio::{sync::Mutex, time::interval};
 
 use crate::{contracts::Contracts, events::Event, ChaindexingRepo, Config, Repo};
 use crate::{ChaindexingRepoRawQueryTxnClient, HasRawQueryClient};
 
 use handle_events::HandleEvents;
-use handled_events::MaybeBacktrackHandledEvents;
 
 #[derive(Clone)]
 pub struct EventHandlerContext<'a> {
@@ -63,7 +62,7 @@ impl EventHandlers {
                 .await;
 
                 let state_migrations = Contracts::get_state_migrations(&config.contracts);
-                MaybeBacktrackHandledEvents::run(
+                maybe_handle_chain_reorg::run(
                     conn.clone(),
                     &mut raw_query_client,
                     &state_migrations,

chaindexing/src/event_handlers/maybe_handle_chain_reorg.rs

@@ -0,0 +1,57 @@
+use std::sync::Arc;
+
+use tokio::sync::Mutex;
+
+use crate::ChaindexingRepo;
+use crate::{
+    ChaindexingRepoConn, ChaindexingRepoRawQueryClient, ExecutesWithRawQuery, HasRawQueryClient,
+    Repo,
+};
+use crate::{ContractStateMigrations, ContractStates};
+use crate::{ReorgedBlock, ReorgedBlocks};
+
+pub async fn run<'a>(
+    conn: Arc<Mutex<ChaindexingRepoConn<'a>>>,
+    raw_query_client: &mut ChaindexingRepoRawQueryClient,
+    state_migrations: &Vec<Arc<dyn ContractStateMigrations>>,
+) {
+    let mut conn = conn.lock().await;
+    let reorged_blocks = ChaindexingRepo::get_unhandled_reorged_blocks(&mut conn).await;
+
+    if !reorged_blocks.is_empty() {
+        let raw_query_txn_client =
+            ChaindexingRepo::get_raw_query_txn_client(raw_query_client).await;
+
+        let reorged_blocks = ReorgedBlocks::only_earliest_per_chain(&reorged_blocks);
+
+        for ReorgedBlock {
+            block_number,
+            chain_id,
+            ..
+        } in &reorged_blocks
+        {
+            ContractStates::backtrack_states(
+                state_migrations,
+                *chain_id,
+                *block_number,
+                &raw_query_txn_client,
+            )
+            .await;
+            ChaindexingRepo::update_every_next_block_number_to_handle_from_in_txn(
+                &raw_query_txn_client,
+                *chain_id,
+                *block_number,
+            )
+            .await
+        }
+
+        let reorged_block_ids = ReorgedBlocks::get_ids(&reorged_blocks);
+        ChaindexingRepo::update_reorged_blocks_as_handled_in_txn(
+            &raw_query_txn_client,
+            &reorged_block_ids,
+        )
+        .await;
+
+        ChaindexingRepo::commit_raw_query_txns(raw_query_txn_client).await;
+    }
+}

chaindexing/src/events_ingester.rs

@@ -1,5 +1,5 @@
 mod ingest_events;
-mod ingested_events;
+mod maybe_handle_chain_reorg;
 
 use std::collections::HashMap;
 use std::sync::Arc;
@@ -16,7 +16,6 @@ use tokio::sync::Mutex;
 use tokio::time::{interval, sleep};
 
 use ingest_events::IngestEvents;
-use ingested_events::MaybeBacktrackIngestedEvents;
 
 use crate::chain_reorg::Execution;
 use crate::contracts::Contract;
@@ -163,7 +162,7 @@ impl EventsIngester {
             )
             .await?;
 
-            MaybeBacktrackIngestedEvents::run(
+            maybe_handle_chain_reorg::run(
                 &mut conn,
                 contract_addresses.clone(),
                 contracts,