cratefield_core/
cooldown.rs

1//! Durable per-subject send cooldown (issue #133): the rate limiter is a
2//! distributed counter whose transport can fail open, so a mail-triggering
3//! route needs a **backstop the database enforces**: at most one send per
4//! subject per window, race-free without a transaction.
5//!
6//! The claim is a guarded `UPDATE` (re-claim only once the window has
7//! passed) followed by an `INSERT ... ON CONFLICT DO NOTHING` (first send
8//! for a never-seen subject) — exactly one of the two statements reports
9//! one affected row per window, on both SQLite/D1 and Postgres. Same
10//! primitive the [`Inbox`] dedup ledger uses, with a time window added.
11//!
12//! ```ignore
13//! let cooldown = SendCooldown::new("waitlist_send_cooldown");
14//! if cooldown.try_acquire(&*db, &subject, &now, &cutoff).await? {
15//!     // this request owns the window: send the mail
16//! } // else: a concurrent or repeat request inside the window: skip the send
17//! ```
18//!
19//! [`Inbox`]: crate::idempotency::Inbox
20
21use crate::ports::{Database, DbError, Statement};
22use sea_query::{Alias, Expr, OnConflict, Query};
23
24fn iden(name: &str) -> Alias {
25    Alias::new(name)
26}
27
28/// A cooldown ledger over the `Database` port. Construct it with the table
29/// the owning module declares (e.g. `"<module>_send_cooldown"`); the
30/// owning module ships [`create_table_sql`](Self::create_table_sql) as a
31/// migration.
32///
33/// Timestamps are RFC 3339 strings from the `Clock` port, the same shape
34/// the house stores in `TEXT` columns everywhere: lexicographic order on
35/// that format is chronological order, which is what the window compare
36/// relies on.
37#[derive(Debug, Clone)]
38pub struct SendCooldown {
39    table: String,
40}
41
42impl SendCooldown {
43    #[must_use]
44    pub fn new(table: impl Into<String>) -> Self {
45        Self {
46            table: table.into(),
47        }
48    }
49
50    /// The portable DDL for the cooldown table (renders identically on
51    /// SQLite/D1 and Postgres, ADR 0004).
52    #[must_use]
53    pub fn create_table_sql(&self) -> String {
54        format!(
55            "CREATE TABLE IF NOT EXISTS {table} (\n    \
56             subject TEXT PRIMARY KEY,\n    \
57             last_sent_at TEXT NOT NULL\n);",
58            table = self.table
59        )
60    }
61
62    /// Claims the send window for `subject`: `true` for the **one** caller
63    /// whose claim lands, `false` for anyone inside the window.
64    ///
65    /// `now` is the RFC 3339 timestamp to record, `cutoff` the instant
66    /// (`now - window`) after which an expired claim may be renewed. The
67    /// update-then-insert pair is race-free: the guarded `UPDATE` can be
68    /// won by exactly one writer, and the conflict-insert covers the row
69    /// not existing (yet).
70    ///
71    /// # Errors
72    ///
73    /// Returns [`DbError`] if either write fails.
74    pub async fn try_acquire(
75        &self,
76        db: &dyn Database,
77        subject: &str,
78        now: &str,
79        cutoff: &str,
80    ) -> Result<bool, DbError> {
81        let mut renew = Query::update();
82        renew
83            .table(iden(&self.table))
84            .value(iden("last_sent_at"), now)
85            .and_where(Expr::col(iden("subject")).eq(subject))
86            .and_where(Expr::col(iden("last_sent_at")).lt(cutoff));
87        if db.execute(&Statement::render(&renew)).await? == 1 {
88            return Ok(true);
89        }
90        let mut first = Query::insert();
91        first
92            .into_table(iden(&self.table))
93            .columns(["subject", "last_sent_at"])
94            .values_panic([subject.to_owned().into(), now.to_owned().into()])
95            .on_conflict(OnConflict::column(iden("subject")).do_nothing().to_owned());
96        Ok(db.execute(&Statement::render(&first)).await? == 1)
97    }
98
99    /// Releases a claim so the window is not consumed by a send that never
100    /// started (e.g. the mailer port answered `NotConfigured`).
101    ///
102    /// # Errors
103    ///
104    /// Returns [`DbError`] if the delete fails.
105    pub async fn release(&self, db: &dyn Database, subject: &str) -> Result<(), DbError> {
106        let mut delete = Query::delete();
107        delete
108            .from_table(iden(&self.table))
109            .and_where(Expr::col(iden("subject")).eq(subject));
110        db.execute(&Statement::render(&delete)).await?;
111        Ok(())
112    }
113
114    /// Deletes claims older than `before` (rows whose window has closed
115    /// anyway). Returns the number removed; a module calls this from its
116    /// scheduled handler.
117    ///
118    /// # Errors
119    ///
120    /// Returns [`DbError`] if the delete fails.
121    pub async fn prune(&self, db: &dyn Database, before: &str) -> Result<u64, DbError> {
122        let mut delete = Query::delete();
123        delete
124            .from_table(iden(&self.table))
125            .and_where(Expr::col(iden("last_sent_at")).lt(before));
126        db.execute(&Statement::render(&delete)).await
127    }
128}
129
130#[cfg(test)]
131mod tests {
132    use super::*;
133
134    #[test]
135    fn create_table_sql_is_portable_ddl() {
136        let sql = SendCooldown::new("waitlist_send_cooldown").create_table_sql();
137        assert!(sql.contains("CREATE TABLE IF NOT EXISTS waitlist_send_cooldown"));
138        assert!(sql.contains("subject TEXT PRIMARY KEY"));
139        assert!(sql.contains("last_sent_at TEXT NOT NULL"));
140    }
141}
cratefield_core/cooldown.rs

cratefield_core/
cooldown.rs
