cratefield_module_webhooks/
store.rs

1//! Every statement this module runs, in one place (on the pattern of
2//! `module-notifications`' `store.rs`). Rows are read back through the
3//! `Database` port's small owned row model, so nothing engine-specific
4//! leaks past this file. A nullable `INTEGER` reads as `Option<i64>`
5//! directly: a SQL `NULL` and a missing column both convert to `None`.
6
7use cratefield_core::{Database, DbError, IdGen, Outbox, Row, Statement, UlidIdGen};
8use sea_query::{Alias, Expr, Order, Query};
9use serde::{Deserialize, Serialize};
10
11/// The delivery targets. Must match `0001_init.sql` and
12/// [`crate::Webhooks::tables`].
13pub(crate) const ENDPOINTS: &str = "webhooks_endpoints";
14/// The core [`Outbox`] table this module owns: one row per
15/// (event, endpoint), so each endpoint retries independently.
16pub(crate) const OUTBOX: &str = "webhooks_outbox";
17/// One row per attempt actually made.
18pub(crate) const DELIVERIES: &str = "webhooks_deliveries";
19/// The terminal state core's outbox does not have (ADR 0016).
20pub(crate) const DEAD_LETTERS: &str = "webhooks_dead_letters";
21
22/// The outbox topic a delivery row carries.
23pub const TOPIC_DELIVER: &str = "webhooks.deliver";
24
25fn iden(name: &str) -> Alias {
26    Alias::new(name)
27}
28
29/// One row of `webhooks_endpoints`, without the secret: a list never hands
30/// signing material back. The secret is returned exactly once, by
31/// [`crate::Webhooks::create_endpoint`].
32// `pub` rather than `pub(crate)`: these are the module's read-back types,
33// re-exported at the crate root; the module itself stays private.
34#[derive(Debug, Clone, PartialEq, Eq)]
35pub struct Endpoint {
36    pub id: String,
37    pub subject: String,
38    pub url: String,
39    pub event_types: String,
40    pub created_at: String,
41}
42
43/// One row of `webhooks_deliveries`.
44#[derive(Debug, Clone, PartialEq, Eq)]
45pub struct Delivery {
46    pub id: String,
47    pub endpoint_id: String,
48    pub subject: String,
49    pub event_id: String,
50    pub event_type: String,
51    pub attempt: i64,
52    pub status_code: Option<i64>,
53    pub error: Option<String>,
54    pub at: String,
55}
56
57/// What a delivery gave up on. `attempts_exhausted` is the ordinary way a
58/// dead letter happens; `rejected` is a failure no retry fixes (an HTTP
59/// 410 Gone, or the `HttpClient` port refusing the destination outright);
60/// `malformed` is an outbox row this module cannot read back — a bug or a
61/// hand-edited database, not weather.
62#[derive(Debug, Clone, Copy, PartialEq, Eq)]
63pub(crate) enum DeadLetterReason {
64    AttemptsExhausted,
65    Rejected,
66    Malformed,
67}
68
69impl DeadLetterReason {
70    /// The value stored in `webhooks_dead_letters.reason`.
71    pub(crate) fn as_str(self) -> &'static str {
72        match self {
73            DeadLetterReason::AttemptsExhausted => "attempts_exhausted",
74            DeadLetterReason::Rejected => "rejected",
75            DeadLetterReason::Malformed => "malformed",
76        }
77    }
78}
79
80/// One row of `webhooks_dead_letters`, as [`crate::Webhooks::dead_letters`]
81/// reads it back and [`crate::Webhooks::replay`] acts on it.
82#[derive(Debug, Clone, PartialEq, Eq)]
83pub struct DeadLetter {
84    pub id: String,
85    pub endpoint_id: String,
86    pub subject: String,
87    pub event_id: String,
88    pub event_type: String,
89    pub data: String,
90    pub attempts: i64,
91    pub reason: String,
92    pub last_error: String,
93    pub status_code: Option<i64>,
94    /// When the event was published — what a replay preserves, because
95    /// the envelope names the event's own time, not a drain's.
96    pub created_at: String,
97    pub failed_at: String,
98}
99
100/// The job one outbox row carries: the event and the endpoint it must
101/// reach. The endpoint's URL and secret are **not** in it — like a push
102/// recipient (ADR 0015), a signing secret is credential material, and an
103/// outbox payload is ordinary data that ends up in exports and
104/// diagnostics; the drain reads the endpoint row instead.
105#[derive(Debug, Clone, Serialize, Deserialize)]
106pub(crate) struct DeliveryJob {
107    pub endpoint_id: String,
108    pub event_id: String,
109    pub event_type: String,
110    pub subject: String,
111    /// When the event was published — the outbox table does not carry a
112    /// `created_at` column, and the envelope and the dead letter both
113    /// name the event's own time, not a drain's.
114    pub created_at: String,
115    /// The caller's event payload, exactly as published.
116    pub data: serde_json::Value,
117}
118
119/// The columns every endpoint read shares.
120const ENDPOINT_COLUMNS: [&str; 5] = ["id", "subject", "url", "event_types", "created_at"];
121
122/// The columns every delivery-log write and read shares (the insert and
123/// the list select must agree).
124const DELIVERY_COLUMNS: [&str; 9] = [
125    "id",
126    "endpoint_id",
127    "subject",
128    "event_id",
129    "event_type",
130    "attempt",
131    "status_code",
132    "error",
133    "at",
134];
135
136/// The columns every dead-letter write and read shares (the insert and
137/// both selects must agree).
138const DEAD_LETTER_COLUMNS: [&str; 12] = [
139    "id",
140    "endpoint_id",
141    "subject",
142    "event_id",
143    "event_type",
144    "data",
145    "attempts",
146    "reason",
147    "last_error",
148    "status_code",
149    "created_at",
150    "failed_at",
151];
152
153/// What the drain needs to reach one endpoint: its URL and the secret it
154/// signs with. The one read that may carry the secret — the listable
155/// [`Endpoint`] never does — so credential material cannot end up in a
156/// list response by construction.
157pub(crate) struct DeliveryTarget {
158    pub url: String,
159    pub secret: String,
160}
161
162/// The one endpoint of `subject` with `endpoint_id`, resolved to
163/// [`DeliveryTarget`] in a single read, or `None`.
164pub(crate) async fn delivery_target(
165    db: &dyn Database,
166    subject: &str,
167    endpoint_id: &str,
168) -> Result<Option<DeliveryTarget>, DbError> {
169    let mut select = Query::select();
170    select
171        .columns(["url", "secret"])
172        .from(iden(ENDPOINTS))
173        .and_where(Expr::col(iden("subject")).eq(subject))
174        .and_where(Expr::col(iden("id")).eq(endpoint_id));
175    Ok(db
176        .query(&Statement::render(&select))
177        .await?
178        .first()
179        .map(|row| DeliveryTarget {
180            url: row.get::<String>("url").unwrap_or_default(),
181            secret: row.get::<String>("secret").unwrap_or_default(),
182        }))
183}
184
185/// Whether the endpoint row still exists — the guard
186/// [`crate::Webhooks::replay`] needs: a dead letter whose endpoint is
187/// gone has nowhere to be re-delivered to, so it must stay put.
188pub(crate) async fn endpoint_exists(
189    db: &dyn Database,
190    subject: &str,
191    endpoint_id: &str,
192) -> Result<bool, DbError> {
193    let mut select = Query::select();
194    select
195        .column(iden("id"))
196        .from(iden(ENDPOINTS))
197        .and_where(Expr::col(iden("subject")).eq(subject))
198        .and_where(Expr::col(iden("id")).eq(endpoint_id));
199    Ok(!db.query(&Statement::render(&select)).await?.rows.is_empty())
200}
201
202/// Every endpoint of one subject, creation order. Disabled and deleted
203/// endpoints are simply not here.
204pub(crate) async fn endpoints_for_subject(
205    db: &dyn Database,
206    subject: &str,
207) -> Result<Vec<Endpoint>, DbError> {
208    let mut select = Query::select();
209    select
210        .columns(ENDPOINT_COLUMNS)
211        .from(iden(ENDPOINTS))
212        .and_where(Expr::col(iden("subject")).eq(subject))
213        // `id` breaks ties: `created_at` has second resolution, and two
214        // endpoints created in the same second must list deterministically.
215        .order_by(iden("created_at"), Order::Asc)
216        .order_by(iden("id"), Order::Asc);
217    let rows = db.query(&Statement::render(&select)).await?;
218    Ok(rows.rows.iter().map(endpoint_from).collect())
219}
220
221fn endpoint_from(row: &Row) -> Endpoint {
222    Endpoint {
223        id: row.get::<String>("id").unwrap_or_default(),
224        subject: row.get::<String>("subject").unwrap_or_default(),
225        url: row.get::<String>("url").unwrap_or_default(),
226        event_types: row.get::<String>("event_types").unwrap_or_default(),
227        created_at: row.get::<String>("created_at").unwrap_or_default(),
228    }
229}
230
231/// Inserts one endpoint. `secret` is bound here and nowhere else; the
232/// caller has already generated it.
233pub(crate) fn insert_endpoint_statement(
234    id: &str,
235    subject: &str,
236    url: &str,
237    secret: &str,
238    event_types: &str,
239    now: &str,
240) -> Statement {
241    let mut insert = Query::insert();
242    insert
243        .into_table(iden(ENDPOINTS))
244        .columns([
245            "id",
246            "subject",
247            "url",
248            "secret",
249            "event_types",
250            "created_at",
251        ])
252        .values_panic([
253            id.to_owned().into(),
254            subject.to_owned().into(),
255            url.to_owned().into(),
256            secret.to_owned().into(),
257            event_types.to_owned().into(),
258            now.to_owned().into(),
259        ]);
260    Statement::render(&insert)
261}
262
263/// Deletes one endpoint of one subject; `true` when a row went. Queued
264/// deliveries for it are left to the drain, which drops them when the
265/// endpoint no longer resolves.
266pub(crate) async fn delete_endpoint(
267    db: &dyn Database,
268    subject: &str,
269    endpoint_id: &str,
270) -> Result<bool, DbError> {
271    let mut delete = Query::delete();
272    delete
273        .from_table(iden(ENDPOINTS))
274        .and_where(Expr::col(iden("subject")).eq(subject))
275        .and_where(Expr::col(iden("id")).eq(endpoint_id));
276    Ok(db.execute(&Statement::render(&delete)).await? == 1)
277}
278
279/// The enqueue for one fan-out target, returned into the caller's own
280/// `db.batch_atomic(..)` — the core [`Outbox`] contract, so the event is
281/// durable exactly when the caller's state change is. Nothing is written
282/// until that batch runs.
283pub(crate) fn enqueue_statement(job: &DeliveryJob, subject: &str, at: &str) -> Statement {
284    let payload = serde_json::to_string(job).unwrap_or_else(|_| "{}".to_owned());
285    Outbox::new(OUTBOX).enqueue_statement(
286        &UlidIdGen.ulid(),
287        TOPIC_DELIVER,
288        &payload,
289        Some(subject),
290        at,
291    )
292}
293
294/// Records one attempt in the delivery log.
295///
296/// Written as its own statement after the attempt, not in the batch that
297/// completes or retries the outbox row: the log is diagnostic, and a crash
298/// between the two costs one double-logged attempt — what at-least-once
299/// delivery looks like in a log.
300pub(crate) async fn record_delivery(
301    db: &dyn Database,
302    delivery_id: &str,
303    job: &DeliveryJob,
304    attempt: i64,
305    status_code: Option<i64>,
306    error: Option<&str>,
307    at: &str,
308) -> Result<(), DbError> {
309    let mut insert = Query::insert();
310    insert
311        .into_table(iden(DELIVERIES))
312        .columns(DELIVERY_COLUMNS)
313        .values_panic([
314            delivery_id.to_owned().into(),
315            job.endpoint_id.clone().into(),
316            job.subject.clone().into(),
317            job.event_id.clone().into(),
318            job.event_type.clone().into(),
319            attempt.into(),
320            status_code.into(),
321            error.map(str::to_owned).into(),
322            at.to_owned().into(),
323        ]);
324    db.execute(&Statement::render(&insert)).await?;
325    Ok(())
326}
327
328/// A subject's delivery log, newest attempt first.
329pub(crate) async fn deliveries(
330    db: &dyn Database,
331    subject: &str,
332    limit: u64,
333) -> Result<Vec<Delivery>, DbError> {
334    let mut select = Query::select();
335    select
336        .columns(DELIVERY_COLUMNS)
337        .from(iden(DELIVERIES))
338        .and_where(Expr::col(iden("subject")).eq(subject))
339        // `id` breaks ties: attempts in the same second keep a stable,
340        // ULID-ordered story instead of an engine-dependent shuffle.
341        .order_by(iden("at"), Order::Desc)
342        .order_by(iden("id"), Order::Desc)
343        .limit(limit);
344    let rows = db.query(&Statement::render(&select)).await?;
345    Ok(rows.rows.iter().map(delivery_from).collect())
346}
347
348fn delivery_from(row: &Row) -> Delivery {
349    Delivery {
350        id: row.get::<String>("id").unwrap_or_default(),
351        endpoint_id: row.get::<String>("endpoint_id").unwrap_or_default(),
352        subject: row.get::<String>("subject").unwrap_or_default(),
353        event_id: row.get::<String>("event_id").unwrap_or_default(),
354        event_type: row.get::<String>("event_type").unwrap_or_default(),
355        attempt: row.get::<i64>("attempt").unwrap_or(0),
356        status_code: row.get::<i64>("status_code"),
357        error: row.get::<Option<String>>("error").flatten(),
358        at: row.get::<String>("at").unwrap_or_default(),
359    }
360}
361
362/// What a delivery gave up on, bound in one place. `last_error` is prose
363/// from a receiver or a transport — never the endpoint's secret, which is
364/// in no error path at all.
365pub(crate) struct GiveUp<'a> {
366    pub outbox_row_id: &'a str,
367    /// Attempts **made**, including the one that just failed.
368    pub attempts: i64,
369    pub reason: DeadLetterReason,
370    pub last_error: &'a str,
371    pub status_code: Option<i64>,
372    /// When the event was enqueued, not when it was abandoned.
373    pub created_at: &'a str,
374    pub failed_at: &'a str,
375}
376
377/// Moves one failed delivery into the dead-letter table and drops its
378/// outbox row, in one batch: a delivery that will never succeed leaves the
379/// work queue and stays replayable.
380pub(crate) async fn dead_letter(
381    db: &dyn Database,
382    dead_letter_id: &str,
383    job: &DeliveryJob,
384    gave_up: &GiveUp<'_>,
385) -> Result<(), DbError> {
386    let mut insert = Query::insert();
387    insert
388        .into_table(iden(DEAD_LETTERS))
389        .columns(DEAD_LETTER_COLUMNS)
390        .values_panic([
391            dead_letter_id.to_owned().into(),
392            job.endpoint_id.clone().into(),
393            job.subject.clone().into(),
394            job.event_id.clone().into(),
395            job.event_type.clone().into(),
396            job.data.to_string().into(),
397            gave_up.attempts.into(),
398            gave_up.reason.as_str().into(),
399            gave_up.last_error.to_owned().into(),
400            gave_up.status_code.into(),
401            gave_up.created_at.to_owned().into(),
402            gave_up.failed_at.to_owned().into(),
403        ]);
404    db.batch_atomic(&[
405        Statement::render(&insert),
406        delete_outbox_statement(gave_up.outbox_row_id),
407    ])
408    .await?;
409    Ok(())
410}
411
412pub(crate) fn delete_outbox_statement(id: &str) -> Statement {
413    let mut delete = Query::delete();
414    delete
415        .from_table(iden(OUTBOX))
416        .and_where(Expr::col(iden("id")).eq(id));
417    Statement::render(&delete)
418}
419
420/// A subject's dead letters, oldest first — the list
421/// [`crate::Webhooks::replay`] takes an id from.
422pub(crate) async fn dead_letters(
423    db: &dyn Database,
424    subject: &str,
425    limit: u64,
426) -> Result<Vec<DeadLetter>, DbError> {
427    let mut select = Query::select();
428    select
429        .columns(DEAD_LETTER_COLUMNS)
430        .from(iden(DEAD_LETTERS))
431        .and_where(Expr::col(iden("subject")).eq(subject))
432        .order_by(iden("failed_at"), Order::Asc)
433        .limit(limit);
434    let rows = db.query(&Statement::render(&select)).await?;
435    Ok(rows.rows.iter().map(dead_letter_from).collect())
436}
437
438fn dead_letter_from(row: &Row) -> DeadLetter {
439    DeadLetter {
440        id: row.get::<String>("id").unwrap_or_default(),
441        endpoint_id: row.get::<String>("endpoint_id").unwrap_or_default(),
442        subject: row.get::<String>("subject").unwrap_or_default(),
443        event_id: row.get::<String>("event_id").unwrap_or_default(),
444        event_type: row.get::<String>("event_type").unwrap_or_default(),
445        data: row.get::<String>("data").unwrap_or_default(),
446        attempts: row.get::<i64>("attempts").unwrap_or(0),
447        reason: row.get::<String>("reason").unwrap_or_default(),
448        last_error: row.get::<String>("last_error").unwrap_or_default(),
449        status_code: row.get::<i64>("status_code"),
450        created_at: row.get::<String>("created_at").unwrap_or_default(),
451        failed_at: row.get::<String>("failed_at").unwrap_or_default(),
452    }
453}
454
455/// The one dead letter of `subject` with `dead_letter_id`, or `None`.
456async fn find_dead_letter(
457    db: &dyn Database,
458    subject: &str,
459    dead_letter_id: &str,
460) -> Result<Option<DeadLetter>, DbError> {
461    let mut select = Query::select();
462    select
463        .columns(DEAD_LETTER_COLUMNS)
464        .from(iden(DEAD_LETTERS))
465        .and_where(Expr::col(iden("subject")).eq(subject))
466        .and_where(Expr::col(iden("id")).eq(dead_letter_id));
467    Ok(db
468        .query(&Statement::render(&select))
469        .await?
470        .first()
471        .map(dead_letter_from))
472}
473
474/// Re-enqueues one dead letter and removes it, in one batch. The new
475/// outbox row starts from zero attempts; the delivery log keeps the
476/// history. `Ok(false)` — the letter left in place — when the id names no
477/// dead letter of this subject, or when the letter's endpoint is gone:
478/// consuming the letter for a delivery the drain would only drop would
479/// destroy the event for good.
480pub(crate) async fn replay(
481    db: &dyn Database,
482    subject: &str,
483    dead_letter_id: &str,
484    at: &str,
485) -> Result<bool, DbError> {
486    let Some(letter) = find_dead_letter(db, subject, dead_letter_id).await? else {
487        return Ok(false);
488    };
489    if !endpoint_exists(db, subject, &letter.endpoint_id).await? {
490        return Ok(false);
491    }
492    let data: serde_json::Value =
493        serde_json::from_str(&letter.data).unwrap_or(serde_json::Value::Null);
494    let job = DeliveryJob {
495        endpoint_id: letter.endpoint_id,
496        event_id: letter.event_id,
497        event_type: letter.event_type,
498        subject: letter.subject,
499        created_at: letter.created_at,
500        data,
501    };
502    db.batch_atomic(&[
503        enqueue_statement(&job, &job.subject, at),
504        delete_dead_letter_statement(&letter.id),
505    ])
506    .await?;
507    Ok(true)
508}
509
510pub(crate) fn delete_dead_letter_statement(id: &str) -> Statement {
511    let mut delete = Query::delete();
512    delete
513        .from_table(iden(DEAD_LETTERS))
514        .and_where(Expr::col(iden("id")).eq(id));
515    Statement::render(&delete)
516}
517
518/// The job every unit test shares, at store scope so sibling test modules
519/// (`deliver`'s) reach it too.
520#[cfg(test)]
521pub(crate) fn test_job() -> DeliveryJob {
522    DeliveryJob {
523        endpoint_id: "ep".to_owned(),
524        event_id: "ev".to_owned(),
525        event_type: "order.paid".to_owned(),
526        subject: "acct".to_owned(),
527        created_at: "2026-09-27T00:00:00Z".to_owned(),
528        data: serde_json::json!({ "amount": 1 }),
529    }
530}
531
532#[cfg(test)]
533mod tests {
534    use super::*;
535
536    /// The migration carries the outbox DDL verbatim; a core change to
537    /// `create_table_sql` must be copied here on purpose, never silently.
538    #[test]
539    fn the_migration_carries_the_outbox_ddl_verbatim() {
540        let sql = std::fs::read_to_string(concat!(
541            env!("CARGO_MANIFEST_DIR"),
542            "/migrations/sqlite/0001_init.sql"
543        ))
544        .expect("the migration is in the crate");
545        assert!(
546            sql.contains(&Outbox::new(OUTBOX).create_table_sql()),
547            "webhooks_outbox must match Outbox::create_table_sql"
548        );
549    }
550
551    #[test]
552    fn a_job_enqueues_one_row_with_its_subject() {
553        let job = test_job();
554        let statement = enqueue_statement(&job, "acct", "2026-09-27T00:00:00Z");
555        assert!(statement.sql.contains("INSERT INTO"));
556        assert!(statement.sql.contains("webhooks_outbox"));
557        assert!(statement.sql.contains("subject"));
558    }
559}
cratefield_module_webhooks/store.rs

cratefield_module_webhooks/
store.rs
