backbone_core/
write_hooks.rs

1//! Write hooks: code that runs inside a write's own transaction, with the change in hand.
2//!
3//! [`GenericCrudService`](crate::service::GenericCrudService) runs every write of an entity
4//! (create, update, patch, delete, restore, hard delete, their bulk forms, and a state-machine
5//! transition) inside one database transaction ([`backbone_orm::in_write_tx`]), in this order:
6//!
7//! 1. the write guard,
8//! 2. [`ServiceLifecycle::before_*`](crate::service::ServiceLifecycle),
9//! 3. [`WriteHooks::before`],
10//! 4. the repository write,
11//! 5. [`ServiceLifecycle::after_*`](crate::service::ServiceLifecycle),
12//! 6. [`WriteHooks::after`],
13//! 7. commit.
14//!
15//! A failure at any step rolls the whole write back: the row, whatever the hooks wrote and every
16//! event they staged. After the commit the service's [`CrudEventPublisher`] keeps its
17//! best-effort in-process role, and nothing else.
18//!
19//! The schema generator emits a `WriteHooks` implementation per model from its declared events
20//! (`on:` in a hook file's `events:`), which stages them with [`WriteCtx::stage`]. Hand code can
21//! implement the trait too, and compose it with [`with_write_hooks`](crate::service::GenericCrudService::with_write_hooks).
22//!
23//! # What a hook may rely on
24//!
25//! - Statements it runs through the ORM's scoped helpers, the repository, or
26//!   [`WriteCtx::stage`] ride the write's transaction and see its uncommitted rows.
27//! - A connection a hook opens itself (a bare `pool.begin()`) is another transaction: it cannot
28//!   see the row being written, and a statement of it that waits for a lock the write holds
29//!   waits for a commit that is waiting for the hook.
30//! - A hook that holds the transaction open for a slow call (HTTP, a queue) holds a database
31//!   connection and the row locks for as long. Stage an event and let the relay do the slow part.
32
33use async_trait::async_trait;
34use serde_json::Value;
35use uuid::Uuid;
36
37pub use backbone_outbox::OutboxRecord;
38
39use crate::persistence::traits::PersistentEntity;
40use crate::service::{ServiceError, ServiceResult};
41
42tokio::task_local! {
43    static CORRELATION: String;
44}
45
46/// Run `f` with `id` as the correlation id of every write it makes: the id the events staged by
47/// those writes carry, so one request's rows, events and logs share it. The generic write routes
48/// do this from the `x-correlation-id` request header (see
49/// [`correlation_middleware`](crate::http::correlation_middleware)).
50pub async fn with_correlation_id<F: std::future::Future>(id: impl Into<String>, f: F) -> F::Output {
51    CORRELATION.scope(id.into(), f).await
52}
53
54/// The correlation id [`with_correlation_id`] bound for the current task.
55pub fn current_correlation_id() -> Option<String> {
56    CORRELATION.try_with(|id| id.clone()).ok()
57}
58
59/// What a write hook knows about the write it runs in.
60#[derive(Debug, Clone)]
61#[non_exhaustive]
62pub struct WriteCtx {
63    /// The correlation id of the request (its `x-correlation-id` header, else the audit
64    /// attribution's, else one minted for the write). Staged events carry it.
65    pub correlation_id: String,
66    /// The id of the event, trigger or workflow step that caused this write, when one did.
67    /// Staged events carry it as their `causation_id`.
68    pub causation_id: Option<String>,
69    /// Who is writing: the request's audit attribution; `None` for a job or an unscoped caller.
70    pub actor: Option<String>,
71    /// The company the request is scoped to, when it is. Events of an entity with no company of
72    /// its own are staged under it.
73    pub company_id: Option<Uuid>,
74    /// The Postgres schema whose `outbox_events` table [`stage`](Self::stage) writes to, as the
75    /// hooks named it ([`WriteHooks::outbox_schema`]).
76    pub outbox_schema: Option<String>,
77}
78
79impl WriteCtx {
80    /// The context of a write made now, in the current task.
81    pub fn current(outbox_schema: Option<&str>) -> Self {
82        let audit = backbone_orm::audit_context::current_request_audit();
83        let actor = audit
84            .as_ref()
85            .map(|a| a.actor.clone())
86            .filter(|a| !a.is_empty());
87        let correlation_id = current_correlation_id()
88            .or_else(|| audit.map(|a| a.correlation_id).filter(|c| !c.is_empty()))
89            .unwrap_or_else(|| Uuid::new_v4().to_string());
90        Self {
91            correlation_id,
92            causation_id: None,
93            actor,
94            company_id: backbone_orm::current_company(),
95            outbox_schema: outbox_schema.map(str::to_string),
96        }
97    }
98
99    /// The company to stage an event under when the entity names none: the request's, else the
100    /// nil uuid. The outbox table requires a value, and under org-unit tenancy the fence is the
101    /// org unit the table fills from the acting unit, not this column.
102    pub fn company_or_nil(&self) -> Uuid {
103        self.company_id.unwrap_or_else(Uuid::nil)
104    }
105
106    /// The company an event of `row` is staged under: the row's own `company_id`, else the
107    /// request's, else the nil uuid.
108    pub fn company_of(&self, row: &Value) -> Uuid {
109        row.get("company_id")
110            .and_then(Value::as_str)
111            .and_then(|id| Uuid::parse_str(id).ok())
112            .unwrap_or_else(|| self.company_or_nil())
113    }
114
115    /// Stage `record` on the write's transaction. It commits with the write and vanishes if
116    /// the write rolls back. The record takes this context's correlation id (and causation id)
117    /// unless it already carries one.
118    ///
119    /// # Errors
120    /// A [`ServiceError`] when the hooks named no outbox schema, when no write transaction is
121    /// active (the repository has no pool, so the service ran the write without one), or when
122    /// the insert fails — which rolls the write back, since a write whose event cannot be staged
123    /// must not commit.
124    pub async fn stage(&self, mut record: OutboxRecord) -> ServiceResult<()> {
125        let Some(schema) = self.outbox_schema.as_deref() else {
126            return Err(ServiceError::Internal(
127                "an event was staged by hooks that name no outbox schema".into(),
128            ));
129        };
130        if record.correlation_id.is_none() {
131            record.correlation_id = Some(self.correlation_id.clone());
132        }
133        if record.causation_id.is_none() {
134            record.causation_id = self.causation_id.clone();
135        }
136        backbone_outbox::outbox::stage_in_write_tx(schema, &record)
137            .await
138            .map(|_| ())
139            .map_err(|e| {
140                ServiceError::Internal(format!("staging {} failed: {e}", record.event_type))
141            })
142    }
143}
144
145/// A fact the module declares it publishes (`events:` in a schema file), as the generated typed
146/// struct of its payload.
147///
148/// The schema generator implements this for every declared event, in the module's
149/// `domain::event::declared`. An event with an `on:` trigger is staged by the generated write
150/// hooks; one without is staged by hand code, which builds its row with [`record`](Self::record)
151/// and stages it in its own transaction (`backbone_outbox::outbox::stage`) or on the write's
152/// ([`WriteCtx::stage`]).
153pub trait DeclaredEvent: serde::Serialize {
154    /// The event type name the outbox row carries (`TicketOpened`).
155    const NAME: &'static str;
156    /// The declared schema version of the payload.
157    const VERSION: i32;
158    /// The aggregate the event belongs to (`Ticket`), or `""` when the declaration names none.
159    const AGGREGATE: &'static str;
160
161    /// The outbox row of this event for aggregate `aggregate_id`, under the company of `ctx`
162    /// (or the nil uuid when the request has none). A fresh event id, now as the time it
163    /// occurred, the declared version, and the payload serialized as written.
164    fn record(&self, aggregate_id: &str, ctx: &WriteCtx) -> OutboxRecord {
165        self.record_for(aggregate_id, ctx.company_or_nil(), ctx)
166    }
167
168    /// [`record`](Self::record) for an aggregate that carries its own company.
169    fn record_for(&self, aggregate_id: &str, company_id: Uuid, ctx: &WriteCtx) -> OutboxRecord {
170        let payload = serde_json::to_value(self).unwrap_or(Value::Null);
171        let mut record = OutboxRecord::new(
172            Self::NAME,
173            Self::AGGREGATE,
174            aggregate_id,
175            company_id,
176            payload,
177            chrono::Utc::now(),
178        )
179        .with_version(Self::VERSION)
180        .with_correlation_id(ctx.correlation_id.clone());
181        if let Some(cause) = &ctx.causation_id {
182            record = record.with_causation_id(cause.clone());
183        }
184        record
185    }
186}
187
188/// The payload of a declared event, read out of the row the write stored: the fields listed,
189/// by their serialized names, deserialized into the event's struct. A listed field the row
190/// lacks is left out, so an optional one reads as `None` and a required one is an error — the
191/// generator only lists fields the aggregate has.
192pub fn payload_from_row<T: serde::de::DeserializeOwned>(
193    row: &Value,
194    fields: &[&str],
195) -> ServiceResult<T> {
196    let mut picked = serde_json::Map::new();
197    if let Value::Object(map) = row {
198        for field in fields {
199            if let Some(value) = map.get(*field) {
200                picked.insert((*field).to_string(), value.clone());
201            }
202        }
203    }
204    serde_json::from_value(Value::Object(picked)).map_err(|e| {
205        ServiceError::Internal(format!(
206            "an event payload could not be read from the row: {e}"
207        ))
208    })
209}
210
211/// Create the outbox and inbox tables of `schema` (idempotent), where the events its hooks stage
212/// are written. The composing service runs it at startup, with a role that may create tables;
213/// the generated module offers it for each schema it stages in (`declared::migrate_outbox`).
214pub async fn migrate_outbox(pool: &backbone_orm::PgPool, schema: &str) -> anyhow::Result<()> {
215    backbone_outbox::outbox::migrate(pool, schema).await?;
216    Ok(())
217}
218
219/// The kind of a [`Change`], without its rows.
220#[derive(Debug, Clone, Copy, PartialEq, Eq)]
221pub enum ChangeKind {
222    /// A new row.
223    Created,
224    /// An existing row changed by a generic write.
225    Updated,
226    /// An existing row moved by a state-machine verb.
227    Transitioned,
228    /// A row deleted (soft, or for good).
229    Deleted,
230    /// A soft-deleted row restored.
231    Restored,
232}
233
234/// One row change, as a write hook sees it.
235///
236/// In [`WriteHooks::before`] the row to be stored is mutable: a hook may fill a derived field.
237/// In [`WriteHooks::after`] it is the row as stored. Matching on it needs a wildcard arm; kinds
238/// may be added.
239#[derive(Debug)]
240#[non_exhaustive]
241pub enum Change<'a, E> {
242    /// A row is created.
243    Created {
244        /// The row to store, or as stored.
245        after: &'a mut E,
246    },
247    /// A row is changed by a generic write (PUT, PATCH or their bulk forms).
248    Updated {
249        /// The row as it was read.
250        before: &'a E,
251        /// The row to store, or as stored.
252        after: &'a mut E,
253    },
254    /// A row is moved by a state-machine verb.
255    Transitioned {
256        /// The row as it was read.
257        before: &'a E,
258        /// The row to store, or as stored.
259        after: &'a mut E,
260        /// The verb (`close`, `approve`).
261        transition: String,
262        /// The state field that moved, in the serialized naming (`status`).
263        field: String,
264        /// The state the row left.
265        from: Option<String>,
266        /// The state the row entered.
267        to: Option<String>,
268    },
269    /// A row is deleted: soft-deleted, or removed for good when `hard`.
270    Deleted {
271        /// The row as it was read.
272        before: &'a E,
273        /// Whether the row is removed for good (from the trash) rather than moved to it.
274        hard: bool,
275    },
276    /// A soft-deleted row is restored.
277    Restored {
278        /// The row: as it was in the trash before the write, as restored after it.
279        entity: &'a E,
280    },
281}
282
283impl<'a, E> Change<'a, E> {
284    /// Which kind of change this is.
285    pub fn kind(&self) -> ChangeKind {
286        match self {
287            Change::Created { .. } => ChangeKind::Created,
288            Change::Updated { .. } => ChangeKind::Updated,
289            Change::Transitioned { .. } => ChangeKind::Transitioned,
290            Change::Deleted { .. } => ChangeKind::Deleted,
291            Change::Restored { .. } => ChangeKind::Restored,
292        }
293    }
294
295    /// The row as it was read before the write; `None` for a create.
296    pub fn before(&self) -> Option<&E> {
297        match self {
298            Change::Updated { before, .. }
299            | Change::Transitioned { before, .. }
300            | Change::Deleted { before, .. } => Some(before),
301            Change::Created { .. } | Change::Restored { .. } => None,
302        }
303    }
304
305    /// The row the write stores; `None` for a delete.
306    pub fn after(&self) -> Option<&E> {
307        match self {
308            Change::Created { after }
309            | Change::Updated { after, .. }
310            | Change::Transitioned { after, .. } => Some(after),
311            Change::Restored { entity } => Some(entity),
312            Change::Deleted { .. } => None,
313        }
314    }
315
316    /// The row an event of this change is about: what the write stores, or the row deleted.
317    pub fn subject(&self) -> &E {
318        match self {
319            Change::Deleted { before, .. } => before,
320            other => other
321                .after()
322                .expect("every change but a delete stores a row"),
323        }
324    }
325}
326
327impl<'a, E: serde::Serialize> Change<'a, E> {
328    /// Whether the serialized field `name` differs between the row before and after. A create
329    /// changes every field it sets (non-null); a delete changes none.
330    pub fn field_changed(&self, name: &str) -> bool {
331        let after = match self.after().and_then(|a| serde_json::to_value(a).ok()) {
332            Some(Value::Object(map)) => map,
333            _ => return false,
334        };
335        let now = after.get(name).filter(|v| !v.is_null());
336        match self.before().and_then(|b| serde_json::to_value(b).ok()) {
337            Some(Value::Object(before)) => before.get(name).filter(|v| !v.is_null()) != now,
338            _ => now.is_some(),
339        }
340    }
341}
342
343impl<'a, E: PersistentEntity> Change<'a, E> {
344    /// The change an update writes: a [`Change::Transitioned`] when a state-machine verb performs
345    /// it ([`as_transition`](crate::write_guard::as_transition)), else a [`Change::Updated`].
346    pub(crate) fn update(before: &'a E, after: &'a mut E) -> Self {
347        let Some(transition) = crate::write_guard::current_transition() else {
348            return Change::Updated { before, after };
349        };
350        let before_obj = serde_json::to_value(before).ok();
351        let after_obj = serde_json::to_value(&*after).ok();
352        let state = |obj: &Option<Value>, field: &str| {
353            obj.as_ref()
354                .and_then(|o| o.get(field))
355                .and_then(Value::as_str)
356                .map(str::to_string)
357        };
358        // The state field the verb moved: the first whose value differs, else the entity's first.
359        let moved = E::state_fields()
360            .iter()
361            .find(|f| state(&before_obj, f) != state(&after_obj, f))
362            .or_else(|| E::state_fields().first())
363            .copied();
364        match moved {
365            Some(field) => Change::Transitioned {
366                from: state(&before_obj, field),
367                to: state(&after_obj, field),
368                field: field.to_string(),
369                transition,
370                before,
371                after,
372            },
373            // A verb on an entity with no state field: nothing to say about a move.
374            None => Change::Updated { before, after },
375        }
376    }
377}
378
379/// Code that runs inside a write's transaction, before and after the repository write.
380///
381/// Both methods default to doing nothing. An error from either rolls the write back and answers
382/// as the service answers any failed write.
383#[async_trait]
384pub trait WriteHooks<E: PersistentEntity>: Send + Sync {
385    /// The Postgres schema whose `outbox_events` table [`WriteCtx::stage`] writes to. `None`
386    /// (the default) means these hooks stage nothing.
387    fn outbox_schema(&self) -> Option<&str> {
388        None
389    }
390
391    /// Before the repository writes `change`: may refuse it, and may fill fields of the row to
392    /// be stored. Runs after the guard and [`ServiceLifecycle::before_*`](crate::service::ServiceLifecycle).
393    async fn before(&self, ctx: &WriteCtx, change: &mut Change<'_, E>) -> ServiceResult<()> {
394        let _ = (ctx, change);
395        Ok(())
396    }
397
398    /// After the repository wrote `change`, still inside the transaction: the place to stage
399    /// events and to write rows that must commit with it. Runs after
400    /// [`ServiceLifecycle::after_*`](crate::service::ServiceLifecycle).
401    async fn after(&self, ctx: &WriteCtx, change: &Change<'_, E>) -> ServiceResult<()> {
402        let _ = (ctx, change);
403        Ok(())
404    }
405}
406
407/// The hooks of a service that has none: every write runs in its transaction and nothing more.
408#[derive(Debug, Default, Clone, Copy)]
409pub struct NoWriteHooks;
410
411#[async_trait]
412impl<E: PersistentEntity> WriteHooks<E> for NoWriteHooks {}
413
414#[cfg(test)]
415mod tests {
416    use super::*;
417    use chrono::{DateTime, Utc};
418    use serde::{Deserialize, Serialize};
419
420    #[derive(Debug, Clone, Serialize, Deserialize)]
421    struct Row {
422        id: String,
423        status: String,
424        note: Option<String>,
425        created_at: DateTime<Utc>,
426        updated_at: DateTime<Utc>,
427        deleted_at: Option<DateTime<Utc>>,
428    }
429
430    impl PersistentEntity for Row {
431        fn state_fields() -> &'static [&'static str] {
432            &["status"]
433        }
434        fn entity_id(&self) -> String {
435            self.id.clone()
436        }
437        fn set_entity_id(&mut self, id: String) {
438            self.id = id;
439        }
440        fn created_at(&self) -> Option<DateTime<Utc>> {
441            Some(self.created_at)
442        }
443        fn set_created_at(&mut self, ts: DateTime<Utc>) {
444            self.created_at = ts;
445        }
446        fn updated_at(&self) -> Option<DateTime<Utc>> {
447            Some(self.updated_at)
448        }
449        fn set_updated_at(&mut self, ts: DateTime<Utc>) {
450            self.updated_at = ts;
451        }
452        fn deleted_at(&self) -> Option<DateTime<Utc>> {
453            self.deleted_at
454        }
455        fn set_deleted_at(&mut self, ts: Option<DateTime<Utc>>) {
456            self.deleted_at = ts;
457        }
458    }
459
460    fn row(status: &str, note: Option<&str>) -> Row {
461        Row {
462            id: "1".into(),
463            status: status.into(),
464            note: note.map(str::to_string),
465            created_at: Utc::now(),
466            updated_at: Utc::now(),
467            deleted_at: None,
468        }
469    }
470
471    #[tokio::test]
472    async fn an_update_is_a_transition_only_inside_a_verb() {
473        let before = row("open", None);
474        let mut after = row("closed", None);
475        assert_eq!(
476            Change::update(&before, &mut after).kind(),
477            ChangeKind::Updated
478        );
479
480        let kind = crate::write_guard::as_transition("close", async {
481            let mut after = row("closed", None);
482            match Change::update(&before, &mut after) {
483                Change::Transitioned {
484                    transition,
485                    field,
486                    from,
487                    to,
488                    ..
489                } => (transition, field, from, to),
490                other => panic!("expected a transition, got {:?}", other.kind()),
491            }
492        })
493        .await;
494        assert_eq!(
495            kind,
496            (
497                "close".to_string(),
498                "status".to_string(),
499                Some("open".to_string()),
500                Some("closed".to_string())
501            )
502        );
503    }
504
505    #[test]
506    fn a_field_counts_as_changed_only_when_its_value_differs() {
507        let before = row("open", Some("a"));
508        let mut same = row("open", Some("a"));
509        let mut note_changed = row("open", Some("b"));
510        let mut created = row("open", None);
511        assert!(!Change::Updated {
512            before: &before,
513            after: &mut same
514        }
515        .field_changed("note"));
516        assert!(Change::Updated {
517            before: &before,
518            after: &mut note_changed
519        }
520        .field_changed("note"));
521        assert!(!Change::Updated {
522            before: &before,
523            after: &mut note_changed
524        }
525        .field_changed("status"));
526        // A create sets what it sets: a null field is not a change, a filled one is.
527        assert!(Change::Created {
528            after: &mut created
529        }
530        .field_changed("status"));
531        assert!(!Change::Created {
532            after: &mut row("open", None)
533        }
534        .field_changed("note"));
535    }
536
537    #[tokio::test]
538    async fn the_correlation_id_comes_from_the_request_scope_else_is_minted() {
539        let minted = WriteCtx::current(None);
540        assert!(!minted.correlation_id.is_empty());
541        let scoped =
542            with_correlation_id("req-7", async { WriteCtx::current(Some("billing")) }).await;
543        assert_eq!(scoped.correlation_id, "req-7");
544        assert_eq!(scoped.outbox_schema.as_deref(), Some("billing"));
545    }
546
547    async fn correlation_seen_by(header: Option<&str>) -> String {
548        use axum::{body::Body, http::Request, routing::get, Router};
549        use tower::ServiceExt;
550        let app = Router::new()
551            .route(
552                "/",
553                get(|| async { WriteCtx::current(None).correlation_id }),
554            )
555            .layer(axum::middleware::from_fn(
556                crate::http::correlation_middleware,
557            ));
558        let mut req = Request::builder().uri("/");
559        if let Some(h) = header {
560            req = req.header("x-correlation-id", h);
561        }
562        let res = app.oneshot(req.body(Body::empty()).unwrap()).await.unwrap();
563        let bytes = axum::body::to_bytes(res.into_body(), usize::MAX)
564            .await
565            .unwrap();
566        String::from_utf8(bytes.to_vec()).unwrap()
567    }
568
569    #[tokio::test]
570    async fn the_request_header_is_the_correlation_id_when_usable() {
571        assert_eq!(correlation_seen_by(Some("req-abc-1")).await, "req-abc-1");
572        for unusable in ["", "has space", &"x".repeat(129)] {
573            let id = correlation_seen_by(Some(unusable)).await;
574            assert!(Uuid::parse_str(&id).is_ok(), "{unusable:?} gave {id}");
575        }
576        assert!(Uuid::parse_str(&correlation_seen_by(None).await).is_ok());
577    }
578
579    #[derive(Serialize, Deserialize, Debug, PartialEq)]
580    struct RowOpened {
581        id: String,
582        note: Option<String>,
583    }
584
585    impl DeclaredEvent for RowOpened {
586        const NAME: &'static str = "RowOpened";
587        const VERSION: i32 = 3;
588        const AGGREGATE: &'static str = "Row";
589    }
590
591    #[test]
592    fn a_payload_is_picked_from_the_row_by_field_name() {
593        let row = serde_json::json!({ "id": "1", "note": null, "status": "open" });
594        let picked: RowOpened = payload_from_row(&row, &["id", "note"]).unwrap();
595        assert_eq!(
596            picked,
597            RowOpened {
598                id: "1".into(),
599                note: None
600            }
601        );
602        // An optional field the row lacks reads as none; a required one is an error.
603        let thin = serde_json::json!({ "id": "1" });
604        assert_eq!(
605            payload_from_row::<RowOpened>(&thin, &["id", "note"])
606                .unwrap()
607                .note,
608            None
609        );
610        assert!(payload_from_row::<RowOpened>(&thin, &["note"]).is_err());
611    }
612
613    #[test]
614    fn a_declared_event_builds_its_outbox_row_from_its_declaration_and_the_context() {
615        let ctx = WriteCtx {
616            correlation_id: "req-9".into(),
617            causation_id: Some("cause-1".into()),
618            actor: None,
619            company_id: Some(Uuid::from_u128(7)),
620            outbox_schema: Some("billing".into()),
621        };
622        let event = RowOpened {
623            id: "1".into(),
624            note: Some("n".into()),
625        };
626        let record = event.record("agg-1", &ctx);
627        assert_eq!(
628            (
629                record.event_type.as_str(),
630                record.aggregate_type.as_str(),
631                record.aggregate_id.as_str(),
632                record.version
633            ),
634            ("RowOpened", "Row", "agg-1", 3)
635        );
636        assert_eq!(record.company_id, Uuid::from_u128(7));
637        assert_eq!(record.correlation_id.as_deref(), Some("req-9"));
638        assert_eq!(record.causation_id.as_deref(), Some("cause-1"));
639        assert_eq!(record.payload["note"], "n");
640        // An aggregate that carries its own company is staged under it.
641        let own = Uuid::from_u128(9);
642        let row = serde_json::json!({ "company_id": own.to_string() });
643        assert_eq!(ctx.company_of(&row), own);
644        assert_eq!(ctx.company_of(&serde_json::json!({})), Uuid::from_u128(7));
645    }
646}
backbone_core/write_hooks.rs

backbone_core/
write_hooks.rs
