fix(su): su assign now checking regular messages along with baselayer #…

…717

servers/su/src/domain/clients/gateway.rs

@@ -1,8 +1,9 @@
 use crate::domain::config::AoConfig;
-use crate::domain::core::dal::{Gateway, NetworkInfo, TxStatus};
+use crate::domain::core::dal::{Gateway, GatewayTx, NetworkInfo, TxStatus};
 use arweave_rs::network::NetworkInfoClient;
 use async_trait::async_trait;
 use reqwest::{Client, Url};
+use serde_derive::Deserialize;
 use std::sync::Arc;
 use tokio::sync::Mutex;
 use tokio::time::{sleep, Duration};
@@ -17,6 +18,8 @@ pub struct ArweaveGateway {
 pub enum GatewayErrorType {
     CheckHeadError(String),
     StatusError(String),
+    GraphQLError(String),
+    JsonParseError(String),
 }
 
 impl From<GatewayErrorType> for String {
@@ -25,6 +28,36 @@ impl From<GatewayErrorType> for String {
     }
 }
 
+/*
+  Right now we dont need all the fields
+  but later we can add to these types to
+  pull more data from gql responses
+*/
+#[derive(Deserialize, Debug, Clone)]
+struct Node {
+    id: String,
+}
+
+#[derive(Deserialize, Debug)]
+struct Edge {
+    node: Node,
+}
+
+#[derive(Deserialize, Debug)]
+struct Transactions {
+    edges: Vec<Edge>,
+}
+
+#[derive(Deserialize, Debug)]
+struct Data {
+    transactions: Transactions,
+}
+
+#[derive(Deserialize, Debug)]
+struct GraphQLResponse {
+    data: Data,
+}
+
 impl ArweaveGateway {
     pub async fn new() -> Result<Self, String> {
         let network_info = ArweaveGateway::network_info_fetch().await?;
@@ -169,4 +202,46 @@ impl Gateway for ArweaveGateway {
             ))
         }
     }
+
+    async fn gql_tx(&self, tx_id: &String) -> Result<GatewayTx, String> {
+        let config = AoConfig::new(Some("su".to_string())).expect("Failed to read configuration");
+        let gateway_url = config.gateway_url;
+        let client = Client::new();
+
+        let query = serde_json::json!({
+            "query": format!(
+                "query {{ transactions (ids: [\"{}\"]){{ edges {{ node {{ id }} }} }} }}",
+                tx_id
+            ),
+            "variables": {}
+        });
+
+        let query_string = serde_json::to_string(&query)
+            .map_err(|e| GatewayErrorType::GraphQLError(e.to_string()))?;
+
+        let response = client
+            .post(format!("{}/graphql", gateway_url))
+            .header("Content-Type", "application/json")
+            .body(query_string)
+            .send()
+            .await
+            .map_err(|e| GatewayErrorType::GraphQLError(e.to_string()))?;
+
+        if response.status().is_success() {
+            let body: GraphQLResponse = response
+                .json()
+                .await
+                .map_err(|e| GatewayErrorType::JsonParseError(e.to_string()))?;
+
+            if let Some(edge) = body.data.transactions.edges.get(0) {
+                Ok(GatewayTx {
+                    id: edge.node.clone().id,
+                })
+            } else {
+                Err("Transaction not found".to_string())
+            }
+        } else {
+            Err(format!("Failed to fetch transaction: {}", response.status()).to_string())
+        }
+    }
 }

servers/su/src/domain/clients/store.rs

@@ -69,7 +69,7 @@ impl StoreClient {
         let database_url = config.database_url;
         let database_read_url = match config.database_read_url {
             Some(u) => u,
-            None => database_url.clone()
+            None => database_url.clone(),
         };
         let manager = ConnectionManager::<PgConnection>::new(database_url);
         let read_manager = ConnectionManager::<PgConnection>::new(database_read_url);
@@ -84,9 +84,10 @@ impl StoreClient {
             .test_on_check_out(true)
             .build(read_manager)
             .map_err(|_| {
-                StoreErrorType::DatabaseError("Failed to initialize read connection pool.".to_string())
+                StoreErrorType::DatabaseError(
+                    "Failed to initialize read connection pool.".to_string(),
+                )
             })?;
-
 
         Ok(StoreClient { pool, read_pool })
     }

servers/su/src/domain/config.rs

@@ -24,9 +24,7 @@ impl AoConfig {
         };
         let database_read_url = match env::var("DATABASE_READ_URL") {
             Ok(val) => Some(val),
-            Err(_e) => {
-                None
-            }
+            Err(_e) => None,
         };
         Ok(AoConfig {
             database_url: env::var("DATABASE_URL")?,

servers/su/src/domain/core/builder.rs

@@ -268,15 +268,22 @@ impl<'a> Builder<'a> {
                     )),
                 }
             }
-            None => Ok(()),
+            None => {
+                /*
+                    If this throws an error then the tx
+                    is not available on the gateway
+                */
+                self.gateway.gql_tx(&tx_id).await?;
+                Ok(())
+            }
         }
     }
 }
 
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::domain::core::dal::NetworkInfo;
+    use crate::domain::core::dal::{GatewayTx, NetworkInfo};
     use async_trait::async_trait;
     use std::sync::Arc;
 
@@ -300,6 +307,12 @@ mod tests {
                 number_of_confirmations: 0,
             })
         }
+
+        async fn gql_tx(&self, _tx_id: &String) -> Result<GatewayTx, String> {
+            Ok(GatewayTx {
+                id: "id".to_string(),
+            })
+        }
     }
 
     struct MockSigner;

servers/su/src/domain/core/dal.rs

@@ -21,11 +21,17 @@ pub struct TxStatus {
     pub number_of_confirmations: i32,
 }
 
+#[derive(Deserialize)]
+pub struct GatewayTx {
+    pub id: String,
+}
+
 #[async_trait]
 pub trait Gateway: Send + Sync {
     async fn check_head(&self, tx_id: String) -> Result<bool, String>;
     async fn network_info(&self) -> Result<NetworkInfo, String>;
     async fn status(&self, tx_id: &String) -> Result<TxStatus, String>;
+    async fn gql_tx(&self, tx_id: &String) -> Result<GatewayTx, String>;
 }
 
 pub trait Wallet: Send + Sync {

servers/su/src/domain/core/flows.rs

@@ -280,4 +280,4 @@ pub async fn health(deps: Arc<Deps>) -> Result<String, String> {
         }
         Err(e) => Err(format!("{:?}", e)),
     }
-}
+}