chore: emit audit logs from ingress-rpc (#48)

Author: danyalprout

.env.example

@@ -3,8 +3,10 @@ TIPS_INGRESS_ADDRESS=0.0.0.0
 TIPS_INGRESS_PORT=8080
 TIPS_INGRESS_RPC_MEMPOOL=http://localhost:2222
 TIPS_INGRESS_DUAL_WRITE_MEMPOOL=false
-TIPS_INGRESS_KAFKA_INGRESS_PROPERTIES_FILE=/app/docker/ingress-kafka-properties
+TIPS_INGRESS_KAFKA_INGRESS_PROPERTIES_FILE=/app/docker/ingress-bundles-kafka-properties
 TIPS_INGRESS_KAFKA_INGRESS_TOPIC=tips-ingress
+TIPS_INGRESS_KAFKA_AUDIT_PROPERTIES_FILE=/app/docker/ingress-audit-kafka-properties
+TIPS_INGRESS_KAFKA_AUDIT_TOPIC=tips-audit
 TIPS_INGRESS_LOG_LEVEL=info
 TIPS_INGRESS_SEND_TRANSACTION_DEFAULT_LIFETIME_SECONDS=10800
 

Cargo.lock

@@ -25,16 +25,15 @@ alloy-primitives = { version = "1.3.1", default-features = false, features = [
     "map-foldhash",
     "serde",
 ] }
-alloy-rpc-types = { version = "1.0.35", default-features = false }
-alloy-consensus = { version = "1.0.35" }
-alloy-provider = { version = "1.0.35" }
-alloy-rpc-types-mev = "1.0.35"
+alloy-consensus = { version = "1.0.37" }
+alloy-provider = { version = "1.0.37" }
 alloy-serde = "1.0.41"
 
 # op-alloy
-op-alloy-network = { version = "0.21.0", default-features = false }
-op-alloy-consensus = { version = "0.21.0", features = ["k256"] }
-op-alloy-rpc-types = { version = "0.21.0", default-features = true}
+op-alloy-network = { version = "0.20.0", default-features = false }
+op-alloy-consensus = { version = "0.20.0", features = ["k256"] }
+op-alloy-rpc-types = { version = "0.20.0", default-features = true}
+op-alloy-flz = {  version = "0.13.1" }
 
 tokio = { version = "1.47.1", features = ["full"] }
 tracing = "0.1.41"

Cargo.toml

@@ -36,12 +36,7 @@ pub struct TransactionMetadata {
 #[derive(Debug, Clone, Serialize, Deserialize)]
 #[serde(tag = "event", content = "data")]
 pub enum BundleHistoryEvent {
-    Created {
-        key: String,
-        timestamp: i64,
-        bundle: Bundle,
-    },
-    Updated {
+    Received {
         key: String,
         timestamp: i64,
         bundle: Bundle,
@@ -73,8 +68,7 @@ pub enum BundleHistoryEvent {
 impl BundleHistoryEvent {
     pub fn key(&self) -> &str {
         match self {
-            BundleHistoryEvent::Created { key, .. } => key,
-            BundleHistoryEvent::Updated { key, .. } => key,
+            BundleHistoryEvent::Received { key, .. } => key,
             BundleHistoryEvent::Cancelled { key, .. } => key,
             BundleHistoryEvent::BuilderIncluded { key, .. } => key,
             BundleHistoryEvent::BlockIncluded { key, .. } => key,
@@ -106,12 +100,7 @@ fn update_bundle_history_transform(
     }
 
     let history_event = match &event.event {
-        BundleEvent::Created { bundle, .. } => BundleHistoryEvent::Created {
-            key: event.key.clone(),
-            timestamp: event.timestamp,
-            bundle: bundle.clone(),
-        },
-        BundleEvent::Updated { bundle, .. } => BundleHistoryEvent::Updated {
+        BundleEvent::Received { bundle, .. } => BundleHistoryEvent::Received {
             key: event.key.clone(),
             timestamp: event.timestamp,
             bundle: bundle.clone(),
@@ -396,7 +385,7 @@ mod tests {
         let bundle_history = BundleHistory { history: vec![] };
         let bundle = create_test_bundle();
         let bundle_id = Uuid::new_v4();
-        let bundle_event = BundleEvent::Created {
+        let bundle_event = BundleEvent::Received {
             bundle_id,
             bundle: bundle.clone(),
         };
@@ -409,7 +398,7 @@ mod tests {
         assert_eq!(bundle_history.history.len(), 1);
 
         match &bundle_history.history[0] {
-            BundleHistoryEvent::Created {
+            BundleHistoryEvent::Received {
                 key,
                 timestamp: ts,
                 bundle: b,
@@ -424,7 +413,7 @@ mod tests {
 
     #[test]
     fn test_update_bundle_history_transform_skips_duplicate_key() {
-        let existing_event = BundleHistoryEvent::Created {
+        let existing_event = BundleHistoryEvent::Received {
             key: "duplicate-key".to_string(),
             timestamp: 1111111111,
             bundle: create_test_bundle(),
@@ -435,7 +424,7 @@ mod tests {
 
         let bundle = create_test_bundle();
         let bundle_id = Uuid::new_v4();
-        let bundle_event = BundleEvent::Updated { bundle_id, bundle };
+        let bundle_event = BundleEvent::Received { bundle_id, bundle };
         let event = create_test_event("duplicate-key", 1234567890, bundle_event);
 
         let result = update_bundle_history_transform(bundle_history, &event);
@@ -449,24 +438,16 @@ mod tests {
         let bundle_id = Uuid::new_v4();
 
         let bundle = create_test_bundle();
-        let bundle_event = BundleEvent::Created {
+        let bundle_event = BundleEvent::Received {
             bundle_id,
             bundle: bundle.clone(),
         };
         let event = create_test_event("test-key", 1234567890, bundle_event);
         let result = update_bundle_history_transform(bundle_history.clone(), &event);
         assert!(result.is_some());
 
-        let bundle_event = BundleEvent::Updated {
-            bundle_id,
-            bundle: bundle.clone(),
-        };
-        let event = create_test_event("test-key-2", 1234567890, bundle_event);
-        let result = update_bundle_history_transform(bundle_history.clone(), &event);
-        assert!(result.is_some());
-
         let bundle_event = BundleEvent::Cancelled { bundle_id };
-        let event = create_test_event("test-key-3", 1234567890, bundle_event);
+        let event = create_test_event("test-key-2", 1234567890, bundle_event);
         let result = update_bundle_history_transform(bundle_history.clone(), &event);
         assert!(result.is_some());
 
@@ -476,7 +457,7 @@ mod tests {
             block_number: 12345,
             flashblock_index: 1,
         };
-        let event = create_test_event("test-key-4", 1234567890, bundle_event);
+        let event = create_test_event("test-key-3", 1234567890, bundle_event);
         let result = update_bundle_history_transform(bundle_history.clone(), &event);
         assert!(result.is_some());
 
@@ -485,15 +466,15 @@ mod tests {
             block_number: 12345,
             block_hash: TxHash::from([1u8; 32]),
         };
-        let event = create_test_event("test-key-5", 1234567890, bundle_event);
+        let event = create_test_event("test-key-4", 1234567890, bundle_event);
         let result = update_bundle_history_transform(bundle_history.clone(), &event);
         assert!(result.is_some());
 
         let bundle_event = BundleEvent::Dropped {
             bundle_id,
             reason: DropReason::TimedOut,
         };
-        let event = create_test_event("test-key-6", 1234567890, bundle_event);
+        let event = create_test_event("test-key-5", 1234567890, bundle_event);
         let result = update_bundle_history_transform(bundle_history, &event);
         assert!(result.is_some());
     }