add event notification integration points

d4mr · d4mr · commit 67a7ab7eb5fe · 2025-07-12T07:00:17.000+05:30
diff --git a/executors/src/eoa/events.rs b/executors/src/eoa/events.rs
@@ -0,0 +1,119 @@
+use std::fmt::Display;
+
+use engine_core::error::EngineError;
+use serde::{Deserialize, Serialize};
+use twmq::job::RequeuePosition;
+
+use crate::{
+    eoa::{
+        store::{SubmittedTransaction, TransactionData},
+        worker::ConfirmedTransactionWithRichReceipt,
+    },
+    webhook::envelope::{
+        BareWebhookNotificationEnvelope, SerializableNackData, SerializableSuccessData, StageEvent,
+    },
+};
+
+pub struct EoaExecutorEvent {
+    pub transaction_data: TransactionData,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct EoaSendAttemptNackData {
+    pub nonce: u64,
+    pub error: EngineError,
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
+pub enum EoaExecutorStage {
+    SendAttempt,
+    SendAttemptNack,
+    TransactionReplaced,
+    TransactionConfirmed,
+}
+
+impl Display for EoaExecutorStage {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        match self {
+            EoaExecutorStage::SendAttempt => write!(f, "send_attempt"),
+            EoaExecutorStage::SendAttemptNack => write!(f, "send_attempt_nack"),
+            EoaExecutorStage::TransactionReplaced => write!(f, "transaction_replaced"),
+            EoaExecutorStage::TransactionConfirmed => write!(f, "transaction_confirmed"),
+        }
+    }
+}
+
+const EXECUTOR_NAME: &str = "eoa";
+
+impl EoaExecutorEvent {
+    pub fn send_attempt_success_envelopes(
+        &self,
+        submitted_transaction: SubmittedTransaction,
+    ) -> BareWebhookNotificationEnvelope<SerializableSuccessData<SubmittedTransaction>> {
+        BareWebhookNotificationEnvelope {
+            transaction_id: self.transaction_data.transaction_id.clone(),
+            executor_name: EXECUTOR_NAME.to_string(),
+            stage_name: EoaExecutorStage::SendAttempt.to_string(),
+            event_type: StageEvent::Success,
+            payload: SerializableSuccessData {
+                result: submitted_transaction.clone(),
+            },
+        }
+    }
+
+    pub fn send_attempt_nack_envelopes(
+        &self,
+        nonce: u64,
+        error: EngineError,
+        attempt_number: u32,
+    ) -> BareWebhookNotificationEnvelope<SerializableNackData<EoaSendAttemptNackData>> {
+        BareWebhookNotificationEnvelope {
+            transaction_id: self.transaction_data.transaction_id.clone(),
+            executor_name: EXECUTOR_NAME.to_string(),
+            stage_name: EoaExecutorStage::SendAttemptNack.to_string(),
+            event_type: StageEvent::Nack,
+            payload: SerializableNackData {
+                error: EoaSendAttemptNackData {
+                    nonce,
+                    error: error.clone(),
+                },
+                delay_ms: None,
+                position: RequeuePosition::Last,
+                attempt_number,
+                max_attempts: None,
+                next_retry_at: None,
+            },
+        }
+    }
+
+    pub fn transaction_replaced_envelopes(
+        &self,
+        replaced_transaction: SubmittedTransaction,
+    ) -> BareWebhookNotificationEnvelope<SerializableSuccessData<SubmittedTransaction>> {
+        BareWebhookNotificationEnvelope {
+            transaction_id: self.transaction_data.transaction_id.clone(),
+            executor_name: EXECUTOR_NAME.to_string(),
+            stage_name: EoaExecutorStage::TransactionReplaced.to_string(),
+            event_type: StageEvent::Success,
+            payload: SerializableSuccessData {
+                result: replaced_transaction.clone(),
+            },
+        }
+    }
+
+    pub fn transaction_confirmed_envelopes(
+        &self,
+        confirmed_transaction: ConfirmedTransactionWithRichReceipt,
+    ) -> BareWebhookNotificationEnvelope<SerializableSuccessData<ConfirmedTransactionWithRichReceipt>>
+    {
+        BareWebhookNotificationEnvelope {
+            transaction_id: self.transaction_data.transaction_id.clone(),
+            executor_name: EXECUTOR_NAME.to_string(),
+            stage_name: EoaExecutorStage::TransactionConfirmed.to_string(),
+            event_type: StageEvent::Success,
+            payload: SerializableSuccessData {
+                result: confirmed_transaction.clone(),
+            },
+        }
+    }
+}
diff --git a/executors/src/eoa/mod.rs b/executors/src/eoa/mod.rs
@@ -1,4 +1,5 @@
 pub mod error_classifier;
+pub mod events;
 pub mod store;
 pub mod worker;
 
diff --git a/executors/src/eoa/store/submitted.rs b/executors/src/eoa/store/submitted.rs
@@ -1,12 +1,14 @@
 use std::collections::{BTreeMap, HashMap, HashSet};
 
+use serde::{Deserialize, Serialize};
 use twmq::redis::{AsyncCommands, Pipeline, aio::ConnectionManager};
 
 use crate::eoa::store::{
     ConfirmedTransaction, EoaExecutorStoreKeys, TransactionStoreError, atomic::SafeRedisTransaction,
 };
 
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
 pub struct SubmittedTransaction {
     pub nonce: u64,
     pub hash: String,
diff --git a/executors/src/eoa/worker.rs b/executors/src/eoa/worker.rs
@@ -21,6 +21,7 @@ use hex;
 use serde::{Deserialize, Serialize};
 use std::{sync::Arc, time::Duration};
 use tokio::time::sleep;
+use twmq::Queue;
 use twmq::redis::AsyncCommands;
 use twmq::redis::aio::ConnectionManager;
 use twmq::{
@@ -35,6 +36,7 @@ use crate::eoa::store::{
     EoaExecutorStore, EoaExecutorStoreKeys, EoaHealth, EoaTransactionRequest, ReplacedTransaction,
     SubmittedTransaction, TransactionData, TransactionStoreError,
 };
+use crate::webhook::WebhookJobHandler;
 
 // ========== SPEC-COMPLIANT CONSTANTS ==========
 const MAX_INFLIGHT_PER_EOA: u64 = 100; // Default from spec
@@ -236,8 +238,9 @@ fn is_retryable_rpc_error(kind: &RpcErrorKind) -> bool {
     }
 }
 
-#[derive(Debug, Clone)]
-struct ConfirmedTransactionWithRichReceipt {
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(rename_all = "camelCase")]
+pub struct ConfirmedTransactionWithRichReceipt {
     pub nonce: u64,
     pub hash: String,
     pub transaction_id: String,
@@ -273,6 +276,8 @@ where
     CS: ChainService + Send + Sync + 'static,
 {
     pub chain_service: Arc<CS>,
+    pub webhook_queue: Arc<Queue<WebhookJobHandler>>,
+
     pub redis: ConnectionManager,
     pub namespace: Option<String>,
 
diff --git a/executors/src/webhook/envelope.rs b/executors/src/webhook/envelope.rs
@@ -41,6 +41,35 @@ pub struct WebhookNotificationEnvelope<T> {
     pub delivery_target_url: Option<String>,
 }
 
+#[derive(Serialize, Deserialize, Debug, Clone)]
+#[serde(rename_all = "camelCase")]
+pub struct BareWebhookNotificationEnvelope<T: Clone> {
+    pub transaction_id: String,
+    pub event_type: StageEvent,
+    pub executor_name: String,
+    pub stage_name: String,
+    pub payload: T,
+}
+
+impl<T: Serialize + Clone> BareWebhookNotificationEnvelope<T> {
+    pub fn into_webhook_notification_envelope(
+        self,
+        timestamp: u64,
+        delivery_target_url: String,
+    ) -> WebhookNotificationEnvelope<T> {
+        WebhookNotificationEnvelope {
+            notification_id: Uuid::new_v4().to_string(),
+            transaction_id: self.transaction_id,
+            timestamp,
+            executor_name: self.executor_name,
+            stage_name: self.stage_name,
+            event_type: self.event_type,
+            payload: self.payload,
+            delivery_target_url: Some(delivery_target_url),
+        }
+    }
+}
+
 // --- Serializable Hook Data Wrappers ---
 // These wrap the hook data to make them serializable (removing lifetimes)
 #[derive(Serialize, Deserialize, Debug, Clone)]
diff --git a/executors/src/webhook/mod.rs b/executors/src/webhook/mod.rs
@@ -1,15 +1,20 @@
 use std::collections::HashMap;
+use std::env;
 use std::sync::Arc;
 use std::time::{Duration, SystemTime, UNIX_EPOCH};
 
+use engine_core::execution_options::WebhookOptions;
 use hex;
 use hmac::{Hmac, Mac};
 use reqwest::header::{HeaderMap, HeaderName, HeaderValue};
 use serde::{Deserialize, Serialize};
 use twmq::error::TwmqError;
 use twmq::hooks::TransactionContext;
 use twmq::job::{BorrowedJob, JobError, JobResult, RequeuePosition, ToJobResult};
-use twmq::{DurableExecution, FailHookData, NackHookData, SuccessHookData, UserCancellable};
+use twmq::{DurableExecution, FailHookData, NackHookData, Queue, SuccessHookData, UserCancellable};
+use uuid::Uuid;
+
+use crate::webhook::envelope::{BareWebhookNotificationEnvelope, WebhookNotificationEnvelope};
 
 pub mod envelope;
 
@@ -113,7 +118,10 @@ impl DurableExecution for WebhookJobHandler {
     type JobData = WebhookJobPayload;
 
     #[tracing::instrument(skip_all, fields(queue = "webhook", job_id = job.job.id))]
-    async fn process(&self, job: &BorrowedJob<Self::JobData>) -> JobResult<Self::Output, Self::ErrorData> {
+    async fn process(
+        &self,
+        job: &BorrowedJob<Self::JobData>,
+    ) -> JobResult<Self::Output, Self::ErrorData> {
         let payload = &job.job.data;
         let mut request_headers = HeaderMap::new();
 
@@ -423,3 +431,77 @@ impl DurableExecution for WebhookJobHandler {
         );
     }
 }
+
+pub fn queue_webhook_envelopes<T: Serialize + Clone>(
+    envelope: BareWebhookNotificationEnvelope<T>,
+    webhook_options: Vec<WebhookOptions>,
+    tx: &mut TransactionContext<'_>,
+    webhook_queue: Arc<Queue<WebhookJobHandler>>,
+) -> Result<(), TwmqError> {
+    let now = chrono::Utc::now().timestamp().min(0) as u64;
+    let serialised_webhook_envelopes =
+        webhook_options
+            .iter()
+            .map(|webhook_option| {
+                let webhook_notification_envelope = envelope
+                    .clone()
+                    .into_webhook_notification_envelope(now, webhook_option.url.clone());
+                let serialised_envelope = serde_json::to_string(&webhook_notification_envelope)?;
+                Ok((
+                    serialised_envelope,
+                    webhook_notification_envelope,
+                    webhook_option.clone(),
+                ))
+            })
+            .collect::<Result<
+                Vec<(String, WebhookNotificationEnvelope<T>, WebhookOptions)>,
+                serde_json::Error,
+            >>()?;
+
+    let webhook_payloads = serialised_webhook_envelopes
+        .into_iter()
+        .map(
+            |(serialised_envelope, webhook_notification_envelope, webhook_option)| {
+                let payload = WebhookJobPayload {
+                    url: webhook_option.url,
+                    body: serialised_envelope,
+                    headers: Some(
+                        [
+                            ("Content-Type".to_string(), "application/json".to_string()),
+                            (
+                                "User-Agent".to_string(),
+                                format!("{}/{}", envelope.executor_name, envelope.stage_name),
+                            ),
+                        ]
+                        .into_iter()
+                        .collect(),
+                    ),
+                    hmac_secret: webhook_option.secret, // TODO: Add HMAC support if needed
+                    http_method: Some("POST".to_string()),
+                };
+                return (payload, webhook_notification_envelope);
+            },
+        )
+        .collect::<Vec<_>>();
+
+    for (payload, webhook_notification_envelope) in webhook_payloads {
+        let mut webhook_job = webhook_queue.clone().job(payload);
+        webhook_job.options.id = format!(
+            "{}_{}_webhook",
+            webhook_notification_envelope.transaction_id,
+            webhook_notification_envelope.notification_id
+        );
+
+        tx.queue_job(webhook_job)?;
+        tracing::info!(
+            transaction_id = %webhook_notification_envelope.transaction_id,
+            executor = %webhook_notification_envelope.executor_name,
+            stage = %webhook_notification_envelope.stage_name,
+            event = ?webhook_notification_envelope.event_type,
+            notification_id = %webhook_notification_envelope.notification_id,
+            "Queued webhook notification"
+        );
+    }
+
+    Ok(())
+}
diff --git a/server/src/queue/manager.rs b/server/src/queue/manager.rs
@@ -5,7 +5,7 @@ use alloy::transports::http::reqwest;
 use engine_core::error::EngineError;
 use engine_executors::{
     eip7702_executor::{confirm::Eip7702ConfirmationHandler, send::Eip7702SendHandler},
-    eoa::{EoaExecutorStore, EoaExecutorWorker},
+    eoa::EoaExecutorWorker,
     external_bundler::{
         confirm::UserOpConfirmationHandler,
         deployment::{RedisDeploymentCache, RedisDeploymentLock},
@@ -16,10 +16,7 @@ use engine_executors::{
 };
 use twmq::{Queue, queue::QueueOptions, shutdown::ShutdownHandle};
 
-use crate::{
-    chains::ThirdwebChainService,
-    config::{QueueConfig, RedisConfig},
-};
+use crate::{chains::ThirdwebChainService, config::QueueConfig};
 
 pub struct QueueManager {
     pub webhook_queue: Arc<Queue<WebhookJobHandler>>,
@@ -212,6 +209,7 @@ impl QueueManager {
         let eoa_executor_handler = EoaExecutorWorker {
             chain_service: chain_service.clone(),
             eoa_signer: eoa_signer.clone(),
+            webhook_queue: webhook_queue.clone(),
             namespace: queue_config.execution_namespace.clone(),
             redis: redis_client.get_connection_manager().await?,
             max_inflight: 100,

Original file line number	Diff line number	Diff line change
`@@ -1,4 +1,5 @@`
`1`	`1`	`pub mod error_classifier;`
	`2`	`+pub mod events;`
`2`	`3`	`pub mod store;`
`3`	`4`	`pub mod worker;`
`4`	`5`