assay_workflow/
events.rs

1//! Typed workflow event layer on top of `assay_domain::events`.
2//!
3//! Every state-mutating method in `assay-workflow` that previously
4//! called `ctx.broadcast(...)` or relied on a PG trigger for
5//! `pg_notify` now emits a `WorkflowBusEvent` via `WorkflowEventBus`.
6
7use std::sync::Arc;
8
9use anyhow::Result;
10use serde::{Deserialize, Serialize};
11
12use assay_domain::events::{
13    CursorGoneError, EngineEventBus, Event, EventFilter, NewEvent, Subsystem,
14};
15
16/// Every event kind the workflow subsystem emits. Each variant
17/// serialises to a `workflow_*` / `activity_*` kind string + a JSON
18/// payload. Fields are chosen to be small and self-contained; we never
19/// dump whole rows.
20#[derive(Debug, Clone, Serialize, Deserialize)]
21#[serde(tag = "kind", rename_all = "snake_case")]
22pub enum WorkflowBusEvent {
23    WorkflowCreated {
24        workflow_id: String,
25        workflow_type: String,
26        task_queue: String,
27        status: String,
28    },
29    WorkflowStatusChanged {
30        workflow_id: String,
31        old_status: String,
32        new_status: String,
33        task_queue: String,
34    },
35    WorkflowNeedsDispatch {
36        workflow_id: String,
37        task_queue: String,
38    },
39    WorkflowStarted {
40        workflow_id: String,
41    },
42    WorkflowRunning {
43        workflow_id: String,
44    },
45    WorkflowCompleted {
46        workflow_id: String,
47    },
48    WorkflowFailed {
49        workflow_id: String,
50    },
51    WorkflowCancelled {
52        workflow_id: String,
53    },
54    WorkflowTerminated {
55        workflow_id: String,
56    },
57    ActivityInserted {
58        activity_id: i64,
59        workflow_id: String,
60        task_queue: String,
61        name: String,
62    },
63    ActivityStatusChanged {
64        activity_id: i64,
65        workflow_id: String,
66        old_status: String,
67        new_status: String,
68    },
69    SignalReceived {
70        workflow_id: String,
71        signal_name: String,
72    },
73    TimerFired {
74        workflow_id: String,
75        seq: i32,
76    },
77}
78
79impl WorkflowBusEvent {
80    /// The kind string written to `engine_events.kind`. Matches the
81    /// serde `rename_all = "snake_case"` tag for each variant.
82    pub fn kind(&self) -> &'static str {
83        match self {
84            WorkflowBusEvent::WorkflowCreated { .. } => "workflow_created",
85            WorkflowBusEvent::WorkflowStatusChanged { .. } => "workflow_status_changed",
86            WorkflowBusEvent::WorkflowNeedsDispatch { .. } => "workflow_needs_dispatch",
87            WorkflowBusEvent::WorkflowStarted { .. } => "workflow_started",
88            WorkflowBusEvent::WorkflowRunning { .. } => "workflow_running",
89            WorkflowBusEvent::WorkflowCompleted { .. } => "workflow_completed",
90            WorkflowBusEvent::WorkflowFailed { .. } => "workflow_failed",
91            WorkflowBusEvent::WorkflowCancelled { .. } => "workflow_cancelled",
92            WorkflowBusEvent::WorkflowTerminated { .. } => "workflow_terminated",
93            WorkflowBusEvent::ActivityInserted { .. } => "activity_inserted",
94            WorkflowBusEvent::ActivityStatusChanged { .. } => "activity_status_changed",
95            WorkflowBusEvent::SignalReceived { .. } => "signal_received",
96            WorkflowBusEvent::TimerFired { .. } => "timer_fired",
97        }
98    }
99
100    /// Serialise the variant's fields to JSON for `engine_events.payload`.
101    /// Strips the `kind` serde tag from the payload since the kind
102    /// lives in its own column.
103    pub fn payload(&self) -> serde_json::Value {
104        let v = serde_json::to_value(self).expect("WorkflowBusEvent serialisable");
105        match v {
106            serde_json::Value::Object(mut m) => {
107                m.remove("kind");
108                serde_json::Value::Object(m)
109            }
110            other => other,
111        }
112    }
113}
114
115/// Typed wrapper around an `EngineEventBus`. Per-subsystem wrappers
116/// (workflow, auth, secrets) share the underlying bus instance at the
117/// engine level but produce/consume their own typed events.
118#[derive(Clone)]
119pub struct WorkflowEventBus {
120    inner: Arc<dyn EngineEventBus>,
121}
122
123impl WorkflowEventBus {
124    pub fn new(inner: Arc<dyn EngineEventBus>) -> Self {
125        Self { inner }
126    }
127
128    /// Publish a typed workflow event. `namespace` is the owning
129    /// workflow's namespace.
130    pub async fn publish(&self, namespace: &str, ev: WorkflowBusEvent) -> Result<i64> {
131        let kind = ev.kind();
132        let payload = ev.payload();
133        self.inner
134            .publish_committed(NewEvent {
135                namespace,
136                subsystem: Subsystem::Workflow,
137                kind,
138                payload,
139            })
140            .await
141    }
142
143    /// Read a cursor's worth of events for this namespace (any
144    /// subsystem — SSE uses this for the replay phase).
145    pub async fn read_since(
146        &self,
147        namespace: &str,
148        after: Option<i64>,
149        filter: &EventFilter,
150        limit: u32,
151    ) -> std::result::Result<Vec<Event>, CursorGoneError> {
152        self.inner.read_since(namespace, after, filter, limit).await
153    }
154
155    /// Expose the underlying bus so the scheduler + SSE can subscribe
156    /// at the generic level and filter for any subsystem.
157    pub fn inner(&self) -> Arc<dyn EngineEventBus> {
158        Arc::clone(&self.inner)
159    }
160}
161
162#[cfg(test)]
163mod tests {
164    use super::*;
165
166    #[test]
167    fn kind_tag_stripped_from_payload() {
168        let e = WorkflowBusEvent::WorkflowCreated {
169            workflow_id: "wf-1".into(),
170            workflow_type: "greet".into(),
171            task_queue: "main".into(),
172            status: "PENDING".into(),
173        };
174        assert_eq!(e.kind(), "workflow_created");
175        let payload = e.payload();
176        assert_eq!(payload["workflow_id"], "wf-1");
177        assert_eq!(payload["workflow_type"], "greet");
178        assert!(payload.get("kind").is_none(), "kind tag must be stripped");
179    }
180
181    #[test]
182    fn activity_inserted_payload() {
183        let e = WorkflowBusEvent::ActivityInserted {
184            activity_id: 42,
185            workflow_id: "wf-1".into(),
186            task_queue: "main".into(),
187            name: "send_email".into(),
188        };
189        let p = e.payload();
190        assert_eq!(p["activity_id"], 42);
191        assert_eq!(p["workflow_id"], "wf-1");
192    }
193}
assay_workflow/events.rs

assay_workflow/
events.rs
