sova_idempotency/
lib.rs

1//! Inbound `Idempotency-Key` middleware for mutating HTTP methods.
2
3use bytes::Bytes;
4use http::Method;
5use serde::{Deserialize, Serialize};
6use sova_core::extend::{named, MwEntry};
7use sova_core::{with_state, App, Plugin, Request, Response};
8use sova_store::{AppStore, KvStore};
9use std::sync::Arc;
10use std::time::Duration;
11
12const DEFAULT_TTL: Duration = Duration::from_secs(24 * 60 * 60);
13const DEFAULT_MAX_BODY: usize = 256 * 1024;
14
15/// Replay cached successful responses for the same `Idempotency-Key`.
16pub struct Idempotency {
17    store: Arc<dyn KvStore>,
18    ttl: Duration,
19    max_body: usize,
20    prefix: String,
21}
22
23#[derive(Serialize, Deserialize)]
24struct CachedResponse {
25    status: u16,
26    content_type: Option<String>,
27    body: Vec<u8>,
28}
29
30impl Idempotency {
31    pub fn from_store(store: Arc<dyn KvStore>) -> Self {
32        Self {
33            store,
34            ttl: DEFAULT_TTL,
35            max_body: DEFAULT_MAX_BODY,
36            prefix: "idem:".into(),
37        }
38    }
39
40    pub fn ttl(mut self, ttl: Duration) -> Self {
41        self.ttl = ttl;
42        self
43    }
44
45    pub fn max_body(mut self, bytes: usize) -> Self {
46        self.max_body = bytes.max(1);
47        self
48    }
49
50    pub fn prefix(mut self, prefix: impl Into<String>) -> Self {
51        self.prefix = prefix.into();
52        self
53    }
54
55    /// Route / mount middleware (same as plugin install).
56    pub fn middleware(self) -> MwEntry {
57        named(
58            "idempotency",
59            with_state(self, |rt, req, next| async move {
60                run(rt, req, next).await
61            }),
62        )
63    }
64
65    /// Prefer an installed [`AppStore`], else fall back to `store`.
66    pub fn from_app_or(app: &App, store: Arc<dyn KvStore>) -> Self {
67        let store = app
68            .try_state::<AppStore>()
69            .map(|s| Arc::clone(&s.inner))
70            .unwrap_or(store);
71        Self::from_store(store)
72    }
73}
74
75impl Plugin for Idempotency {
76    fn id(&self) -> &'static str {
77        "idempotency"
78    }
79
80    fn meta(&self) -> sova_core::PluginMeta {
81        sova_core::PluginMeta::new("Idempotency")
82            .description("Replay 2xx responses for Idempotency-Key on mutating methods")
83            .version(env!("CARGO_PKG_VERSION"))
84    }
85
86    fn install(self, app: &mut App) {
87        app.use_middleware(self.middleware());
88    }
89}
90
91fn is_mutating(method: &Method) -> bool {
92    matches!(
93        *method,
94        Method::POST | Method::PUT | Method::PATCH | Method::DELETE
95    )
96}
97
98async fn run(
99    rt: Arc<Idempotency>,
100    req: Request,
101    next: sova_core::Next,
102) -> Response {
103    if !is_mutating(&req.method) {
104        return next(req).await;
105    }
106    let Some(key) = req
107        .header("idempotency-key")
108        .map(str::trim)
109        .filter(|s| !s.is_empty())
110        .map(|s| s.to_string())
111    else {
112        return next(req).await;
113    };
114
115    let cache_key = format!("{}{}", rt.prefix, key);
116    if let Some(bytes) = rt.store.get(&cache_key).await {
117        if let Ok(cached) = serde_json::from_slice::<CachedResponse>(&bytes) {
118            let mut res = Response::bytes(cached.body, cached.content_type.as_deref().unwrap_or("application/octet-stream"))
119                .status(cached.status)
120                .header("x-idempotency-replay", "true");
121            if let Some(ct) = cached.content_type {
122                res = res.header("content-type", ct);
123            }
124            return res;
125        }
126    }
127
128    let res = next(req).await;
129    let status = res.status_code().as_u16();
130    if !(200..300).contains(&status) {
131        return res;
132    }
133
134    let Some(body) = res.body_bytes().map(|b| b.to_vec()) else {
135        return res;
136    };
137    if body.len() > rt.max_body {
138        return res;
139    }
140
141    let content_type = res
142        .headers()
143        .get(http::header::CONTENT_TYPE)
144        .and_then(|v| v.to_str().ok())
145        .map(|s| s.to_string());
146
147    let payload = CachedResponse {
148        status,
149        content_type,
150        body,
151    };
152    if let Ok(raw) = serde_json::to_vec(&payload) {
153        rt.store
154            .set(&cache_key, Bytes::from(raw), Some(rt.ttl))
155            .await;
156    }
157    res.header("x-idempotency-replay", "false")
158}
sova_idempotency/lib.rs

sova_idempotency/
lib.rs
