Fix missing replace

crates/oracle/tests/oracle_blocks.rs

@@ -12,12 +12,26 @@ use void_oracle_types::config::{
 };
 
 #[tokio::test]
-async fn test_e2e() {
-    let db = Db::memory();
+async fn test_e2e_memory() {
+    let send_db = Db::memory();
+    let recv_db = Db::memory();
+    run_e2e(send_db, recv_db).await;
+}
+
+#[tokio::test]
+#[cfg(feature = "sqlite")]
+async fn test_e2e_sqlite() {
+    let temp_dir = tempfile::TempDir::new().unwrap();
+    let send_db = Db::sqlite(temp_dir.path().join("send.db")).unwrap();
+    let recv_db = Db::sqlite(temp_dir.path().join("recv.db")).unwrap();
+    run_e2e(send_db, recv_db).await;
+}
+
+async fn run_e2e(send_db: Db, recv_db: Db) {
     let priv_key = PrivateKeySigner::random();
     let pub_key = priv_key.address();
 
-    let state = blocks_sse::State::new(db.clone()).with_signer(priv_key);
+    let state = blocks_sse::State::new(send_db.clone()).with_signer(priv_key);
 
     let bind_address = "127.0.0.1:3000".parse().unwrap();
     let conn_limit = 100;
@@ -28,7 +42,6 @@ async fn test_e2e() {
     tokio::spawn(run_block_sse_server(bind_address, conn_limit, state));
 
     tokio::spawn({
-        let db = db.clone();
         async move {
             let config = Config {
                 query: QueryConfig {
@@ -42,7 +55,7 @@ async fn test_e2e() {
                 },
             };
 
-            watch_blocks(config, 0, db)
+            watch_blocks(config, 0, send_db)
                 .for_each(|block| {
                     new_blocks.send(()).ok();
                     async {
@@ -59,8 +72,7 @@ async fn test_e2e() {
         public_key: Some(format!("{}", pub_key)),
     };
 
-    let db = Db::memory();
-    recv_blocks(config, db)
+    recv_blocks(config, recv_db)
         .scan(0, |i, block| {
             *i += 1;
             let i = *i;
@@ -75,12 +87,28 @@ async fn test_e2e() {
 }
 
 #[tokio::test]
-async fn test_relay() {
-    let db = Db::memory();
+async fn test_relay_memory() {
+    let publisher_db = Db::memory();
+    let relay_db = Db::memory();
+    let observer_db = Db::memory();
+    run_relay(publisher_db, relay_db, observer_db).await;
+}
+
+#[tokio::test]
+#[cfg(feature = "sqlite")]
+async fn test_relay_sqlite() {
+    let temp_dir = tempfile::TempDir::new().unwrap();
+    let publisher_db = Db::sqlite(temp_dir.path().join("publisher.db")).unwrap();
+    let relay_db = Db::sqlite(temp_dir.path().join("relay.db")).unwrap();
+    let observer_db = Db::sqlite(temp_dir.path().join("observer.db")).unwrap();
+    run_relay(publisher_db, relay_db, observer_db).await;
+}
+
+async fn run_relay(publisher_db: Db, relay_db: Db, observer_db: Db) {
     let priv_key = PrivateKeySigner::random();
     let pub_key = priv_key.address();
 
-    let state = blocks_sse::State::new(db.clone()).with_signer(priv_key);
+    let state = blocks_sse::State::new(publisher_db.clone()).with_signer(priv_key);
 
     let bind_address = "127.0.0.1:3300".parse().unwrap();
     let conn_limit = 100;
@@ -91,7 +119,6 @@ async fn test_relay() {
     tokio::spawn(run_block_sse_server(bind_address, conn_limit, state));
 
     tokio::spawn({
-        let db = db.clone();
         async move {
             let config = Config {
                 query: QueryConfig {
@@ -105,7 +132,7 @@ async fn test_relay() {
                 },
             };
 
-            watch_blocks(config, 0, db)
+            watch_blocks(config, 0, publisher_db)
                 .for_each(|block| {
                     new_blocks.send(()).ok();
                     async {
@@ -122,17 +149,15 @@ async fn test_relay() {
         public_key: Some(format!("{}", pub_key)),
     };
 
-    let db = Db::memory();
-
-    let state = blocks_sse::State::new(db.clone()).forward_signatures();
+    let state = blocks_sse::State::new(relay_db.clone()).forward_signatures();
 
     let bind_address = "127.0.0.1:3030".parse().unwrap();
     let conn_limit = 100;
     let new_blocks = state.new_blocks.clone();
 
     tokio::spawn(run_block_sse_server(bind_address, conn_limit, state));
 
-    tokio::spawn(recv_signed_blocks(config, db).for_each(move |_| {
+    tokio::spawn(recv_signed_blocks(config, relay_db).for_each(move |_| {
         new_blocks.send(()).ok();
         std::future::ready(())
     }));
@@ -143,8 +168,7 @@ async fn test_relay() {
         public_key: Some(format!("{}", pub_key)),
     };
 
-    let db = Db::memory();
-    recv_blocks(config, db)
+    recv_blocks(config, observer_db)
         .scan(0, |i, block| {
             *i += 1;
             let i = *i;

crates/void-oracle-db-sql/sql/insert/overwrite_block.sql

@@ -1,3 +1,4 @@
-INSERT INTO blocks (id, block_number, timestamp_ms)
+INSERT
+    OR REPLACE INTO blocks (id, block_number, timestamp_ms)
 VALUES
     (1, ?, 0)

crates/void-oracle-db-sql/sql/insert/overwrite_events.sql

@@ -1,2 +1,4 @@
-INSERT INTO events (hash, block_number, stream_type, data)
-VALUES (?, ?, ?, ?);
+INSERT
+    OR REPLACE INTO EVENTS (hash, block_number, stream_type, data)
+VALUES
+    (?, ?, ?, ?);

crates/void-oracle-db/src/sqlite.rs

@@ -225,7 +225,7 @@ pub fn get_signed_range(conn: &mut Connection, from: u64) -> Result<Vec<SignedBl
         let block_number: i64 = row.get(0)?;
         let block_number = block_number as u64;
 
-        let signature: [u8; 65] = row.get("1")?;
+        let signature: [u8; 65] = row.get(1)?;
         let Some(block) = blocks.next() else {
             break;
         };