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    /// Deliver a signal to a workflow. The signal row, its `SignalReceived`
12    /// history event and the dispatch arming land in one store transaction,
13    /// so a stored signal is never invisible to the workflow that has to
14    /// react to it.
15    pub async fn send_signal(
16        &self,
17        workflow_id: &str,
18        name: &str,
19        payload: Option<&str>,
20    ) -> Result<()> {
21        let now = timestamp_now();
22        // Parse the incoming payload string back to a JSON value so the
23        // event payload nests cleanly (otherwise the recorded payload is
24        // a stringified JSON-inside-JSON and Lua workers would have to
25        // double-decode).
26        let payload_value: serde_json::Value = payload
27            .and_then(|s| serde_json::from_str(s).ok())
28            .unwrap_or(serde_json::Value::Null);
29        let event_payload =
30            serde_json::json!({ "signal": name, "payload": payload_value }).to_string();
31
32        self.store
33            .deliver_signal(
34                &WorkflowSignal {
35                    id: None,
36                    workflow_id: workflow_id.to_string(),
37                    name: name.to_string(),
38                    payload: payload.map(String::from),
39                    consumed: false,
40                    received_at: now,
41                },
42                &event_payload,
43            )
44            .await?;
45
46        // so the worker can replay and notice the signal in history.
47        self.emit_needs_dispatch(workflow_id).await;
48
49        // Emit so the dashboard can refresh the run's row (signal
50        // count bump, log-tail tick, etc.).
51        let ns = self
52            .store
53            .get_workflow(workflow_id)
54            .await?
55            .map(|w| w.namespace)
56            .unwrap_or_default();
57        self.emit(
58            &ns,
59            WorkflowBusEvent::SignalReceived {
60                workflow_id: workflow_id.to_string(),
61                signal_name: name.to_string(),
62            },
63        )
64        .await;
65
66        Ok(())
67    }
68
69    pub async fn get_events(&self, workflow_id: &str) -> Result<Vec<WorkflowEvent>> {
70        self.store.list_events(workflow_id).await
71    }
72
73    pub async fn get_events_page(
74        &self,
75        workflow_id: &str,
76        cursor: Option<i32>,
77        limit: i64,
78        descending: bool,
79    ) -> Result<Vec<WorkflowEvent>> {
80        self.store
81            .list_events_page(workflow_id, cursor, limit, descending)
82            .await
83    }
84}
assay_workflow/signals.rs

assay_workflow/
signals.rs
