feat: recreate pangea client

Author: vecheslav

forge/src/pangea/indexer.rs

@@ -19,8 +19,8 @@ use crate::{
 const BATCH_SIZE: u64 = 100_000;
 
 pub struct PangeaIndexer {
-    pangea_client: Client<WsProvider>,
-    fuel_provider: Provider,
+    pangea_host: String,
+    provider: Provider,
     operation_tx: Sender<OperationMessage>,
     chain_id: ChainId,
     contract_h256: H256,
@@ -36,37 +36,42 @@ impl PangeaIndexer {
             "FUEL" => ChainId::FUEL,
             _ => ChainId::FUELTESTNET,
         };
-
-        let username = env::var("PANGEA_USERNAME").unwrap();
-        let password = env::var("PANGEA_PASSWORD").unwrap();
-
-        let pangea_client = ClientBuilder::default()
-            .endpoint(&config.pangea_host)
-            .credential(username, password)
-            .build::<WsProvider>()
-            .await?;
+        let pangea_host = config.pangea_host.clone();
 
         let provider_url = match chain_id {
             ChainId::FUEL => Ok("mainnet.fuel.network"),
             ChainId::FUELTESTNET => Ok("testnet.fuel.network"),
             _ => Err(Error::InvalidChainId),
         }?;
-        let fuel_provider = Provider::connect(provider_url).await?;
+        let provider = Provider::connect(provider_url).await?;
 
         log::info!("CHAIN: {:?}, PROVIDER: {:?}", chain_id, provider_url);
 
         Ok(Self {
-            pangea_client,
-            fuel_provider,
+            pangea_host,
+            provider,
             operation_tx,
             chain_id,
             contract_h256: H256::from_str(market_id)?,
         })
     }
 
+    pub async fn create_pangea_client(&self) -> Result<Client<WsProvider>, Error> {
+        let username = env::var("PANGEA_USERNAME").unwrap();
+        let password = env::var("PANGEA_PASSWORD").unwrap();
+
+        let pangea_client = ClientBuilder::default()
+            .endpoint(&self.pangea_host)
+            .credential(username, password)
+            .build::<WsProvider>()
+            .await?;
+
+        Ok(pangea_client)
+    }
+
     pub async fn start(&self, latest_processed_block: i64) -> Result<(), Error> {
         // Get latest block number from blockchain
-        let latest_block = self.fuel_provider.latest_block_height().await.unwrap() as i64;
+        let latest_block = self.provider.latest_block_height().await.unwrap() as i64;
 
         log::info!("Prune newest orders & trades");
         self.prune(latest_processed_block).await?;
@@ -107,6 +112,8 @@ impl PangeaIndexer {
         mut latest_processed_block: i64,
         to_block: i64,
     ) -> Result<i64, Error> {
+        let client = self.create_pangea_client().await?;
+
         while latest_processed_block < to_block {
             let to_block = (latest_processed_block + BATCH_SIZE as i64).min(to_block);
 
@@ -118,8 +125,7 @@ impl PangeaIndexer {
                 ..Default::default()
             };
 
-            let stream = self
-                .pangea_client
+            let stream = client
                 .get_fuel_spark_orders_by_format(batch_request, Format::JsonStream, false)
                 .await
                 .expect("Failed to get fuel spark orders batch");
@@ -168,46 +174,59 @@ impl PangeaIndexer {
         let max_backoff = Duration::from_secs(32);
 
         loop {
-            let deltas_request = GetSparkOrderRequest {
-                from_block: Bound::Exact(latest_processed_block + 1),
-                to_block: Bound::Subscribe,
-                market_id__in: HashSet::from([self.contract_h256]),
-                chains: HashSet::from([self.chain_id]),
-                ..Default::default()
-            };
-
-            match self
-                .pangea_client
-                .get_fuel_spark_orders_by_format(deltas_request, Format::JsonStream, true)
-                .await
-            {
-                Ok(stream) => {
-                    backoff = Duration::from_secs(1);
-                    futures::pin_mut!(stream);
-
-                    while let Some(data) = stream.next().await {
-                        match data {
-                            Ok(data) => {
-                                let data = String::from_utf8(data)?;
-                                let event = serde_json::from_str::<PangeaEvent>(&data)?;
-                                latest_processed_block = event.block_number;
-
-                                log::debug!("LATEST_PROCESSED_BLOCK: {}", latest_processed_block);
-
-                                self.handle_event(&event).await;
-                                self.operation_tx
-                                    .send(OperationMessage::Dispatch(latest_processed_block))
-                                    .unwrap();
-                            }
-                            Err(e) => {
-                                log::error!("Error in the stream of new orders (deltas): {e}");
-                                break;
+            match self.create_pangea_client().await {
+                Ok(client) => {
+                    let deltas_request = GetSparkOrderRequest {
+                        from_block: Bound::Exact(latest_processed_block + 1),
+                        to_block: Bound::Subscribe,
+                        market_id__in: HashSet::from([self.contract_h256]),
+                        chains: HashSet::from([self.chain_id]),
+                        ..Default::default()
+                    };
+
+                    match client
+                        .get_fuel_spark_orders_by_format(deltas_request, Format::JsonStream, true)
+                        .await
+                    {
+                        Ok(stream) => {
+                            backoff = Duration::from_secs(1);
+                            futures::pin_mut!(stream);
+
+                            while let Some(data) = stream.next().await {
+                                match data {
+                                    Ok(data) => {
+                                        let data = String::from_utf8(data)?;
+                                        let event = serde_json::from_str::<PangeaEvent>(&data)?;
+                                        latest_processed_block = event.block_number;
+
+                                        log::debug!(
+                                            "LATEST_PROCESSED_BLOCK: {}",
+                                            latest_processed_block
+                                        );
+
+                                        self.handle_event(&event).await;
+                                        self.operation_tx
+                                            .send(OperationMessage::Dispatch(
+                                                latest_processed_block,
+                                            ))
+                                            .unwrap();
+                                    }
+                                    Err(e) => {
+                                        log::error!(
+                                            "Error in the stream of new orders (deltas): {e}"
+                                        );
+                                        break;
+                                    }
+                                }
                             }
                         }
+                        Err(e) => {
+                            log::error!("Failed to get fuel spark deltas: {e}");
+                        }
                     }
                 }
                 Err(e) => {
-                    log::error!("Failed to get fuel spark deltas: {e}");
+                    log::error!("Failed to create pangea client: {e}");
                 }
             }