Insert/get methods

Author: danyalprout

.env.example

@@ -0,0 +1 @@
+DATABASE_URL=postgresql://postgres:postgres@localhost:5432/postgres

.gitignore

@@ -21,4 +21,4 @@ Thumbs.db
 
 # Backup files
 *~
-*.bak
+*.bak

CLAUDE.md

@@ -3,6 +3,7 @@
 ## Notes
 - 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
 

Cargo.toml

@@ -16,7 +16,13 @@ tracing-subscriber = "0.3.20"
 anyhow = "1.0.99"
 clap = { version = "4.5.47", features = ["derive", "env"] }
 url = "2.5.7"
-sqlx = { version = "0.8", features = ["runtime-tokio-rustls", "postgres", "uuid", "chrono", "migrate"] }
+sqlx = { version = "0.8.6", features = [
+    "runtime-tokio-native-tls",
+    "postgres",
+    "uuid",
+    "chrono",
+    "json",
+]}
 uuid = { version = "1.18.1", features = ["v4", "serde"] }
 serde = { version = "1.0.219", features = ["derive"] }
 chrono = { version = "0.4.42", features = ["serde"] }

crates/datastore/migrations/1757444171_create_bundles_table.sql

@@ -1,12 +1,18 @@
 -- Create bundles table
 CREATE TABLE IF NOT EXISTS bundles (
     id UUID PRIMARY KEY,
+
+    senders CHAR(42)[],
+    minimum_base_fee BIGINT, -- todo validate it's large enough
+    txn_hashes CHAR(66)[],
+
     txs TEXT[] NOT NULL,
-    block_number BIGINT NOT NULL,
+    reverting_tx_hashes CHAR(66)[],
+    dropping_tx_hashes CHAR(66)[],
+
+    block_number BIGINT,
     min_timestamp BIGINT,
     max_timestamp BIGINT,
-    reverting_tx_hashes TEXT[],
-    replacement_uuid TEXT,
     created_at TIMESTAMPTZ NOT NULL,
     updated_at TIMESTAMPTZ NOT NULL
 );

crates/datastore/src/postgres.rs

@@ -1,4 +1,6 @@
 use crate::traits::BundleDatastore;
+use alloy_primitives::TxHash;
+use alloy_primitives::hex::{FromHex, ToHexExt};
 use alloy_rpc_types_mev::EthSendBundle;
 use anyhow::Result;
 use sqlx::PgPool;
