graphrefly/cqrs/
messaging.rs

1//! Optional CQRS-over-messageBus recipe (D350/D351/D353).
2//!
3//! Retained delivery is lowered to CQRS command facts. Ack commands are emitted
4//! only after a graph-visible accepted/rejected status or lowering issue exists;
5//! receiving a retained message is never itself an ack.
6
7use std::collections::BTreeMap;
8use std::rc::Rc;
9
10use crate::cqrs::{CqrsCommand, CqrsEvent, CqrsStatus, CqrsStatusState};
11use crate::ctx::Ctx;
12use crate::graph::{Graph, GraphNodeOpts};
13use crate::identity::{canonical_tuple_key, compound_tuple_key};
14use crate::messaging::{DataIssue, MessageBusAvailablePage, MessageBusCommand, MessageEnvelope};
15use crate::node::Node;
16
17#[derive(Debug, Clone, PartialEq, Eq)]
18/// `MessageBusDelivery` data container.
19pub struct MessageBusDelivery {
20    /// `topic` field for topic.
21    pub topic: String,
22    /// `seq` field for seq.
23    pub seq: u64,
24    /// `subscription_id` field for subscription id.
25    pub subscription_id: String,
26    /// `command_id` field for command id.
27    pub command_id: String,
28}
29
30#[derive(Debug, Clone, PartialEq)]
31/// `CqrsDeliveredCommand` data container.
32pub struct CqrsDeliveredCommand<TCommand> {
33    /// `command` field for command.
34    pub command: CqrsCommand<TCommand>,
35    /// `delivery` field for delivery.
36    pub delivery: MessageBusDelivery,
37}
38
39/// `CqrsMessageCommandFn` type alias.
40pub type CqrsMessageCommandFn<TPayload, TCommand> =
41    Rc<dyn Fn(&MessageEnvelope<TPayload>, &MessageBusDelivery) -> Option<CqrsCommand<TCommand>>>;
42
43#[derive(Clone)]
44/// `CqrsMessagingPolicy` data container.
45pub struct CqrsMessagingPolicy<TPayload, TCommand> {
46    /// `command` field for command.
47    pub command: CqrsMessageCommandFn<TPayload, TCommand>,
48    /// `ack_rejected` field for ack rejected.
49    pub ack_rejected: bool,
50    /// `outbox_topic` field for outbox topic.
51    pub outbox_topic: Option<String>,
52}
53
54impl<TPayload, TCommand> CqrsMessagingPolicy<TPayload, TCommand> {
55    /// Creates or computes `new`.
56    pub fn new(
57        command: impl Fn(&MessageEnvelope<TPayload>, &MessageBusDelivery) -> Option<CqrsCommand<TCommand>>
58            + 'static,
59    ) -> Self {
60        Self {
61            command: Rc::new(command),
62            ack_rejected: true,
63            outbox_topic: None,
64        }
65    }
66
67    /// Updates or reads `ack_rejected`.
68    pub fn ack_rejected(mut self, ack_rejected: bool) -> Self {
69        self.ack_rejected = ack_rejected;
70        self
71    }
72
73    /// Updates or reads `with_outbox_topic`.
74    pub fn with_outbox_topic(mut self, topic: impl Into<String>) -> Self {
75        self.outbox_topic = Some(topic.into());
76        self
77    }
78}
79
80#[derive(Clone)]
81/// `CqrsMessagingRecipeOptions` data container.
82pub struct CqrsMessagingRecipeOptions<TPayload, TCommand, TEvent> {
83    /// `name` field for name.
84    pub name: String,
85    /// `deliveries` field for deliveries.
86    pub deliveries: Node<MessageBusAvailablePage<TPayload>>,
87    /// `status` field for status.
88    pub status: Node<CqrsStatus>,
89    /// `events` field for events.
90    pub events: Option<Node<CqrsEvent<TEvent>>>,
91    /// `policy` field for policy.
92    pub policy: CqrsMessagingPolicy<TPayload, TCommand>,
93}
94
95impl<TPayload, TCommand, TEvent> CqrsMessagingRecipeOptions<TPayload, TCommand, TEvent> {
96    /// Creates or computes `new`.
97    pub fn new(
98        deliveries: Node<MessageBusAvailablePage<TPayload>>,
99        status: Node<CqrsStatus>,
100        policy: CqrsMessagingPolicy<TPayload, TCommand>,
101    ) -> Self {
102        Self {
103            name: "cqrsMessaging".to_owned(),
104            deliveries,
105            status,
106            events: None,
107            policy,
108        }
109    }
110
111    /// Updates or reads `named`.
112    pub fn named(mut self, name: impl Into<String>) -> Self {
113        self.name = name.into();
114        self
115    }
116
117    /// Updates or reads `with_events`.
118    pub fn with_events(mut self, events: Node<CqrsEvent<TEvent>>) -> Self {
119        self.events = Some(events);
120        self
121    }
122}
123
124#[derive(Clone)]
125/// `CqrsMessagingRecipeBundle` data container.
126pub struct CqrsMessagingRecipeBundle<TCommand, TEvent> {
127    /// `delivered_commands` field for delivered commands.
128    pub delivered_commands: Node<CqrsDeliveredCommand<TCommand>>,
129    /// `commands` field for commands.
130    pub commands: Node<CqrsCommand<TCommand>>,
131    /// `ack_commands` field for ack commands.
132    pub ack_commands: Node<MessageBusCommand<TCommand>>,
133    /// `outbox_commands` field for outbox commands.
134    pub outbox_commands: Option<Node<MessageBusCommand<CqrsEvent<TEvent>>>>,
135    /// `issues` field for issues.
136    pub issues: Node<DataIssue>,
137}
138
139#[derive(Clone)]
140enum CqrsMessagingFact<TCommand> {
141    Command(CqrsDeliveredCommand<TCommand>),
142    Issue(DataIssue),
143}
144
145#[derive(Clone, Default)]
146struct AckState {
147    deliveries: BTreeMap<String, Vec<MessageBusDelivery>>,
148}
149
150/// Creates or computes `cqrs_messaging_recipe`.
151pub fn cqrs_messaging_recipe<
152    TPayload: Clone + 'static,
153    TCommand: Clone + 'static,
154    TEvent: Clone + 'static,
155>(
156    graph: &Graph,
157    opts: CqrsMessagingRecipeOptions<TPayload, TCommand, TEvent>,
158) -> CqrsMessagingRecipeBundle<TCommand, TEvent> {
159    let name = opts.name.clone();
160    let policy = opts.policy.clone();
161    let runtime = graph.node_opts::<CqrsMessagingFact<TCommand>, _>(
162        vec![opts.deliveries.erased()],
163        move |ctx| {
164            for page in ctx.batch::<MessageBusAvailablePage<TPayload>>(0) {
165                for message in &page.messages {
166                    let delivery = message_delivery(page.as_ref(), message);
167                    if let Some(command) = (policy.command)(message, &delivery) {
168                        ctx.emit(CqrsMessagingFact::Command(CqrsDeliveredCommand {
169                            command,
170                            delivery,
171                        }));
172                    } else {
173                        ctx.emit(CqrsMessagingFact::<TCommand>::Issue(message_issue(
174                            &delivery,
175                        )));
176                    }
177                }
178            }
179        },
180        GraphNodeOpts::named(format!("{name}/runtime")),
181    );
182    let delivered_commands = project(
183        graph,
184        &runtime,
185        format!("{name}/deliveredCommands"),
186        |fact| match fact {
187            CqrsMessagingFact::Command(delivered) => Some(delivered.clone()),
188            CqrsMessagingFact::Issue(_) => None,
189        },
190    );
191    let commands = project(
192        graph,
193        &runtime,
194        format!("{name}/commands"),
195        |fact| match fact {
196            CqrsMessagingFact::Command(delivered) => Some(delivered.command.clone()),
197            CqrsMessagingFact::Issue(_) => None,
198        },
199    );
200    let issues = project(
201        graph,
202        &runtime,
203        format!("{name}/issues"),
204        |fact| match fact {
205            CqrsMessagingFact::Issue(issue) => Some(issue.clone()),
206            CqrsMessagingFact::Command(_) => None,
207        },
208    );
209    let ack_commands = cqrs_message_ack_commands(
210        graph,
211        CqrsMessageAckOptions {
212            name: format!("{name}/ackCommands"),
213            delivered_commands: delivered_commands.clone(),
214            status: opts.status,
215            issues: Some(issues.clone()),
216            ack_rejected: opts.policy.ack_rejected,
217        },
218    );
219    let outbox_commands = opts.events.and_then(|events| {
220        opts.policy.outbox_topic.map(|topic| {
221            cqrs_event_outbox_commands(graph, events, topic, format!("{name}/outboxCommands"))
222        })
223    });
224    CqrsMessagingRecipeBundle {
225        delivered_commands,
226        commands,
227        ack_commands,
228        outbox_commands,
229        issues,
230    }
231}
232
233/// `CqrsMessageAckOptions` data container.
234pub struct CqrsMessageAckOptions<TCommand> {
235    /// `name` field for name.
236    pub name: String,
237    /// `delivered_commands` field for delivered commands.
238    pub delivered_commands: Node<CqrsDeliveredCommand<TCommand>>,
239    /// `status` field for status.
240    pub status: Node<CqrsStatus>,
241    /// `issues` field for issues.
242    pub issues: Option<Node<DataIssue>>,
243    /// `ack_rejected` field for ack rejected.
244    pub ack_rejected: bool,
245}
246
247/// Creates or computes `cqrs_message_ack_commands`.
248pub fn cqrs_message_ack_commands<TCommand: Clone + 'static>(
249    graph: &Graph,
250    opts: CqrsMessageAckOptions<TCommand>,
251) -> Node<MessageBusCommand<TCommand>> {
252    let mut deps = vec![opts.delivered_commands.erased(), opts.status.erased()];
253    if let Some(issues) = &opts.issues {
254        deps.push(issues.erased());
255    }
256    graph.node_opts::<MessageBusCommand<TCommand>, _>(
257        deps,
258        move |ctx| {
259            let mut state = ctx
260                .state_get::<AckState>()
261                .map(|state| (*state).clone())
262                .unwrap_or_default();
263            for delivered in ctx.batch::<CqrsDeliveredCommand<TCommand>>(0) {
264                state
265                    .deliveries
266                    .entry(delivered.command.id.clone())
267                    .or_default()
268                    .push(delivered.delivery.clone());
269            }
270            for status in ctx.batch::<CqrsStatus>(1) {
271                if status.state == CqrsStatusState::Rejected && !opts.ack_rejected {
272                    continue;
273                }
274                let Some(command_id) = &status.command_id else {
275                    continue;
276                };
277                if let Some(delivery) = shift_delivery(&mut state.deliveries, command_id) {
278                    ctx.emit(ack_command::<TCommand>(
279                        &delivery,
280                        cqrs_ack_id(&delivery, "status-ack"),
281                    ));
282                }
283            }
284            if opts.issues.is_some() {
285                for issue in ctx.batch::<DataIssue>(2) {
286                    if let Some(delivery) = issue_delivery(&issue) {
287                        ctx.emit(ack_command::<TCommand>(
288                            &delivery,
289                            cqrs_ack_id(&delivery, "issue-ack"),
290                        ));
291                    }
292                }
293            }
294            ctx.state_set(state);
295            ctx.state_persist(true);
296        },
297        GraphNodeOpts::named(opts.name),
298    )
299}
300
301fn cqrs_ack_id(delivery: &MessageBusDelivery, reason: &str) -> String {
302    compound_tuple_key(
303        "cqrs-message-ack",
304        &[
305            &delivery.topic,
306            &delivery.subscription_id,
307            &delivery.seq.to_string(),
308            reason,
309        ],
310    )
311}
312
313/// Creates or computes `cqrs_event_outbox_commands`.
314pub fn cqrs_event_outbox_commands<TEvent: Clone + 'static>(
315    graph: &Graph,
316    events: Node<CqrsEvent<TEvent>>,
317    topic: impl Into<String>,
318    name: impl Into<String>,
319) -> Node<MessageBusCommand<CqrsEvent<TEvent>>> {
320    let topic = topic.into();
321    graph.node_opts::<MessageBusCommand<CqrsEvent<TEvent>>, _>(
322        vec![events.erased()],
323        move |ctx| {
324            for event in ctx.batch::<CqrsEvent<TEvent>>(0) {
325                ctx.emit(MessageBusCommand::Publish {
326                    topic: topic.clone(),
327                    payload: (*event).clone(),
328                    key: event.aggregate_id.clone(),
329                    command_id: Some(compound_tuple_key("cqrs-outbox", &[&event.id])),
330                    idempotency_key: Some(event.id.clone()),
331                });
332            }
333        },
334        GraphNodeOpts::named(name.into()),
335    )
336}
337
338fn project<TIn: Clone + 'static, TOut: 'static>(
339    graph: &Graph,
340    source: &Node<TIn>,
341    name: String,
342    pick: impl Fn(&TIn) -> Option<TOut> + 'static,
343) -> Node<TOut> {
344    graph.node_opts::<TOut, _>(
345        vec![source.erased()],
346        move |ctx: &Ctx| {
347            for fact in ctx.batch::<TIn>(0) {
348                if let Some(value) = pick(&fact) {
349                    ctx.emit(value);
350                }
351            }
352        },
353        GraphNodeOpts::named(name),
354    )
355}
356
357fn message_delivery<T>(
358    page: &MessageBusAvailablePage<T>,
359    message: &MessageEnvelope<T>,
360) -> MessageBusDelivery {
361    MessageBusDelivery {
362        topic: page.topic.clone(),
363        seq: message.seq,
364        subscription_id: page.subscription_id.clone(),
365        command_id: message
366            .command_id
367            .clone()
368            .unwrap_or_else(|| canonical_tuple_key(&[&page.topic, &message.seq.to_string()])),
369    }
370}
371
372fn message_issue(delivery: &MessageBusDelivery) -> DataIssue {
373    DataIssue {
374        kind: "issue".to_owned(),
375        code: "cqrs-message-lowering-rejected".to_owned(),
376        message: "CQRS messaging recipe could not lower retained message to a command fact"
377            .to_owned(),
378        severity: "error".to_owned(),
379        source: "cqrs.messaging".to_owned(),
380        topic: Some(delivery.topic.clone()),
381        details: Some(delivery_details(delivery)),
382    }
383}
384
385fn issue_delivery(issue: &DataIssue) -> Option<MessageBusDelivery> {
386    issue.details.as_deref().and_then(parse_delivery_details)
387}
388
389fn ack_command<T>(delivery: &MessageBusDelivery, command_id: String) -> MessageBusCommand<T> {
390    MessageBusCommand::Ack {
391        topic: delivery.topic.clone(),
392        subscription_id: delivery.subscription_id.clone(),
393        seq: delivery.seq,
394        command_id: Some(command_id),
395    }
396}
397
398fn shift_delivery(
399    deliveries: &mut BTreeMap<String, Vec<MessageBusDelivery>>,
400    command_id: &str,
401) -> Option<MessageBusDelivery> {
402    let queue = deliveries.get_mut(command_id)?;
403    if queue.is_empty() {
404        return None;
405    }
406    let first = queue.remove(0);
407    if queue.is_empty() {
408        deliveries.remove(command_id);
409    }
410    Some(first)
411}
412
413fn delivery_details(delivery: &MessageBusDelivery) -> String {
414    format!(
415        "messageBus:topic={};subscription_id={};seq={};command_id={}",
416        delivery.topic, delivery.subscription_id, delivery.seq, delivery.command_id
417    )
418}
419
420fn parse_delivery_details(details: &str) -> Option<MessageBusDelivery> {
421    let rest = details.strip_prefix("messageBus:")?;
422    let mut topic = None;
423    let mut subscription_id = None;
424    let mut seq = None;
425    let mut command_id = None;
426    for part in rest.split(';') {
427        let (key, value) = part.split_once('=')?;
428        match key {
429            "topic" => topic = Some(value.to_owned()),
430            "subscription_id" => subscription_id = Some(value.to_owned()),
431            "seq" => seq = value.parse::<u64>().ok(),
432            "command_id" => command_id = Some(value.to_owned()),
433            _ => {}
434        }
435    }
436    Some(MessageBusDelivery {
437        topic: topic?,
438        subscription_id: subscription_id?,
439        seq: seq?,
440        command_id: command_id?,
441    })
442}
graphrefly/cqrs/messaging.rs

graphrefly/cqrs/
messaging.rs
