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    /// Use installed [`AppStore`] (`idem` namespace). Panics if SharedStore is missing.
41    pub fn from_app(app: &App) -> Self {
42        let store = app.try_state::<AppStore>().unwrap_or_else(|| {
43            panic!("Idempotency::from_app requires SharedStore / AppStore installed first")
44        });
45        Self::from_store(store.namespaced("idem"))
46    }
47
48    pub fn ttl(mut self, ttl: Duration) -> Self {
49        self.ttl = ttl;
50        self
51    }
52
53    pub fn max_body(mut self, bytes: usize) -> Self {
54        self.max_body = bytes.max(1);
55        self
56    }
57
58    pub fn prefix(mut self, prefix: impl Into<String>) -> Self {
59        self.prefix = prefix.into();
60        self
61    }
62
63    /// Route / mount middleware (same as plugin install).
64    pub fn middleware(self) -> MwEntry {
65        named(
66            "idempotency",
67            with_state(self, |rt, req, next| async move {
68                run(rt, req, next).await
69            }),
70        )
71    }
72
73    /// Prefer an installed [`AppStore`], else fall back to `store`.
74    pub fn from_app_or(app: &App, store: Arc<dyn KvStore>) -> Self {
75        let store = app
76            .try_state::<AppStore>()
77            .map(|s| Arc::clone(&s.inner))
78            .unwrap_or(store);
79        Self::from_store(store)
80    }
81}
82
83impl Plugin for Idempotency {
84    fn id(&self) -> &'static str {
85        "idempotency"
86    }
87
88    fn meta(&self) -> sova_core::PluginMeta {
89        sova_core::PluginMeta::new("Idempotency")
90            .description("Replay 2xx responses for Idempotency-Key on mutating methods")
91            .version(env!("CARGO_PKG_VERSION"))
92    }
93
94    fn install(self, app: &mut App) {
95        app.use_middleware(self.middleware());
96    }
97}
98
99fn is_mutating(method: &Method) -> bool {
100    matches!(
101        *method,
102        Method::POST | Method::PUT | Method::PATCH | Method::DELETE
103    )
104}
105
106async fn run(
107    rt: Arc<Idempotency>,
108    req: Request,
109    next: sova_core::Next,
110) -> Response {
111    if !is_mutating(&req.method) {
112        return next(req).await;
113    }
114    let Some(key) = req
115        .header("idempotency-key")
116        .map(str::trim)
117        .filter(|s| !s.is_empty())
118        .map(|s| s.to_string())
119    else {
120        return next(req).await;
121    };
122
123    let cache_key = format!("{}{}", rt.prefix, key);
124    if let Some(bytes) = rt.store.get(&cache_key).await {
125        if let Ok(cached) = serde_json::from_slice::<CachedResponse>(&bytes) {
126            let mut res = Response::bytes(cached.body, cached.content_type.as_deref().unwrap_or("application/octet-stream"))
127                .status(cached.status)
128                .header("x-idempotency-replay", "true");
129            if let Some(ct) = cached.content_type {
130                res = res.header("content-type", ct);
131            }
132            return res;
133        }
134    }
135
136    let res = next(req).await;
137    let status = res.status_code().as_u16();
138    if !(200..300).contains(&status) {
139        return res;
140    }
141
142    let Some(body) = res.body_bytes().map(|b| b.to_vec()) else {
143        return res;
144    };
145    if body.len() > rt.max_body {
146        return res;
147    }
148
149    let content_type = res
150        .headers()
151        .get(http::header::CONTENT_TYPE)
152        .and_then(|v| v.to_str().ok())
153        .map(|s| s.to_string());
154
155    let payload = CachedResponse {
156        status,
157        content_type,
158        body,
159    };
160    if let Ok(raw) = serde_json::to_vec(&payload) {
161        rt.store
162            .set(&cache_key, Bytes::from(raw), Some(rt.ttl))
163            .await;
164    }
165    res.header("x-idempotency-replay", "false")
166}
sova_idempotency/lib.rs

sova_idempotency/
lib.rs
