assay_workflow/
signals.rs

1//! Signal and event-history methods.
2
3use anyhow::Result;
4
5use crate::ctx::{WorkflowCtx, timestamp_now};
6use crate::events::WorkflowBusEvent;
7use crate::store::WorkflowStore;
8use crate::types::*;
9
10impl<S: WorkflowStore> WorkflowCtx<S> {
11    pub async fn send_signal(
12        &self,
13        workflow_id: &str,
14        name: &str,
15        payload: Option<&str>,
16    ) -> Result<()> {
17        let now = timestamp_now();
18
19        self.store
20            .send_signal(&WorkflowSignal {
21                id: None,
22                workflow_id: workflow_id.to_string(),
23                name: name.to_string(),
24                payload: payload.map(String::from),
25                consumed: false,
26                received_at: now,
27            })
28            .await?;
29
30        let seq = self.store.get_event_count(workflow_id).await? as i32 + 1;
31        // Parse the incoming payload string back to a JSON value so the
32        // event payload nests cleanly (otherwise the recorded payload is
33        // a stringified JSON-inside-JSON and Lua workers would have to
34        // double-decode).
35        let payload_value: serde_json::Value = payload
36            .and_then(|s| serde_json::from_str(s).ok())
37            .unwrap_or(serde_json::Value::Null);
38        self.store
39            .append_event(&WorkflowEvent {
40                id: None,
41                workflow_id: workflow_id.to_string(),
42                seq,
43                event_type: "SignalReceived".to_string(),
44                payload: Some(
45                    serde_json::json!({ "signal": name, "payload": payload_value }).to_string(),
46                ),
47                timestamp: now,
48            })
49            .await?;
50
51        // so the worker can replay and notice the signal in history. The
52        // helper also emits WorkflowNeedsDispatch on the engine event bus.
53        self.mark_and_emit_needs_dispatch(workflow_id).await?;
54
55        // Emit so the dashboard can refresh the run's row (signal
56        // count bump, log-tail tick, etc.).
57        let ns = self
58            .store
59            .get_workflow(workflow_id)
60            .await?
61            .map(|w| w.namespace)
62            .unwrap_or_default();
63        self.emit(
64            &ns,
65            WorkflowBusEvent::SignalReceived {
66                workflow_id: workflow_id.to_string(),
67                signal_name: name.to_string(),
68            },
69        )
70        .await;
71
72        Ok(())
73    }
74
75    pub async fn get_events(&self, workflow_id: &str) -> Result<Vec<WorkflowEvent>> {
76        self.store.list_events(workflow_id).await
77    }
78}
assay_workflow/signals.rs

assay_workflow/
signals.rs
