assay_workflow/api/
events.rs

1//! SSE endpoint for the engine-events outbox.
2//!
3//! Endpoint: `GET /api/v1/engine/workflow/events/stream`
4//!   Query params:
5//!     ?ns=<namespace>                       default "main"
6//!     ?subsystem=<workflow|auth|secrets|system>   repeatable
7//!     ?workflow_id=<id>                     optional
8//!     ?kind=<kind>                          repeatable
9//!   Header: `Last-Event-ID: <i64>`          optional cursor for replay
10//!
11//! Behaviour:
12//!   1. Replay phase reads `engine_events` since `Last-Event-ID` (or
13//!      from the beginning if absent) and emits matching frames.
14//!   2. If the cursor is older than the namespace's oldest retained
15//!      id, returns HTTP 410 Gone so the client knows to snapshot +
16//!      resync.
17//!   3. Live phase subscribes to the node-local broadcast; emits the
18//!      frames that pass the filter.
19//!   4. `broadcast::Lagged` force-closes the stream — the client
20//!      reconnects with the last id it saw and replays the gap via
21//!      `engine_events`.
22
23use std::convert::Infallible;
24use std::sync::Arc;
25use std::time::Duration;
26
27use assay_domain::events::{EventFilter, Subsystem};
28use axum::Router;
29use axum::extract::{Query, State};
30use axum::http::{HeaderMap, StatusCode};
31use axum::response::sse::{Event as SseEvent, KeepAlive, Sse};
32use axum::response::{IntoResponse, Response};
33use axum::routing::get;
34use futures_util::StreamExt;
35use futures_util::stream::{self, Stream};
36use serde::Deserialize;
37use tokio_stream::wrappers::BroadcastStream;
38
39use crate::ctx::WorkflowCtx;
40use crate::store::WorkflowStore;
41
42const REPLAY_PAGE_LIMIT: u32 = 500;
43
44#[derive(Deserialize, Default)]
45struct StreamQuery {
46    #[serde(default)]
47    ns: Option<String>,
48    #[serde(default)]
49    subsystem: Option<Vec<String>>,
50    #[serde(default)]
51    workflow_id: Option<String>,
52    #[serde(default)]
53    kind: Option<Vec<String>>,
54}
55
56pub fn router<S: WorkflowStore + 'static>() -> Router<Arc<WorkflowCtx<S>>> {
57    Router::new().route("/events/stream", get(event_stream))
58}
59
60async fn event_stream<S: WorkflowStore>(
61    State(state): State<Arc<WorkflowCtx<S>>>,
62    Query(q): Query<StreamQuery>,
63    headers: HeaderMap,
64) -> Response {
65    let Some(wf_bus) = state.bus() else {
66        return (StatusCode::SERVICE_UNAVAILABLE, "event bus not configured").into_response();
67    };
68
69    let namespace = q.ns.clone().unwrap_or_else(|| "main".to_string());
70    let filter = EventFilter {
71        subsystems: q
72            .subsystem
73            .clone()
74            .unwrap_or_default()
75            .into_iter()
76            .filter_map(|s| match s.as_str() {
77                "workflow" => Some(Subsystem::Workflow),
78                "auth" => Some(Subsystem::Auth),
79                "secrets" => Some(Subsystem::Secrets),
80                "system" => Some(Subsystem::System),
81                _ => None,
82            })
83            .collect(),
84        kinds: q.kind.clone().unwrap_or_default(),
85        workflow_id: q.workflow_id.clone(),
86    };
87
88    // EventSource spec: browser auto-sends `Last-Event-ID` on reconnect.
89    let last_id: Option<i64> = headers
90        .get("last-event-id")
91        .and_then(|h| h.to_str().ok())
92        .and_then(|s| s.parse().ok());
93
94    let inner = wf_bus.inner();
95
96    let replay_events = match inner
97        .read_since(&namespace, last_id, &filter, REPLAY_PAGE_LIMIT)
98        .await
99    {
100        Ok(evs) => evs,
101        Err(gone) => {
102            return (
103                StatusCode::GONE,
104                format!(
105                    "cursor {} older than retention (oldest {}); resync via point queries then reconnect without Last-Event-ID",
106                    gone.after, gone.oldest
107                ),
108            )
109                .into_response();
110        }
111    };
112
113    let replay_stream = stream::iter(
114        replay_events
115            .into_iter()
116            .map(|e| Ok::<_, Infallible>(event_to_sse(&e))),
117    );
118
119    // Subscribe for the live phase. sqlite's broadcast is global; pg's
120    // is per-node but fed by a per-namespace LISTEN bridge. Filter
121    // client-side on namespace + caller-supplied predicates.
122    let rx = inner.subscribe(&namespace);
123    let ns_for_live = namespace.clone();
124    let filter_for_live = filter.clone();
125    let live_stream = BroadcastStream::new(rx).filter_map(move |result| {
126        let ns = ns_for_live.clone();
127        let f = filter_for_live.clone();
128        async move {
129            match result {
130                Ok(arc_ev) => {
131                    if arc_ev.namespace != ns {
132                        return None;
133                    }
134                    if !f.matches(&arc_ev) {
135                        return None;
136                    }
137                    Some(Ok::<_, Infallible>(event_to_sse(&arc_ev)))
138                }
139                Err(tokio_stream::wrappers::errors::BroadcastStreamRecvError::Lagged(n)) => {
140                    tracing::warn!(lagged = n, "SSE client lagged; forcing close");
141                    None
142                }
143            }
144        }
145    });
146
147    let combined: std::pin::Pin<Box<dyn Stream<Item = Result<SseEvent, Infallible>> + Send>> =
148        Box::pin(replay_stream.chain(live_stream));
149    Sse::new(combined)
150        .keep_alive(KeepAlive::new().interval(Duration::from_secs(15)))
151        .into_response()
152}
153
154fn event_to_sse(e: &assay_domain::events::Event) -> SseEvent {
155    let data = serde_json::json!({
156        "id": e.id,
157        "ts": e.ts,
158        "namespace": e.namespace,
159        "subsystem": e.subsystem,
160        "kind": e.kind,
161        "payload": e.payload,
162    })
163    .to_string();
164    SseEvent::default()
165        .id(e.id.to_string())
166        .event(&e.kind)
167        .data(data)
168}
assay_workflow/api/events.rs

assay_workflow/api/
events.rs
