77 lines
2.6 KiB
Rust
77 lines
2.6 KiB
Rust
//! Live SSE event stream + history endpoints.
|
|
|
|
use std::convert::Infallible;
|
|
|
|
use axum::Json;
|
|
use axum::extract::{Query, State};
|
|
use axum::response::sse::{Event, KeepAlive, Sse};
|
|
use serde::Deserialize;
|
|
use tokio_stream::{Stream, StreamExt, wrappers::BroadcastStream};
|
|
|
|
use super::AppState;
|
|
|
|
/// Query params for the paginated history endpoint.
|
|
#[derive(Debug, Deserialize)]
|
|
pub(super) struct HistoryParams {
|
|
/// Cursor: only return events with sqlite row id < `before`.
|
|
/// Omit for the initial (most-recent) page.
|
|
before: Option<i64>,
|
|
/// Page size (default 100, capped at `HISTORY_CAPACITY`).
|
|
limit: Option<usize>,
|
|
}
|
|
|
|
pub(super) async fn events_history(
|
|
State(state): State<AppState>,
|
|
Query(params): Query<HistoryParams>,
|
|
) -> Json<serde_json::Value> {
|
|
use crate::events::HISTORY_CAPACITY;
|
|
let limit = params.limit.unwrap_or(100).min(HISTORY_CAPACITY);
|
|
let before = params.before;
|
|
let is_initial = before.is_none();
|
|
|
|
// Capture seq *before* the read on initial loads so the SSE dedupe
|
|
// window is "drop buffered events you've already seen in history",
|
|
// never "lose an event that fired between the read and the seq."
|
|
// On paginated loads (`before` is set) seq is not needed.
|
|
let seq = if is_initial {
|
|
Some(state.bus.current_seq())
|
|
} else {
|
|
None
|
|
};
|
|
|
|
let (events, min_id, has_more) = state.bus.history_page(before, limit);
|
|
let mut resp = serde_json::json!({
|
|
"events": events,
|
|
"min_id": min_id,
|
|
"has_more": has_more,
|
|
});
|
|
if let Some(s) = seq {
|
|
resp["seq"] = serde_json::json!(s);
|
|
}
|
|
Json(resp)
|
|
}
|
|
|
|
pub(super) async fn events_stream(
|
|
State(state): State<AppState>,
|
|
) -> Sse<impl Stream<Item = Result<Event, Infallible>>> {
|
|
tracing::info!("sse: client subscribed");
|
|
let rx = state.bus.subscribe();
|
|
// Prime THIS connection with a one-off "hello" so it can clear the
|
|
// connecting placeholder immediately. Injected into this subscriber's own
|
|
// stream rather than emitted to the bus — a bus emit would spam every
|
|
// already-connected client with a spurious note each time anyone opens
|
|
// the stream.
|
|
let hello = Event::default().data(
|
|
serde_json::to_string(&crate::events::LiveEvent::Note {
|
|
text: "live stream attached".into(),
|
|
})
|
|
.unwrap_or_default(),
|
|
);
|
|
let live = BroadcastStream::new(rx).filter_map(|res| {
|
|
let ev = res.ok()?;
|
|
let json = serde_json::to_string(&ev).ok()?;
|
|
Some(Ok(Event::default().data(json)))
|
|
});
|
|
let stream = tokio_stream::once(Ok(hello)).chain(live);
|
|
Sse::new(stream).keep_alive(KeepAlive::default())
|
|
}
|