assay_workflow/
signals.rs

1//! Signal and event-history methods.
2
3use anyhow::Result;
4
5use crate::ctx::{timestamp_now, WorkflowCtx};
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        // Phase 9: a workflow waiting on this signal needs to be re-dispatched
52        // so the worker can replay and notice the signal in history. The
53        // helper also emits WorkflowNeedsDispatch on the engine event bus.
54        self.mark_and_emit_needs_dispatch(workflow_id).await?;
55
56        // Emit so the dashboard can refresh the run's row (signal
57        // count bump, log-tail tick, etc.).
58        let ns = self
59            .store
60            .get_workflow(workflow_id)
61            .await?
62            .map(|w| w.namespace)
63            .unwrap_or_default();
64        self.emit(
65            &ns,
66            WorkflowBusEvent::SignalReceived {
67                workflow_id: workflow_id.to_string(),
68                signal_name: name.to_string(),
69            },
70        )
71        .await;
72
73        Ok(())
74    }
75
76    pub async fn get_events(&self, workflow_id: &str) -> Result<Vec<WorkflowEvent>> {
77        self.store.list_events(workflow_id).await
78    }
79}
assay_workflow/signals.rs

assay_workflow/
signals.rs
