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: the primary key lets exactly one
9//! caller record the event, so exactly one caller sees `true` and does the
10//! work. When the effect is database writes, claim with
11//! [`claim_with`](Inbox::claim_with) — or put
12//! [`claim_statement`](Inbox::claim_statement) first in your own
13//! `db.batch_atomic(..)`, the [`Outbox`](crate::Outbox) rule mirrored: the
14//! claim and the effect commit together or not at all, so a worker that dies
15//! mid-batch leaves the key unclaimed and the redelivery re-runs both.
16//!
17//! ```ignore
18//! let inbox = Inbox::new("billing_inbox");
19//! let event = payments.verify_webhook(sig, body).await?; // verified first
20//! if inbox.claim_with(db, &event.id, &now, &grant_stmts).await? {
21//! // first time: the claim AND the effect just committed together
22//! } // else: a duplicate delivery — already handled, do nothing
23//! ```
24//!
25//! The owning module includes [`Inbox::create_table_sql`] as a migration (it
26//! owns the table, per the module contract). Prune old rows on whatever horizon
27//! the provider retries within; the inbox is a dedup ledger, not a record.
28
29use crate::ports::{Database, DbError, Statement};
30use sea_query::{Alias, Expr, InsertStatement, OnConflict, Query};
31
32fn iden(name: &str) -> Alias {
33 Alias::new(name)
34}
35
36/// A dedup ledger over the `Database` port. Construct it with the table the
37/// owning module declares (e.g. `"<module>_inbox"`).
38#[derive(Debug, Clone)]
39pub struct Inbox {
40 table: String,
41}
42
43impl Inbox {
44 #[must_use]
45 pub fn new(table: impl Into<String>) -> Self {
46 Self {
47 table: table.into(),
48 }
49 }
50
51 /// The portable DDL for the inbox table. The owning module ships this as a
52 /// forward-only migration (it renders identically on SQLite/D1 and
53 /// Postgres, ADR 0004).
54 #[must_use]
55 pub fn create_table_sql(&self) -> String {
56 format!(
57 "CREATE TABLE IF NOT EXISTS {table} (\n \
58 event_key TEXT PRIMARY KEY,\n \
59 seen_at TEXT NOT NULL\n);",
60 table = self.table
61 )
62 }
63
64 /// Records `event_key`, returning `true` the **first** time it is seen
65 /// (proceed and apply the effect) and `false` for any later delivery of the
66 /// same id (a duplicate/replay/concurrent race — skip).
67 ///
68 /// The insert is a single `ON CONFLICT DO NOTHING` statement, so it is
69 /// atomic on both D1 and Postgres: under concurrent claims of the same key
70 /// exactly one insert takes, and only that caller gets `true`. `seen_at` is
71 /// an RFC 3339 timestamp the caller reads from the `Clock` port.
72 ///
73 /// The claim commits **on its own**: if the worker dies after this returns
74 /// `true` but before the effect is applied, the redelivery sees the key and
75 /// skips — the event is lost (at-most-once). When the effect is itself
76 /// database writes, use [`claim_with`](Self::claim_with), or put
77 /// [`claim_statement`](Self::claim_statement) first in the same
78 /// `db.batch_atomic(..)` as the effect, so the two commit together.
79 ///
80 /// # Errors
81 ///
82 /// Returns [`DbError`] if the write fails.
83 pub async fn claim(
84 &self,
85 db: &dyn Database,
86 event_key: &str,
87 seen_at: &str,
88 ) -> Result<bool, DbError> {
89 let mut insert = self.claim_insert(event_key, seen_at);
90 insert.on_conflict(
91 OnConflict::column(iden("event_key"))
92 .do_nothing()
93 .to_owned(),
94 );
95 let affected = db.execute(&Statement::render(&insert)).await?;
96 Ok(affected == 1)
97 }
98
99 /// The bare claim insert both public paths render — [`claim`](Self::claim)
100 /// extends it with `ON CONFLICT DO NOTHING`,
101 /// [`claim_statement`](Self::claim_statement) renders it as-is. Shared so
102 /// the two cannot drift.
103 fn claim_insert(&self, event_key: &str, seen_at: &str) -> InsertStatement {
104 let mut insert = Query::insert();
105 insert
106 .into_table(iden(&self.table))
107 .columns(["event_key", "seen_at"])
108 .values_panic([event_key.to_owned().into(), seen_at.to_owned().into()]);
109 insert
110 }
111
112 /// The claim as a bare `INSERT` with **no** `ON CONFLICT` clause, to run
113 /// as the first statement of the same `db.batch_atomic(..)` that carries
114 /// the effect's statements (see [`claim_with`](Self::claim_with) for the
115 /// ready-made wrapper).
116 ///
117 /// Inside the batch the primary key does the first-writer-wins work: a
118 /// key that is already claimed fails the insert, and the whole batch —
119 /// effect included — rolls back, so a duplicate delivery applies nothing
120 /// a second time. The claim and the effect commit together or not at all:
121 /// a worker that dies before the batch commits leaves the key unclaimed,
122 /// and the redelivery re-runs both.
123 #[must_use]
124 pub fn claim_statement(&self, event_key: &str, seen_at: &str) -> Statement {
125 Statement::render(&self.claim_insert(event_key, seen_at))
126 }
127
128 /// Claims `event_key` and applies `effect` in one atomic batch, returning
129 /// `true` when **this** delivery committed them (the convention of
130 /// [`claim`](Self::claim)) and `false` when the key was already claimed —
131 /// a duplicate, a replay, or a concurrent delivery that won the race — so
132 /// the effect did not run again.
133 ///
134 /// The [`claim_statement`](Self::claim_statement) goes first, then
135 /// `effect`, all through [`batch_atomic`](Database::batch_atomic): either
136 /// everything commits or nothing does, and a worker that dies mid-batch
137 /// leaves the key unclaimed for the retry.
138 ///
139 /// # Errors
140 ///
141 /// Returns [`DbError`] when the batch fails and the key is *not* claimed
142 /// afterwards: nothing committed and the effect did not apply — the caller
143 /// (or its queue) retries the whole delivery. A batch that failed because
144 /// a concurrent delivery claimed the key first is not an error; it returns
145 /// `Ok(false)`, told apart by re-reading the key rather than parsing the
146 /// engine's error text.
147 pub async fn claim_with(
148 &self,
149 db: &dyn Database,
150 event_key: &str,
151 seen_at: &str,
152 effect: &[Statement],
153 ) -> Result<bool, DbError> {
154 if self.seen(db, event_key).await? {
155 return Ok(false);
156 }
157 let mut batch = Vec::with_capacity(effect.len() + 1);
158 batch.push(self.claim_statement(event_key, seen_at));
159 batch.extend_from_slice(effect);
160 match db.batch_atomic(&batch).await {
161 Ok(()) => Ok(true),
162 Err(batch_error) => {
163 // The failed batch committed nothing, but a concurrent
164 // delivery may have claimed the key meanwhile — the re-check
165 // is what tells the two apart (no port guarantees the shape
166 // of the engine's error text). If the re-check itself fails,
167 // surface the original batch error.
168 match self.seen(db, event_key).await {
169 Ok(true) => Ok(false),
170 _ => Err(batch_error),
171 }
172 }
173 }
174 }
175
176 /// Whether `event_key` has already been recorded (a read-only check; a
177 /// handler still uses [`claim`](Self::claim) or
178 /// [`claim_with`](Self::claim_with) to act, since only those are
179 /// race-free).
180 ///
181 /// # Errors
182 ///
183 /// Returns [`DbError`] if the read fails.
184 pub async fn seen(&self, db: &dyn Database, event_key: &str) -> Result<bool, DbError> {
185 let mut select = Query::select();
186 select
187 .expr(Expr::val(1))
188 .from(iden(&self.table))
189 .and_where(Expr::col(iden("event_key")).eq(event_key))
190 .limit(1);
191 let rows = db.query(&Statement::render(&select)).await?;
192 Ok(!rows.is_empty())
193 }
194}
195
196#[cfg(test)]
197mod tests {
198 use super::*;
199
200 #[test]
201 fn create_table_sql_is_portable_ddl() {
202 let sql = Inbox::new("billing_inbox").create_table_sql();
203 assert!(sql.contains("CREATE TABLE IF NOT EXISTS billing_inbox"));
204 assert!(sql.contains("event_key TEXT PRIMARY KEY"));
205 assert!(sql.contains("seen_at TEXT NOT NULL"));
206 }
207
208 #[test]
209 fn claim_renders_insert_on_conflict_do_nothing() {
210 // Render the statement the claim issues, without a database, and assert
211 // its shape: an insert into the table with the two columns and an
212 // ON CONFLICT DO NOTHING clause (the atomic first-writer-wins).
213 let inbox = Inbox::new("billing_inbox");
214 let mut insert = Query::insert();
215 insert
216 .into_table(iden(&inbox.table))
217 .columns(["event_key", "seen_at"])
218 .values_panic([
219 "evt_1".to_owned().into(),
220 "2026-09-07T00:00:00Z".to_owned().into(),
221 ])
222 .on_conflict(
223 OnConflict::column(iden("event_key"))
224 .do_nothing()
225 .to_owned(),
226 );
227 let stmt = Statement::render(&insert);
228 assert!(stmt.sql.contains("INSERT INTO"));
229 assert!(stmt.sql.contains("billing_inbox"));
230 assert!(stmt.sql.to_uppercase().contains("ON CONFLICT"));
231 assert!(stmt.sql.to_uppercase().contains("DO NOTHING"));
232 assert_eq!(stmt.values.0.len(), 2);
233 }
234
235 #[test]
236 fn claim_statement_renders_bare_insert_without_on_conflict() {
237 // The in-batch claim must NOT carry ON CONFLICT DO NOTHING: inside
238 // batch_atomic a duplicate key has to fail the insert so the whole
239 // batch (claim + effect) rolls back together.
240 let inbox = Inbox::new("billing_inbox");
241 let stmt = inbox.claim_statement("evt_1", "2026-09-07T00:00:00Z");
242 assert!(stmt.sql.contains("INSERT INTO"));
243 assert!(stmt.sql.contains("billing_inbox"));
244 assert!(stmt.sql.contains("event_key"));
245 assert!(stmt.sql.contains("seen_at"));
246 assert!(
247 !stmt.sql.to_uppercase().contains("ON CONFLICT"),
248 "a duplicate key must fail the batch, not be swallowed: {}",
249 stmt.sql
250 );
251 assert_eq!(stmt.values.0.len(), 2);
252 }
253}