feat: verify L1 message index (#241)

* feat: verify L1 message index

* fix: propagate errors

* fix: apply suggestions

* fix: clippy

---------

Co-authored-by: greg <[email protected]>
Co-authored-by: Gregory Edison <[email protected]>

Author: varun-doshi

crates/database/db/src/db.rs

@@ -298,6 +298,46 @@ mod test {
             .unwrap()
             .unwrap();
         assert_eq!(l1_message_1, l1_message_1_from_db);
+        let l1_message_2_from_db = db
+            .get_l1_message_by_index(l1_message_2.transaction.queue_index)
+            .await
+            .unwrap()
+            .unwrap();
+
+        assert_eq!(l1_message_2, l1_message_2_from_db);
+    }
+
+    #[tokio::test]
+    async fn test_database_duplicate_l1_index() {
+        // Setup the test database.
+        let db = setup_test_db().await;
+
+        // Generate unstructured bytes.
+        let mut bytes = [0u8; 2048];
+        rand::rng().fill(bytes.as_mut_slice());
+        let mut u = Unstructured::new(&bytes);
+
+        // Generate 2 random L1Messages.
+        let mut l1_message_1 = L1MessageEnvelope::arbitrary(&mut u).unwrap();
+        let original_l1_message_1 = l1_message_1.clone();
+        let l1_message_2 = L1MessageEnvelope::arbitrary(&mut u).unwrap();
+
+        // Insert the L1Messages into the database in a transaction.
+        let tx = db.tx().await.unwrap();
+        tx.insert_l1_message(l1_message_1.clone()).await.unwrap();
+        tx.insert_l1_message(l1_message_2.clone()).await.unwrap();
+        // Modify l1_block_number of l1_message_1 and attempt to insert again
+        l1_message_1.l1_block_number = 1000;
+        tx.insert_l1_message(l1_message_1.clone()).await.unwrap();
+        tx.commit().await.unwrap();
+
+        // Check that the L1Messages are in the database.
+        let l1_message_1_from_db = db
+            .get_l1_message_by_index(l1_message_1.transaction.queue_index)
+            .await
+            .unwrap()
+            .unwrap();
+        assert_eq!(original_l1_message_1, l1_message_1_from_db);
         let l1_message_2_from_db = db
             .get_l1_message_by_index(l1_message_2.transaction.queue_index)
             .await

crates/database/db/src/operations.rs

@@ -9,8 +9,8 @@ use rollup_node_primitives::{
 use scroll_alloy_rpc_types_engine::BlockDataHint;
 use sea_orm::{
     sea_query::{Expr, OnConflict},
-    ActiveModelTrait, ColumnTrait, Condition, EntityTrait, QueryFilter, QueryOrder, QuerySelect,
-    Set,
+    ActiveModelTrait, ColumnTrait, Condition, DbErr, EntityTrait, QueryFilter, QueryOrder,
+    QuerySelect, Set,
 };
 
 /// The [`DatabaseOperations`] trait provides methods for interacting with the database.
@@ -170,26 +170,21 @@ pub trait DatabaseOperations: DatabaseConnectionProvider {
 
     /// Insert an [`L1MessageEnvelope`] into the database.
     async fn insert_l1_message(&self, l1_message: L1MessageEnvelope) -> Result<(), DatabaseError> {
-        tracing::trace!(target: "scroll::db", queue_index = l1_message.transaction.queue_index, "Inserting L1 message into database.");
+        let l1_index = l1_message.transaction.queue_index;
+        tracing::trace!(target: "scroll::db", queue_index = l1_index, "Inserting L1 message into database.");
+
         let l1_message: models::l1_message::ActiveModel = l1_message.into();
-        models::l1_message::Entity::insert(l1_message)
-            .on_conflict(
-                OnConflict::column(models::l1_message::Column::QueueIndex)
-                    .update_columns(vec![
-                        models::l1_message::Column::QueueHash,
-                        models::l1_message::Column::Hash,
-                        models::l1_message::Column::L1BlockNumber,
-                        models::l1_message::Column::GasLimit,
-                        models::l1_message::Column::To,
-                        models::l1_message::Column::Value,
-                        models::l1_message::Column::Sender,
-                        models::l1_message::Column::Input,
-                    ])
-                    .to_owned(),
-            )
+        let result = models::l1_message::Entity::insert(l1_message)
+            .on_conflict_do_nothing()
             .exec(self.get_connection())
-            .await?;
-        Ok(())
+            .await;
+
+        if matches!(result, Err(DbErr::RecordNotInserted)) {
+            tracing::error!(target: "scroll::db", queue_index = l1_index, "L1 message already exists");
+            Ok(())
+        } else {
+            Ok(result.map(|_| ())?)
+        }
     }
 
     /// Delete all [`L1MessageEnvelope`]s with a block number greater than the provided block