assay_workflow/
signals.rs

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

assay_workflow/
signals.rs
