Update deposit and withdraw polling logic

crates/common/src/local_db/query/fetch_transaction_by_hash/mod.rs

@@ -0,0 +1,55 @@
+use crate::local_db::{
+    query::{SqlStatement, SqlValue},
+    OrderbookIdentifier,
+};
+use alloy::primitives::{Address, B256};
+use serde::{Deserialize, Serialize};
+
+const QUERY_TEMPLATE: &str = include_str!("query.sql");
+
+/// Transaction info returned from local DB query.
+#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
+#[serde(rename_all = "camelCase")]
+pub struct LocalDbTransaction {
+    pub transaction_hash: B256,
+    pub block_number: u64,
+    pub block_timestamp: u64,
+    pub sender: Address,
+}
+
+/// Builds a SQL statement to fetch transaction info by transaction hash
+/// from the deposits and withdrawals tables.
+pub fn build_fetch_transaction_by_hash_stmt(
+    ob_id: &OrderbookIdentifier,
+    tx_hash: B256,
+) -> SqlStatement {
+    SqlStatement::new_with_params(
+        QUERY_TEMPLATE,
+        vec![
+            SqlValue::from(ob_id.chain_id),
+            SqlValue::from(ob_id.orderbook_address),
+            SqlValue::from(tx_hash),
+        ],
+    )
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use alloy::primitives::{address, b256};
+
+    #[test]
+    fn builds_correct_sql_with_params() {
+        let orderbook = address!("0x1234567890123456789012345678901234567890");
+        let tx_hash = b256!("0xabcdef1234567890abcdef1234567890abcdef1234567890abcdef1234567890");
+        let ob_id = OrderbookIdentifier::new(1, orderbook);
+
+        let stmt = build_fetch_transaction_by_hash_stmt(&ob_id, tx_hash);
+
+        assert!(stmt.sql.contains("SELECT"));
+        assert!(stmt.sql.contains("FROM deposits"));
+        assert!(stmt.sql.contains("FROM withdrawals"));
+        assert!(stmt.sql.contains("transaction_hash"));
+        assert_eq!(stmt.params.len(), 3);
+    }
+}

crates/common/src/local_db/query/fetch_transaction_by_hash/query.sql

@@ -0,0 +1,40 @@
+WITH params AS (
+  SELECT
+    ?1 AS chain_id,
+    ?2 AS orderbook_address,
+    ?3 AS transaction_hash
+),
+combined AS (
+  SELECT
+    d.transaction_hash,
+    d.block_number,
+    d.block_timestamp,
+    d.sender,
+    d.log_index
+  FROM deposits d
+  JOIN params p
+    ON p.chain_id = d.chain_id
+   AND p.orderbook_address = d.orderbook_address
+   AND p.transaction_hash = d.transaction_hash
+  UNION ALL
+  SELECT
+    w.transaction_hash,
+    w.block_number,
+    w.block_timestamp,
+    w.sender,
+    w.log_index
+  FROM withdrawals w
+  JOIN params p
+    ON p.chain_id = w.chain_id
+   AND p.orderbook_address = w.orderbook_address
+   AND p.transaction_hash = w.transaction_hash
+)
+SELECT
+  transaction_hash AS transactionHash,
+  block_number AS blockNumber,
+  block_timestamp AS blockTimestamp,
+  sender
+FROM combined
+ORDER BY log_index ASC
+LIMIT 1;
+

crates/common/src/local_db/query/mod.rs

@@ -12,6 +12,7 @@ pub mod fetch_orders;
 pub mod fetch_store_addresses;
 pub mod fetch_tables;
 pub mod fetch_target_watermark;
+pub mod fetch_transaction_by_hash;
 pub mod fetch_vault_balance_changes;
 pub mod fetch_vaults;
 pub mod insert_db_metadata;

crates/common/src/raindex_client/add_orders.rs

