openzeppelin_relayer/jobs/handlers/
notification_handler.rs

1//! Notification handling worker implementation.
2//!
3//! This module implements the notification handling worker that processes
4//! notification jobs from the queue.
5
6use actix_web::web::ThinData;
7use eyre::Result;
8use tracing::{debug, instrument};
9
10use crate::{
11    constants::WORKER_NOTIFICATION_SENDER_RETRIES,
12    jobs::{handle_result, Job, NotificationSend},
13    models::DefaultAppState,
14    observability::request_id::set_request_id,
15    queues::{HandlerError, WorkerContext},
16    repositories::Repository,
17    services::WebhookNotificationService,
18};
19
20/// Handles incoming notification jobs from the queue.
21///
22/// # Arguments
23/// * `job` - The notification job containing recipient and message details
24/// * `context` - Application state containing notification services
25///
26/// # Returns
27/// * `Result<(), HandlerError>` - Success or failure of notification processing
28#[instrument(
29    level = "debug",
30    skip(job, context),
31    fields(
32        request_id = ?job.request_id,
33        job_id = %job.message_id,
34        job_type = %job.job_type.to_string(),
35        attempt = %ctx.attempt,
36        task_id = %ctx.task_id,
37        notification_id = %job.data.notification_id,
38    )
39)]
40pub async fn notification_handler(
41    job: Job<NotificationSend>,
42    context: ThinData<DefaultAppState>,
43    ctx: WorkerContext,
44) -> Result<(), HandlerError> {
45    if let Some(request_id) = job.request_id.clone() {
46        set_request_id(request_id);
47    }
48
49    debug!(
50        notification_id = %job.data.notification_id,
51        "handling notification"
52    );
53
54    let result = handle_request(job.data, &context).await;
55
56    handle_result(
57        result,
58        &ctx,
59        "Notification",
60        WORKER_NOTIFICATION_SENDER_RETRIES,
61    )
62}
63
64async fn handle_request(
65    request: NotificationSend,
66    context: &ThinData<DefaultAppState>,
67) -> Result<()> {
68    debug!(
69        notification_id = %request.notification_id,
70        "sending notification"
71    );
72    let notification = context
73        .notification_repository
74        .get_by_id(request.notification_id.clone())
75        .await?;
76
77    let notification_service =
78        WebhookNotificationService::new(notification.url, notification.signing_key);
79
80    notification_service
81        .send_notification(request.notification)
82        .await?;
83
84    debug!(
85        notification_id = %request.notification_id,
86        "notification sent successfully"
87    );
88
89    Ok(())
90}
91
92#[cfg(test)]
93mod tests {
94    use super::*;
95    use crate::models::{
96        EvmTransactionResponse, NetworkType, RelayerDisabledPayload, RelayerEvmPolicy,
97        RelayerNetworkPolicyResponse, RelayerResponse, TransactionResponse, TransactionStatus,
98        WebhookNotification, WebhookPayload, U256,
99    };
100
101    #[tokio::test]
102    async fn test_notification_job_creation() {
103        // Create a basic notification webhook payload
104        let payload = WebhookPayload::Transaction(TransactionResponse::Evm(Box::new(
105            EvmTransactionResponse {
106                id: "tx123".to_string(),
107                hash: Some("0x123".to_string()),
108                status: TransactionStatus::Confirmed,
109                status_reason: None,
110                created_at: "2025-01-27T15:31:10.777083+00:00".to_string(),
111                sent_at: Some("2025-01-27T15:31:10.777083+00:00".to_string()),
112                confirmed_at: Some("2025-01-27T15:31:10.777083+00:00".to_string()),
113                gas_price: Some(1000000000),
114                gas_limit: Some(21000),
115                nonce: Some(1),
116                value: U256::from(1000000000000000000_u64),
117                from: "0xabc".to_string(),
118                to: Some("0xdef".to_string()),
119                relayer_id: "relayer-1".to_string(),
120                data: None,
121                max_fee_per_gas: None,
122                max_priority_fee_per_gas: None,
123                signature: None,
124                speed: None,
125                is_canceled: None,
126            },
127        )));
128
129        // Create a notification
130        let notification = WebhookNotification::new("test_event".to_string(), payload);
131        let notification_job =
132            NotificationSend::new("notification-1".to_string(), notification.clone());
133
134        // Create the job
135        let job = Job::new(crate::jobs::JobType::NotificationSend, notification_job);
136
137        // Test the job structure
138        assert_eq!(job.data.notification_id, "notification-1");
139        assert_eq!(job.data.notification.event, "test_event");
140    }
141
142    #[tokio::test]
143    async fn test_notification_job_with_different_payloads() {
144        // Test with different payload types
145
146        let transaction_payload = WebhookPayload::Transaction(TransactionResponse::Evm(Box::new(
147            EvmTransactionResponse {
148                id: "tx123".to_string(),
149                hash: Some("0x123".to_string()),
150                status: TransactionStatus::Confirmed,
151                status_reason: None,
152                created_at: "2025-01-27T15:31:10.777083+00:00".to_string(),
153                sent_at: Some("2025-01-27T15:31:10.777083+00:00".to_string()),
154                confirmed_at: Some("2025-01-27T15:31:10.777083+00:00".to_string()),
155                gas_price: Some(1000000000),
156                gas_limit: Some(21000),
157                nonce: Some(1),
158                value: U256::from(1000000000000000000_u64),
159                from: "0xabc".to_string(),
160                to: Some("0xdef".to_string()),
161                relayer_id: "relayer-1".to_string(),
162                data: None,
163                max_fee_per_gas: None,
164                max_priority_fee_per_gas: None,
165                signature: None,
166                speed: None,
167                is_canceled: None,
168            },
169        )));
170
171        let string_notification =
172            WebhookNotification::new("transaction_payload".to_string(), transaction_payload);
173        let job = NotificationSend::new("notification-string".to_string(), string_notification);
174        assert_eq!(job.notification.event, "transaction_payload");
175
176        let relayer_disabled = WebhookPayload::RelayerDisabled(Box::new(RelayerDisabledPayload {
177            relayer: RelayerResponse {
178                id: "relayer-1".to_string(),
179                name: "relayer-1".to_string(),
180                network: "ethereum".to_string(),
181                network_type: NetworkType::Evm,
182                paused: false,
183                policies: Some(RelayerNetworkPolicyResponse::Evm(
184                    RelayerEvmPolicy {
185                        gas_price_cap: None,
186                        whitelist_receivers: None,
187                        eip1559_pricing: None,
188                        private_transactions: Some(false),
189                        min_balance: Some(0),
190                        gas_limit_estimation: None,
191                    }
192                    .into(),
193                )),
194                signer_id: "signer-1".to_string(),
195                notification_id: None,
196                custom_rpc_urls: None,
197                address: Some("0xabc".to_string()),
198                system_disabled: Some(false),
199                ..Default::default()
200            },
201            disable_reason: "test".to_string(),
202        }));
203        let object_notification =
204            WebhookNotification::new("object_event".to_string(), relayer_disabled);
205        let job = NotificationSend::new("notification-object".to_string(), object_notification);
206        assert_eq!(job.notification.event, "object_event");
207    }
208}
openzeppelin_relayer/jobs/handlers/notification_handler.rs

openzeppelin_relayer/jobs/handlers/
notification_handler.rs
