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}