sqlite migration

m30m · m30m · commit 7b2cdaf44d06 · 2025-05-13T11:31:44.000-07:00
diff --git a/apps/fortuna/migrations/20250502164500_init.up.sql b/apps/fortuna/migrations/20250502164500_init.up.sql
@@ -1,5 +1,5 @@
 CREATE TABLE log(
-        id INTEGER PRIMARY KEY AUTOINCREMENT,
+        id INTEGER PRIMARY KEY AUTOINCREMENT NOT NULL,
         chain_id VARCHAR(255) NOT NULL,
         sequence INTEGER NOT NULL,
         timestamp DATETIME NOT NULL,
diff --git a/apps/fortuna/src/api/explorer.rs b/apps/fortuna/src/api/explorer.rs
@@ -1,5 +1,5 @@
 use crate::api::{ChainId, RestError};
-use crate::history::RequestJournal;
+use crate::history::RequestLog;
 use axum::extract::{Query, State};
 use axum::Json;
 use ethers::types::TxHash;
@@ -38,7 +38,7 @@ pub enum ExplorerQueryParamsMode {
 pub async fn explorer(
     State(state): State<crate::api::ApiState>,
     Query(query_params): Query<ExplorerQueryParams>,
-) -> anyhow::Result<Json<Vec<RequestJournal>>, RestError> {
+) -> anyhow::Result<Json<Vec<RequestLog>>, RestError> {
     let result = match query_params.mode {
         ExplorerQueryParamsMode::TxHash => {
             let tx_hash = query_params.tx_hash.ok_or(RestError::BadFilterParameters(
@@ -49,6 +49,7 @@ pub async fn explorer(
                 .read()
                 .await
                 .get_request_logs_by_tx_hash(tx_hash)
+                .await
         }
         ExplorerQueryParamsMode::ChainAndSequence => {
             let chain_id = query_params.chain_id.ok_or(RestError::BadFilterParameters(
@@ -64,6 +65,7 @@ pub async fn explorer(
                 .read()
                 .await
                 .get_request_logs(&(chain_id, sequence_id))
+                .await
                 .into_iter()
                 .collect()
         }
diff --git a/apps/fortuna/src/command/run.rs b/apps/fortuna/src/command/run.rs
@@ -40,7 +40,7 @@ pub async fn run_api(
     components(
     schemas(
     crate::api::GetRandomValueResponse,
-    crate::history::RequestJournal,
+    crate::history::RequestLog,
     crate::api::ExplorerQueryParamsMode,
     crate::api::Blob,
     crate::api::BinaryEncoding,
diff --git a/apps/fortuna/src/history.rs b/apps/fortuna/src/history.rs
@@ -1,82 +1,70 @@
 use crate::api::ChainId;
 use chrono::{DateTime, NaiveDateTime};
+use ethers::abi::AbiEncode;
 use ethers::prelude::TxHash;
 use ethers::types::BlockNumber;
 use serde::Serialize;
 use sqlx::{Pool, Sqlite, SqlitePool};
 use std::collections::{BTreeMap, HashMap};
 use utoipa::ToSchema;
 
-#[derive(Clone, Debug, Serialize)]
-pub enum JournalLog {
+#[derive(Clone, Debug, Serialize, PartialEq)]
+pub enum RequestLogType {
     Observed { tx_hash: TxHash },
     FailedToReveal { reason: String },
     Revealed { tx_hash: TxHash },
     Landed { block_number: u64 },
 }
 
-impl JournalLog {
+impl RequestLogType {
     pub fn get_tx_hash(&self) -> Option<TxHash> {
         match self {
-            JournalLog::Observed { tx_hash } => Some(*tx_hash),
-            JournalLog::FailedToReveal { .. } => None,
-            JournalLog::Revealed { tx_hash } => Some(*tx_hash),
-            JournalLog::Landed { .. } => None,
+            RequestLogType::Observed { tx_hash } => Some(*tx_hash),
+            RequestLogType::FailedToReveal { .. } => None,
+            RequestLogType::Revealed { tx_hash } => Some(*tx_hash),
+            RequestLogType::Landed { .. } => None,
         }
     }
 
     pub fn get_info(&self) -> Option<String> {
         match self {
-            JournalLog::Observed { tx_hash } => None,
-            JournalLog::FailedToReveal { reason } => Some(reason.clone()),
-            JournalLog::Revealed { tx_hash } => None,
-            JournalLog::Landed { block_number } => None,
+            RequestLogType::Observed { tx_hash } => None,
+            RequestLogType::FailedToReveal { reason } => Some(reason.clone()),
+            RequestLogType::Revealed { tx_hash } => None,
+            RequestLogType::Landed { block_number } => None,
         }
     }
     pub fn get_type(&self) -> String {
         match self {
-            JournalLog::Observed { .. } => "Observed".to_string(),
-            JournalLog::FailedToReveal { .. } => "FailedToReveal".to_string(),
-            JournalLog::Revealed { .. } => "Revealed".to_string(),
-            JournalLog::Landed { .. } => "Landed".to_string(),
+            RequestLogType::Observed { .. } => "Observed".to_string(),
+            RequestLogType::FailedToReveal { .. } => "FailedToReveal".to_string(),
+            RequestLogType::Revealed { .. } => "Revealed".to_string(),
+            RequestLogType::Landed { .. } => "Landed".to_string(),
         }
     }
 
     pub fn get_block_number(&self) -> Option<u64> {
         match self {
-            JournalLog::Observed { .. } => None,
-            JournalLog::FailedToReveal { .. } => None,
-            JournalLog::Revealed { .. } => None,
-            JournalLog::Landed { block_number } => Some(*block_number),
+            RequestLogType::Observed { .. } => None,
+            RequestLogType::FailedToReveal { .. } => None,
+            RequestLogType::Revealed { .. } => None,
+            RequestLogType::Landed { block_number } => Some(*block_number),
         }
     }
 }
 
-#[derive(Clone, Debug, Serialize)]
-pub struct TimedJournalLog {
-    pub timestamp: DateTime<chrono::Utc>,
-    pub log: JournalLog,
-}
-
-impl TimedJournalLog {
-    pub fn with_current_time(log: JournalLog) -> Self {
-        TimedJournalLog {
-            timestamp: chrono::Utc::now(),
-            log,
-        }
-    }
-}
-
-#[derive(Clone, Debug, Serialize, ToSchema)]
-pub struct RequestJournal {
+#[derive(Clone, Debug, Serialize, ToSchema, PartialEq)]
+pub struct RequestLog {
     pub chain_id: ChainId,
     pub sequence: u64,
-    pub journal: Vec<TimedJournalLog>,
+    pub timestamp: DateTime<chrono::Utc>,
+    pub log: RequestLogType,
 }
 
 type RequestKey = (ChainId, u64);
 
 struct LogRow {
+    id: i64,
     chain_id: String,
     sequence: i64,
     timestamp: NaiveDateTime,
@@ -86,47 +74,59 @@ struct LogRow {
     tx_hash: Option<String>,
 }
 
+impl From<LogRow> for RequestLog {
+    fn from(row: LogRow) -> Self {
+        let chain_id = row.chain_id;
+        let sequence = row.sequence as u64;
+        let timestamp = row.timestamp.and_utc();
+        let log_type = match row.r#type.as_str() {
+            "Observed" => RequestLogType::Observed {
+                tx_hash: row.tx_hash.unwrap_or_default().parse().unwrap(),
+            },
+            "FailedToReveal" => RequestLogType::FailedToReveal {
+                reason: row.info.unwrap_or_default(),
+            },
+            "Revealed" => RequestLogType::Revealed {
+                tx_hash: row.tx_hash.unwrap_or_default().parse().unwrap(),
+            },
+            "Landed" => RequestLogType::Landed {
+                block_number: row.block_number.unwrap_or_default() as u64,
+            },
+            _ => panic!("Unknown log type"),
+        };
+        Self {
+            chain_id,
+            sequence,
+            timestamp,
+            log: log_type,
+        }
+    }
+}
+
 pub struct History {
-    pub by_hash: HashMap<TxHash, Vec<RequestKey>>,
-    pub by_chain_and_time: BTreeMap<(ChainId, DateTime<chrono::Utc>), RequestKey>,
-    pub by_time: BTreeMap<DateTime<chrono::Utc>, RequestKey>,
-    pub by_request_key: HashMap<RequestKey, RequestJournal>,
     pool: Pool<Sqlite>,
 }
 
 impl History {
     const MAX_HISTORY: usize = 1_000_000;
     pub async fn new() -> Self {
         let pool = SqlitePool::connect("sqlite:fortuna.db").await.unwrap();
-        Self {
-            by_hash: HashMap::new(),
-            by_chain_and_time: BTreeMap::new(),
-            by_time: BTreeMap::new(),
-            by_request_key: HashMap::new(),
-            pool,
-        }
+        Self { pool }
     }
 
-    pub async fn add_to_db(
-        &self,
-        (chain_id, sequence): RequestKey,
-        request_journal_log: TimedJournalLog,
-    ) {
-        let sequence = sequence as i64;
-        let log_type = request_journal_log.log.get_type();
-        let block_number = request_journal_log
+    pub async fn add_to_db(&self, log: RequestLog) {
+        let sequence = log.sequence as i64;
+        let log_type = log.log.get_type();
+        let block_number = log
             .log
             .get_block_number()
             .map(|block_number| block_number as i64); // sqlite does not support u64
-        let tx_hash = request_journal_log
-            .log
-            .get_tx_hash()
-            .map(|tx_hash| tx_hash.to_string());
-        let info = request_journal_log.log.get_info();
+        let tx_hash = log.log.get_tx_hash().map(|tx_hash| tx_hash.encode_hex());
+        let info = log.log.get_info();
         sqlx::query!("INSERT INTO log (chain_id, sequence, timestamp, type, block_number, info, tx_hash) VALUES (?, ?, ?, ?, ?, ?, ?)",
-            chain_id,
+            log.chain_id,
             sequence,
-            request_journal_log.timestamp,
+            log.timestamp,
             log_type,
             block_number,
             info,
@@ -136,81 +136,66 @@ impl History {
             .unwrap();
     }
 
-    pub async fn get_from_db(&self, (chain_id, sequence): RequestKey) -> Option<TimedJournalLog> {
+    pub async fn get_from_db(&self, (chain_id, sequence): RequestKey) -> Vec<RequestLog> {
         let sequence = sequence as i64;
-        let row =  sqlx::query_as!(LogRow, "SELECT chain_id, sequence, timestamp, type, block_number, info, tx_hash FROM log WHERE chain_id = ? AND sequence = ?", chain_id, sequence)
-            .fetch_optional(&self.pool)
-            .await
-            .unwrap();
-        if let Some(row) = row {
-            let ts = row.timestamp;
-            Some(TimedJournalLog {
-                timestamp: ts.and_utc(),
-                log: JournalLog::Observed {
-                    tx_hash: TxHash::zero(),
-                },
-            })
-        } else {
-            None
-        }
+        let row = sqlx::query_as!(
+            LogRow,
+            "SELECT * FROM log WHERE chain_id = ? AND sequence = ?",
+            chain_id,
+            sequence
+        )
+        .fetch_all(&self.pool)
+        .await
+        .unwrap();
+        row.into_iter().map(|row| row.into()).collect()
     }
 
-    pub fn add(&mut self, (chain_id, sequence): RequestKey, request_journal_log: TimedJournalLog) {
-        self.add_to_db((chain_id, sequence), request_journal_log);
+    pub fn add(&mut self, log: RequestLog) {
+        self.add_to_db(log);
     }
 
-    pub fn get_request_logs(&self, request_key: &RequestKey) -> Option<RequestJournal> {
-        self.by_request_key.get(request_key).cloned()
+    pub async fn get_request_logs(&self, request_key: &RequestKey) -> Vec<RequestLog> {
+        self.get_from_db(request_key.clone()).await
     }
 
-    pub fn get_request_logs_by_tx_hash(&self, tx_hash: TxHash) -> Vec<RequestJournal> {
-        self.by_hash
-            .get(&tx_hash)
-            .map(|request_keys| {
-                request_keys
-                    .iter()
-                    .map(|request_key| self.by_request_key.get(request_key).unwrap().clone())
-                    .collect()
-            })
-            .unwrap_or_default()
+    pub async fn get_request_logs_by_tx_hash(&self, tx_hash: TxHash) -> Vec<RequestLog> {
+        let tx_hash = tx_hash.encode_hex();
+        let rows = sqlx::query_as!(LogRow, "SELECT * FROM log WHERE tx_hash = ?", tx_hash)
+            .fetch_all(&self.pool)
+            .await
+            .unwrap();
+        rows.into_iter().map(|row| row.into()).collect()
     }
 
-    pub fn get_latest_requests(
+    pub async fn get_latest_requests(
         &self,
         chain_id: Option<&ChainId>,
         limit: u64,
         min_timestamp: Option<DateTime<chrono::Utc>>,
         max_timestamp: Option<DateTime<chrono::Utc>>,
-    ) -> Vec<RequestJournal> {
-        match chain_id {
+    ) -> Vec<RequestLog> {
+        let limit = limit as i64;
+        let rows = match chain_id {
             Some(chain_id) => {
-                let range = self.by_chain_and_time.range(
-                    (
-                        chain_id.clone(),
-                        min_timestamp.unwrap_or(DateTime::<chrono::Utc>::MIN_UTC),
-                    )
-                        ..(
-                            chain_id.clone(),
-                            max_timestamp.unwrap_or(DateTime::<chrono::Utc>::MAX_UTC),
-                        ),
-                );
-                range
-                    .rev()
-                    .take(limit as usize)
-                    .map(|(_, request_key)| self.by_request_key.get(request_key).unwrap().clone())
-                    .collect()
+                let chain_id = chain_id.to_string();
+                let min_timestamp = min_timestamp.unwrap_or(DateTime::<chrono::Utc>::MIN_UTC);
+                let max_timestamp = max_timestamp.unwrap_or(DateTime::<chrono::Utc>::MAX_UTC);
+                sqlx::query_as!(LogRow, "SELECT * FROM log WHERE chain_id = ? AND timestamp >= ? AND timestamp <= ? ORDER BY timestamp DESC LIMIT ?",
+                    chain_id,
+                    min_timestamp,
+                    max_timestamp,
+                    limit).fetch_all(&self.pool).await
             }
-            None => self
-                .by_time
-                .range(
-                    min_timestamp.unwrap_or(DateTime::<chrono::Utc>::MIN_UTC)
-                        ..max_timestamp.unwrap_or(DateTime::<chrono::Utc>::MAX_UTC),
-                )
-                .rev()
-                .take(limit as usize)
-                .map(|(_time, request_key)| self.by_request_key.get(request_key).unwrap().clone())
-                .collect::<Vec<_>>(),
-        }
+            None => {
+                let min_timestamp = min_timestamp.unwrap_or(DateTime::<chrono::Utc>::MIN_UTC);
+                let max_timestamp = max_timestamp.unwrap_or(DateTime::<chrono::Utc>::MAX_UTC);
+                sqlx::query_as!(LogRow, "SELECT * FROM log WHERE timestamp >= ? AND timestamp <= ? ORDER BY timestamp DESC LIMIT ?",
+                    min_timestamp,
+                    max_timestamp,
+                    limit).fetch_all(&self.pool).await
+            }
+        };
+        rows.unwrap().into_iter().map(|row| row.into()).collect()
     }
 }
 
@@ -219,20 +204,17 @@ mod tests {
 
     #[sqlx::test]
     async fn test_history(pool: Pool<Sqlite>) {
-        let history = History {
-            by_hash: HashMap::new(),
-            by_chain_and_time: BTreeMap::new(),
-            by_time: BTreeMap::new(),
-            by_request_key: HashMap::new(),
-            pool,
+        let history = History { pool };
+        let log = RequestLog {
+            chain_id: "ethereum".to_string(),
+            sequence: 1,
+            timestamp: chrono::Utc::now(),
+            log: RequestLogType::Observed {
+                tx_hash: TxHash::zero(),
+            },
         };
-        history
-            .add_to_db(
-                ("ethereum".to_string(), 1),
-                TimedJournalLog::with_current_time(JournalLog::Observed {
-                    tx_hash: TxHash::zero(),
-                }),
-            )
-            .await;
+        history.add_to_db(log.clone()).await;
+        let logs = history.get_request_logs(&("ethereum".to_string(), 1)).await;
+        assert_eq!(logs, vec![log.clone()]);
     }
 }
diff --git a/apps/fortuna/src/keeper/process_event.rs b/apps/fortuna/src/keeper/process_event.rs
