Bundle complete

Author: danyalprout

CLAUDE.md

@@ -1,11 +1,12 @@
 # Project: Tips - Transaction Inclusion Pipeline Services
 
 ## Notes
+- DO NOT ADD COMMENTS UNLESS INSTRUCTED
+- Put imports at the top of the file, never in functions
 - Always run `just ci` before claiming a task is complete and fix any issues
 - Use `just fix` to fix formatting and warnings
-- Only add comments when the implementation logic is unclear, i.e. do not comment insert item into database when the code is db.insert(item)
 - Always add dependencies to the cargo.toml in the root and reference them in the crate cargo files
-- Use https://crates.io/ to find dependency versions when adding new deps
+- Always use the latest dependency versions. Use https://crates.io/ to find dependency versions when adding new deps
 
 ## Project Structure
 ```

Cargo.toml

@@ -4,12 +4,23 @@ resolver = "2"
 
 [workspace.dependencies]
 jsonrpsee = { version = "0.26.0", features = ["server", "macros"] }
+
+# alloy
+alloy-primitives = { version = "1.3.1", default-features = false, features = [
+    "map-foldhash",
+] }
+alloy-rpc-types = { version = "1.0.30", default-features = false }
+alloy-consensus = { version = "1.0.30" }
+alloy-provider = { version = "1.0.30" }
+alloy-rpc-client = { version = "1.0.30" }
 alloy-rpc-types-mev = "1.0.30"
-alloy-rpc-types = "1.0.30"
-alloy-primitives = "1.3.1"
-alloy-provider = "1.0.30"
 alloy-transport-http = "1.0.30"
-alloy-rpc-client = "1.0.30"
+alloy-rlp = "0.3.12"
+
+# op-alloy
+op-alloy-rpc-types = { version = "0.19.0" }
+op-alloy-consensus = { version = "0.19.0", features = ["k256"] }
+
 tokio = { version = "1.47.1", features = ["full"] }
 tracing = "0.1.41"
 tracing-subscriber = "0.3.20"

crates/datastore/Cargo.toml

@@ -4,19 +4,18 @@ version = "0.1.0"
 edition = "2024"
 
 [dependencies]
-sqlx = { workspace = true }
-uuid = { workspace = true }
-serde = { workspace = true }
-chrono = { workspace = true }
-tokio = { workspace = true }
-anyhow = { workspace = true }
-async-trait = { workspace = true }
-alloy-rpc-types-mev = { workspace = true }
-alloy-primitives = { workspace = true }
-serde_json = { workspace = true }
-eyre = { workspace = true }
-tracing = { workspace = true }
+sqlx.workspace = true
+uuid.workspace = true
+tokio.workspace = true
+anyhow.workspace = true
+async-trait.workspace = true
+alloy-rpc-types-mev.workspace = true
+alloy-primitives.workspace = true
+alloy-consensus.workspace = true
+op-alloy-consensus.workspace = true
+eyre.workspace = true
+tracing.workspace = true
 
 [dev-dependencies]
-testcontainers = { workspace = true }
-testcontainers-modules = { workspace = true }
+testcontainers.workspace = true
+testcontainers-modules.workspace = true

crates/datastore/migrations/1757444171_create_bundles_table.sql

@@ -3,7 +3,7 @@ CREATE TABLE IF NOT EXISTS bundles (
     id UUID PRIMARY KEY,
 
     senders CHAR(42)[],
-    minimum_base_fee BIGINT, -- todo validate it's large enough
+    minimum_base_fee BIGINT, -- todo find a larger type
     txn_hashes CHAR(66)[],
 
     txs TEXT[] NOT NULL,

crates/datastore/src/postgres.rs

@@ -1,12 +1,25 @@
 use crate::traits::BundleDatastore;
-use alloy_primitives::TxHash;
+use alloy_consensus::Transaction;
+use alloy_consensus::private::alloy_eips::Decodable2718;
+use alloy_consensus::transaction::SignerRecoverable;
 use alloy_primitives::hex::{FromHex, ToHexExt};
+use alloy_primitives::{Address, TxHash};
 use alloy_rpc_types_mev::EthSendBundle;
 use anyhow::Result;
+use op_alloy_consensus::OpTxEnvelope;
 use sqlx::PgPool;
 use tracing::info;
 use uuid::Uuid;
 
+/// Extended bundle data that includes the original bundle plus extracted metadata
+#[derive(Debug, Clone)]
+pub struct BundleWithMetadata {
+    pub bundle: EthSendBundle,
+    pub txn_hashes: Vec<TxHash>,
+    pub senders: Vec<Address>,
+    pub min_base_fee: i64,
+}
+
 /// PostgreSQL implementation of the BundleDatastore trait
 pub struct PostgresDatastore {
     pool: PgPool,
@@ -28,11 +41,46 @@ impl PostgresDatastore {
     }
 }
 
+impl PostgresDatastore {
+    fn extract_bundle_metadata(
+        &self,
+        bundle: &EthSendBundle,
+    ) -> Result<(Vec<String>, i64, Vec<String>)> {
+        let mut senders = Vec::new();
+        let mut txn_hashes = Vec::new();
+
+        let mut min_base_fee = i64::MAX;
+
+        for tx_bytes in &bundle.txs {
+            let envelope = OpTxEnvelope::decode_2718_exact(tx_bytes)?;
+            txn_hashes.push(envelope.hash().encode_hex_with_prefix());
+
+            let sender = match envelope.recover_signer() {
+                Ok(signer) => signer,
+                Err(err) => return Err(err.into()),
+            };
+
+            senders.push(sender.encode_hex_with_prefix());
+            min_base_fee = min_base_fee.min(envelope.max_fee_per_gas() as i64); // todo type and todo not right
+        }
+
+        let minimum_base_fee = if min_base_fee == i64::MAX {
+            0
+        } else {
+            min_base_fee
+        };
+
+        Ok((senders, minimum_base_fee, txn_hashes))
+    }
+}
+
 #[async_trait::async_trait]
 impl BundleDatastore for PostgresDatastore {
     async fn insert_bundle(&self, bundle: EthSendBundle) -> Result<Uuid> {
         let id = Uuid::new_v4();
 
+        let (senders, minimum_base_fee, txn_hashes) = self.extract_bundle_metadata(&bundle)?;
+
         let txs: Vec<String> = bundle
             .txs
             .iter()
@@ -52,13 +100,17 @@ impl BundleDatastore for PostgresDatastore {
         sqlx::query!(
             r#"
             INSERT INTO bundles (
-                id, txs, reverting_tx_hashes, dropping_tx_hashes, 
+                id, senders, minimum_base_fee, txn_hashes, 
+                txs, reverting_tx_hashes, dropping_tx_hashes, 
                 block_number, min_timestamp, max_timestamp,
                 created_at, updated_at
             )
-            VALUES ($1, $2, $3, $4, $5, $6, $7, NOW(), NOW())
+            VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, NOW(), NOW())
             "#,
             id,
+            &senders,
+            minimum_base_fee,
+            &txn_hashes,
             &txs,
             &reverting_tx_hashes,
             &dropping_tx_hashes,
@@ -72,11 +124,11 @@ impl BundleDatastore for PostgresDatastore {
         Ok(id)
     }
 
-    async fn get_bundle(&self, id: Uuid) -> Result<Option<EthSendBundle>> {
+    async fn get_bundle(&self, id: Uuid) -> Result<Option<BundleWithMetadata>> {
         let result = sqlx::query!(
             r#"
-            SELECT txs, reverting_tx_hashes, dropping_tx_hashes, 
-                   block_number, min_timestamp, max_timestamp
+            SELECT senders, minimum_base_fee, txn_hashes, txs, reverting_tx_hashes, 
+                   dropping_tx_hashes, block_number, min_timestamp, max_timestamp
             FROM bundles 
             WHERE id = $1
             "#,
@@ -104,7 +156,7 @@ impl BundleDatastore for PostgresDatastore {
                     .map(TxHash::from_hex)
                     .collect();
 
-                Ok(Some(EthSendBundle {
+                let bundle = EthSendBundle {
                     txs: txs?,
                     block_number: row.block_number.unwrap_or(0) as u64,
                     min_timestamp: row.min_timestamp.map(|t| t as u64),
@@ -116,6 +168,27 @@ impl BundleDatastore for PostgresDatastore {
                     refund_recipient: None,
                     refund_tx_hashes: Vec::new(),
                     extra_fields: Default::default(),
+                };
+
+                let txn_hashes: Result<Vec<TxHash>, _> = row
+                    .txn_hashes
+                    .unwrap_or_default()
+                    .into_iter()
+                    .map(TxHash::from_hex)
+                    .collect();
+
+                let senders: Result<Vec<Address>, _> = row
+                    .senders
+                    .unwrap_or_default()
+                    .into_iter()
+                    .map(Address::from_hex)
+                    .collect();
+
+                Ok(Some(BundleWithMetadata {
+                    bundle,
+                    txn_hashes: txn_hashes?,
+                    senders: senders?,
+                    min_base_fee: row.minimum_base_fee.unwrap_or(0),
                 }))
             }
             None => Ok(None),

crates/datastore/src/traits.rs

@@ -1,3 +1,4 @@
+use crate::postgres::BundleWithMetadata;
 use alloy_rpc_types_mev::EthSendBundle;
 use anyhow::Result;
 use uuid::Uuid;
@@ -8,8 +9,8 @@ pub trait BundleDatastore: Send + Sync {
     /// Insert a new bundle into the datastore
     async fn insert_bundle(&self, bundle: EthSendBundle) -> Result<Uuid>;
 
-    /// Fetch a bundle by its ID
-    async fn get_bundle(&self, id: Uuid) -> Result<Option<EthSendBundle>>;
+    /// Fetch a bundle with metadata by its ID
+    async fn get_bundle(&self, id: Uuid) -> Result<Option<BundleWithMetadata>>;
 
     /// Cancel a bundle by UUID
     async fn cancel_bundle(&self, id: Uuid) -> Result<()>;

crates/datastore/tests/datastore.rs

@@ -1,4 +1,4 @@
-use alloy_primitives::{Bytes, TxHash};
+use alloy_primitives::{Address, Bytes, TxHash};
 use alloy_rpc_types_mev::EthSendBundle;
 use datastore::{PostgresDatastore, traits::BundleDatastore};
 use sqlx::PgPool;
@@ -35,34 +35,40 @@ async fn insert_and_get() -> eyre::Result<()> {
     let harness = setup_datastore().await?;
     let test_bundle = EthSendBundle {
         txs: vec![
-            "0x02f86f0102843b9aca0085029e7822d68298f094d2c8e0b2e8f2a8e8f2a8e8f2a8e8f2a8e8f2a880b844a9059cbb000000000000000000000000d2c8e0b2e8f2a8e8f2a8e8f2a8e8f2a8e8f2a80000000000000000000000000000000000000000000000000de0b6b3a7640000c080a0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef0a0fedcba9876543210fedcba9876543210fedcba9876543210fedcba9876543210".parse::<Bytes>()?,
+            "0x02f8bf8221058304f8c782038c83d2a76b833d0900942e85c218afcdeb3d3b3f0f72941b4861f915bbcf80b85102000e0000000bb800001010c78c430a094eb7ae67d41a7cca25cdb9315e63baceb03bf4529e57a6b1b900010001f4000a101010110111101111011011faa7efc8e6aa13b029547eecbf5d370b4e1e52eec080a009fc02a6612877cec7e1223f0a14f9a9507b82ef03af41fcf14bf5018ccf2242a0338b46da29a62d28745c828077327588dc82c03a4b0210e3ee1fd62c608f8fcd".parse::<Bytes>()?,
         ],
         block_number: 12345,
         min_timestamp: Some(1640995200),
         max_timestamp: Some(1640995260),
         reverting_tx_hashes: vec![
-            "0x1234567890abcdef1234567890abcdef1234567890abcdef1234567890abcdef".parse::<TxHash>()?,
+            "0x3ea7e1482485387e61150ee8e5c8cad48a14591789ac02cc2504046d96d0a5f4".parse::<TxHash>()?,
         ],
         replacement_uuid: None,
-        dropping_tx_hashes: vec![
-            "0xfedcba0987654321fedcba0987654321fedcba0987654321fedcba0987654321".parse::<TxHash>()?,
-        ],
+        dropping_tx_hashes: vec![],
         refund_percent: None,
         refund_recipient: None,
         refund_tx_hashes: vec![],
         extra_fields: Default::default(),
     };
 
     let insert_result = harness.data_store.insert_bundle(test_bundle.clone()).await;
+    if let Err(ref err) = insert_result {
+        eprintln!("Insert failed with error: {:?}", err);
+    }
     assert!(insert_result.is_ok());
     let bundle_id = insert_result.unwrap();
 
     let query_result = harness.data_store.get_bundle(bundle_id).await;
     assert!(query_result.is_ok());
-    let retrieved_bundle = query_result.unwrap();
+    let retrieved_bundle_with_metadata = query_result.unwrap();
+
+    assert!(
+        retrieved_bundle_with_metadata.is_some(),
+        "Bundle should be found"
+    );
+    let metadata = retrieved_bundle_with_metadata.unwrap();
+    let retrieved_bundle = &metadata.bundle;
 
-    assert!(retrieved_bundle.is_some(), "Bundle should be found");
-    let retrieved_bundle = retrieved_bundle.unwrap();
     assert_eq!(retrieved_bundle.txs.len(), test_bundle.txs.len());
     assert_eq!(retrieved_bundle.block_number, test_bundle.block_number);
     assert_eq!(retrieved_bundle.min_timestamp, test_bundle.min_timestamp);
@@ -76,5 +82,31 @@ async fn insert_and_get() -> eyre::Result<()> {
         test_bundle.dropping_tx_hashes.len()
     );
 
+    assert!(
+        !metadata.txn_hashes.is_empty(),
+        "Transaction hashes should not be empty"
+    );
+    assert!(!metadata.senders.is_empty(), "Senders should not be empty");
+    assert_eq!(
+        metadata.txn_hashes.len(),
+        1,
+        "Should have one transaction hash"
+    );
+    assert_eq!(metadata.senders.len(), 1, "Should have one sender");
+    assert!(
+        metadata.min_base_fee >= 0,
+        "Min base fee should be non-negative"
+    );
+
+    let expected_hash: TxHash =
+        "0x3ea7e1482485387e61150ee8e5c8cad48a14591789ac02cc2504046d96d0a5f4".parse()?;
+    let expected_sender: Address = "0x24ae36512421f1d9f6e074f00ff5b8393f5dd925".parse()?;
+
+    assert_eq!(
+        metadata.txn_hashes[0], expected_hash,
+        "Transaction hash should match"
+    );
+    assert_eq!(metadata.senders[0], expected_sender, "Sender should match");
+
     Ok(())
 }