fraiseql_core/runtime/
after_mutation.rs

1//! The engine's notice that a mutation committed (#1340, #1440).
2//!
3//! `after:mutation` functions run once a write has committed. Deciding *that* a write
4//! committed, and *what* it wrote, is the engine's knowledge: it holds the parsed outcome
5//! of the mutation function. A transport holds only the response it is about to serve.
6//!
7//! Dispatch used to live in the transports, reading the response, and that placement was
8//! wrong three ways:
9//!
10//! - **Only two transports dispatched.** GraphQL and REST did; MCP, gRPC and the functions bridge
11//!   commit through the same executor and fired nothing (#1440).
12//! - **A failure looked like a success.** Under `auto_error_union` a failed write is *data*: an
13//!   error member of the union, not a GraphQL error. A dispatcher reading the response could not
14//!   tell the two apart unless the client happened to select `__typename` (#1340).
15//! - **The row was the client's selection.** The event's row image was whatever fields the client
16//!   selected, so a `when` predicate on an unselected field never matched.
17//!
18//! The executor therefore calls an [`AfterMutationObserver`] from the mutation runner, the
19//! single chokepoint every transport converges on. It does so once per committed success,
20//! with the type the write produced and the full entity the function returned. A failed
21//! write, a dry run and a refused write never reach it. Like
22//! [`BeforeMutationGate`](crate::security::BeforeMutationGate), the observer is a trait
23//! object the application supplies, because `fraiseql-core` knows nothing about function
24//! runtimes. `fraiseql-server` installs one that dispatches into the compiled schema's
25//! `functions` section.
26
27use std::future::Future;
28
29use crate::{schema::MutationOperation, security::SecurityContext};
30
31/// A mutation that committed successfully, as the engine saw it.
32#[derive(Debug, Clone, Copy)]
33pub struct CommittedMutation<'a> {
34    /// The mutation's name, as declared.
35    pub mutation_name:  &'a str,
36    /// The type the write produced: the function's `entity_type` stamp, or else the one type
37    /// a success of this mutation can be.
38    ///
39    /// Never the name of a synthesized error union or a cascade payload: the produced
40    /// entity, which is what an `after:mutation:<Type>` trigger names.
41    pub entity_type:    &'a str,
42    /// The declared operation (insert, update, delete, or a custom function).
43    pub operation:      &'a MutationOperation,
44    /// The full entity the mutation function returned, independent of what any client
45    /// selected. For a delete it is the removed row.
46    pub entity:         &'a serde_json::Value,
47    /// The mutated entity's id, when the function reported one.
48    pub entity_id:      Option<&'a str>,
49    /// The principal the write ran as, or `None` for an unauthenticated write.
50    pub security_ctx:   Option<&'a SecurityContext>,
51    /// How deep in a dispatch chain the write was made: `0` for a write a request or a
52    /// scheduled source made, `n` for one made by a function that a depth-`n - 1` event
53    /// dispatched. See [`dispatched_at`].
54    pub dispatch_depth: u8,
55}
56
57/// Receives every committed mutation, from every transport.
58///
59/// Called on the request path, after the commit and before the response is returned, so an
60/// implementation must not block: hand the work to a task and return. Whatever it does
61/// cannot change the outcome of a write that has already committed, which is why the method
62/// returns nothing.
63///
64/// # Example
65///
66/// ```
67/// use std::sync::Arc;
68///
69/// use fraiseql_core::runtime::{AfterMutationObserver, CommittedMutation, RuntimeConfig};
70///
71/// /// Log every committed write.
72/// struct AuditLog;
73///
74/// impl AfterMutationObserver for AuditLog {
75///     fn on_committed(&self, mutation: &CommittedMutation<'_>) {
76///         tracing::info!(mutation = mutation.mutation_name, entity = mutation.entity_type);
77///     }
78/// }
79///
80/// let config = RuntimeConfig::default().with_after_mutation_observer(Arc::new(AuditLog));
81/// ```
82pub trait AfterMutationObserver: Send + Sync {
83    /// A mutation committed successfully.
84    fn on_committed(&self, mutation: &CommittedMutation<'_>);
85}
86
87tokio::task_local! {
88    static DISPATCH_DEPTH: u8;
89}
90
91/// Run `future` as the work of a function dispatched `depth` levels deep, so every write it
92/// makes is observed with that [`CommittedMutation::dispatch_depth`].
93///
94/// A dispatched function's writes pass the same chokepoint as a request's, so an
95/// `after:mutation` function that writes the entity it is triggered by would trigger itself
96/// without end. The depth is what lets the observer stop such a chain. It is carried in a
97/// task-local rather than on the [`SecurityContext`] because a client controls neither, and
98/// cannot set this one: only code that wraps its own executor calls in this scope can.
99pub async fn dispatched_at<F: Future>(depth: u8, future: F) -> F::Output {
100    DISPATCH_DEPTH.scope(depth, future).await
101}
102
103/// The dispatch depth of the current task: `0` outside any [`dispatched_at`] scope.
104#[must_use]
105pub fn current_dispatch_depth() -> u8 {
106    DISPATCH_DEPTH.try_with(|depth| *depth).unwrap_or(0)
107}
fraiseql_core/runtime/after_mutation.rs

fraiseql_core/runtime/
after_mutation.rs
