rustlavel_payment/
receiver.rs

1//! The endpoint a gateway calls.
2//!
3//! Three things, in order, and the order is the point:
4//!
5//! 1. Verify the signature. Anything that fails is `401` and is not parsed
6//!    further — a body that did not come from the gateway is not looked at.
7//! 2. Record the event under the gateway's id. A duplicate is answered `200`
8//!    *without* running the handler: the gateway retried because it did not
9//!    hear the first acknowledgement, and it needs to hear this one.
10//! 3. Run the application's handler. If it fails, answer `500` so the gateway
11//!    retries — and undo the record, so the retry is not then dropped as a
12//!    duplicate of an event nothing handled.
13//!
14//! Step 3's undo is what makes "at least once" from the gateway into "exactly
15//! once" at the application: a handler that crashed half-way is retried, and
16//! a handler that finished is never run again.
17
18use crate::gateway::Gateway;
19use crate::webhook::{WebhookEvent, WebhookLog};
20use rustlavel_core::{Json, Result};
21use rustlavel_http::{BoxFuture, Request, Response, Status};
22use std::sync::Arc;
23
24/// What the receiver did with a delivery.
25#[derive(Debug, Clone, PartialEq, Eq)]
26pub enum Handled {
27    /// Verified, new, and the handler ran.
28    Processed,
29    /// Verified, but seen before. The handler did not run.
30    Duplicate,
31    /// The signature did not check out. Nothing was parsed.
32    Refused,
33    /// Verified and new, but the handler failed. The record was withdrawn so
34    /// the gateway's retry is processed.
35    HandlerFailed(String),
36}
37
38pub type Handler =
39    Arc<dyn Fn(WebhookEvent) -> BoxFuture<Result<()>> + Send + Sync + 'static>;
40
41pub struct Receiver {
42    gateway: Arc<dyn Gateway>,
43    log: Arc<dyn WebhookLog>,
44    handler: Handler,
45}
46
47impl Receiver {
48    pub fn new(
49        gateway: Arc<dyn Gateway>,
50        log: Arc<dyn WebhookLog>,
51        handler: impl Fn(WebhookEvent) -> BoxFuture<Result<()>> + Send + Sync + 'static,
52    ) -> Receiver {
53        Receiver { gateway, log, handler: Arc::new(handler) }
54    }
55
56    /// Process one delivery. Separated from the HTTP handler so it can be
57    /// tested without a request, and so a replay tool can call it.
58    pub async fn deliver(&self, headers: &rustlavel_http::Headers, body: &[u8]) -> Handled {
59        let event = match self.gateway.verify_webhook(headers, body) {
60            Ok(event) => event,
61            Err(error) => {
62                // Warn, not error: a refused webhook is usually a
63                // misconfigured secret or a probe, and neither is an incident.
64                // The gateway's name says which secret to check.
65                rustlavel_core::warn!(
66                    "payment: refused a webhook for {}: {error}",
67                    self.gateway.name()
68                );
69                return Handled::Refused;
70            }
71        };
72
73        match self.log.record(&event).await {
74            Ok(true) => {}
75            Ok(false) => {
76                rustlavel_core::debug!("payment: duplicate {} ignored", event.dedup_key());
77                return Handled::Duplicate;
78            }
79            Err(error) => {
80                // The log is down. Failing the delivery is the safe direction:
81                // the gateway retries, and nothing is credited without a
82                // record of it.
83                return Handled::HandlerFailed(format!("the webhook log failed: {error}"));
84            }
85        }
86
87        let key = event.dedup_key();
88        match (self.handler)(event).await {
89            Ok(()) => Handled::Processed,
90            Err(error) => {
91                rustlavel_core::error!("payment: the handler for {key} failed: {error}");
92                // Withdrawn, so the retry the 500 provokes is processed. If
93                // the withdrawal itself fails the event stays recorded and
94                // the retry will be dropped — said loudly, because that is a
95                // payment nobody will credit.
96                if let Err(inner) = self.log.withdraw(&key).await {
97                    rustlavel_core::error!(
98                        "payment: could not withdraw {key} after a failed handler; the gateway's \
99                         retry will be treated as a duplicate: {inner}"
100                    );
101                }
102                Handled::HandlerFailed(error.to_string())
103            }
104        }
105    }
106
107    /// The route handler: mount it at whatever path the gateway was given.
108    ///
109    /// ```ignore
110    /// let receiver = Arc::new(Receiver::new(gateway, log, |event| Box::pin(async move { … })));
111    /// r.post("/webhooks/payment", move |req: Request| {
112    ///     let receiver = Arc::clone(&receiver);
113    ///     async move { receiver.handle(req).await }
114    /// });
115    /// ```
116    ///
117    /// Behind no CSRF and no session middleware: the caller is a server with
118    /// a signature, and a CSRF check would refuse it every time.
119    pub async fn handle(&self, request: Request) -> Response {
120        match self.deliver(request.headers(), request.body()).await {
121            Handled::Processed | Handled::Duplicate => Response::ok(),
122            Handled::Refused => Response::new(Status::UNAUTHORIZED)
123                .with_json(Json::object([("message", Json::from("the signature did not verify"))])),
124            // 500 so the gateway retries. The record was not kept, so the
125            // retry is processed rather than dropped as a duplicate.
126            Handled::HandlerFailed(_) => Response::new(Status::INTERNAL_ERROR),
127        }
128    }
129}
rustlavel_payment/receiver.rs

rustlavel_payment/
receiver.rs
