minco_plugin_idempotency/
lib.rs

1//! Default idempotency primitives and a deterministic in-memory implementation for tests/local use.
2#![forbid(unsafe_code)]
3
4use async_trait::async_trait;
5use chrono::{DateTime, Utc};
6use minco_core::{
7    CapabilityProvision, Plugin, PluginContext, PluginDescriptor, PluginError, PluginId,
8};
9use semver::Version;
10use serde::{Deserialize, Serialize};
11use sha2::{Digest, Sha256};
12use std::{collections::BTreeMap, sync::Arc};
13use tokio::sync::RwLock;
14
15#[derive(Debug, Clone, PartialEq, Eq, PartialOrd, Ord, Serialize, Deserialize)]
16#[serde(transparent)]
17pub struct IdempotencyKey(String);
18impl IdempotencyKey {
19    pub fn parse(value: impl Into<String>) -> Result<Self, IdempotencyError> {
20        let value = value.into();
21        if value.trim().is_empty() || value.len() > 200 || value.chars().any(char::is_control) {
22            return Err(IdempotencyError::InvalidKey);
23        }
24        Ok(Self(value))
25    }
26    pub fn as_str(&self) -> &str {
27        &self.0
28    }
29}
30
31#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
32#[serde(transparent)]
33pub struct RequestFingerprint(String);
34impl RequestFingerprint {
35    pub fn from_serializable<T: Serialize>(value: &T) -> Result<Self, IdempotencyError> {
36        let bytes = serde_json::to_vec(value).map_err(IdempotencyError::Serialization)?;
37        Ok(Self(format!("{:x}", Sha256::digest(bytes))))
38    }
39}
40
41#[derive(Debug, Clone, PartialEq, Eq, Serialize, Deserialize)]
42pub struct IdempotencyRecord {
43    pub fingerprint: RequestFingerprint,
44    pub response: serde_json::Value,
45    pub created_at: DateTime<Utc>,
46}
47
48#[async_trait]
49pub trait IdempotencyStore: Send + Sync + std::fmt::Debug {
50    async fn get(
51        &self,
52        key: &IdempotencyKey,
53    ) -> Result<Option<IdempotencyRecord>, IdempotencyError>;
54    async fn put_if_absent(
55        &self,
56        key: IdempotencyKey,
57        record: IdempotencyRecord,
58    ) -> Result<PutOutcome, IdempotencyError>;
59}
60
61#[derive(Debug, Clone, PartialEq, Eq)]
62pub enum PutOutcome {
63    Inserted,
64    Existing(IdempotencyRecord),
65}
66
67#[derive(Debug, Default)]
68pub struct MemoryIdempotencyStore {
69    records: RwLock<BTreeMap<IdempotencyKey, IdempotencyRecord>>,
70}
71
72#[async_trait]
73impl IdempotencyStore for MemoryIdempotencyStore {
74    async fn get(
75        &self,
76        key: &IdempotencyKey,
77    ) -> Result<Option<IdempotencyRecord>, IdempotencyError> {
78        Ok(self.records.read().await.get(key).cloned())
79    }
80    async fn put_if_absent(
81        &self,
82        key: IdempotencyKey,
83        record: IdempotencyRecord,
84    ) -> Result<PutOutcome, IdempotencyError> {
85        let mut records = self.records.write().await;
86        let outcome = match records.entry(key) {
87            std::collections::btree_map::Entry::Occupied(existing) => {
88                PutOutcome::Existing(existing.get().clone())
89            }
90            std::collections::btree_map::Entry::Vacant(entry) => {
91                entry.insert(record);
92                PutOutcome::Inserted
93            }
94        };
95        drop(records);
96        Ok(outcome)
97    }
98}
99
100#[derive(Debug, Clone, Default)]
101pub struct IdempotencyPlugin;
102impl Plugin for IdempotencyPlugin {
103    fn descriptor(&self) -> PluginDescriptor {
104        let mut descriptor = PluginDescriptor::new(
105            PluginId::new("idempotency").expect("static id"),
106            Version::new(1, 0, 0),
107            "Idempotency keys, fingerprints and storage port",
108        );
109        descriptor.default_enabled = true;
110        descriptor.provides.push(CapabilityProvision {
111            name: "http.idempotency".into(),
112            version: Version::new(1, 0, 0),
113        });
114        descriptor
115    }
116    fn install(&self, context: &mut PluginContext<'_>) -> Result<(), PluginError> {
117        context
118            .services()
119            .insert(Arc::new(MemoryIdempotencyStore::default()))?;
120        Ok(())
121    }
122}
123
124#[derive(Debug, thiserror::Error)]
125pub enum IdempotencyError {
126    #[error("idempotency key must contain 1-200 visible characters")]
127    InvalidKey,
128    #[error("failed to serialize request fingerprint: {0}")]
129    Serialization(serde_json::Error),
130    #[error("idempotency store failed: {0}")]
131    Store(String),
132}
133
134#[cfg(test)]
135mod tests {
136    use super::*;
137    #[tokio::test]
138    async fn put_if_absent_returns_the_original_record() {
139        let store = MemoryIdempotencyStore::default();
140        let key = IdempotencyKey::parse("request-1").unwrap();
141        let record = IdempotencyRecord {
142            fingerprint: RequestFingerprint::from_serializable(&serde_json::json!({"a":1}))
143                .unwrap(),
144            response: serde_json::json!({"ok":true}),
145            created_at: Utc::now(),
146        };
147        assert_eq!(
148            store
149                .put_if_absent(key.clone(), record.clone())
150                .await
151                .unwrap(),
152            PutOutcome::Inserted
153        );
154        assert_eq!(
155            store.put_if_absent(key, record.clone()).await.unwrap(),
156            PutOutcome::Existing(record)
157        );
158    }
159}
minco_plugin_idempotency/lib.rs

minco_plugin_idempotency/
lib.rs
