update

Author: m30m

apps/fortuna/src/api.rs

@@ -1,6 +1,4 @@
 use crate::history::History;
-use serde::Serialize;
-use utoipa::ToSchema;
 use {
     crate::{
         chain::reader::{BlockNumber, BlockStatus, EntropyReader},

apps/fortuna/src/api/explorer.rs

@@ -2,30 +2,19 @@ use crate::api::{ChainId, RestError};
 use crate::history::RequestStatus;
 use axum::extract::{Query, State};
 use axum::Json;
-use ethers::types::TxHash;
+use ethers::types::{Address, TxHash};
+use std::str::FromStr;
 use utoipa::{IntoParams, ToSchema};
 
 #[derive(Debug, serde::Serialize, serde::Deserialize, IntoParams)]
 #[into_params(parameter_in=Query)]
 pub struct ExplorerQueryParams {
-    pub mode: ExplorerQueryParamsMode,
-
     pub min_timestamp: Option<u64>,
     pub max_timestamp: Option<u64>,
-    pub sequence_id: Option<u64>,
-    #[param(value_type = Option<String>)]
-    pub tx_hash: Option<TxHash>,
+    pub query: Option<String>,
     #[param(value_type = Option<String>)]
     pub chain_id: Option<ChainId>,
 }
-#[derive(Debug, serde::Serialize, serde::Deserialize, ToSchema)]
-#[serde(rename_all = "kebab-case")]
-pub enum ExplorerQueryParamsMode {
-    TxHash,
-    ChainAndSequence,
-    ChainAndTimestamp,
-    Timestamp,
-}
 
 #[utoipa::path(
     get,
@@ -39,35 +28,27 @@ pub async fn explorer(
     State(state): State<crate::api::ApiState>,
     Query(query_params): Query<ExplorerQueryParams>,
 ) -> anyhow::Result<Json<Vec<RequestStatus>>, RestError> {
-    let result = match query_params.mode {
-        ExplorerQueryParamsMode::TxHash => {
-            let tx_hash = query_params.tx_hash.ok_or(RestError::BadFilterParameters(
-                "tx_hash is required when mode=tx-hash".to_string(),
-            ))?;
-            state.history.get_request_logs_by_tx_hash(tx_hash).await
-        }
-        ExplorerQueryParamsMode::ChainAndSequence => {
-            let chain_id = query_params.chain_id.ok_or(RestError::BadFilterParameters(
-                "chain_id is required when mode=chain-and-sequence".to_string(),
-            ))?;
-            let sequence_id = query_params
-                .sequence_id
-                .ok_or(RestError::BadFilterParameters(
-                    "sequence_id is required when mode=chain-and-sequence".to_string(),
-                ))?;
-            state
-                .history
-                .get_request_logs(&(chain_id, sequence_id))
-                .await
-                .into_iter()
-                .collect()
+    if let Some(query) = query_params.query {
+        if let Ok(tx_hash) = TxHash::from_str(&query) {
+            return Ok(Json(state.history.get_requests_by_tx_hash(tx_hash).await));
         }
-        ExplorerQueryParamsMode::ChainAndTimestamp => {
-            vec![]
+        if let Ok(sender) = Address::from_str(&query) {
+            return Ok(Json(
+                state
+                    .history
+                    .get_requests_by_sender(sender, query_params.chain_id)
+                    .await,
+            ));
         }
-        ExplorerQueryParamsMode::Timestamp => {
-            vec![]
+        if let Ok(sequence_number) = u64::from_str(&query) {
+            return Ok(Json(
+                state
+                    .history
+                    .get_requests_by_sequence(sequence_number, query_params.chain_id)
+                    .await,
+            ));
         }
-    };
-    Ok(Json(result))
+    }
+    //TODO: handle more types of queries
+    Ok(Json(vec![]))
 }

apps/fortuna/src/chain/reader.rs

@@ -1,5 +1,4 @@
 use ethers::prelude::LogMeta;
-use ethers::types::TxHash;
 use {
     anyhow::Result,
     axum::async_trait,

apps/fortuna/src/command/run.rs

@@ -41,7 +41,6 @@ pub async fn run_api(
     schemas(
     crate::api::GetRandomValueResponse,
     crate::history::RequestStatus,
-    crate::api::ExplorerQueryParamsMode,
     crate::api::Blob,
     crate::api::BinaryEncoding,
     )

apps/fortuna/src/eth_utils/utils.rs

@@ -157,7 +157,6 @@ pub async fn submit_tx_with_backoff<T: Middleware + NonceManaged + 'static>(
     call: ContractCall<T, ()>,
     gas_limit: U256,
     escalation_policy: EscalationPolicy,
-    history: Arc<History>,
 ) -> Result<SubmitTxResult> {
     let start_time = std::time::Instant::now();
 

apps/fortuna/src/history.rs

@@ -93,11 +93,10 @@ impl From<RequestRow> for RequestStatus {
 pub struct History {
     pool: Pool<Sqlite>,
     write_queue: mpsc::Sender<RequestStatus>,
-    writer_thread: Arc<tokio::task::JoinHandle<()>>,
+    _writer_thread: Arc<tokio::task::JoinHandle<()>>,
 }
 
 impl History {
-    const MAX_HISTORY: usize = 1_000_000;
     const MAX_WRITE_QUEUE: usize = 1_000;
     pub async fn new() -> Self {
         Self::new_with_url("sqlite:fortuna.db").await
@@ -124,14 +123,14 @@ impl History {
         Self {
             pool,
             write_queue: sender,
-            writer_thread: Arc::new(writer_thread),
+            _writer_thread: Arc::new(writer_thread),
         }
     }
 
     async fn update_request_status(pool: &Pool<Sqlite>, new_status: RequestStatus) {
         let sequence = new_status.sequence as i64;
         let chain_id = new_status.chain_id;
-        let state = match new_status.state {
+        match new_status.state {
             RequestEntryState::Pending => {
                 let block_number = new_status.request_block_number as i64;
                 let request_tx_hash: String = new_status.request_tx_hash.encode_hex();
@@ -204,7 +203,7 @@ impl History {
         self.get_from_db(request_key.clone()).await
     }
 
-    pub async fn get_request_logs_by_tx_hash(&self, tx_hash: TxHash) -> Vec<RequestStatus> {
+    pub async fn get_requests_by_tx_hash(&self, tx_hash: TxHash) -> Vec<RequestStatus> {
         let tx_hash: String = tx_hash.encode_hex();
         let rows = sqlx::query_as!(
             RequestRow,
@@ -218,6 +217,58 @@ impl History {
         rows.into_iter().map(|row| row.into()).collect()
     }
 
+    pub async fn get_requests_by_sender(
+        &self,
+        sender: Address,
+        chain_id: Option<ChainId>,
+    ) -> Vec<RequestStatus> {
+        let sender: String = sender.encode_hex();
+        let rows = match chain_id {
+            Some(chain_id) => sqlx::query_as!(
+                RequestRow,
+                "SELECT * FROM request WHERE sender = ? AND chain_id = ?",
+                sender,
+                chain_id,
+            )
+            .fetch_all(&self.pool)
+            .await
+            .unwrap(),
+            None => sqlx::query_as!(RequestRow, "SELECT * FROM request WHERE sender = ?", sender,)
+                .fetch_all(&self.pool)
+                .await
+                .unwrap(),
+        };
+        rows.into_iter().map(|row| row.into()).collect()
+    }
+
+    pub async fn get_requests_by_sequence(
+        &self,
+        sequence: u64,
+        chain_id: Option<ChainId>,
+    ) -> Vec<RequestStatus> {
+        let sequence = sequence as i64;
+        let rows = match chain_id {
+            Some(chain_id) => sqlx::query_as!(
+                RequestRow,
+                "SELECT * FROM request WHERE sequence = ? AND chain_id = ?",
+                sequence,
+                chain_id,
+            )
+            .fetch_all(&self.pool)
+            .await
+            .unwrap(),
+            None => sqlx::query_as!(
+                RequestRow,
+                "SELECT * FROM request WHERE sequence = ?",
+                sequence,
+            )
+            .fetch_all(&self.pool)
+            .await
+            .unwrap(),
+        };
+        rows.into_iter().map(|row| row.into()).collect()
+    }
+
     pub async fn get_latest_requests(
         &self,
         chain_id: Option<&ChainId>,
@@ -251,7 +302,8 @@ impl History {
 }
 
 mod tests {
-    use super::*;
+    use crate::history::{History, RequestEntryState, RequestStatus};
+    use ethers::types::{Address, TxHash};
     use tokio::time::sleep;
 
     #[tokio::test]
@@ -274,7 +326,7 @@ mod tests {
 
     #[tokio::test(flavor = "multi_thread")]
     async fn test_writer_thread() {
-        let mut history = History::new_in_memory().await;
+        let history = History::new_in_memory().await;
         let status = RequestStatus {
             chain_id: "ethereum".to_string(),
             sequence: 1,

apps/fortuna/src/keeper/process_event.rs

@@ -74,7 +74,6 @@ pub async fn process_event_with_backoff(
         contract_call,
         gas_limit,
         escalation_policy,
-        history.clone(),
     )
     .await;