crates/gunnar-ui-web/src/feed.rs
//! The live feed, over server-sent events.
//!
//! One subtlety that is easy to get wrong: an `EventSource` reconnects on its
//! own after a network drop, but **not** after an HTTP error status. When the
//! session ends the server answers 401, the browser gives up permanently, and
//! the feed would just stop with no explanation. So `onerror` with a closed
//! ready state is treated as an authentication failure and sends the operator
//! to the login page.
//!
//! # The stream does not start empty any more
//!
//! It used to, and the overview panel said so in words — "nothing has happened
//! since the page was opened" — beside a stats panel reading "Events 147
//! published". The console was blind, not the forge quiet.
//!
//! A connection now begins with the history the BFF still holds, oldest first,
//! and then continues live; the two halves arrive as ordinary `feed` messages
//! and nothing here has to tell them apart. That is deliberate. The join is
//! made exact on the server, under one lock (`gunnar_ui_server::sse`), so this
//! file cannot be handed a duplicate to de-duplicate or a hole to detect —
//! which is the only place the correctness could have lived, since a browser
//! has no way to know what it was not sent.
//!
//! Two consequences worth knowing here. [`KEEP`] is what the console displays
//! and the server's backlog is bounded to the same number, so a fresh
//! connection arrives full rather than filling up over the following hours. And
//! a replayed `gap` or `restart` line bumps [`Feed::reread`] exactly as a live
//! one does — the lists are re-read once more than strictly necessary on such a
//! page open, which is the cheap side of that trade.
use leptos::prelude::*;
use serde::Deserialize;
use wasm_bindgen::prelude::*;
use wasm_bindgen::JsCast;
/// How many lines the console keeps in view.
const KEEP: usize = 200;
/// One line of the feed.
#[derive(Debug, Clone, Deserialize, PartialEq)]
pub struct FeedItem {
pub seq: u64,
pub at_unix_ms: i64,
pub kind: String,
pub repo: Option<String>,
pub summary: String,
pub dropped: Option<u64>,
}
impl FeedItem {
/// True when this line marks events nobody will ever see.
pub fn is_gap(&self) -> bool {
self.kind == "gap"
}
/// True when the feed is not continuous across this line, so the lists it
/// affects have to be re-read rather than patched.
///
/// Two different facts, one consequence. A `gap` is a COUNTED loss: gunnar
/// dropped n events and says which. A `restart` is an UNCOUNTABLE one:
/// gunnar's sequence began again, so the console's cursor means nothing and
/// nobody can say how much happened in between. Both mean "stop trusting
/// the incremental view"; only the first has a number, which is why they
/// are separate kinds and not one with an optional count.
pub fn breaks_continuity(&self) -> bool {
self.is_gap() || self.kind == "restart"
}
}
/// Whether the stream is up.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum Connection {
Connecting,
Live,
/// The stream ended and will not come back on its own.
Ended,
}
/// What a page gets from the feed.
#[derive(Clone, Copy)]
pub struct Feed {
pub items: RwSignal<Vec<FeedItem>>,
pub connection: RwSignal<Connection>,
/// Bumped whenever the feed stops being continuous — a gap, or a restart —
/// so lists know to re-read.
///
/// Either one means the console cannot know what changed while it was not
/// listening, so the only correct response is to ask again.
pub reread: RwSignal<u32>,
}
/// Open a stream for as long as the calling component lives: the reader's
/// whole feed — the server decides whose events that is (an administrator's:
/// the whole server's; anybody else's: their own account's).
pub fn use_feed() -> Feed {
use_feed_at("")
}
/// The same, narrowed to one repository (its Activity tab): `?repo=<ns>/<name>`
/// on the feed's address. The server never widens a feed for it.
pub fn use_repo_feed(namespace: &str, name: &str) -> Feed {
let path = format!("{namespace}/{name}");
use_feed_at(&format!("?repo={}", crate::auth::percent_encode(&path)))
}
fn use_feed_at(query: &str) -> Feed {
let feed = Feed {
items: RwSignal::new(Vec::new()),
connection: RwSignal::new(Connection::Connecting),
reread: RwSignal::new(0),
};
let source = match web_sys::EventSource::new(&format!("{}/api/events{query}", crate::api::base())) {
Ok(source) => source,
Err(_) => {
feed.connection.set(Connection::Ended);
return feed;
}
};
// `open` rather than the first message. A server with nothing to say is the
// normal case for a git forge at three in the morning, and waiting for an
// event to prove the stream works leaves "connecting …" on screen for
// hours, which reads as broken.
let on_open = Closure::<dyn FnMut(web_sys::Event)>::new({
move |_: web_sys::Event| feed.connection.set(Connection::Live)
});
let _ = source.add_event_listener_with_callback("open", on_open.as_ref().unchecked_ref());
let on_message = Closure::<dyn FnMut(web_sys::MessageEvent)>::new({
move |event: web_sys::MessageEvent| {
feed.connection.set(Connection::Live);
let Some(text) = event.data().as_string() else {
return;
};
let Ok(item) = serde_json::from_str::<FeedItem>(&text) else {
return;
};
if item.breaks_continuity() {
feed.reread.update(|n| *n = n.wrapping_add(1));
}
feed.items.update(|items| {
items.insert(0, item);
items.truncate(KEEP);
});
}
});
let _ = source.add_event_listener_with_callback("feed", on_message.as_ref().unchecked_ref());
// The server sends this before closing a stream whose session ended, which
// is the case EventSource itself cannot report usefully.
let on_expired = Closure::<dyn FnMut(web_sys::MessageEvent)>::new({
move |_: web_sys::MessageEvent| {
feed.connection.set(Connection::Ended);
go_to_login();
}
});
let _ = source.add_event_listener_with_callback("expired", on_expired.as_ref().unchecked_ref());
let on_error = Closure::<dyn FnMut(web_sys::Event)>::new({
let source = source.clone();
move |_: web_sys::Event| {
// CLOSED means the browser will not retry: an HTTP error status,
// which for this endpoint means the session is gone.
if source.ready_state() == web_sys::EventSource::CLOSED {
feed.connection.set(Connection::Ended);
go_to_login();
} else {
feed.connection.set(Connection::Connecting);
}
}
});
let _ = source.add_event_listener_with_callback("error", on_error.as_ref().unchecked_ref());
// `on_cleanup` wants something it can send between threads, and neither an
// `EventSource` nor a `Closure` is. Nothing here ever leaves the browser's
// single thread, so the wrapper is the honest way to say so — and keeping
// the closures owned here rather than `forget`ing them means a page that is
// opened and closed repeatedly does not leak one set per visit.
let owned = send_wrapper::SendWrapper::new((source, on_open, on_message, on_expired, on_error));
on_cleanup(move || {
let (source, ..) = &*owned;
source.close();
drop(owned);
});
feed
}
/// Send the operator to log in again, keeping where they were.
fn go_to_login() {
let Some(window) = web_sys::window() else {
return;
};
let here = window
.location()
.pathname()
.unwrap_or_else(|_| "/".to_owned());
if here == "/login" {
return;
}
let _ = window.location().set_href(&crate::auth::login_url(&here));
}