cratefield_core/
idempotency.rs

1//! Idempotent event handling (issue #134): an **inbox** that records each
2//! external event's id once, so a duplicate, replayed, concurrent or
3//! out-of-order delivery applies its effects exactly once.
4//!
5//! Verifying a Stripe webhook's signature and timestamp authenticates a
6//! *delivery* — it does not stop Stripe from delivering the same event twice
7//! (it retries), or two workers from processing the same event at once. A
8//! module wraps its handler in a [`claim`](Inbox::claim): the primary key plus
9//! `ON CONFLICT DO NOTHING` lets exactly one caller insert the row, so exactly
10//! one caller sees `true` and does the work.
11//!
12//! ```ignore
13//! let inbox = Inbox::new("billing_inbox");
14//! let event = payments.verify_webhook(sig, body).await?; // verified first
15//! if inbox.claim(db, &event.id, &now).await? {
16//!     // first time: apply the effect (grant the entitlement, ...)
17//! } // else: a duplicate delivery — already handled, do nothing
18//! ```
19//!
20//! The owning module includes [`Inbox::create_table_sql`] as a migration (it
21//! owns the table, per the module contract). Prune old rows on whatever horizon
22//! the provider retries within; the inbox is a dedup ledger, not a record.
23
24use crate::ports::{Database, DbError, Statement};
25use sea_query::{Alias, Expr, OnConflict, Query};
26
27fn iden(name: &str) -> Alias {
28    Alias::new(name)
29}
30
31/// A dedup ledger over the `Database` port. Construct it with the table the
32/// owning module declares (e.g. `"<module>_inbox"`).
33#[derive(Debug, Clone)]
34pub struct Inbox {
35    table: String,
36}
37
38impl Inbox {
39    #[must_use]
40    pub fn new(table: impl Into<String>) -> Self {
41        Self {
42            table: table.into(),
43        }
44    }
45
46    /// The portable DDL for the inbox table. The owning module ships this as a
47    /// forward-only migration (it renders identically on SQLite/D1 and
48    /// Postgres, ADR 0004).
49    #[must_use]
50    pub fn create_table_sql(&self) -> String {
51        format!(
52            "CREATE TABLE IF NOT EXISTS {table} (\n    \
53             event_key TEXT PRIMARY KEY,\n    \
54             seen_at TEXT NOT NULL\n);",
55            table = self.table
56        )
57    }
58
59    /// Records `event_key`, returning `true` the **first** time it is seen
60    /// (proceed and apply the effect) and `false` for any later delivery of the
61    /// same id (a duplicate/replay/concurrent race — skip).
62    ///
63    /// The insert is a single `ON CONFLICT DO NOTHING` statement, so it is
64    /// atomic on both D1 and Postgres: under concurrent claims of the same key
65    /// exactly one insert takes, and only that caller gets `true`. `seen_at` is
66    /// an RFC 3339 timestamp the caller reads from the `Clock` port.
67    ///
68    /// # Errors
69    ///
70    /// Returns [`DbError`] if the write fails.
71    pub async fn claim(
72        &self,
73        db: &dyn Database,
74        event_key: &str,
75        seen_at: &str,
76    ) -> Result<bool, DbError> {
77        let mut insert = Query::insert();
78        insert
79            .into_table(iden(&self.table))
80            .columns(["event_key", "seen_at"])
81            .values_panic([event_key.to_owned().into(), seen_at.to_owned().into()])
82            .on_conflict(
83                OnConflict::column(iden("event_key"))
84                    .do_nothing()
85                    .to_owned(),
86            );
87        let affected = db.execute(&Statement::render(&insert)).await?;
88        Ok(affected == 1)
89    }
90
91    /// Whether `event_key` has already been recorded (a read-only check; a
92    /// handler still uses [`claim`](Self::claim) to act, since only `claim` is
93    /// race-free).
94    ///
95    /// # Errors
96    ///
97    /// Returns [`DbError`] if the read fails.
98    pub async fn seen(&self, db: &dyn Database, event_key: &str) -> Result<bool, DbError> {
99        let mut select = Query::select();
100        select
101            .expr(Expr::val(1))
102            .from(iden(&self.table))
103            .and_where(Expr::col(iden("event_key")).eq(event_key))
104            .limit(1);
105        let rows = db.query(&Statement::render(&select)).await?;
106        Ok(!rows.is_empty())
107    }
108}
109
110#[cfg(test)]
111mod tests {
112    use super::*;
113
114    #[test]
115    fn create_table_sql_is_portable_ddl() {
116        let sql = Inbox::new("billing_inbox").create_table_sql();
117        assert!(sql.contains("CREATE TABLE IF NOT EXISTS billing_inbox"));
118        assert!(sql.contains("event_key TEXT PRIMARY KEY"));
119        assert!(sql.contains("seen_at TEXT NOT NULL"));
120    }
121
122    #[test]
123    fn claim_renders_insert_on_conflict_do_nothing() {
124        // Render the statement the claim issues, without a database, and assert
125        // its shape: an insert into the table with the two columns and an
126        // ON CONFLICT DO NOTHING clause (the atomic first-writer-wins).
127        let inbox = Inbox::new("billing_inbox");
128        let mut insert = Query::insert();
129        insert
130            .into_table(iden(&inbox.table))
131            .columns(["event_key", "seen_at"])
132            .values_panic([
133                "evt_1".to_owned().into(),
134                "2026-09-07T00:00:00Z".to_owned().into(),
135            ])
136            .on_conflict(
137                OnConflict::column(iden("event_key"))
138                    .do_nothing()
139                    .to_owned(),
140            );
141        let stmt = Statement::render(&insert);
142        assert!(stmt.sql.contains("INSERT INTO"));
143        assert!(stmt.sql.contains("billing_inbox"));
144        assert!(stmt.sql.to_uppercase().contains("ON CONFLICT"));
145        assert!(stmt.sql.to_uppercase().contains("DO NOTHING"));
146        assert_eq!(stmt.values.0.len(), 2);
147    }
148}
cratefield_core/idempotency.rs

cratefield_core/
idempotency.rs
