graphrefly/process/
messaging.rs

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

graphrefly/process/
messaging.rs