@@ -123,7 +123,7 @@ impl RaindexClient {
                     Ok(orders) => return Ok(orders),
                     Err(RetryError::Operation(PollError::Inner(e))) => return Err(e),
                     Err(RetryError::InvalidMaxAttempts) => {
-                        return Err(RaindexError::SubgraphIndexingTimeout { tx_hash, attempts })
+                        return Err(RaindexError::TransactionIndexingTimeout { tx_hash, attempts })
                     }
                     Err(RetryError::Operation(PollError::Empty)) => {
                         // Local DB exhausted, fall through to subgraph
@@ -173,7 +173,7 @@ impl RaindexClient {
             Ok(orders) => Ok(orders),
             Err(RetryError::Operation(PollError::Inner(e))) => Err(e),
             Err(RetryError::Operation(PollError::Empty)) | Err(RetryError::InvalidMaxAttempts) => {
-                Err(RaindexError::SubgraphIndexingTimeout { tx_hash, attempts })
+                Err(RaindexError::TransactionIndexingTimeout { tx_hash, attempts })
             }
         }
     }
@@ -588,7 +588,7 @@ mod tests {
                 .unwrap_err();
 
             match err {
-                RaindexError::SubgraphIndexingTimeout { attempts, .. } => {
+                RaindexError::TransactionIndexingTimeout { attempts, .. } => {
                     assert_eq!(attempts, DEFAULT_ADD_ORDER_POLL_ATTEMPTS);
                 }
                 other => panic!("expected timeout error, got {other:?}"),

crates/common/src/raindex_client/local_db/mod.rs

@@ -16,6 +16,7 @@ pub mod executor;
 pub mod orders;
 pub mod pipeline;
 pub mod query;
+pub mod transactions;
 pub mod vaults;
 
 type ExecuteBatchFn =

crates/common/src/raindex_client/local_db/query/fetch_transaction_by_hash.rs

@@ -0,0 +1,85 @@
+use crate::local_db::query::fetch_transaction_by_hash::{
+    build_fetch_transaction_by_hash_stmt, LocalDbTransaction,
+};
+use crate::local_db::query::{LocalDbQueryError, LocalDbQueryExecutor};
+use crate::local_db::OrderbookIdentifier;
+use alloy::primitives::B256;
+
+pub async fn fetch_transaction_by_hash<E: LocalDbQueryExecutor + ?Sized>(
+    exec: &E,
+    ob_id: &OrderbookIdentifier,
+    tx_hash: B256,
+) -> Result<Vec<LocalDbTransaction>, LocalDbQueryError> {
+    let stmt = build_fetch_transaction_by_hash_stmt(ob_id, tx_hash);
+    exec.query_json(&stmt).await
+}
+
+#[cfg(all(test, target_family = "wasm"))]
+mod wasm_tests {
+    use super::*;
+    use crate::raindex_client::local_db::executor::tests::create_sql_capturing_callback;
+    use crate::raindex_client::local_db::executor::JsCallbackExecutor;
+    use alloy::primitives::{address, b256, Address};
+    use std::cell::RefCell;
+    use std::rc::Rc;
+    use wasm_bindgen_test::*;
+    use wasm_bindgen_utils::prelude::*;
+
+    #[wasm_bindgen_test]
+    async fn wrapper_uses_builder_sql_exactly() {
+        let tx_hash = b256!("0x0000000000000000000000000000000000000000000000000000000000000abc");
+        let orderbook = Address::from([0x51; 20]);
+        let expected_stmt =
+            build_fetch_transaction_by_hash_stmt(&OrderbookIdentifier::new(1, orderbook), tx_hash);
+
+        let store = Rc::new(RefCell::new((String::new(), JsValue::UNDEFINED)));
+        let callback = create_sql_capturing_callback("[]", store.clone());
+        let exec = JsCallbackExecutor::from_ref(&callback);
+
+        let res = super::fetch_transaction_by_hash(
+            &exec,
+            &OrderbookIdentifier::new(1, orderbook),
+            tx_hash,
+        )
+        .await;
+        assert!(res.is_ok());
+        assert_eq!(store.borrow().clone().0, expected_stmt.sql);
+    }
+
+    #[wasm_bindgen_test]
+    async fn wrapper_returns_rows_when_present() {
+        let tx_hash = b256!("0x0000000000000000000000000000000000000000000000000000000000000abc");
+        let orderbook = address!("0x5151515151515151515151515151515151515151");
+        let sender = address!("0x1111111111111111111111111111111111111111");
+        let expected_stmt =
+            build_fetch_transaction_by_hash_stmt(&OrderbookIdentifier::new(1, orderbook), tx_hash);
+
+        let row_json = format!(
+            r#"[{{
+                "transactionHash":"{}",
+                "blockNumber":100,
+                "blockTimestamp":999,
+                "sender":"{}"
+            }}]"#,
+            tx_hash, sender
+        );
+
+        let store = Rc::new(RefCell::new((
+            String::new(),
+            wasm_bindgen::JsValue::UNDEFINED,
+        )));
+        let callback = create_sql_capturing_callback(&row_json, store.clone());
+        let exec = JsCallbackExecutor::from_ref(&callback);
+
+        let res = super::fetch_transaction_by_hash(
+            &exec,
+            &OrderbookIdentifier::new(1, orderbook),
+            tx_hash,
+        )
+        .await;
+        assert!(res.is_ok());
+        let rows = res.unwrap();
+        assert_eq!(rows.len(), 1);
+        assert_eq!(store.borrow().clone().0, expected_stmt.sql);
+    }
+}

crates/common/src/raindex_client/local_db/query/mod.rs

@@ -7,6 +7,7 @@ pub mod fetch_order_trades_count;
 pub mod fetch_orders;
 pub mod fetch_store_addresses;
 pub mod fetch_tables;
+pub mod fetch_transaction_by_hash;
 pub mod fetch_vault_balance_changes;
 pub mod fetch_vaults;
 pub mod update_last_synced_block;

crates/common/src/raindex_client/local_db/transactions.rs

@@ -0,0 +1,128 @@
+use super::super::transactions::RaindexTransaction;
+use super::super::RaindexError;
+use super::LocalDb;
+use crate::local_db::query::fetch_transaction_by_hash::{
+    build_fetch_transaction_by_hash_stmt, LocalDbTransaction,
+};
+use crate::local_db::query::LocalDbQueryExecutor;
+use crate::local_db::OrderbookIdentifier;
+use alloy::primitives::B256;
+
+pub struct LocalDbTransactions<'a> {
+    pub(crate) db: &'a LocalDb,
+}
+
+impl<'a> LocalDbTransactions<'a> {
+    pub(crate) fn new(db: &'a LocalDb) -> Self {
+        Self { db }
+    }
+
+    /// Fetch transaction info by transaction hash from the local DB.
+    /// Returns None if no transaction with that hash is found.
+    pub async fn get_by_tx_hash(
+        &self,
+        ob_id: &OrderbookIdentifier,
+        tx_hash: B256,
+    ) -> Result<Option<RaindexTransaction>, RaindexError> {
+        let stmt = build_fetch_transaction_by_hash_stmt(ob_id, tx_hash);
+        let results: Vec<LocalDbTransaction> = self.db.query_json(&stmt).await?;
+
+        if let Some(local_tx) = results.into_iter().next() {
+            let tx = RaindexTransaction::from_local_parts(
+                local_tx.transaction_hash,
+                local_tx.sender,
+                local_tx.block_number,
+                local_tx.block_timestamp,
+            )?;
+            return Ok(Some(tx));
+        }
+
+        Ok(None)
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    #[cfg(target_family = "wasm")]
+    use super::*;
+
+    #[cfg(target_family = "wasm")]
+    mod wasm_tests {
+        use super::*;
+        use crate::raindex_client::local_db::executor::JsCallbackExecutor;
+        use crate::raindex_client::local_db::LocalDb;
+        use alloy::primitives::{address, b256};
+        use serde_json::json;
+        use wasm_bindgen_test::wasm_bindgen_test;
+        use wasm_bindgen_utils::prelude::*;
+
+        fn create_mock_callback(response_json: &str) -> js_sys::Function {
+            let json_str = response_json.to_string();
+            let result = WasmEncodedResult::Success::<String> {
+                value: json_str,
+                error: None,
+            };
+            let payload = js_sys::JSON::stringify(&serde_wasm_bindgen::to_value(&result).unwrap())
+                .unwrap()
+                .as_string()
+                .unwrap();
+
+            let closure =
+                Closure::wrap(Box::new(move |_sql: String, _params: JsValue| -> JsValue {
+                    js_sys::JSON::parse(&payload).unwrap()
+                })
+                    as Box<dyn Fn(String, JsValue) -> JsValue>);
+
+            closure.into_js_value().dyn_into().unwrap()
+        }
+
+        #[wasm_bindgen_test]
+        async fn test_get_by_tx_hash_returns_transaction_when_found() {
+            let tx_hash =
+                b256!("0xabcdef1234567890abcdef1234567890abcdef1234567890abcdef1234567890");
+            let sender = address!("0x1111111111111111111111111111111111111111");
+            let orderbook = address!("0x2222222222222222222222222222222222222222");
+
+            let tx_json = json!([{
+                "transactionHash": tx_hash.to_string(),
+                "blockNumber": 12345,
+                "blockTimestamp": 1700000000,
+                "sender": sender.to_string()
+            }]);
+
+            let callback = create_mock_callback(&tx_json.to_string());
+            let exec = JsCallbackExecutor::from_ref(&callback);
+            let local_db = LocalDb::new(exec);
+
+            let transactions = LocalDbTransactions::new(&local_db);
+            let ob_id = OrderbookIdentifier::new(1, orderbook);
+
+            let result = transactions.get_by_tx_hash(&ob_id, tx_hash).await;
+
+            assert!(result.is_ok());
+            let tx = result.unwrap();
+            assert!(tx.is_some());
+            let tx = tx.unwrap();
+            assert_eq!(tx.id().to_lowercase(), tx_hash.to_string().to_lowercase());
+        }
+
+        #[wasm_bindgen_test]
+        async fn test_get_by_tx_hash_returns_none_when_not_found() {
+            let tx_hash =
+                b256!("0xabcdef1234567890abcdef1234567890abcdef1234567890abcdef1234567890");
+            let orderbook = address!("0x2222222222222222222222222222222222222222");
+
+            let callback = create_mock_callback("[]");
+            let exec = JsCallbackExecutor::from_ref(&callback);
+            let local_db = LocalDb::new(exec);
+
+            let transactions = LocalDbTransactions::new(&local_db);
+            let ob_id = OrderbookIdentifier::new(1, orderbook);
+
+            let result = transactions.get_by_tx_hash(&ob_id, tx_hash).await;
+
+            assert!(result.is_ok());
+            assert!(result.unwrap().is_none());
+        }
+    }
+}

crates/common/src/raindex_client/mod.rs

@@ -250,8 +250,8 @@ pub enum RaindexError {
     NoNetworksConfigured,
     #[error("Subgraph not configured for chain ID: {0}")]
     SubgraphNotConfigured(String),
-    #[error("Subgraph did not index transaction {tx_hash:#x} after {attempts} attempts")]
-    SubgraphIndexingTimeout { tx_hash: B256, attempts: usize },
+    #[error("Transaction {tx_hash:#x} was not indexed after {attempts} attempts")]
+    TransactionIndexingTimeout { tx_hash: B256, attempts: usize },
     #[error(transparent)]
     YamlError(#[from] YamlError),
     #[error(transparent)]
@@ -354,9 +354,9 @@ impl RaindexError {
             RaindexError::SubgraphNotConfigured(chain_id) => {
                 format!("No subgraph is configured for chain ID '{}'.", chain_id)
             }
-            RaindexError::SubgraphIndexingTimeout { tx_hash, attempts } => {
+            RaindexError::TransactionIndexingTimeout { tx_hash, attempts } => {
                 format!(
-                    "Timeout waiting for the subgraph to index transaction {tx_hash:#x} after {attempts} attempts."
+                    "Timeout waiting for transaction {tx_hash:#x} to be indexed after {attempts} attempts."
                 )
             }
             RaindexError::YamlError(err) => format!(