@@ -28,12 +30,96 @@ impl PostgresDatastore {
 
 #[async_trait::async_trait]
 impl BundleDatastore for PostgresDatastore {
-    async fn insert_bundle(&self, _bundle: EthSendBundle) -> Result<Uuid> {
-        todo!()
+    async fn insert_bundle(&self, bundle: EthSendBundle) -> Result<Uuid> {
+        let id = Uuid::new_v4();
+
+        let txs: Vec<String> = bundle
+            .txs
+            .iter()
+            .map(|tx| tx.encode_hex_upper_with_prefix())
+            .collect();
+        let reverting_tx_hashes: Vec<String> = bundle
+            .reverting_tx_hashes
+            .iter()
+            .map(|h| h.encode_hex_with_prefix())
+            .collect();
+        let dropping_tx_hashes: Vec<String> = bundle
+            .dropping_tx_hashes
+            .iter()
+            .map(|h| h.encode_hex_with_prefix())
+            .collect();
+
+        sqlx::query!(
+            r#"
+            INSERT INTO bundles (
+                id, 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())
+            "#,
+            id,
+            &txs,
+            &reverting_tx_hashes,
+            &dropping_tx_hashes,
+            bundle.block_number as i64,
+            bundle.min_timestamp.map(|t| t as i64),
+            bundle.max_timestamp.map(|t| t as i64),
+        )
+        .execute(&self.pool)
+        .await?;
+
+        Ok(id)
     }
 
-    async fn get_bundle(&self, _id: Uuid) -> Result<Option<EthSendBundle>> {
-        todo!()
+    async fn get_bundle(&self, id: Uuid) -> Result<Option<EthSendBundle>> {
+        let result = sqlx::query!(
+            r#"
+            SELECT txs, reverting_tx_hashes, dropping_tx_hashes, 
+                   block_number, min_timestamp, max_timestamp
+            FROM bundles 
+            WHERE id = $1
+            "#,
+            id
+        )
+        .fetch_optional(&self.pool)
+        .await?;
+
+        match result {
+            Some(row) => {
+                let txs: Result<Vec<alloy_primitives::Bytes>, _> =
+                    row.txs.into_iter().map(|tx_hex| tx_hex.parse()).collect();
+
+                let reverting_tx_hashes: Result<Vec<TxHash>, _> = row
+                    .reverting_tx_hashes
+                    .unwrap_or_default()
+                    .into_iter()
+                    .map(TxHash::from_hex)
+                    .collect();
+
+                let dropping_tx_hashes: Result<Vec<TxHash>, _> = row
+                    .dropping_tx_hashes
+                    .unwrap_or_default()
+                    .into_iter()
+                    .map(TxHash::from_hex)
+                    .collect();
+
+                Ok(Some(EthSendBundle {
+                    txs: txs?,
+                    block_number: row.block_number.unwrap_or(0) as u64,
+                    min_timestamp: row.min_timestamp.map(|t| t as u64),
+                    max_timestamp: row.max_timestamp.map(|t| t as u64),
+                    reverting_tx_hashes: reverting_tx_hashes?,
+                    replacement_uuid: None,
+                    dropping_tx_hashes: dropping_tx_hashes?,
+                    refund_percent: None,
+                    refund_recipient: None,
+                    refund_tx_hashes: Vec::new(),
+                    extra_fields: Default::default(),
+                }))
+            }
+            None => Ok(None),
+        }
     }
 
     async fn cancel_bundle(&self, _id: Uuid) -> Result<()> {

crates/datastore/tests/datastore.rs

@@ -0,0 +1,80 @@
+use alloy_primitives::{Bytes, TxHash};
+use alloy_rpc_types_mev::EthSendBundle;
+use datastore::{PostgresDatastore, traits::BundleDatastore};
+use sqlx::PgPool;
+use testcontainers_modules::{
+    postgres,
+    testcontainers::{ContainerAsync, runners::AsyncRunner},
+};
+
+struct TestHarness {
+    _postgres_instance: ContainerAsync<postgres::Postgres>,
+    data_store: PostgresDatastore,
+}
+
+async fn setup_datastore() -> eyre::Result<TestHarness> {
+    let postgres_instance = postgres::Postgres::default().start().await?;
+    let connection_string = format!(
+        "postgres://postgres:postgres@{}:{}/postgres",
+        postgres_instance.get_host().await?,
+        postgres_instance.get_host_port_ipv4(5432).await?
+    );
+
+    let pool = PgPool::connect(&connection_string).await?;
+    let data_store = PostgresDatastore::new(pool);
+
+    assert!(data_store.run_migrations().await.is_ok());
+    Ok(TestHarness {
+        _postgres_instance: postgres_instance,
+        data_store,
+    })
+}
+
+#[tokio::test]
+async fn insert_and_get() -> eyre::Result<()> {
+    let harness = setup_datastore().await?;
+    let test_bundle = EthSendBundle {
+        txs: vec![
+            "0x02f86f0102843b9aca0085029e7822d68298f094d2c8e0b2e8f2a8e8f2a8e8f2a8e8f2a8e8f2a880b844a9059cbb000000000000000000000000d2c8e0b2e8f2a8e8f2a8e8f2a8e8f2a8e8f2a80000000000000000000000000000000000000000000000000de0b6b3a7640000c080a0123456789abcdef0123456789abcdef0123456789abcdef0123456789abcdef0a0fedcba9876543210fedcba9876543210fedcba9876543210fedcba9876543210".parse::<Bytes>()?,
+        ],
+        block_number: 12345,
+        min_timestamp: Some(1640995200),
+        max_timestamp: Some(1640995260),
+        reverting_tx_hashes: vec![
+            "0x1234567890abcdef1234567890abcdef1234567890abcdef1234567890abcdef".parse::<TxHash>()?,
+        ],
+        replacement_uuid: None,
+        dropping_tx_hashes: vec![
+            "0xfedcba0987654321fedcba0987654321fedcba0987654321fedcba0987654321".parse::<TxHash>()?,
+        ],
+        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;
+    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();
+
+    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);
+    assert_eq!(retrieved_bundle.max_timestamp, test_bundle.max_timestamp);
+    assert_eq!(
+        retrieved_bundle.reverting_tx_hashes.len(),
+        test_bundle.reverting_tx_hashes.len()
+    );
+    assert_eq!(
+        retrieved_bundle.dropping_tx_hashes.len(),
+        test_bundle.dropping_tx_hashes.len()
+    );
+
+    Ok(())
+}

crates/datastore/tests/integration_tests.rs

@@ -1,6 +1,19 @@
 # Run all CI checks locally
 ci: check test fmt clippy build
 
+db:
+    #!/usr/bin/env bash
+    set -euxo pipefail
+    docker container stop tips-db
+    docker container rm tips-db
+    docker run -d --name tips-db -e POSTGRES_PASSWORD=postgres -p 5432:5432 postgres
+    sleep 2
+    for file in ./crates/datastore/migrations/*.sql; do
+      echo $file
+      psql -d postgres://postgres:postgres@localhost:5432/postgres -f $file
+    done
+
+
 create-migration name:
     touch crates/datastore/migrations/$(date +%s)_{{ name }}.sql