autumn_web/
webhook_outbound.rs

1#![allow(
2    clippy::significant_drop_tightening,
3    clippy::missing_panics_doc,
4    clippy::missing_errors_doc
5)]
6//! Outbound signed webhook delivery with retries, DLQ, and subscription management.
7
8use chrono::{DateTime, Utc};
9use serde::{Deserialize, Serialize};
10use std::collections::HashMap;
11use std::future::Future;
12use std::pin::Pin;
13use std::sync::{Arc, RwLock};
14
15use crate::http_client::Client;
16use crate::{AppState, AutumnError, AutumnResult};
17
18const MAX_LOGGED_RESPONSE_BODY_BYTES: usize = 16 * 1024;
19const TRUNCATED_RESPONSE_BODY_SUFFIX: &str = "\n[truncated]";
20
21/// The status of a webhook subscription.
22#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
23#[serde(rename_all = "lowercase")]
24pub enum WebhookSubscriptionStatus {
25    Active,
26    Disabled,
27    Failed,
28}
29
30impl WebhookSubscriptionStatus {
31    /// Return the lower-case status label.
32    #[must_use]
33    pub const fn as_str(self) -> &'static str {
34        match self {
35            Self::Active => "active",
36            Self::Disabled => "disabled",
37            Self::Failed => "failed",
38        }
39    }
40}
41
42impl std::fmt::Display for WebhookSubscriptionStatus {
43    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
44        f.write_str(self.as_str())
45    }
46}
47
48/// A registered webhook subscription targeting a consumer endpoint.
49#[derive(Debug, Clone, Serialize, Deserialize)]
50pub struct WebhookSubscription {
51    pub id: String,
52    pub target_url: String,
53    pub event_topics: Vec<String>,
54    pub secret: String,
55    pub status: WebhookSubscriptionStatus,
56    pub consecutive_failures: u32,
57}
58
59/// A structured log of an outbound webhook delivery attempt.
60#[derive(Debug, Clone, Serialize, Deserialize)]
61pub struct WebhookDeliveryLog {
62    pub id: String,
63    pub subscription_id: String,
64    pub topic: String,
65    pub payload: String,
66    pub request_headers: HashMap<String, String>,
67    pub response_status: Option<u16>,
68    pub response_body: Option<String>,
69    pub elapsed_ms: u64,
70    pub attempt: u32,
71    pub max_attempts: u32,
72    pub is_dlq: bool,
73    pub last_error: Option<String>,
74    pub timestamp: DateTime<Utc>,
75}
76
77/// Pluggable handler interface for outbound webhook subscriptions and delivery logs.
78pub trait OutboundWebhookHandler: Send + Sync + 'static {
79    /// Retrieve active subscriptions registered for a specific event topic.
80    fn get_subscriptions(
81        &self,
82        topic: &str,
83    ) -> Pin<Box<dyn Future<Output = AutumnResult<Vec<WebhookSubscription>>> + Send>>;
84
85    /// Log a webhook delivery attempt and handle failure counters/statuses.
86    fn log_delivery(
87        &self,
88        log: WebhookDeliveryLog,
89    ) -> Pin<Box<dyn Future<Output = AutumnResult<()>> + Send>>;
90
91    /// Replace a stored delivery log without treating it as a new delivery outcome.
92    ///
93    /// Implementations must perform a plain record replacement. This must not
94    /// update subscription failure counters or auto-failure state.
95    fn replace_delivery_log(
96        &self,
97        log: WebhookDeliveryLog,
98    ) -> Pin<Box<dyn Future<Output = AutumnResult<()>> + Send>>;
99
100    /// Retrieve a specific webhook subscription by ID (regardless of status/active state).
101    fn get_subscription(
102        &self,
103        id: &str,
104    ) -> Pin<Box<dyn Future<Output = AutumnResult<Option<WebhookSubscription>>> + Send>>;
105
106    /// Optional: List only permanently failed delivery attempts archived in the Dead Letter Queue.
107    fn get_dlq_logs(
108        &self,
109    ) -> Pin<Box<dyn Future<Output = AutumnResult<Vec<WebhookDeliveryLog>>> + Send>> {
110        Box::pin(async { Ok(Vec::new()) })
111    }
112
113    /// Get a specific delivery log by ID.
114    fn get_delivery_log(
115        &self,
116        id: &str,
117    ) -> Pin<Box<dyn Future<Output = AutumnResult<Option<WebhookDeliveryLog>>> + Send>>;
118
119    /// Optional: Reset consecutive failures for a subscription.
120    fn reset_subscription_failures(
121        &self,
122        _id: &str,
123    ) -> Pin<Box<dyn Future<Output = AutumnResult<()>> + Send>> {
124        Box::pin(async { Ok(()) })
125    }
126
127    /// Optional: Reactivate a subscription that was auto-marked as failed.
128    ///
129    /// Manual DLQ replays need to bypass the automatic failure guard without
130    /// re-enabling subscriptions that an operator explicitly disabled.
131    fn reactivate_failed_subscription(
132        &self,
133        id: &str,
134    ) -> Pin<Box<dyn Future<Output = AutumnResult<()>> + Send>> {
135        self.reset_subscription_failures(id)
136    }
137}
138
139/// Legacy alias for backward compatibility.
140pub use OutboundWebhookHandler as OutboundWebhookStore;
141
142/// Bounded, thread-safe, process-local in-memory implementation of the outbound webhook handler.
143#[derive(Debug, Default)]
144pub struct InMemoryOutboundWebhookHandler {
145    subscriptions: RwLock<HashMap<String, WebhookSubscription>>,
146    logs: RwLock<HashMap<String, WebhookDeliveryLog>>,
147}
148
149/// Legacy alias for backward compatibility.
150pub type InMemoryOutboundWebhookStore = InMemoryOutboundWebhookHandler;
151
152impl InMemoryOutboundWebhookHandler {
153    /// Create a new, empty in-memory handler.
154    #[must_use]
155    pub fn new() -> Self {
156        Self::default()
157    }
158
159    /// Helper to register a subscription in memory for testing/dev.
160    #[allow(clippy::unused_async)]
161    pub async fn create_subscription(
162        &self,
163        sub: WebhookSubscription,
164    ) -> AutumnResult<WebhookSubscription> {
165        let mut subs = self
166            .subscriptions
167            .write()
168            .expect("subscriptions write lock poisoned");
169        subs.insert(sub.id.clone(), sub.clone());
170        Ok(sub)
171    }
172
173    /// Helper to retrieve logged deliveries for testing/dev.
174    #[allow(clippy::unused_async)]
175    pub async fn get_delivery_logs(&self) -> AutumnResult<Vec<WebhookDeliveryLog>> {
176        let logs = self.logs.read().expect("logs read lock poisoned");
177        let mut list: Vec<WebhookDeliveryLog> = logs.values().cloned().collect();
178        list.sort_by_key(|l| l.timestamp);
179        list.reverse();
180        Ok(list)
181    }
182
183    /// Helper to fetch a single subscription.
184    #[allow(clippy::unused_async)]
185    pub async fn get_subscription(&self, id: &str) -> AutumnResult<Option<WebhookSubscription>> {
186        let subs = self
187            .subscriptions
188            .read()
189            .expect("subscriptions read lock poisoned");
190        Ok(subs.get(id).cloned())
191    }
192}
193
194impl OutboundWebhookHandler for InMemoryOutboundWebhookHandler {
195    fn get_subscriptions(
196        &self,
197        topic: &str,
198    ) -> Pin<Box<dyn Future<Output = AutumnResult<Vec<WebhookSubscription>>> + Send>> {
199        let subs = self
200            .subscriptions
201            .read()
202            .expect("subscriptions read lock poisoned");
203        let topic = topic.to_owned();
204        let list: Vec<WebhookSubscription> = subs
205            .values()
206            .filter(|sub| {
207                sub.event_topics.iter().any(|t| t == &topic)
208                    && sub.status == WebhookSubscriptionStatus::Active
209            })
210            .cloned()
211            .collect();
212        Box::pin(async move { Ok(list) })
213    }
214
215    fn get_subscription(
216        &self,
217        id: &str,
218    ) -> Pin<Box<dyn Future<Output = AutumnResult<Option<WebhookSubscription>>> + Send>> {
219        let subs = self
220            .subscriptions
221            .read()
222            .expect("subscriptions read lock poisoned");
223        let sub = subs.get(id).cloned();
224        Box::pin(async move { Ok(sub) })
225    }
226
227    fn log_delivery(
228        &self,
229        log: WebhookDeliveryLog,
230    ) -> Pin<Box<dyn Future<Output = AutumnResult<()>> + Send>> {
231        let mut logs = self.logs.write().expect("logs write lock poisoned");
232        logs.insert(log.id.clone(), log.clone());
233
234        // Manage subscription consecutive failures and auto-disabling state
235        let mut subs = self
236            .subscriptions
237            .write()
238            .expect("subscriptions write lock poisoned");
239        if let Some(sub) = subs.get_mut(&log.subscription_id) {
240            let is_active = sub.status == WebhookSubscriptionStatus::Active;
241            if is_active {
242                if let Some(status) = log.response_status {
243                    if (200..300).contains(&status) {
244                        sub.consecutive_failures = 0;
245                    } else {
246                        sub.consecutive_failures = sub.consecutive_failures.saturating_add(1);
247                        if sub.consecutive_failures >= 50 {
248                            sub.status = WebhookSubscriptionStatus::Failed;
249                            tracing::warn!(subscription_id = %sub.id, "Webhook subscription auto-disabled due to 50 consecutive failures");
250                        }
251                    }
252                } else if log.last_error.is_some() {
253                    sub.consecutive_failures = sub.consecutive_failures.saturating_add(1);
254                    if sub.consecutive_failures >= 50 {
255                        sub.status = WebhookSubscriptionStatus::Failed;
256                        tracing::warn!(subscription_id = %sub.id, "Webhook subscription auto-disabled due to 50 consecutive failures");
257                    }
258                }
259            }
260        }
261
262        Box::pin(async move { Ok(()) })
263    }
264
265    fn replace_delivery_log(
266        &self,
267        log: WebhookDeliveryLog,
268    ) -> Pin<Box<dyn Future<Output = AutumnResult<()>> + Send>> {
269        let mut logs = self.logs.write().expect("logs write lock poisoned");
270        logs.insert(log.id.clone(), log);
271        Box::pin(async move { Ok(()) })
272    }
273
274    fn get_dlq_logs(
275        &self,
276    ) -> Pin<Box<dyn Future<Output = AutumnResult<Vec<WebhookDeliveryLog>>> + Send>> {
277        let list = {
278            let logs = self.logs.read().expect("logs read lock poisoned");
279            let mut list: Vec<WebhookDeliveryLog> =
280                logs.values().filter(|l| l.is_dlq).cloned().collect();
281            list.sort_by_key(|l| l.timestamp);
282            list.reverse();
283            list
284        };
285        Box::pin(async move { Ok(list) })
286    }
287
288    fn get_delivery_log(
289        &self,
290        id: &str,
291    ) -> Pin<Box<dyn Future<Output = AutumnResult<Option<WebhookDeliveryLog>>> + Send>> {
292        let log = self
293            .logs
294            .read()
295            .expect("logs read lock poisoned")
296            .get(id)
297            .cloned();
298        Box::pin(async move { Ok(log) })
299    }
300
301    fn reset_subscription_failures(
302        &self,
303        id: &str,
304    ) -> Pin<Box<dyn Future<Output = AutumnResult<()>> + Send>> {
305        {
306            let mut subs = self
307                .subscriptions
308                .write()
309                .expect("subscriptions write lock poisoned");
310            if let Some(sub) = subs.get_mut(id) {
311                sub.consecutive_failures = 0;
312            }
313        }
314        Box::pin(async move { Ok(()) })
315    }
316
317    fn reactivate_failed_subscription(
318        &self,
319        id: &str,
320    ) -> Pin<Box<dyn Future<Output = AutumnResult<()>> + Send>> {
321        {
322            let mut subs = self
323                .subscriptions
324                .write()
325                .expect("subscriptions write lock poisoned");
326            if let Some(sub) = subs.get_mut(id) {
327                sub.consecutive_failures = 0;
328                if sub.status == WebhookSubscriptionStatus::Failed {
329                    sub.status = WebhookSubscriptionStatus::Active;
330                }
331            }
332        }
333        Box::pin(async move { Ok(()) })
334    }
335}
336
337/// A runtime delegation callback type to bridge core autumn to autumn-harvest dynamically.
338pub type WebhookDelegate = Arc<
339    dyn Fn(
340            &AppState,
341            WebhookSubscription,
342            WebhookDeliveryLog,
343        ) -> Pin<Box<dyn Future<Output = AutumnResult<()>> + Send>>
344        + Send
345        + Sync,
346>;
347
348/// `AppState` extension for the runtime delegation hook.
349#[derive(Clone)]
350pub struct WebhookDelegateExt(pub WebhookDelegate);
351
352/// The runtime manager for outbound webhooks.
353#[derive(Clone)]
354pub struct WebhookOutboundManager {
355    handler: Arc<dyn OutboundWebhookHandler>,
356    client: Client,
357    initial_backoff_ms: u64,
358}
359
360impl WebhookOutboundManager {
361    /// Create a new webhook manager with a handler.
362    pub fn new(handler: Arc<dyn OutboundWebhookHandler>) -> Self {
363        Self {
364            handler,
365            client: Client::new(),
366            initial_backoff_ms: 1000,
367        }
368    }
369
370    /// Set a custom initial backoff for retries.
371    #[must_use]
372    pub const fn with_initial_backoff_ms(mut self, ms: u64) -> Self {
373        self.initial_backoff_ms = ms;
374        self
375    }
376
377    fn with_client_from_state(mut self, state: &AppState) -> Self {
378        self.client = Client::from_state(state);
379        self
380    }
381
382    /// Access the underlying webhook handler (compatibility/actuator support).
383    #[must_use]
384    pub fn store(&self) -> &Arc<dyn OutboundWebhookHandler> {
385        &self.handler
386    }
387
388    /// Access the underlying http client.
389    #[must_use]
390    pub const fn client(&self) -> &Client {
391        &self.client
392    }
393
394    /// Dispatch a signed webhook payload to all subscriptions interested in `topic`.
395    ///
396    /// # Errors
397    ///
398    /// Returns [`AutumnError`] if payload serialization or queueing fails.
399    pub async fn dispatch<T: Serialize + Sync>(
400        &self,
401        state: &AppState,
402        topic: &str,
403        payload: &T,
404    ) -> AutumnResult<()> {
405        let serialized = serde_json::to_string(payload).map_err(|e| {
406            AutumnError::internal_server_error_msg(format!("failed to serialize payload: {e}"))
407        })?;
408
409        let mut errors = Vec::new();
410        let subs = self.handler.get_subscriptions(topic).await?;
411        for sub in subs {
412            if sub.status == WebhookSubscriptionStatus::Disabled {
413                continue;
414            }
415
416            let log_id = uuid::Uuid::new_v4().to_string();
417            let log = WebhookDeliveryLog {
418                id: log_id.clone(),
419                subscription_id: sub.id.clone(),
420                topic: topic.to_owned(),
421                payload: serialized.clone(),
422                request_headers: HashMap::new(),
423                response_status: None,
424                response_body: None,
425                elapsed_ms: 0,
426                attempt: 1,
427                max_attempts: 5,
428                is_dlq: false,
429                last_error: None,
430                timestamp: Utc::now(),
431            };
432
433            // Register the initial attempt in local storage
434            if let Err(e) = self.handler.log_delivery(log.clone()).await {
435                errors.push(e);
436                continue;
437            }
438
439            // If a delegate extension is registered, run it (delegates to Harvest workflow)
440            if let Some(delegate_ext) = state.extension::<WebhookDelegateExt>() {
441                tracing::info!(subscription_id = %sub.id, "WebhookOutboundManager::dispatch: delegating webhook delivery via runtime hook");
442                if let Err(e) = (delegate_ext.0)(state, sub, log).await {
443                    errors.push(e);
444                }
445            } else {
446                // Fallback: enqueue a standard background job
447                tracing::debug!(subscription_id = %sub.id, "WebhookOutboundManager::dispatch: enqueuing fallback webhook delivery job");
448                if let Some(job_client) = crate::job::global_job_client() {
449                    let job_payload = serde_json::json!({
450                        "log_id": log.id.clone(),
451                    });
452                    if let Err(e) = job_client
453                        .enqueue("autumn_webhook_delivery", job_payload)
454                        .await
455                    {
456                        errors.push(
457                            self.record_delivery_enqueue_failure(log, e.to_string())
458                                .await,
459                        );
460                    }
461                } else {
462                    errors.push(
463                        self.record_delivery_enqueue_failure(
464                            log,
465                            "Global job client is unavailable; fallback webhook delivery job not enqueued"
466                                .to_owned(),
467                        )
468                        .await,
469                    );
470                }
471            }
472        }
473
474        if !errors.is_empty() {
475            return Err(errors.remove(0));
476        }
477
478        Ok(())
479    }
480
481    async fn record_delivery_enqueue_failure(
482        &self,
483        mut log: WebhookDeliveryLog,
484        message: String,
485    ) -> AutumnError {
486        log.is_dlq = true;
487        log.last_error = Some(message.clone());
488        log.timestamp = Utc::now();
489
490        if let Err(e) = self.handler.replace_delivery_log(log).await {
491            tracing::error!(
492                error = %e,
493                "Failed to mark webhook delivery log as DLQ after enqueue failure"
494            );
495            return e;
496        }
497
498        AutumnError::internal_server_error_msg(message)
499    }
500}
501
502fn install_outbound_webhook_manager(
503    state: &AppState,
504    store: Arc<dyn OutboundWebhookHandler>,
505    initial_backoff_ms: u64,
506) {
507    let manager = WebhookOutboundManager::new(store)
508        .with_initial_backoff_ms(initial_backoff_ms)
509        .with_client_from_state(state);
510    state.insert_extension(manager);
511}
512
513/// Asynchronous background job that delivers a webhook payload (legacy fallback).
514#[must_use]
515#[allow(clippy::redundant_closure_for_method_calls, clippy::too_many_lines)]
516pub fn deliver_webhook_job(
517    state: AppState,
518    payload: serde_json::Value,
519) -> Pin<Box<dyn Future<Output = AutumnResult<()>> + Send + 'static>> {
520    Box::pin(async move {
521        let is_replay = payload
522            .get("replay")
523            .and_then(serde_json::Value::as_bool)
524            .unwrap_or(false);
525        let manager = state.extension::<WebhookOutboundManager>().ok_or_else(|| {
526            AutumnError::internal_server_error_msg("WebhookOutboundManager not found in extensions")
527        })?;
528
529        // Support both self-contained payload structure and legacy log_id lookup (for replays)
530        let (sub, mut log) = if let Some(sub_val) = payload.get("subscription") {
531            let _payload_sub: WebhookSubscription = serde_json::from_value(sub_val.clone())
532                .map_err(|e| {
533                    AutumnError::bad_request_msg(format!("failed to parse subscription: {e}"))
534                })?;
535            let mut log: WebhookDeliveryLog = serde_json::from_value(
536                payload
537                    .get("log")
538                    .cloned()
539                    .ok_or_else(|| AutumnError::bad_request_msg("missing log in job payload"))?,
540            )
541            .map_err(|e| AutumnError::bad_request_msg(format!("failed to parse log: {e}")))?;
542
543            // If this log has already been attempted (i.e. is running a retry from the job runner),
544            // increment the attempt counter and write the pre-send log.
545            if log.response_status.is_some() || log.last_error.is_some() {
546                log.attempt = log.attempt.saturating_add(1);
547                log.response_status = None;
548                log.response_body = None;
549                log.last_error = None;
550                manager.store().log_delivery(log.clone()).await?;
551            }
552
553            let sub = load_current_subscription(&manager, &log).await?;
554            (sub, log)
555        } else {
556            let log_id = payload
557                .get("log_id")
558                .and_then(|v| v.as_str())
559                .ok_or_else(|| AutumnError::bad_request_msg("missing log_id in job payload"))?;
560
561            tracing::debug!(log_id = %log_id, "deliver_webhook_job: starting webhook delivery via log lookup");
562
563            let log_opt = manager.store().get_delivery_log(log_id).await?;
564            let mut log = log_opt.ok_or_else(|| {
565                AutumnError::not_found_msg(format!("delivery log {log_id} not found"))
566            })?;
567
568            // If this log has already been attempted (i.e. is running a retry from the job runner),
569            // increment the attempt counter and write the pre-send log.
570            if log.response_status.is_some() || log.last_error.is_some() {
571                log.attempt = log.attempt.saturating_add(1);
572                log.response_status = None;
573                log.response_body = None;
574                log.last_error = None;
575                manager.store().log_delivery(log.clone()).await?;
576            }
577
578            // Load latest subscription state to respect emergency rotations/disable
579            let sub = load_current_subscription(&manager, &log).await?;
580            (sub, log)
581        };
582
583        if sub.status == WebhookSubscriptionStatus::Disabled {
584            tracing::info!(subscription_id = %sub.id, "Webhook subscription is disabled; skipping delivery");
585            log.last_error = Some("Subscription is disabled".to_owned());
586            log.timestamp = Utc::now();
587            if is_replay {
588                log.is_dlq = true;
589            }
590            manager.store().log_delivery(log).await?;
591            return Ok(());
592        }
593
594        if sub.status == WebhookSubscriptionStatus::Failed && !is_replay {
595            tracing::info!(subscription_id = %sub.id, "Webhook subscription has failed; skipping delivery");
596            log.last_error = Some("Subscription has failed due to consecutive errors".to_owned());
597            log.timestamp = Utc::now();
598            manager.store().log_delivery(log).await?;
599            return Ok(());
600        }
601        if sub.status == WebhookSubscriptionStatus::Failed {
602            tracing::info!(subscription_id = %sub.id, "Replaying webhook delivery for failed subscription");
603        }
604
605        // Stripe-style payload signing: t=<timestamp>,v1=<signature>
606        let timestamp = Utc::now().timestamp();
607        let signing_payload = format!("{timestamp}.{}", log.payload);
608        let signature = crate::security::config::hmac_sha256_hex(
609            sub.secret.as_bytes(),
610            signing_payload.as_bytes(),
611        );
612        let signature_header = format!("t={timestamp},v1={signature}");
613
614        let mut request_headers = HashMap::new();
615        request_headers.insert("Content-Type".to_owned(), "application/json".to_owned());
616        request_headers.insert("Autumn-Signature".to_owned(), signature_header.clone());
617
618        let start = std::time::Instant::now();
619        let req = manager
620            .client
621            .named(&sub.target_url)
622            .post(&sub.target_url)
623            .header("Content-Type", "application/json")
624            .header("Autumn-Signature", signature_header)
625            .text_body(log.payload.clone());
626
627        let response = req.send().await;
628        let elapsed = u64::try_from(start.elapsed().as_millis()).unwrap_or(u64::MAX);
629
630        tracing::debug!(
631            log_id = %log.id,
632            status = ?response.as_ref().map(|r| r.status()),
633            "deliver_webhook_job: webhook HTTP request finished"
634        );
635
636        log.elapsed_ms = elapsed;
637        log.timestamp = Utc::now();
638        log.request_headers = request_headers;
639
640        match response {
641            Ok(res) => {
642                let status = res.status();
643                log.response_status = Some(status.as_u16());
644                let is_success = res.is_success();
645                let body_str = cap_logged_response_body(res.text());
646                log.response_body = Some(body_str);
647
648                if is_success {
649                    log.last_error = None;
650                    manager.store().log_delivery(log).await?;
651                    reset_subscription_after_success(&manager, &sub).await;
652                    Ok(())
653                } else {
654                    let status_err = format!("server returned status: {status}");
655                    log.last_error = Some(status_err.clone());
656                    if log.attempt < log.max_attempts {
657                        manager.store().log_delivery(log.clone()).await?;
658                    }
659                    handle_delivery_failure(&manager, &sub, log, status_err).await
660                }
661            }
662            Err(e) => {
663                let error_str = e.to_string();
664                log.last_error = Some(error_str.clone());
665                if log.attempt < log.max_attempts {
666                    manager.store().log_delivery(log.clone()).await?;
667                }
668                handle_delivery_failure(&manager, &sub, log, error_str).await
669            }
670        }
671    })
672}
673
674async fn load_current_subscription(
675    manager: &WebhookOutboundManager,
676    log: &WebhookDeliveryLog,
677) -> AutumnResult<WebhookSubscription> {
678    manager
679        .store()
680        .get_subscription(&log.subscription_id)
681        .await?
682        .ok_or_else(|| {
683            AutumnError::not_found_msg(format!("subscription {} not found", log.subscription_id))
684        })
685}
686
687fn cap_logged_response_body(mut body: String) -> String {
688    if body.len() <= MAX_LOGGED_RESPONSE_BODY_BYTES {
689        return body;
690    }
691
692    let body_budget =
693        MAX_LOGGED_RESPONSE_BODY_BYTES.saturating_sub(TRUNCATED_RESPONSE_BODY_SUFFIX.len());
694    let mut cutoff = body_budget.min(body.len());
695    while cutoff > 0 && !body.is_char_boundary(cutoff) {
696        cutoff -= 1;
697    }
698    body.truncate(cutoff);
699    body.push_str(TRUNCATED_RESPONSE_BODY_SUFFIX);
700    body
701}
702
703async fn reset_subscription_after_success(
704    manager: &WebhookOutboundManager,
705    sub: &WebhookSubscription,
706) {
707    if let Err(e) = manager
708        .store()
709        .reactivate_failed_subscription(&sub.id)
710        .await
711    {
712        tracing::warn!(
713            subscription_id = %sub.id,
714            "Webhook delivery succeeded but subscription failure state could not be reset: {}",
715            e
716        );
717    }
718}
719
720async fn handle_delivery_failure(
721    manager: &WebhookOutboundManager,
722    sub: &WebhookSubscription,
723    mut log: WebhookDeliveryLog,
724    error_msg: String,
725) -> AutumnResult<()> {
726    if log.attempt < log.max_attempts {
727        // Return an error to signal the background job runner to retry this job
728        Err(AutumnError::internal_server_error_msg(format!(
729            "delivery attempt {} failed, scheduled retry: {error_msg}",
730            log.attempt
731        )))
732    } else {
733        log.is_dlq = true;
734        manager.store().log_delivery(log).await?;
735        // Return Ok(()) to mark the permanently failed job as complete and send to DLQ
736        tracing::warn!(subscription_id = %sub.id, "Webhook delivery failed permanently; sent to DLQ: {}", error_msg);
737        Ok(())
738    }
739}
740
741/// `AppBuilder` plugin for outbound signed webhook delivery infrastructure.
742pub struct OutboundWebhookPlugin {
743    store: Arc<dyn OutboundWebhookHandler>,
744    initial_backoff_ms: u64,
745}
746
747impl OutboundWebhookPlugin {
748    /// Create a new outbound webhook plugin using the specified store.
749    #[must_use]
750    pub fn new(store: Arc<dyn OutboundWebhookHandler>) -> Self {
751        Self {
752            store,
753            initial_backoff_ms: 1000,
754        }
755    }
756
757    /// Override the initial backoff retry delay.
758    #[must_use]
759    pub const fn with_initial_backoff_ms(mut self, ms: u64) -> Self {
760        self.initial_backoff_ms = ms;
761        self
762    }
763}
764
765impl crate::plugin::Plugin for OutboundWebhookPlugin {
766    fn build(self, app: crate::app::AppBuilder) -> crate::app::AppBuilder {
767        let store = self.store;
768        let initial_backoff_ms = self.initial_backoff_ms;
769
770        app.state_initializer(move |state| {
771            install_outbound_webhook_manager(state, store.clone(), initial_backoff_ms);
772        })
773        .jobs(vec![crate::job::JobInfo {
774            name: "autumn_webhook_delivery".to_string(),
775            max_attempts: 10, // Retries are handled durably via the background job engine
776            initial_backoff_ms,
777            queue: "default".to_string(),
778            uniqueness: None,
779            concurrency: None,
780            version: 1,
781            handler: deliver_webhook_job,
782        }])
783    }
784}
785
786#[cfg(test)]
787mod tests {
788    use super::*;
789    use crate::http_client::{HttpMockRegistryExt, MockRegistry, MockSetupBuilder};
790    use std::sync::Arc;
791    use std::sync::atomic::{AtomicUsize, Ordering};
792
793    fn mock_builder(registry: Arc<MockRegistry>, alias: &str) -> MockSetupBuilder {
794        MockSetupBuilder {
795            registry,
796            alias: alias.to_owned(),
797            method: None,
798            path: None,
799        }
800    }
801
802    fn sample_subscription(
803        id: &str,
804        target_url: &str,
805        status: WebhookSubscriptionStatus,
806    ) -> WebhookSubscription {
807        WebhookSubscription {
808            id: id.to_owned(),
809            target_url: target_url.to_owned(),
810            event_topics: vec!["orders.created".to_owned()],
811            secret: "my_webhook_signing_secret_32_bytes!!".to_owned(),
812            status,
813            consecutive_failures: if status == WebhookSubscriptionStatus::Failed {
814                50
815            } else {
816                0
817            },
818        }
819    }
820
821    fn sample_log(id: &str, subscription_id: &str) -> WebhookDeliveryLog {
822        WebhookDeliveryLog {
823            id: id.to_owned(),
824            subscription_id: subscription_id.to_owned(),
825            topic: "orders.created".to_owned(),
826            payload: serde_json::json!({ "order_id": "ord_123" }).to_string(),
827            request_headers: HashMap::new(),
828            response_status: None,
829            response_body: None,
830            elapsed_ms: 0,
831            attempt: 1,
832            max_attempts: 5,
833            is_dlq: false,
834            last_error: None,
835            timestamp: Utc::now(),
836        }
837    }
838
839    #[test]
840    fn outbound_webhook_plugin_installs_manager_without_startup_hook() {
841        let store = Arc::new(InMemoryOutboundWebhookHandler::new());
842        let builder = crate::app().plugin(OutboundWebhookPlugin::new(store));
843
844        assert!(
845            builder.startup_hooks.is_empty(),
846            "webhook manager must be installed before job workers start, not from a startup hook"
847        );
848        assert_eq!(builder.state_initializers.len(), 1);
849    }
850
851    #[tokio::test]
852    async fn replay_job_sends_failed_subscription_instead_of_skipping() {
853        let state = AppState::for_test();
854        let store = Arc::new(InMemoryOutboundWebhookHandler::new());
855        let registry = Arc::new(MockRegistry::new());
856        let mock = mock_builder(registry.clone(), "http://mock-receiver/webhooks/replay")
857            .post("/webhooks/replay")
858            .respond_with(200, serde_json::json!({ "received": true }));
859        state.insert_extension(HttpMockRegistryExt(registry));
860        install_outbound_webhook_manager(&state, store.clone(), 1);
861
862        let sub = sample_subscription(
863            "sub_failed",
864            "http://mock-receiver/webhooks/replay",
865            WebhookSubscriptionStatus::Failed,
866        );
867        store.create_subscription(sub).await.unwrap();
868        store
869            .replace_delivery_log(sample_log("log_replay", "sub_failed"))
870            .await
871            .unwrap();
872
873        deliver_webhook_job(
874            state,
875            serde_json::json!({
876                "log_id": "log_replay",
877                "replay": true,
878            }),
879        )
880        .await
881        .unwrap();
882
883        mock.expect_called(1);
884        let log = store
885            .get_delivery_log("log_replay")
886            .await
887            .unwrap()
888            .expect("log should remain stored");
889        assert_eq!(log.response_status, Some(200));
890        assert!(!log.is_dlq);
891        assert!(log.last_error.is_none());
892
893        let updated_sub = store
894            .get_subscription("sub_failed")
895            .await
896            .unwrap()
897            .expect("subscription should remain stored");
898        assert_eq!(updated_sub.status, WebhookSubscriptionStatus::Active);
899        assert_eq!(updated_sub.consecutive_failures, 0);
900    }
901
902    #[tokio::test]
903    async fn replay_job_keeps_disabled_subscription_log_in_dlq() {
904        let state = AppState::for_test();
905        let store = Arc::new(InMemoryOutboundWebhookHandler::new());
906        let registry = Arc::new(MockRegistry::new());
907        let mock = mock_builder(registry.clone(), "http://mock-receiver/webhooks/disabled")
908            .post("/webhooks/disabled")
909            .respond_with(200, serde_json::json!({ "received": true }));
910        state.insert_extension(HttpMockRegistryExt(registry));
911        install_outbound_webhook_manager(&state, store.clone(), 1);
912
913        let sub = sample_subscription(
914            "sub_disabled",
915            "http://mock-receiver/webhooks/disabled",
916            WebhookSubscriptionStatus::Disabled,
917        );
918        store.create_subscription(sub).await.unwrap();
919        store
920            .replace_delivery_log(sample_log("log_disabled_replay", "sub_disabled"))
921            .await
922            .unwrap();
923
924        deliver_webhook_job(
925            state,
926            serde_json::json!({
927                "log_id": "log_disabled_replay",
928                "replay": true,
929            }),
930        )
931        .await
932        .unwrap();
933
934        mock.expect_called(0);
935        let log = store
936            .get_delivery_log("log_disabled_replay")
937            .await
938            .unwrap()
939            .expect("log should remain stored");
940        assert!(log.is_dlq, "disabled replay must remain visible in DLQ");
941        assert_eq!(log.last_error.as_deref(), Some("Subscription is disabled"));
942        assert_eq!(log.response_status, None);
943    }
944
945    #[tokio::test]
946    async fn self_contained_delivery_uses_latest_subscription_state() {
947        let state = AppState::for_test();
948        let store = Arc::new(InMemoryOutboundWebhookHandler::new());
949        let registry = Arc::new(MockRegistry::new());
950        let stale_mock = mock_builder(registry.clone(), "http://mock-receiver/webhooks/stale")
951            .post("/webhooks/stale")
952            .respond_with(200, serde_json::json!({ "received": true }));
953        state.insert_extension(HttpMockRegistryExt(registry));
954        install_outbound_webhook_manager(&state, store.clone(), 1);
955
956        let stored_sub = sample_subscription(
957            "sub_refresh",
958            "http://mock-receiver/webhooks/current-disabled",
959            WebhookSubscriptionStatus::Disabled,
960        );
961        store.create_subscription(stored_sub).await.unwrap();
962        let stale_sub = sample_subscription(
963            "sub_refresh",
964            "http://mock-receiver/webhooks/stale",
965            WebhookSubscriptionStatus::Active,
966        );
967        let log = sample_log("log_refresh", "sub_refresh");
968
969        deliver_webhook_job(
970            state,
971            serde_json::json!({
972                "subscription": stale_sub,
973                "log": log,
974            }),
975        )
976        .await
977        .unwrap();
978
979        stale_mock.expect_called(0);
980        let stored = store
981            .get_delivery_log("log_refresh")
982            .await
983            .unwrap()
984            .expect("delivery log should exist");
985        assert_eq!(stored.response_status, None);
986        assert_eq!(
987            stored.last_error.as_deref(),
988            Some("Subscription is disabled")
989        );
990    }
991
992    #[tokio::test]
993    async fn dispatch_marks_log_dlq_when_fallback_enqueue_fails() {
994        let _guard = crate::job::global_job_runtime_test_lock().lock().await;
995        crate::job::clear_global_job_client();
996
997        let state = AppState::for_test();
998        let store = Arc::new(InMemoryOutboundWebhookHandler::new());
999        let manager = WebhookOutboundManager::new(store.clone()).with_initial_backoff_ms(1);
1000        let sub = sample_subscription(
1001            "sub_enqueue_missing",
1002            "http://mock-receiver/webhooks/enqueue-missing",
1003            WebhookSubscriptionStatus::Active,
1004        );
1005        store.create_subscription(sub).await.unwrap();
1006
1007        let err = manager
1008            .dispatch(&state, "orders.created", &serde_json::json!({ "id": 42 }))
1009            .await
1010            .expect_err("dispatch should report the missing fallback job runtime");
1011        assert!(
1012            err.to_string().contains("not enqueued"),
1013            "error should describe the enqueue failure: {err}"
1014        );
1015
1016        let logs = store.get_delivery_logs().await.unwrap();
1017        assert_eq!(logs.len(), 1);
1018        let log = &logs[0];
1019        assert!(
1020            log.is_dlq,
1021            "enqueue failure must leave a replayable DLQ record"
1022        );
1023        assert!(
1024            log.last_error
1025                .as_deref()
1026                .is_some_and(|msg| msg.contains("not enqueued")),
1027            "DLQ log should record enqueue failure: {:?}",
1028            log.last_error
1029        );
1030        assert_eq!(log.response_status, None);
1031
1032        let sub = store
1033            .get_subscription("sub_enqueue_missing")
1034            .await
1035            .unwrap()
1036            .expect("subscription should remain stored");
1037        assert_eq!(sub.consecutive_failures, 0);
1038    }
1039
1040    #[tokio::test]
1041    async fn delivery_log_response_body_is_capped() {
1042        let state = AppState::for_test();
1043        let store = Arc::new(InMemoryOutboundWebhookHandler::new());
1044        let registry = Arc::new(MockRegistry::new());
1045        let large_body = "x".repeat(MAX_LOGGED_RESPONSE_BODY_BYTES + 1024);
1046        let _mock = mock_builder(
1047            registry.clone(),
1048            "http://mock-receiver/webhooks/large-error",
1049        )
1050        .post("/webhooks/large-error")
1051        .respond_with(500, serde_json::json!({ "error": large_body }));
1052        state.insert_extension(HttpMockRegistryExt(registry));
1053        install_outbound_webhook_manager(&state, store.clone(), 1);
1054
1055        let sub = sample_subscription(
1056            "sub_large_error",
1057            "http://mock-receiver/webhooks/large-error",
1058            WebhookSubscriptionStatus::Active,
1059        );
1060        store.create_subscription(sub.clone()).await.unwrap();
1061        let mut log = sample_log("log_large_error", "sub_large_error");
1062        log.max_attempts = 1;
1063
1064        deliver_webhook_job(
1065            state,
1066            serde_json::json!({
1067                "subscription": sub,
1068                "log": log,
1069            }),
1070        )
1071        .await
1072        .unwrap();
1073
1074        let stored = store
1075            .get_delivery_log("log_large_error")
1076            .await
1077            .unwrap()
1078            .expect("delivery log should exist");
1079        let body = stored
1080            .response_body
1081            .expect("response body should be logged");
1082        assert!(
1083            body.len() <= MAX_LOGGED_RESPONSE_BODY_BYTES,
1084            "stored response body should be capped, got {} bytes",
1085            body.len()
1086        );
1087        assert!(body.ends_with("[truncated]"));
1088    }
1089
1090    struct CountingReplacementStore {
1091        log_delivery_calls: AtomicUsize,
1092    }
1093
1094    impl CountingReplacementStore {
1095        fn new() -> Self {
1096            Self {
1097                log_delivery_calls: AtomicUsize::new(0),
1098            }
1099        }
1100
1101        fn log_delivery_count(&self) -> usize {
1102            self.log_delivery_calls.load(Ordering::SeqCst)
1103        }
1104    }
1105
1106    impl OutboundWebhookHandler for CountingReplacementStore {
1107        fn get_subscriptions(
1108            &self,
1109            _topic: &str,
1110        ) -> Pin<Box<dyn Future<Output = AutumnResult<Vec<WebhookSubscription>>> + Send>> {
1111            Box::pin(async { Ok(Vec::new()) })
1112        }
1113
1114        fn log_delivery(
1115            &self,
1116            _log: WebhookDeliveryLog,
1117        ) -> Pin<Box<dyn Future<Output = AutumnResult<()>> + Send>> {
1118            self.log_delivery_calls.fetch_add(1, Ordering::SeqCst);
1119            Box::pin(async { Ok(()) })
1120        }
1121
1122        fn replace_delivery_log(
1123            &self,
1124            _log: WebhookDeliveryLog,
1125        ) -> Pin<Box<dyn Future<Output = AutumnResult<()>> + Send>> {
1126            Box::pin(async { Ok(()) })
1127        }
1128
1129        fn get_subscription(
1130            &self,
1131            _id: &str,
1132        ) -> Pin<Box<dyn Future<Output = AutumnResult<Option<WebhookSubscription>>> + Send>>
1133        {
1134            Box::pin(async { Ok(None) })
1135        }
1136
1137        fn get_delivery_log(
1138            &self,
1139            _id: &str,
1140        ) -> Pin<Box<dyn Future<Output = AutumnResult<Option<WebhookDeliveryLog>>> + Send>>
1141        {
1142            Box::pin(async { Ok(None) })
1143        }
1144    }
1145
1146    #[tokio::test]
1147    async fn replace_delivery_log_is_not_a_delivery_outcome() {
1148        let store = CountingReplacementStore::new();
1149        let mut log = sample_log("log_replace", "sub_replace");
1150        log.response_status = Some(500);
1151        log.last_error = Some("server returned status: 500 Internal Server Error".to_owned());
1152        log.is_dlq = true;
1153
1154        store.replace_delivery_log(log).await.unwrap();
1155
1156        assert_eq!(
1157            store.log_delivery_count(),
1158            0,
1159            "plain delivery-log replacement must not call log_delivery"
1160        );
1161    }
1162
1163    struct ResetFailingStore {
1164        inner: InMemoryOutboundWebhookHandler,
1165    }
1166
1167    impl ResetFailingStore {
1168        fn new() -> Self {
1169            Self {
1170                inner: InMemoryOutboundWebhookHandler::new(),
1171            }
1172        }
1173
1174        async fn create_subscription(&self, sub: WebhookSubscription) {
1175            self.inner.create_subscription(sub).await.unwrap();
1176        }
1177
1178        async fn delivery_log(&self, id: &str) -> WebhookDeliveryLog {
1179            self.inner
1180                .get_delivery_log(id)
1181                .await
1182                .unwrap()
1183                .expect("delivery log should exist")
1184        }
1185    }
1186
1187    impl OutboundWebhookHandler for ResetFailingStore {
1188        fn get_subscriptions(
1189            &self,
1190            topic: &str,
1191        ) -> Pin<Box<dyn Future<Output = AutumnResult<Vec<WebhookSubscription>>> + Send>> {
1192            <InMemoryOutboundWebhookHandler as OutboundWebhookHandler>::get_subscriptions(
1193                &self.inner,
1194                topic,
1195            )
1196        }
1197
1198        fn log_delivery(
1199            &self,
1200            log: WebhookDeliveryLog,
1201        ) -> Pin<Box<dyn Future<Output = AutumnResult<()>> + Send>> {
1202            <InMemoryOutboundWebhookHandler as OutboundWebhookHandler>::log_delivery(
1203                &self.inner,
1204                log,
1205            )
1206        }
1207
1208        fn replace_delivery_log(
1209            &self,
1210            log: WebhookDeliveryLog,
1211        ) -> Pin<Box<dyn Future<Output = AutumnResult<()>> + Send>> {
1212            <InMemoryOutboundWebhookHandler as OutboundWebhookHandler>::replace_delivery_log(
1213                &self.inner,
1214                log,
1215            )
1216        }
1217
1218        fn get_subscription(
1219            &self,
1220            id: &str,
1221        ) -> Pin<Box<dyn Future<Output = AutumnResult<Option<WebhookSubscription>>> + Send>>
1222        {
1223            <InMemoryOutboundWebhookHandler as OutboundWebhookHandler>::get_subscription(
1224                &self.inner,
1225                id,
1226            )
1227        }
1228
1229        fn get_delivery_log(
1230            &self,
1231            id: &str,
1232        ) -> Pin<Box<dyn Future<Output = AutumnResult<Option<WebhookDeliveryLog>>> + Send>>
1233        {
1234            <InMemoryOutboundWebhookHandler as OutboundWebhookHandler>::get_delivery_log(
1235                &self.inner,
1236                id,
1237            )
1238        }
1239
1240        fn reset_subscription_failures(
1241            &self,
1242            _id: &str,
1243        ) -> Pin<Box<dyn Future<Output = AutumnResult<()>> + Send>> {
1244            Box::pin(async {
1245                Err(AutumnError::internal_server_error_msg(
1246                    "reset backend unavailable",
1247                ))
1248            })
1249        }
1250    }
1251
1252    #[tokio::test]
1253    async fn successful_delivery_does_not_retry_when_failure_reset_fails() {
1254        let state = AppState::for_test();
1255        let store = Arc::new(ResetFailingStore::new());
1256        let registry = Arc::new(MockRegistry::new());
1257        let mock = mock_builder(registry.clone(), "http://mock-receiver/webhooks/success")
1258            .post("/webhooks/success")
1259            .respond_with(200, serde_json::json!({ "received": true }));
1260        state.insert_extension(HttpMockRegistryExt(registry));
1261        install_outbound_webhook_manager(&state, store.clone(), 1);
1262
1263        let sub = sample_subscription(
1264            "sub_success",
1265            "http://mock-receiver/webhooks/success",
1266            WebhookSubscriptionStatus::Active,
1267        );
1268        store.create_subscription(sub.clone()).await;
1269        let log = sample_log("log_success", "sub_success");
1270
1271        deliver_webhook_job(
1272            state,
1273            serde_json::json!({
1274                "subscription": sub,
1275                "log": log,
1276            }),
1277        )
1278        .await
1279        .expect("accepted webhook delivery must not be retried because counter reset failed");
1280
1281        mock.expect_called(1);
1282        let persisted = store.delivery_log("log_success").await;
1283        assert_eq!(persisted.response_status, Some(200));
1284        assert!(persisted.last_error.is_none());
1285    }
1286
1287    #[tokio::test]
1288    async fn webhook_manager_uses_http_client_config_base_urls() {
1289        let _guard = crate::job::global_job_runtime_test_lock().lock().await;
1290        crate::job::clear_global_job_client();
1291
1292        let store = Arc::new(InMemoryOutboundWebhookHandler::new());
1293        let plugin = OutboundWebhookPlugin::new(store.clone()).with_initial_backoff_ms(1);
1294        let mut config = crate::config::AutumnConfig::default();
1295        config.http.client.base_urls.insert(
1296            "hook-service".to_owned(),
1297            "http://mock-receiver/base".to_owned(),
1298        );
1299
1300        let mut app_builder = crate::test::TestApp::new().config(config).plugin(plugin);
1301        let mock = app_builder
1302            .http_mock("hook-service")
1303            .post("/base/hook-service")
1304            .respond_with(200, serde_json::json!({ "received": true }));
1305        let app = app_builder.build();
1306        let state = app.state();
1307
1308        let sub = sample_subscription(
1309            "sub_config",
1310            "hook-service",
1311            WebhookSubscriptionStatus::Active,
1312        );
1313        store.create_subscription(sub.clone()).await.unwrap();
1314        let log = sample_log("log_config", "sub_config");
1315
1316        deliver_webhook_job(
1317            state.clone(),
1318            serde_json::json!({
1319                "subscription": sub,
1320                "log": log,
1321            }),
1322        )
1323        .await
1324        .unwrap();
1325
1326        mock.expect_called(1);
1327        crate::job::clear_global_job_client();
1328    }
1329}
autumn_web/webhook_outbound.rs

autumn_web/
webhook_outbound.rs
