stateset_db/sqlite/
http_idempotency.rs

1//! SQLite implementation of the durable HTTP idempotency store.
2
3use chrono::{DateTime, Utc};
4use r2d2::Pool;
5use r2d2_sqlite::SqliteConnectionManager;
6use rusqlite::OptionalExtension as _;
7use stateset_core::{CommerceError, Result};
8
9use crate::{HttpIdempotencyRecord, HttpIdempotencyRepository};
10
11/// SQLite repository over the `http_idempotency_keys` table.
12#[derive(Debug)]
13pub struct SqliteHttpIdempotencyRepository {
14    pool: Pool<SqliteConnectionManager>,
15}
16
17impl SqliteHttpIdempotencyRepository {
18    #[must_use]
19    pub const fn new(pool: Pool<SqliteConnectionManager>) -> Self {
20        Self { pool }
21    }
22
23    fn conn(&self) -> Result<r2d2::PooledConnection<SqliteConnectionManager>> {
24        self.pool.get().map_err(|e| CommerceError::DatabaseError(e.to_string()))
25    }
26}
27
28fn map_err(e: rusqlite::Error) -> CommerceError {
29    CommerceError::DatabaseError(e.to_string())
30}
31
32impl HttpIdempotencyRepository for SqliteHttpIdempotencyRepository {
33    fn get(
34        &self,
35        tenant: &str,
36        key: &str,
37        expired_before: DateTime<Utc>,
38    ) -> Result<Option<HttpIdempotencyRecord>> {
39        let conn = self.conn()?;
40        // Lazy cleanup: drop an expired row for this key before reading.
41        // `<=` matches the in-memory expiry (`age >= ttl`): an entry created
42        // exactly at the cutoff is expired. With `<` and a zero TTL, a row
43        // written in the same millisecond as the lookup replayed after it
44        // should have expired (flaky under coverage instrumentation).
45        conn.execute(
46            "DELETE FROM http_idempotency_keys
47             WHERE tenant = ?1 AND idempotency_key = ?2 AND created_at <= ?3",
48            rusqlite::params![tenant, key, expired_before.timestamp_millis()],
49        )
50        .map_err(map_err)?;
51
52        conn.query_row(
53            "SELECT request_fingerprint, response_status, content_type, response_body, created_at
54             FROM http_idempotency_keys
55             WHERE tenant = ?1 AND idempotency_key = ?2",
56            rusqlite::params![tenant, key],
57            |row| {
58                Ok((
59                    row.get::<_, String>(0)?,
60                    row.get::<_, u16>(1)?,
61                    row.get::<_, Option<String>>(2)?,
62                    row.get::<_, Vec<u8>>(3)?,
63                    row.get::<_, i64>(4)?,
64                ))
65            },
66        )
67        .optional()
68        .map_err(map_err)?
69        .map(|(request_fingerprint, response_status, content_type, response_body, created_ms)| {
70            let created_at =
71                DateTime::<Utc>::from_timestamp_millis(created_ms).ok_or_else(|| {
72                    CommerceError::DatabaseError(format!(
73                        "http_idempotency_keys.created_at out of range: {created_ms}"
74                    ))
75                })?;
76            Ok(HttpIdempotencyRecord {
77                tenant: tenant.to_string(),
78                idempotency_key: key.to_string(),
79                request_fingerprint,
80                response_status,
81                content_type,
82                response_body,
83                created_at,
84            })
85        })
86        .transpose()
87    }
88
89    fn put(&self, record: &HttpIdempotencyRecord) -> Result<bool> {
90        let conn = self.conn()?;
91        let inserted = conn
92            .execute(
93                "INSERT INTO http_idempotency_keys
94                 (tenant, idempotency_key, request_fingerprint, response_status,
95                  content_type, response_body, created_at)
96                 VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)
97                 ON CONFLICT(tenant, idempotency_key) DO NOTHING",
98                rusqlite::params![
99                    record.tenant,
100                    record.idempotency_key,
101                    record.request_fingerprint,
102                    record.response_status,
103                    record.content_type,
104                    record.response_body,
105                    record.created_at.timestamp_millis(),
106                ],
107            )
108            .map_err(map_err)?;
109        Ok(inserted > 0)
110    }
111
112    fn purge_expired(&self, expired_before: DateTime<Utc>) -> Result<u64> {
113        let conn = self.conn()?;
114        let deleted = conn
115            .execute(
116                "DELETE FROM http_idempotency_keys WHERE created_at < ?1",
117                rusqlite::params![expired_before.timestamp_millis()],
118            )
119            .map_err(map_err)?;
120        Ok(deleted as u64)
121    }
122}
123
124#[cfg(test)]
125mod tests {
126    use super::*;
127    use crate::{DatabaseConfig, SqliteDatabase};
128    use chrono::Duration;
129
130    fn repo() -> SqliteHttpIdempotencyRepository {
131        let db = SqliteDatabase::new(&DatabaseConfig::in_memory()).expect("in-memory db");
132        db.http_idempotency()
133    }
134
135    fn record(key: &str, created_at: DateTime<Utc>) -> HttpIdempotencyRecord {
136        HttpIdempotencyRecord {
137            tenant: "tenant-a".into(),
138            idempotency_key: key.into(),
139            request_fingerprint: "f".repeat(64),
140            response_status: 201,
141            content_type: Some("application/json".into()),
142            response_body: b"{\"id\":\"order-1\"}".to_vec(),
143            created_at,
144        }
145    }
146
147    #[test]
148    fn put_then_get_round_trips() {
149        let repo = repo();
150        let now = Utc::now();
151        let rec = record("k1", now);
152        assert!(repo.put(&rec).unwrap());
153
154        let cutoff = now - Duration::hours(1);
155        let loaded = repo.get("tenant-a", "k1", cutoff).unwrap().expect("record");
156        assert_eq!(loaded.request_fingerprint, rec.request_fingerprint);
157        assert_eq!(loaded.response_status, 201);
158        assert_eq!(loaded.content_type.as_deref(), Some("application/json"));
159        assert_eq!(loaded.response_body, rec.response_body);
160        // Millisecond precision round trip.
161        assert_eq!(loaded.created_at.timestamp_millis(), now.timestamp_millis());
162        // Other tenants and keys see nothing.
163        assert!(repo.get("tenant-b", "k1", cutoff).unwrap().is_none());
164        assert!(repo.get("tenant-a", "other", cutoff).unwrap().is_none());
165    }
166
167    #[test]
168    fn put_is_first_write_wins() {
169        let repo = repo();
170        let now = Utc::now();
171        assert!(repo.put(&record("k1", now)).unwrap());
172
173        let mut second = record("k1", now);
174        second.response_body = b"different".to_vec();
175        // Duplicate insert is ignored, not overwritten.
176        assert!(!repo.put(&second).unwrap());
177        let loaded = repo.get("tenant-a", "k1", now - Duration::hours(1)).unwrap().expect("record");
178        assert_eq!(loaded.response_body, b"{\"id\":\"order-1\"}");
179    }
180
181    #[test]
182    fn get_lazily_deletes_expired_rows() {
183        let repo = repo();
184        let created = Utc::now() - Duration::hours(25);
185        assert!(repo.put(&record("k1", created)).unwrap());
186
187        // With a 24h TTL the row is expired: get() returns None and deletes it.
188        let cutoff = Utc::now() - Duration::hours(24);
189        assert!(repo.get("tenant-a", "k1", cutoff).unwrap().is_none());
190        // Even a permissive cutoff now finds nothing — the row is gone.
191        assert!(repo.get("tenant-a", "k1", DateTime::<Utc>::MIN_UTC).unwrap().is_none());
192        // The key is reusable after expiry.
193        assert!(repo.put(&record("k1", Utc::now())).unwrap());
194    }
195
196    #[test]
197    fn purge_expired_sweeps_only_old_rows() {
198        let repo = repo();
199        let now = Utc::now();
200        assert!(repo.put(&record("old-1", now - Duration::hours(30))).unwrap());
201        assert!(repo.put(&record("old-2", now - Duration::hours(25))).unwrap());
202        assert!(repo.put(&record("fresh", now)).unwrap());
203
204        let purged = repo.purge_expired(now - Duration::hours(24)).unwrap();
205        assert_eq!(purged, 2);
206        assert!(repo.get("tenant-a", "fresh", now - Duration::hours(24)).unwrap().is_some());
207    }
208    #[test]
209    fn entry_created_exactly_at_cutoff_is_expired() {
210        // Boundary contract with the in-memory cache (`age >= ttl` = expired):
211        // an entry whose created_at equals the cutoff must NOT be returned.
212        // With a zero TTL (cutoff == now), a same-millisecond write used to
213        // survive the strict `<` cleanup and replay a response that the memory
214        // path had already declared expired.
215        let repo = repo();
216        let created = Utc::now();
217        assert!(repo.put(&record("boundary", created)).unwrap());
218
219        let hit = repo.get("tenant-a", "boundary", created).unwrap();
220        assert!(hit.is_none(), "created_at == cutoff must be expired: {hit:?}");
221    }
222}
stateset_db/sqlite/http_idempotency.rs

stateset_db/sqlite/
http_idempotency.rs
