Tahinli/forge

crates/forge-web/assets/nav.jsmodified+153 -4

		@@ -9,6 +9,8 @@
9	9
10	10	var root = document.documentElement;
11	11	var latest = 0; // id of the newest request; older responses are dropped
	12	+ var userBusy = 0; // navigations and form posts on the wire (not live refreshes)
	13	+ var liveDirty = false; // a live change came while a navigation was on the wire
12	14	var inFlight = 0; // requests on the wire, drives aria-busy
13	15	var pending = new WeakSet(); // forms with a request on the wire: a second submit is ignored
14	16	var shown = location.pathname + location.search;
		@@ -142,6 +144,8 @@
142	144	old.replaceWith(s);
143	145	});
144	146	if (kept) { restoreFields(kept); }
	147	+ syncUnread();
	148	+ syncLive();
145	149	}
146	150
147	151	function focusMain() {
		@@ -161,19 +165,28 @@
161	165	}
162	166
163	167	// how: 'nav' (link or GET form: push, top), 'pop' (history: push nothing, restore scroll),
164		- // 'post' (form post: replace, keep scroll), 'query' (in place: replace, keep scroll).
	168	+ // 'post' (form post: replace, keep scroll), 'query' (in place: replace, keep scroll), 'live' (a
	169	+ // live page's change: no history entry, keep scroll, no fragment jump).
165	170	function swap(html, dest, how, keepFields) {
166	171	var doc = new DOMParser().parseFromString(html, 'text/html');
167	172	if (!doc.body) { location.assign(dest.href); return; }
168	173	var x = window.scrollX, y = window.scrollY;
169	174	var saved = history.state;
	175	+ // A live swap keeps the log tail where the reader had it: at the end it stays at the end.
	176	+ var tail = how === 'live' ? document.querySelector('[data-live-log]') : null;
	177	+ var tailAt = tail ? { top: tail.scrollTop, end: tail.scrollTop + tail.clientHeight >= tail.scrollHeight - 8 } : null;
	178	+ closeDropdowns(null);
170	179	adopt(doc, keepFields);
	180	+ if (tailAt) {
	181	+ var nextTail = document.querySelector('[data-live-log]');
	182	+ if (nextTail) { nextTail.scrollTop = tailAt.end ? nextTail.scrollHeight : tailAt.top; }
	183	+ }
171	184	if (how === 'nav') { history.pushState(null, '', dest.href); }
172	185	else if (how === 'post' \|\| how === 'query') { history.replaceState(null, '', dest.href); }
173	186	shown = dest.pathname + dest.search;
174	187	if (how === 'nav' \|\| how === 'pop') { focusMain(); }
175		- if (placeHash(dest.hash)) { return; }
176		- if (how === 'post' \|\| how === 'query') { window.scrollTo(x, y); }
	188	+ if (how !== 'live' && placeHash(dest.hash)) { return; }
	189	+ if (how === 'post' \|\| how === 'query' \|\| how === 'live') { window.scrollTo(x, y); }
177	190	else if (how === 'pop') { window.scrollTo(0, saved && typeof saved.y === 'number' ? saved.y : 0); }
178	191	else { window.scrollTo(0, 0); }
179	192	}
		@@ -194,6 +207,8 @@
194	207	// how, keep (keep fields on a re-render of the same path), post, and the form for a POST or GET form.
195	208	function load(o) {
196	209	var mark = ++latest;
	210	+ var user = o.how !== 'live';
	211	+ if (user) { userBusy += 1; }
197	212	busyStart();
198	213	fetch(o.target.href, o.init).then(function (r) {
199	214	var final = new URL(r.url \|\| o.target.href);
		@@ -209,11 +224,15 @@
209	224	dest.hash = o.dest.hash;
210	225	swap(res.html, dest, o.how, !!o.keep && res.url.pathname === o.target.pathname);
211	226	}).catch(function () {
212		- if (mark !== latest) { return; }
	227	+ if (mark !== latest \|\| o.how === 'live') { return; }
213	228	if (o.post) { o.form.submit(); } else { location.assign(o.dest.href); }
214	229	}).then(function () {
215	230	busyEnd();
216	231	if (o.form) { pending.delete(o.form); }
	232	+ if (user) {
	233	+ userBusy -= 1;
	234	+ if (liveDirty) { refreshLive(); }
	235	+ }
217	236	});
218	237	}
219	238
		@@ -221,6 +240,39 @@
221	240	try { history.replaceState({ y: window.scrollY }, '', location.href); } catch (err) { }
222	241	}
223	242
	243	+ // Dropdown menus: the <details> that float over the page (avatar menu, branch selector, clone
	244	+ // popover, phone menu). Inline disclosures are not listed here and keep their state.
	245	+ var DROPDOWN = 'details.menu, details.ref-select, details.clone, details.nav-toggle';
	246	+
	247	+ // Closes every open dropdown except `keep` (pass null to close them all).
	248	+ function closeDropdowns(keep) {
	249	+ document.querySelectorAll(DROPDOWN).forEach(function (d) {
	250	+ if (d.open && d !== keep) { d.open = false; }
	251	+ });
	252	+ }
	253	+
	254	+ // A click outside an open dropdown closes it. A click inside one (a select, a form field) keeps
	255	+ // it open. Opening a dropdown by its summary closes the others: the browser toggles it after this.
	256	+ document.addEventListener('click', function (e) {
	257	+ var inside = e.target && e.target.closest ? e.target.closest(DROPDOWN) : null;
	258	+ closeDropdowns(inside);
	259	+ }, false);
	260	+
	261	+ // Escape closes the open dropdowns and returns focus to the summary of the one that held it.
	262	+ document.addEventListener('keydown', function (e) {
	263	+ if (e.key !== 'Escape') { return; }
	264	+ var back = null;
	265	+ document.querySelectorAll(DROPDOWN).forEach(function (d) {
	266	+ if (!d.open) { return; }
	267	+ if (d.contains(document.activeElement)) { back = d; }
	268	+ d.open = false;
	269	+ });
	270	+ if (back) {
	271	+ var summary = back.querySelector('summary');
	272	+ if (summary) { summary.focus(); }
	273	+ }
	274	+ }, false);
	275	+
224	276	document.addEventListener('click', function (e) {
225	277	if (e.defaultPrevented \|\| e.button !== 0 \|\| e.metaKey \|\| e.ctrlKey \|\| e.shiftKey \|\| e.altKey) { return; }
226	278	var a = e.target && e.target.closest ? e.target.closest('a[href]') : null;
		@@ -297,6 +349,103 @@
297	349
298	350	// A filtered view asked for in place: the URL is replaced (no history entry per change), fields
299	351	// and caret come through the swap, and the scroll stays put.
	352	+ // Live unread count. The bell is re-rendered by every swap, so it is looked up on each update. One
	353	+ // EventSource serves the page and stays open across swaps. It is opened only while a bell is shown
	354	+ // (signed-in pages) and closed when the bell is gone. The server pushes each change; nothing polls.
	355	+ var unreadStream = null;
	356	+
	357	+ function bell() { return document.querySelector('a.bell'); }
	358	+
	359	+ // Shows the count the way the server renders it: a `.bell-count` span with the number, or no span at zero.
	360	+ function setUnread(n) {
	361	+ var el = bell();
	362	+ if (!el) { return; }
	363	+ var badge = el.querySelector('.bell-count');
	364	+ if (n > 0) {
	365	+ if (!badge) {
	366	+ badge = document.createElement('span');
	367	+ badge.className = 'bell-count';
	368	+ el.appendChild(badge);
	369	+ }
	370	+ badge.textContent = String(n);
	371	+ } else if (badge) {
	372	+ badge.remove();
	373	+ }
	374	+ }
	375	+
	376	+ function syncUnread() {
	377	+ if (!bell()) {
	378	+ if (unreadStream) { unreadStream.close(); unreadStream = null; }
	379	+ return;
	380	+ }
	381	+ if (unreadStream \|\| !window.EventSource) { return; }
	382	+ unreadStream = new EventSource('/notifications/stream');
	383	+ unreadStream.addEventListener('unread', function (e) {
	384	+ setUnread(parseInt(e.data, 10) \|\| 0);
	385	+ });
	386	+ }
	387	+
	388	+ syncUnread();
	389	+
	390	+ // Live pages. A run or job page that is not finished marks its wrapper with data-live, the URL of
	391	+ // its event stream. One EventSource follows it. A `changed` event re-fetches the page and swaps it in
	392	+ // with no history entry and the scroll kept. A job page also has a log tail (data-live-log) whose
	393	+ // `log` events append the new bytes as they come. The stream opens at the tail's data-offset, and
	394	+ // each swap brings a fresh offset, so the stream is opened again after every swap. The server ends
	395	+ // the stream once the status is terminal, and the swap that shows that page (no data-live) closes it.
	396	+ var liveStream = null;
	397	+ var liveUrl = null;
	398	+
	399	+ function closeLive() {
	400	+ if (liveStream) { liveStream.close(); }
	401	+ liveStream = null;
	402	+ liveUrl = null;
	403	+ liveDirty = false;
	404	+ }
	405	+
	406	+ function syncLive() {
	407	+ var el = document.querySelector('[data-live]');
	408	+ var url = el ? el.getAttribute('data-live') : null;
	409	+ var tail = document.querySelector('[data-live-log]');
	410	+ if (url && tail) {
	411	+ url += (url.indexOf('?') < 0 ? '?' : '&') + 'from=' + encodeURIComponent(tail.getAttribute('data-offset') \|\| '0');
	412	+ }
	413	+ if (!url \|\| !window.EventSource) { closeLive(); return; }
	414	+ if (liveStream && liveUrl === url) { return; }
	415	+ closeLive();
	416	+ liveUrl = url;
	417	+ liveStream = new EventSource(url);
	418	+ liveStream.addEventListener('changed', function (e) {
	419	+ // A job page's stream opens with the status the page already shows: nothing to re-render then.
	420	+ var wrap = document.querySelector('[data-live]');
	421	+ if (document.querySelector('[data-live-log]') && wrap && wrap.getAttribute('data-live-state') === e.data) { return; }
	422	+ refreshLive();
	423	+ });
	424	+ liveStream.addEventListener('log', function (e) { appendLog(e.data); });
	425	+ }
	426	+
	427	+ // Appends streamed log text to the tail as a text node, never as HTML. The tail follows the end
	428	+ // only while the reader is already at the end.
	429	+ function appendLog(text) {
	430	+ var pre = document.querySelector('[data-live-log]');
	431	+ if (!pre \|\| !text) { return; }
	432	+ var atEnd = pre.scrollTop + pre.clientHeight >= pre.scrollHeight - 8;
	433	+ pre.appendChild(document.createTextNode(text));
	434	+ if (atEnd) { pre.scrollTop = pre.scrollHeight; }
	435	+ }
	436	+
	437	+ // A change re-renders the current page. It waits while a navigation is on the wire, since that
	438	+ // response could be superseded: the change is kept and sent after that request completes.
	439	+ function refreshLive() {
	440	+ if (!liveStream) { return; }
	441	+ if (userBusy > 0) { liveDirty = true; return; }
	442	+ liveDirty = false;
	443	+ var here = new URL(location.href);
	444	+ load({ target: withoutHash(here), dest: here, init: GET_INIT, how: 'live', keep: true, post: false, form: null });
	445	+ }
	446	+
	447	+ syncLive();
	448	+
300	449	window.__forgeQuery = function (href) {
301	450	var url = pageUrl(href);
302	451	if (!url) { location.assign(href); return; }

crates/forge-web/src/ci/live.rsadded+163 -0

		@@ -0,0 +1,163 @@
	1	+//! Change signals for CI jobs and runs. [`CiLive`] keeps one generation counter per job and one per
	2	+//! run. A write that changes a job or a run bumps its counter, and every subscriber wakes on the
	3	+//! change. The counter carries no data: a subscriber reads the database or the log file again when
	4	+//! it wakes. Nothing polls. A signal with no receivers left is removed, so the maps hold only the
	5	+//! jobs and runs somebody is watching. The hub is in-process only, like [`NotifyHub`].
	6	+//!
	7	+//! [`NotifyHub`]: crate::notify_hub::NotifyHub
	8	+
	9	+use std::collections::HashMap;
	10	+use std::sync::{Mutex, MutexGuard, PoisonError};
	11	+
	12	+use tokio::sync::watch;
	13	+
	14	+/// The change counters of CI jobs and runs, each keyed by its id.
	15	+#[derive(Default)]
	16	+pub struct CiLive {
	17	+ jobs: Signals,
	18	+ runs: Signals,
	19	+}
	20	+
	21	+impl CiLive {
	22	+ /// Subscribes to changes of job `job_id`. The receiver starts as seen, so a bump after this
	23	+ /// call wakes it. Subscribe before the state you read, so a change in between is not lost.
	24	+ pub fn subscribe_job(&self, job_id: &str) -> watch::Receiver<u64> {
	25	+ self.jobs.subscribe(job_id)
	26	+ }
	27	+
	28	+ /// Subscribes to changes of run `run_id`. Same rules as [`CiLive::subscribe_job`].
	29	+ // No caller yet: the run status stream that waits on this is the next change.
	30	+ #[allow(dead_code)]
	31	+ pub fn subscribe_run(&self, run_id: &str) -> watch::Receiver<u64> {
	32	+ self.runs.subscribe(run_id)
	33	+ }
	34	+
	35	+ /// Wakes the subscribers of job `job_id`: its log grew or its state changed. Does nothing when
	36	+ /// nobody is subscribed.
	37	+ pub fn bump_job(&self, job_id: &str) {
	38	+ self.jobs.bump(job_id);
	39	+ }
	40	+
	41	+ /// Wakes the subscribers of run `run_id`: one of its jobs or its own status changed.
	42	+ pub fn bump_run(&self, run_id: &str) {
	43	+ self.runs.bump(run_id);
	44	+ }
	45	+
	46	+ /// Wakes the subscribers of job `job_id` once more and drops its signal. Call it when the job
	47	+ /// has reached a terminal state: no change follows, so the signal is not kept. A subscriber
	48	+ /// that waits on a closed signal wakes once and then sees it closed.
	49	+ pub fn close_job(&self, job_id: &str) {
	50	+ self.jobs.close(job_id);
	51	+ }
	52	+}
	53	+
	54	+/// One keyed set of counters. A key has an entry only while it has a receiver or a bump is due.
	55	+#[derive(Default)]
	56	+struct Signals {
	57	+ counters: Mutex<HashMap<String, watch::Sender<u64>>>,
	58	+}
	59	+
	60	+impl Signals {
	61	+ fn counters(&self) -> MutexGuard<'_, HashMap<String, watch::Sender<u64>>> {
	62	+ self.counters.lock().unwrap_or_else(PoisonError::into_inner)
	63	+ }
	64	+
	65	+ fn subscribe(&self, key: &str) -> watch::Receiver<u64> {
	66	+ self.counters()
	67	+ .entry(key.to_owned())
	68	+ .or_insert_with(\|\| watch::channel(0).0)
	69	+ .subscribe()
	70	+ }
	71	+
	72	+ /// Counts one change. Without a subscriber there is no entry, so the bump is a no-op. An entry
	73	+ /// whose last receiver is gone is removed here.
	74	+ fn bump(&self, key: &str) {
	75	+ let mut counters = self.counters();
	76	+ let Some(tx) = counters.get(key) else {
	77	+ return;
	78	+ };
	79	+ tx.send_modify(\|generation\| *generation = generation.wrapping_add(1));
	80	+ if tx.receiver_count() == 0 {
	81	+ counters.remove(key);
	82	+ }
	83	+ }
	84	+
	85	+ /// Counts one last change and removes the entry. The sender is dropped afterwards, so each
	86	+ /// subscriber wakes once more and then finds the signal closed.
	87	+ fn close(&self, key: &str) {
	88	+ let removed = self.counters().remove(key);
	89	+ if let Some(tx) = removed {
	90	+ tx.send_modify(\|generation\| *generation = generation.wrapping_add(1));
	91	+ }
	92	+ }
	93	+}
	94	+
	95	+#[cfg(test)]
	96	+mod tests {
	97	+ use std::time::Duration;
	98	+
	99	+ use super::*;
	100	+
	101	+ #[tokio::test]
	102	+ async fn a_bump_wakes_only_the_subscribers_of_that_job() {
	103	+ let live = CiLive::default();
	104	+ let mut job = live.subscribe_job("job-1");
	105	+ let other = live.subscribe_job("job-2");
	106	+ let run = live.subscribe_run("run-1");
	107	+
	108	+ live.bump_job("job-1");
	109	+
	110	+ tokio::time::timeout(Duration::from_secs(1), job.changed())
	111	+ .await
	112	+ .expect("woken")
	113	+ .expect("open");
	114	+ assert_eq!(*job.borrow_and_update(), 1);
	115	+ assert!(!other.has_changed().expect("open"));
	116	+ assert!(!run.has_changed().expect("open"));
	117	+ }
	118	+
	119	+ #[tokio::test]
	120	+ async fn a_bump_before_subscribing_is_not_seen_and_a_bump_after_is() {
	121	+ let live = CiLive::default();
	122	+ let early = live.subscribe_run("run-1");
	123	+ live.bump_run("run-1");
	124	+ let mut late = live.subscribe_run("run-1");
	125	+ assert!(early.has_changed().expect("open"));
	126	+ assert!(!late.has_changed().expect("open"));
	127	+ live.bump_run("run-1");
	128	+ assert!(late.has_changed().expect("open"));
	129	+ late.changed().await.expect("open");
	130	+ assert_eq!(*late.borrow_and_update(), 2);
	131	+ }
	132	+
	133	+ #[test]
	134	+ fn a_bump_without_subscribers_keeps_no_entry() {
	135	+ let live = CiLive::default();
	136	+ live.bump_job("job-1");
	137	+ live.bump_run("run-1");
	138	+ assert!(live.jobs.counters().is_empty());
	139	+ assert!(live.runs.counters().is_empty());
	140	+ }
	141	+
	142	+ #[test]
	143	+ fn a_bump_drops_the_entry_once_its_last_receiver_is_gone() {
	144	+ let live = CiLive::default();
	145	+ let rx = live.subscribe_job("job-1");
	146	+ assert_eq!(live.jobs.counters().len(), 1);
	147	+ drop(rx);
	148	+ live.bump_job("job-1");
	149	+ assert!(live.jobs.counters().is_empty());
	150	+ }
	151	+
	152	+ #[tokio::test]
	153	+ async fn close_wakes_the_subscriber_once_and_then_ends_the_signal() {
	154	+ let live = CiLive::default();
	155	+ let mut rx = live.subscribe_job("job-1");
	156	+ live.close_job("job-1");
	157	+ assert!(live.jobs.counters().is_empty());
	158	+
	159	+ rx.changed().await.expect("the last change is delivered");
	160	+ assert_eq!(*rx.borrow_and_update(), 1);
	161	+ assert!(rx.changed().await.is_err());
	162	+ }
	163	+}

crates/forge-web/src/i18n/misc.rsmodified+23 -4

		@@ -80,10 +80,12 @@ pub enum K {
80	80	// Notifications: subscription box and watch control.
81	81	Subscribe,
82	82	Unsubscribe,
83		- NotificationLevel,
84	83	OptParticipating,
85	84	OptWatching,
86	85	OptIgnoring,
	86	+ WatchDescParticipating,
	87	+ WatchDescWatching,
	88	+ WatchDescIgnoring,
87	89	// Repo security page.
88	90	SecurityOff,
89	91	LockfilesRead,
		@@ -140,6 +142,9 @@ pub enum K {
140	142	BreadcrumbNav,
141	143	ApplyTheme,
142	144	RefLabel,
	145	+ // Top-bar user menu: the viewer's public profile and the settings editor.
	146	+ YourProfile,
	147	+ EditProfile,
143	148	}
144	149
145	150	/// Every variant, for the coverage test below.
		@@ -207,10 +212,12 @@ const ALL: &[K] = &[
207	212	K::Older,
208	213	K::Subscribe,
209	214	K::Unsubscribe,
210		- K::NotificationLevel,
211	215	K::OptParticipating,
212	216	K::OptWatching,
213	217	K::OptIgnoring,
	218	+ K::WatchDescParticipating,
	219	+ K::WatchDescWatching,
	220	+ K::WatchDescIgnoring,
214	221	K::SecurityOff,
215	222	K::LockfilesRead,
216	223	K::NoAlerts,
		@@ -262,6 +269,8 @@ const ALL: &[K] = &[
262	269	K::BreadcrumbNav,
263	270	K::ApplyTheme,
264	271	K::RefLabel,
	272	+ K::YourProfile,
	273	+ K::EditProfile,
265	274	];
266	275
267	276	pub fn t(lang: Lang, key: K) -> &'static str {
		@@ -400,14 +409,20 @@ pub fn t(lang: Lang, key: K) -> &'static str {
400	409	(Subscribe, Tr) => "Abone ol",
401	410	(Unsubscribe, En) => "Unsubscribe",
402	411	(Unsubscribe, Tr) => "Abonelikten çık",
403		- (NotificationLevel, En) => "Notification level",
404		- (NotificationLevel, Tr) => "Bildirim düzeyi",
405	412	(OptParticipating, En) => "Participating",
406	413	(OptParticipating, Tr) => "Katılımcı",
407	414	(OptWatching, En) => "Watching",
408	415	(OptWatching, Tr) => "İzleniyor",
409	416	(OptIgnoring, En) => "Ignoring",
410	417	(OptIgnoring, Tr) => "Yok sayılıyor",
	418	+ (WatchDescParticipating, En) => "Only threads you start, join or are mentioned in.",
	419	+ (WatchDescParticipating, Tr) => {
	420	+ "Yalnızca başlattığın, katıldığın veya adının geçtiği konular."
	421	+ }
	422	+ (WatchDescWatching, En) => "All activity in this repository.",
	423	+ (WatchDescWatching, Tr) => "Bu depodaki tüm etkinlik.",
	424	+ (WatchDescIgnoring, En) => "Nothing from this repository, except mentions.",
	425	+ (WatchDescIgnoring, Tr) => "Bu depodan hiçbir şey, yalnızca adınızın geçtikleri hariç.",
411	426	(SecurityOff, En) => {
412	427	"Security advisories are off on this instance, so no lockfile is scanned. An administrator can turn them on with the [advisories] section of the config."
413	428	}
		@@ -526,6 +541,10 @@ pub fn t(lang: Lang, key: K) -> &'static str {
526	541	(ApplyTheme, Tr) => "Uygula",
527	542	(RefLabel, En) => "Ref",
528	543	(RefLabel, Tr) => "Referans",
	544	+ (YourProfile, En) => "Your profile",
	545	+ (YourProfile, Tr) => "Profilin",
	546	+ (EditProfile, En) => "Edit profile",
	547	+ (EditProfile, Tr) => "Profili düzenle",
529	548	}
530	549	}
531	550

crates/forge-web/src/notify.rsmodified+5 -2

		@@ -17,6 +17,7 @@ use crate::mail::{
17	17	self,
18	18	templates::{self, Verb, Why},
19	19	};
	20	+use crate::notify_hub::publish_unread;
20	21	use crate::render::markdown::mentioned_usernames;
21	22	use crate::state::AppState;
22	23
		@@ -181,6 +182,7 @@ async fn notify_advisory_admins(state: &AppState, repo: &Repo, opened: usize) ->
181	182	.store
182	183	.notify_new_alerts(&admin, &repo.id, count)
183	184	.await?;
	185	+ publish_unread(state, &admin).await;
184	186	}
185	187	Ok(())
186	188	}
		@@ -365,7 +367,7 @@ async fn run(
365	367	}
366	368
367	369	for (user_id, reason) in recipients {
368		- if let Err(e) = store
	370	+ match store
369	371	.upsert_notification(
370	372	&user_id,
371	373	&repo.id,
		@@ -376,7 +378,8 @@ async fn run(
376	378	)
377	379	.await
378	380	{
379		- tracing::error!(error = %e, user_id, issue_id, "notify: inbox write failed");
	381	+ Ok(()) => publish_unread(state, &user_id).await,
	382	+ Err(e) => tracing::error!(error = %e, user_id, issue_id, "notify: inbox write failed"),
380	383	}
381	384	if let Some(plan) = &plan
382	385	&& let Err(e) = queue_email(state, plan, &user_id, reason).await

crates/forge-web/src/notify_hub.rsadded+202 -0

		@@ -0,0 +1,202 @@
	1	+//! Live unread counts for the bell. [`NotifyHub`] keeps one `watch` channel per signed-in user who
	2	+//! has an open `GET /notifications/stream`. A write that changes a user's inbox publishes the new
	3	+//! count, and each open stream of that user sends it. Nothing polls: a stream waits on its channel.
	4	+//! The hub is in-process only, so a second server process would not see the change.
	5	+//!
	6	+//! A channel with no receivers left is removed, so the map holds only users who are connected.
	7	+
	8	+use std::collections::HashMap;
	9	+use std::sync::{Arc, Mutex, MutexGuard, PoisonError};
	10	+
	11	+use tokio::sync::watch;
	12	+
	13	+use crate::state::AppState;
	14	+
	15	+/// The users with an open stream, each with the channel that carries its unread count.
	16	+pub struct NotifyHub {
	17	+ users: Mutex<HashMap<String, watch::Sender<u64>>>,
	18	+ /// Set once on server shutdown. Every open stream ends when it sees this.
	19	+ closing: watch::Sender<bool>,
	20	+}
	21	+
	22	+impl NotifyHub {
	23	+ pub fn new() -> Self {
	24	+ Self {
	25	+ users: Mutex::new(HashMap::new()),
	26	+ closing: watch::channel(false).0,
	27	+ }
	28	+ }
	29	+
	30	+ fn users(&self) -> MutexGuard<'_, HashMap<String, watch::Sender<u64>>> {
	31	+ self.users.lock().unwrap_or_else(PoisonError::into_inner)
	32	+ }
	33	+
	34	+ /// Opens a subscription to the counts published for `user_id`. The first count is not sent
	35	+ /// here: the caller reads it from the store after subscribing, so a change in between is
	36	+ /// seen by the subscription and sent again rather than lost.
	37	+ pub fn subscribe(self: &Arc<Self>, user_id: &str) -> Subscription {
	38	+ let rx = self
	39	+ .users()
	40	+ .entry(user_id.to_owned())
	41	+ .or_insert_with(\|\| watch::channel(0).0)
	42	+ .subscribe();
	43	+ Subscription {
	44	+ rx: Some(rx),
	45	+ closing: self.closing.subscribe(),
	46	+ hub: Arc::clone(self),
	47	+ user_id: user_id.to_owned(),
	48	+ }
	49	+ }
	50	+
	51	+ /// True when `user_id` has an open stream. A write skips the count query for anyone else.
	52	+ pub fn is_watched(&self, user_id: &str) -> bool {
	53	+ self.users()
	54	+ .get(user_id)
	55	+ .is_some_and(\|tx\| tx.receiver_count() > 0)
	56	+ }
	57	+
	58	+ /// Sends `count` to every open stream of `user_id`. Drops the channel when none is left.
	59	+ pub fn publish(&self, user_id: &str, count: u64) {
	60	+ let mut users = self.users();
	61	+ let Some(tx) = users.get(user_id) else {
	62	+ return;
	63	+ };
	64	+ if tx.receiver_count() == 0 {
	65	+ users.remove(user_id);
	66	+ return;
	67	+ }
	68	+ tx.send_replace(count);
	69	+ }
	70	+
	71	+ /// Ends every open stream. Called once when the server shuts down.
	72	+ pub fn close(&self) {
	73	+ self.closing.send_replace(true);
	74	+ }
	75	+
	76	+ /// Removes the user's channel once its last subscription is gone.
	77	+ fn release(&self, user_id: &str) {
	78	+ let mut users = self.users();
	79	+ if users
	80	+ .get(user_id)
	81	+ .is_some_and(\|tx\| tx.receiver_count() == 0)
	82	+ {
	83	+ users.remove(user_id);
	84	+ }
	85	+ }
	86	+}
	87	+
	88	+/// One open stream's view of its user's count. Dropping it releases the user's channel when no
	89	+/// other stream of that user is open.
	90	+pub struct Subscription {
	91	+ /// Taken in `drop` before the hub is told, so the hub counts this receiver as gone.
	92	+ rx: Option<watch::Receiver<u64>>,
	93	+ closing: watch::Receiver<bool>,
	94	+ hub: Arc<NotifyHub>,
	95	+ user_id: String,
	96	+}
	97	+
	98	+impl Subscription {
	99	+ /// Waits for the next published count. `None` once the hub is closing (server shutdown).
	100	+ pub async fn next_count(&mut self) -> Option<u64> {
	101	+ loop {
	102	+ if *self.closing.borrow() {
	103	+ return None;
	104	+ }
	105	+ let rx = self.rx.as_mut()?;
	106	+ tokio::select! {
	107	+ changed = rx.changed() => {
	108	+ changed.ok()?;
	109	+ return Some(*rx.borrow_and_update());
	110	+ }
	111	+ changed = self.closing.changed() => {
	112	+ changed.ok()?;
	113	+ }
	114	+ }
	115	+ }
	116	+ }
	117	+}
	118	+
	119	+impl Drop for Subscription {
	120	+ fn drop(&mut self) {
	121	+ self.rx = None;
	122	+ self.hub.release(&self.user_id);
	123	+ }
	124	+}
	125	+
	126	+/// Publishes `user_id`'s current unread count to that user's open streams. Call it after a write
	127	+/// that changes the inbox. Skips the count query when nobody is listening. Never fails: problems
	128	+/// are logged.
	129	+pub async fn publish_unread(state: &AppState, user_id: &str) {
	130	+ if !state.notify_hub.is_watched(user_id) {
	131	+ return;
	132	+ }
	133	+ match state.store.unread_count(user_id).await {
	134	+ Ok(count) => state.notify_hub.publish(user_id, count),
	135	+ Err(e) => tracing::error!(error = %e, user_id, "notify: unread count failed"),
	136	+ }
	137	+}
	138	+
	139	+#[cfg(test)]
	140	+mod tests {
	141	+ use std::time::Duration;
	142	+
	143	+ use super::*;
	144	+
	145	+ #[tokio::test]
	146	+ async fn publish_reaches_the_subscriber_of_that_user_only() {
	147	+ let hub = Arc::new(NotifyHub::new());
	148	+ let mut alice = hub.subscribe("alice");
	149	+ let mut bob = hub.subscribe("bob");
	150	+
	151	+ hub.publish("alice", 3);
	152	+ assert_eq!(alice.next_count().await, Some(3));
	153	+
	154	+ // Bob heard nothing: his wait times out.
	155	+ let quiet = tokio::time::timeout(Duration::from_millis(50), bob.next_count()).await;
	156	+ assert!(quiet.is_err());
	157	+ }
	158	+
	159	+ #[tokio::test]
	160	+ async fn publish_without_subscribers_is_a_no_op_and_nothing_is_kept() {
	161	+ let hub = Arc::new(NotifyHub::new());
	162	+ assert!(!hub.is_watched("alice"));
	163	+ hub.publish("alice", 1);
	164	+ assert!(!hub.is_watched("alice"));
	165	+ assert!(hub.users().is_empty());
	166	+ }
	167	+
	168	+ #[tokio::test]
	169	+ async fn the_channel_is_released_when_the_last_subscription_drops() {
	170	+ let hub = Arc::new(NotifyHub::new());
	171	+ let first = hub.subscribe("alice");
	172	+ let second = hub.subscribe("alice");
	173	+ assert!(hub.is_watched("alice"));
	174	+
	175	+ drop(first);
	176	+ assert!(hub.is_watched("alice"));
	177	+
	178	+ drop(second);
	179	+ assert!(!hub.is_watched("alice"));
	180	+ assert!(hub.users().is_empty());
	181	+ }
	182	+
	183	+ #[tokio::test]
	184	+ async fn every_open_stream_of_a_user_gets_the_count() {
	185	+ let hub = Arc::new(NotifyHub::new());
	186	+ let mut tab_one = hub.subscribe("alice");
	187	+ let mut tab_two = hub.subscribe("alice");
	188	+
	189	+ hub.publish("alice", 5);
	190	+ assert_eq!(tab_one.next_count().await, Some(5));
	191	+ assert_eq!(tab_two.next_count().await, Some(5));
	192	+ }
	193	+
	194	+ #[tokio::test]
	195	+ async fn close_ends_every_subscription() {
	196	+ let hub = Arc::new(NotifyHub::new());
	197	+ let mut sub = hub.subscribe("alice");
	198	+
	199	+ hub.close();
	200	+ assert_eq!(sub.next_count().await, None);
	201	+ }
	202	+}

crates/forge-web/src/pages/ci_job.rsmodified+218 -28

		@@ -1,11 +1,13 @@
1	1	//! Pages: one CI job (`/{owner}/{repo}/ci/jobs/{job_id}`), its live log (`.../log.txt`) and its
2	2	//! artifacts (`.../artifacts/{name}`). Reading follows repo Read, like the run pages. The job page
3		-//! shows the last 512 KiB of the log with no JavaScript: while the job is not terminal it refreshes
4		-//! itself with a meta tag. `log.txt` streams the whole log and keeps following it until the job is
	3	+//! shows the last 512 KiB of the log. While the job is not terminal the page is live: it carries a
	4	+//! `data-live` marker for its `.../events` stream, and nav.js re-renders it when the job's status
	5	+//! changes. `log.txt` streams the whole log and keeps following it until the job is
5	6	//! terminal and the file is read to its end. Artifacts are blobs under `ci/<artifact id>` and are
6	7	//! always served as attachments.
7	8
8	9	use std::path::{Path, PathBuf};
	10	+use std::sync::Arc;
9	11	use std::time::Duration;
10	12
11	13	use bytes::Bytes;
		@@ -16,19 +18,24 @@ use http::header::{CONTENT_DISPOSITION, CONTENT_LENGTH, CONTENT_TYPE};
16	18	use http_body::Frame;
17	19	use http_body_util::StreamBody;
18	20	use tokio::io::{AsyncReadExt, AsyncSeekExt, SeekFrom};
	21	+use tokio::sync::watch;
19	22	use topcoat::Result;
20	23	use topcoat::context::{Cx, app_context};
	24	+use topcoat::router::content::sse::{Event, last_event_id};
21	25	use topcoat::router::error::not_found;
	26	+use topcoat::router::request::uri;
22	27	use topcoat::router::response::Response;
23	28	use topcoat::router::{Body, HeaderName, HeaderValue, StatusCode, page, path_param, route};
24	29	use topcoat::view::{View, component, view};
25	30
26		-use crate::ci::{artifact_key, log_path};
	31	+use crate::ci::{CiLive, artifact_key, log_path};
27	32	use crate::i18n::{
28	33	self, Lang,
29	34	releases_ci::{self as g, K},
30	35	};
31		-use crate::pages::ci_runs::{job_href, run_href, state_text};
	36	+use crate::pages::ci_runs::{
	37	+ changed_event, events_response, job_href, job_status, run_href, state_text,
	38	+};
32	39	use crate::pages::repo::repo_head;
33	40	use crate::pages::repo_ctx::{RepoCtx, RepoHead, load_repo};
34	41	use crate::state::AppState;
		@@ -43,8 +50,9 @@ path_param!(artifact_name);
43	50	pub const TAIL_BYTES: u64 = 512 * 1024;
44	51	/// Bytes read per chunk while following a log.
45	52	const CHUNK_BYTES: usize = 64 * 1024;
46		-/// How long `log.txt` waits before it looks at the file again.
47		-const FOLLOW_POLL: Duration = Duration::from_secs(1);
	53	+/// The longest a follower waits for a change to its job before it reads again anyway. A change
	54	+/// wakes it at once; this only guards against a missed wake.
	55	+const WAKE_BACKSTOP: Duration = Duration::from_secs(60);
48	56
49	57	/// The job and its run, both checked to belong to this repo. Anything else is 404.
50	58	async fn job_of(state: &AppState, ctx: &RepoCtx, id: &str) -> Result<(CiJob, CiRun)> {
		@@ -60,18 +68,30 @@ async fn job_of(state: &AppState, ctx: &RepoCtx, id: &str) -> Result<(CiJob, CiR
60	68	Ok((job, run))
61	69	}
62	70
63		-/// The last `TAIL_BYTES` of a log, from a line boundary, and whether anything was cut. A log that
64		-/// does not exist yet is empty.
65		-async fn log_tail(path: &Path) -> Result<(String, bool)> {
	71	+/// The length of the longest whole-UTF-8 prefix of `buf`. A character that a read cut short at the
	72	+/// end is left out, so the next read sends it whole. Other invalid bytes are kept.
	73	+fn whole_utf8_len(buf: &[u8]) -> usize {
	74	+ match std::str::from_utf8(buf) {
	75	+ Ok(_) => buf.len(),
	76	+ Err(e) if e.error_len().is_none() => e.valid_up_to(),
	77	+ Err(_) => buf.len(),
	78	+ }
	79	+}
	80	+
	81	+/// The last `TAIL_BYTES` of a log, from a line boundary, whether anything was cut, and the byte
	82	+/// offset just past the text shown. A log that does not exist yet is empty.
	83	+async fn log_tail(path: &Path) -> Result<(String, bool, u64)> {
66	84	let mut file = match tokio::fs::File::open(path).await {
67	85	Ok(file) => file,
68		- Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok((String::new(), false)),
	86	+ Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok((String::new(), false, 0)),
69	87	Err(e) => return Err(forge_core::Error::Io(e).into()),
70	88	};
71	89	let len = file.metadata().await.map_err(forge_core::Error::Io)?.len();
72	90	let truncated = len > TAIL_BYTES;
	91	+ let mut start = 0;
73	92	if truncated {
74		- file.seek(SeekFrom::Start(len - TAIL_BYTES))
	93	+ start = len - TAIL_BYTES;
	94	+ file.seek(SeekFrom::Start(start))
75	95	.await
76	96	.map_err(forge_core::Error::Io)?;
77	97	}
		@@ -83,30 +103,45 @@ async fn log_tail(path: &Path) -> Result<(String, bool)> {
83	103	.map_err(forge_core::Error::Io)?;
84	104	if truncated && let Some(end) = buf.iter().position(\|byte\| *byte == b'\n') {
85	105	buf.drain(..=end);
	106	+ start += end as u64 + 1;
86	107	}
87		- Ok((String::from_utf8_lossy(&buf).into_owned(), truncated))
	108	+ buf.truncate(whole_utf8_len(&buf));
	109	+ let offset = start + buf.len() as u64;
	110	+ Ok((
	111	+ String::from_utf8_lossy(&buf).into_owned(),
	112	+ truncated,
	113	+ offset,
	114	+ ))
88	115	}
89	116
90		-/// Follows a log: yields what the file holds, then polls for new bytes once a second. The stream
91		-/// ends when the job is terminal and a read after seeing that returns nothing, so every byte the
92		-/// runner wrote is sent. A job that is gone, or a read error, ends the stream.
93		-fn follow_log(
	117	+/// Follows a log: yields what the file holds, then sends new bytes each time the job changes (its
	118	+/// log grew or its state did). The stream ends when the job is terminal and a read after seeing
	119	+/// that returns nothing, so every byte the runner wrote is sent. A job that is gone, or a read
	120	+/// error, ends the stream.
	121	+pub(crate) fn follow_log(
94	122	store: Store,
	123	+ live: Arc<CiLive>,
95	124	job_id: String,
96	125	path: PathBuf,
97	126	) -> impl Stream<Item = std::io::Result<Bytes>> + Send + 'static {
98	127	struct Follow {
99	128	store: Store,
	129	+ live: Arc<CiLive>,
100	130	job_id: String,
101	131	path: PathBuf,
102	132	file: Option<tokio::fs::File>,
	133	+ /// Subscribed before the first read, so a change between the read and the wait is seen.
	134	+ wake: watch::Receiver<u64>,
103	135	}
	136	+ let wake = live.subscribe_job(&job_id);
104	137	futures_util::stream::unfold(
105	138	Follow {
106	139	store,
	140	+ live,
107	141	job_id,
108	142	path,
109	143	file: None,
	144	+ wake,
110	145	},
111	146	\|mut follow\| async move {
112	147	loop {
		@@ -139,7 +174,12 @@ fn follow_log(
139	174	if terminal {
140	175	return None;
141	176	}
142		- tokio::time::sleep(FOLLOW_POLL).await;
	177	+ // A closed signal means the job finished, which the next check sees as terminal.
	178	+ // It is subscribed again so this wait does not return at once.
	179	+ if let Ok(Err(_)) = tokio::time::timeout(WAKE_BACKSTOP, follow.wake.changed()).await
	180	+ {
	181	+ follow.wake = follow.live.subscribe_job(&follow.job_id);
	182	+ }
143	183	}
144	184	},
145	185	)
		@@ -161,6 +201,8 @@ fn attachment_name(name: &str) -> String {
161	201	/// The job page's values, already shown as text.
162	202	struct JobView {
163	203	name: String,
	204	+ /// The status word as the events carry it, for the live check.
	205	+ status: String,
164	206	state: String,
165	207	state_class: String,
166	208	runner: String,
		@@ -180,22 +222,22 @@ struct ArtifactRow {
180	222	size: String,
181	223	}
182	224
183		-/// The job page body. The meta refresh is in the body, since the shell has no head hook.
	225	+/// The job page body. A job that is not terminal carries `live`, the URL of its event stream, and
	226	+/// the status the page shows. The log tail carries the offset its text ends at.
184	227	#[component]
185	228	async fn job_body(
186	229	head: RepoHead,
187	230	lang: Lang,
188	231	job: JobView,
189	232	log: String,
	233	+ log_offset: String,
190	234	truncated: bool,
191	235	artifacts: Vec<ArtifactRow>,
192		- refresh: bool,
	236	+ live: Option<String>,
193	237	) -> Result<impl View> {
194	238	Ok(view! {
195	239	repo_head(head: head)
196		- if refresh {
197		- <meta http-equiv="refresh" content="3">
198		- }
	240	+ <div data-live=(live.as_deref()) data-live-state=(job.status.as_str())>
199	241	<p class="muted small"><a href=(job.run_href.as_str())>(g::t(lang, K::RunNumber)) (job.run_number)</a></p>
200	242	<section class="card">
201	243	<h2>(job.name.as_str())</h2>
		@@ -224,7 +266,7 @@ async fn job_body(
224	266	if truncated {
225	267	<p class="muted small">(g::t(lang, K::LogTruncated))</p>
226	268	}
227		- <pre class="ci-log">(log.as_str())</pre>
	269	+ <pre class="ci-log" data-live-log="true" data-offset=(log_offset.as_str())>(log.as_str())</pre>
228	270	</section>
229	271	if !artifacts.is_empty() {
230	272	<section class="card">
		@@ -236,6 +278,7 @@ async fn job_body(
236	278	</ul>
237	279	</section>
238	280	}
	281	+ </div>
239	282	})
240	283	}
241	284
		@@ -249,7 +292,7 @@ async fn job_page(cx: &Cx) -> Result<impl View> {
249	292	let lang = i18n::lang(cx, ctx.viewer.as_ref().map(\|(user, _)\| user));
250	293	let state = app_context::<AppState>(cx);
251	294	let (job, run) = job_of(state, &ctx, &id).await?;
252		- let (log, truncated) = log_tail(&log_path(&state.config.data_dir, &job.id)).await?;
	295	+ let (log, truncated, log_offset) = log_tail(&log_path(&state.config.data_dir, &job.id)).await?;
253	296	let artifacts = state
254	297	.store
255	298	.list_artifacts(&job.id)
		@@ -277,9 +320,10 @@ async fn job_page(cx: &Cx) -> Result<impl View> {
277	320	None => String::new(),
278	321	};
279	322	let title = title_of(&ctx);
280		- let refresh = !job.state.is_terminal();
	323	+ let live = (!job.state.is_terminal()).then(\|\| format!("{}/events", job_href(&title, &job.id)));
281	324	let view = JobView {
282	325	name: job.name.clone(),
	326	+ status: job.state.as_str().to_owned(),
283	327	state: state_text(lang, job.state.as_str()),
284	328	state_class: format!("ci-state ci-{}", job.state.as_str()),
285	329	runner,
		@@ -298,8 +342,8 @@ async fn job_page(cx: &Cx) -> Result<impl View> {
298	342	let page_title = format!("{title} \u{b7} CI \u{b7} {}", view.name);
299	343	Ok(view! {
300	344	shell(title: page_title.as_str(), viewer: ctx.viewer.as_ref(),
301		- job_body(head: head, lang: lang, job: view, log: log, truncated: truncated,
302		- artifacts: artifacts, refresh: refresh))
	345	+ job_body(head: head, lang: lang, job: view, log: log, log_offset: log_offset.to_string(),
	346	+ truncated: truncated, artifacts: artifacts, live: live))
303	347	})
304	348	}
305	349
		@@ -307,6 +351,146 @@ fn title_of(ctx: &RepoCtx) -> String {
307	351	format!("{}/{}", ctx.repo.owner_name, ctx.repo.name)
308	352	}
309	353
	354	+/// Bytes of log per `log` event. A bigger append is sent as several events.
	355	+const LOG_EVENT_BYTES: u64 = 64 * 1024;
	356	+
	357	+/// The job's live stream. It opens with the status, sends a `changed` event when the status changes
	358	+/// and a `log` event with each new run of log bytes. A `log` event's id is the offset past its text,
	359	+/// so a reconnect (`Last-Event-ID`) resumes there. A new stream starts at `?from=`, the offset the
	360	+/// page already shows. The stream ends once the job is terminal and its log is read to the end.
	361	+#[route(GET "/{owner}/{repo}/ci/jobs/{job_id}/events")]
	362	+async fn job_events(cx: &Cx) -> Result<Response> {
	363	+ let owner = path_param::<Owner>(cx);
	364	+ let name = path_param::<Repo>(cx);
	365	+ let id = path_param::<JobId>(cx).to_owned();
	366	+ let ctx = load_repo(cx, owner, name).await?;
	367	+ let state = app_context::<AppState>(cx);
	368	+ let (job, _) = job_of(state, &ctx, &id).await?;
	369	+ let offset = last_event_id(cx)
	370	+ .and_then(\|id\| id.parse::<u64>().ok())
	371	+ .or_else(\|\| from_query(cx))
	372	+ .unwrap_or(0);
	373	+ // Subscribed before the first read, so a change in between is sent.
	374	+ let wake = state.ci_live.subscribe_job(&job.id);
	375	+ let follow = JobFollow {
	376	+ store: state.store.clone(),
	377	+ job_id: job.id.clone(),
	378	+ path: log_path(&state.config.data_dir, &job.id),
	379	+ wake,
	380	+ offset,
	381	+ status: None,
	382	+ terminal: false,
	383	+ more: false,
	384	+ primed: false,
	385	+ pending_status: None,
	386	+ pending_log: None,
	387	+ };
	388	+ events_response(cx, job_stream(follow))
	389	+}
	390	+
	391	+/// The `from` offset of the events URL, if it is a number.
	392	+fn from_query(cx: &Cx) -> Option<u64> {
	393	+ uri(cx).query().and_then(\|query\| {
	394	+ form_urlencoded::parse(query.as_bytes())
	395	+ .find(\|(key, _)\| key == "from")
	396	+ .and_then(\|(_, value)\| value.parse::<u64>().ok())
	397	+ })
	398	+}
	399	+
	400	+/// One job's stream state. Each read of the status and the log happens after a wake, and a read that
	401	+/// sent log bytes is followed by another read at once, so a long append drains in order.
	402	+struct JobFollow {
	403	+ store: Store,
	404	+ job_id: String,
	405	+ path: PathBuf,
	406	+ wake: watch::Receiver<u64>,
	407	+ /// The log offset read so far. The next read starts here.
	408	+ offset: u64,
	409	+ /// The status last read.
	410	+ status: Option<String>,
	411	+ /// The job was terminal at the last read.
	412	+ terminal: bool,
	413	+ /// The last read sent log bytes, so more may be waiting.
	414	+ more: bool,
	415	+ /// False until the first read. Later reads wait for a wake first, unless `more` is set.
	416	+ primed: bool,
	417	+ pending_status: Option<String>,
	418	+ /// The text and the offset past it.
	419	+ pending_log: Option<(String, u64)>,
	420	+}
	421	+
	422	+fn job_stream(follow: JobFollow) -> impl Stream<Item = Result<Event>> + Send + 'static {
	423	+ futures_util::stream::unfold(follow, \|mut f\| async move {
	424	+ loop {
	425	+ if let Some(status) = f.pending_status.take() {
	426	+ return Some((changed_event(status), f));
	427	+ }
	428	+ if let Some((text, end)) = f.pending_log.take() {
	429	+ let event = Event::new().event("log").data(text).id(end.to_string());
	430	+ return Some((Ok(event), f));
	431	+ }
	432	+ if f.terminal && !f.more {
	433	+ return None;
	434	+ }
	435	+ if f.primed && !f.more && f.wake.changed().await.is_err() {
	436	+ return None;
	437	+ }
	438	+ f.primed = true;
	439	+ f.more = false;
	440	+ let (status, terminal) = match job_status(&f.store, &f.job_id).await {
	441	+ Ok(Some(found)) => found,
	442	+ Ok(None) => return None,
	443	+ Err(e) => {
	444	+ tracing::error!(error = %e, "ci: job stream could not read the job");
	445	+ return None;
	446	+ }
	447	+ };
	448	+ f.terminal = terminal;
	449	+ if f.status.as_deref() != Some(status.as_str()) {
	450	+ f.status = Some(status.clone());
	451	+ f.pending_status = Some(status);
	452	+ }
	453	+ match read_log(&f.path, f.offset).await {
	454	+ Ok(Some((text, end))) => {
	455	+ f.offset = end;
	456	+ f.more = true;
	457	+ f.pending_log = Some((text, end));
	458	+ }
	459	+ Ok(None) => {}
	460	+ Err(e) => {
	461	+ tracing::error!(error = %e, "ci: job stream could not read the log");
	462	+ return None;
	463	+ }
	464	+ }
	465	+ }
	466	+ })
	467	+}
	468	+
	469	+/// The whole text of the log from `offset`, at most `LOG_EVENT_BYTES`, and the offset past it. None
	470	+/// when there is nothing new (no file, no bytes past `offset`, or only part of a character).
	471	+async fn read_log(path: &Path, offset: u64) -> std::io::Result<Option<(String, u64)>> {
	472	+ let mut file = match tokio::fs::File::open(path).await {
	473	+ Ok(file) => file,
	474	+ Err(e) if e.kind() == std::io::ErrorKind::NotFound => return Ok(None),
	475	+ Err(e) => return Err(e),
	476	+ };
	477	+ file.seek(SeekFrom::Start(offset)).await?;
	478	+ let mut buf = Vec::new();
	479	+ (&mut file)
	480	+ .take(LOG_EVENT_BYTES)
	481	+ .read_to_end(&mut buf)
	482	+ .await?;
	483	+ let whole = whole_utf8_len(&buf);
	484	+ if whole == 0 {
	485	+ return Ok(None);
	486	+ }
	487	+ buf.truncate(whole);
	488	+ Ok(Some((
	489	+ String::from_utf8_lossy(&buf).into_owned(),
	490	+ offset + whole as u64,
	491	+ )))
	492	+}
	493	+
310	494	/// Streams the job's log as plain text. It sends the file so far and then follows it until the job
311	495	/// is terminal.
312	496	#[route(GET "/{owner}/{repo}/ci/jobs/{job_id}/log.txt")]
		@@ -318,7 +502,13 @@ async fn job_log(cx: &Cx) -> Result<Response> {
318	502	let state = app_context::<AppState>(cx);
319	503	let (job, _) = job_of(state, &ctx, &id).await?;
320	504	let path = log_path(&state.config.data_dir, &job.id);
321		- let chunks = follow_log(state.store.clone(), job.id.clone(), path).map_ok(Frame::data);
	505	+ let chunks = follow_log(
	506	+ state.store.clone(),
	507	+ Arc::clone(&state.ci_live),
	508	+ job.id.clone(),
	509	+ path,
	510	+ )
	511	+ .map_ok(Frame::data);
322	512	let mut res = Response::new(Body::new(StreamBody::new(chunks)));
323	513	*res.status_mut() = StatusCode::OK;
324	514	let headers = res.headers_mut();

crates/forge-web/src/pages/ci_runs.rsmodified+182 -12

		@@ -2,22 +2,30 @@
2	2	//! the repo sees its runs, their jobs and the config error of a run. Starting a manual run, cancelling,
3	3	//! re-running and approving need Write, and the buttons are shown only to writers. Anyone else gets a
4	4	//! 404 for a private repo and a refusal for a change. Every change notifies the job waker and syncs
5		-//! the commit status of each job it changed.
	5	+//! the commit status of each job it changed. A run or job page that is not finished stays live: it
	6	+//! opens a `changed` event stream (`.../events`) and re-renders itself on each event, with no reload
	7	+//! and no timer.
6	8
7	9	use std::collections::HashMap;
	10	+use std::future::Future;
	11	+use std::time::Duration;
8	12
9	13	use forge_core::ci::{CiJob, CiRun, RunState};
10		-use forge_core::{Action, Denied, User, authorize};
	14	+use forge_core::{Action, Denied, Store, User, authorize};
11	15	use forge_git::browse::{RefInfo, RefKind};
	16	+use futures_util::{Stream, stream};
12	17	use serde::Deserialize;
13	18	use time::OffsetDateTime;
14	19	use time::format_description::well_known::Rfc3339;
	20	+use tokio::sync::watch;
15	21	use topcoat::Result;
16	22	use topcoat::context::{Cx, app_context};
17	23	use topcoat::router::content::Form;
	24	+use topcoat::router::content::sse::{Event, KeepAlive, Sse};
18	25	use topcoat::router::error::{SeeOther, forbidden, not_found, see_other};
19	26	use topcoat::router::request::uri;
20		-use topcoat::router::{page, path_param, route};
	27	+use topcoat::router::response::{IntoResponse, Response};
	28	+use topcoat::router::{HeaderValue, page, path_param, route};
21	29	use topcoat::view::{View, component, view};
22	30
23	31	use crate::auth::{current_user, require_csrf};
		@@ -116,6 +124,15 @@ pub fn job_href(title: &str, job_id: &str) -> String {
116	124	format!("/{title}/ci/jobs/{}", enc(job_id))
117	125	}
118	126
	127	+/// The badge image (same origin, default branch) and the README snippet that links it to the CI
	128	+/// runs page. The snippet uses the configured base URL so it works when pasted elsewhere.
	129	+fn badge_info(state: &AppState, title: &str, default_branch: &str) -> BadgeInfo {
	130	+ let base = state.config.base_url.trim_end_matches('/');
	131	+ let image = format!("/{}/badge.svg?branch={}", enc(title), enc(default_branch));
	132	+ let snippet = format!("[![CI]({base}{image})]({base}{})", ci_href(&enc(title)));
	133	+ BadgeInfo { image, snippet }
	134	+}
	135	+
119	136	/// The notice from `?notice=<word>`. Anything else is ignored.
120	137	fn notice_from_query(cx: &Cx, lang: Lang) -> Option<(&'static str, bool)> {
121	138	let query = uri(cx).query()?;
		@@ -230,16 +247,22 @@ async fn run_of(state: &AppState, ctx: &RepoCtx, number: &str) -> Result<CiRun>
230	247	/// Wakes the job queue and syncs the commit status of every job whose state changed since `before`.
231	248	/// A job missing from `before` (a fresh run) counts as changed.
232	249	async fn after_change(state: &AppState, run_id: &str, before: &[CiJob]) -> Result<()> {
233		- state.ci_waker.notify_one();
	250	+ state.ci_waker.notify_waiters();
234	251	let Some(run) = state.store.run_by_id(run_id).await? else {
235	252	return Ok(());
236	253	};
	254	+ state.ci_live.bump_run(run_id);
237	255	for job in state.store.jobs_for_run(run_id).await? {
238	256	let changed = before
239	257	.iter()
240	258	.find(\|old\| old.id == job.id)
241	259	.is_none_or(\|old\| old.state != job.state);
242	260	if changed {
	261	+ if job.state.is_terminal() {
	262	+ state.ci_live.close_job(&job.id);
	263	+ } else {
	264	+ state.ci_live.bump_job(&job.id);
	265	+ }
243	266	crate::ci::sync_status(state, &run, &job).await;
244	267	}
245	268	}
		@@ -266,6 +289,12 @@ struct ManualRun {
266	289	branches: Vec<String>,
267	290	}
268	291
	292	+/// The status badge section: the image path (same origin) and the Markdown for a README.
	293	+struct BadgeInfo {
	294	+ image: String,
	295	+ snippet: String,
	296	+}
	297	+
269	298	/// A job row on the run page.
270	299	struct JobRow {
271	300	href: String,
		@@ -331,6 +360,7 @@ async fn runs_body(
331	360	lang: Lang,
332	361	notice: Option<(&'static str, bool)>,
333	362	manual: Option<ManualRun>,
	363	+ badge: BadgeInfo,
334	364	rows: Vec<RunRow>,
335	365	newer_href: Option<String>,
336	366	older_href: Option<String>,
		@@ -363,6 +393,17 @@ async fn runs_body(
363	393	</form>
364	394	</section>
365	395	}
	396	+ <section class="card">
	397	+ <details>
	398	+ <summary>(g::t(lang, K::BadgeHeading))</summary>
	399	+ <p class="muted small">(g::t(lang, K::BadgeHelp))</p>
	400	+ <p><img src=(badge.image.as_str()) alt="CI"></p>
	401	+ <label class="small">
	402	+ (g::t(lang, K::BadgeSnippet))
	403	+ <textarea readonly="readonly" rows="2" class="ci-badge-snippet">(badge.snippet.as_str())</textarea>
	404	+ </label>
	405	+ </details>
	406	+ </section>
366	407	<section class="card">
367	408	<h2>(g::t(lang, K::Runs))</h2>
368	409	if rows.is_empty() {
		@@ -455,11 +496,12 @@ async fn runs_list_page(cx: &Cx) -> Result<impl View> {
455	496	None
456	497	};
457	498	let notice = notice_from_query(cx, lang);
	499	+ let badge = badge_info(state, &title_of(&ctx), &ctx.repo.default_branch);
458	500	let head = ctx.head(state, "ci").await?;
459	501	let title = format!("{} \u{b7} CI", title_of(&ctx));
460	502	Ok(view! {
461	503	shell(title: title.as_str(), viewer: ctx.viewer.as_ref(),
462		- runs_body(head: head, lang: lang, notice: notice, manual: manual, rows: rows,
	504	+ runs_body(head: head, lang: lang, notice: notice, manual: manual, badge: badge, rows: rows,
463	505	newer_href: newer_href, older_href: older_href, page_label: page_label))
464	506	})
465	507	}
		@@ -544,7 +586,113 @@ fn job_rows(
544	586	.collect()
545	587	}
546	588
547		-/// The run page body: the header, the buttons for writers, the config error, and the jobs.
	589	+/// How often an idle event stream sends a `: keepalive` comment, so a reverse proxy does not drop
	590	+/// the connection. This is not a poll: events follow the change signals.
	591	+const STREAM_KEEP_ALIVE: Duration = Duration::from_secs(30);
	592	+
	593	+/// A run's or job's status word and whether it is terminal. `None` when the row is gone.
	594	+pub(crate) type Status = Option<(String, bool)>;
	595	+
	596	+pub(crate) async fn run_status(store: &Store, run_id: &str) -> Result<Status> {
	597	+ Ok(store
	598	+ .run_by_id(run_id)
	599	+ .await?
	600	+ .map(\|run\| (run.state.as_str().to_owned(), run_finished(&run.state))))
	601	+}
	602	+
	603	+pub(crate) async fn job_status(store: &Store, job_id: &str) -> Result<Status> {
	604	+ Ok(store
	605	+ .job_by_id(job_id)
	606	+ .await?
	607	+ .map(\|job\| (job.state.as_str().to_owned(), job.state.is_terminal())))
	608	+}
	609	+
	610	+/// One `changed` event carrying the status word.
	611	+pub(crate) fn changed_event(status: String) -> Result<Event> {
	612	+ Ok(Event::new().event("changed").data(status))
	613	+}
	614	+
	615	+/// The `changed` events of a CI page. The first event carries the status now. Each later one
	616	+/// follows a wake of `wake`: with `only_on_status`, only when the status differs from the last one
	617	+/// sent. The stream ends after the terminal status is sent, when the row is gone, or when the signal
	618	+/// closes. Subscribe `wake` before the first read, so a change in between is not lost.
	619	+pub(crate) fn changed_stream<F, Fut>(
	620	+ wake: watch::Receiver<u64>,
	621	+ only_on_status: bool,
	622	+ read: F,
	623	+) -> impl Stream<Item = Result<Event>> + Send + 'static
	624	+where
	625	+ F: Fn() -> Fut + Send + 'static,
	626	+ Fut: Future<Output = Result<Status>> + Send,
	627	+{
	628	+ struct Follow<F> {
	629	+ wake: watch::Receiver<u64>,
	630	+ only_on_status: bool,
	631	+ read: F,
	632	+ /// The status last read, sent or not.
	633	+ last: Option<String>,
	634	+ /// Set once a terminal status is read. Nothing follows it.
	635	+ done: bool,
	636	+ /// False until the first read. Every later read waits for a wake first.
	637	+ primed: bool,
	638	+ }
	639	+ stream::unfold(
	640	+ Follow {
	641	+ wake,
	642	+ only_on_status,
	643	+ read,
	644	+ last: None,
	645	+ done: false,
	646	+ primed: false,
	647	+ },
	648	+ \|mut follow\| async move {
	649	+ loop {
	650	+ if follow.done {
	651	+ return None;
	652	+ }
	653	+ if follow.primed && follow.wake.changed().await.is_err() {
	654	+ return None;
	655	+ }
	656	+ follow.primed = true;
	657	+ let (status, terminal) = match (follow.read)().await {
	658	+ Ok(Some(found)) => found,
	659	+ Ok(None) => return None,
	660	+ Err(e) => {
	661	+ tracing::error!(error = %e, "ci: live stream could not read the status");
	662	+ return None;
	663	+ }
	664	+ };
	665	+ follow.done = terminal;
	666	+ let changed = follow.last.as_deref() != Some(status.as_str());
	667	+ let send = changed \|\| !follow.only_on_status;
	668	+ follow.last = Some(status.clone());
	669	+ if send {
	670	+ return Some((changed_event(status), follow));
	671	+ }
	672	+ }
	673	+ },
	674	+ )
	675	+}
	676	+
	677	+/// The response of an event stream for a CI page: keep-alive comments while idle, and no buffering
	678	+/// by a reverse proxy.
	679	+pub(crate) fn events_response(
	680	+ cx: &Cx,
	681	+ events: impl Stream<Item = Result<Event>> + Send + 'static,
	682	+) -> Result<Response> {
	683	+ let events = Sse::new(events).keep_alive(
	684	+ KeepAlive::new()
	685	+ .interval(STREAM_KEEP_ALIVE)
	686	+ .text("keepalive"),
	687	+ );
	688	+ let mut res = events.into_response(cx)?;
	689	+ res.headers_mut()
	690	+ .insert("x-accel-buffering", HeaderValue::from_static("no"));
	691	+ Ok(res)
	692	+}
	693	+
	694	+/// The run page body: the header, the buttons for writers, the config error, and the jobs. A run
	695	+/// that is not finished carries `live`, the URL of its event stream.
548	696	#[component]
549	697	async fn run_body(
550	698	head: RepoHead,
		@@ -553,13 +701,11 @@ async fn run_body(
553	701	run: RunView,
554	702	jobs: Vec<JobRow>,
555	703	writer_csrf: Option<String>,
556		- refresh: bool,
	704	+ live: Option<String>,
557	705	) -> Result<impl View> {
558	706	Ok(view! {
559	707	repo_head(head: head)
560		- if refresh {
561		- <meta http-equiv="refresh" content="3">
562		- }
	708	+ <div data-live=(live.as_deref())>
563	709	if let Some((text, is_error)) = notice {
564	710	if is_error {
565	711	alert(message: text)
		@@ -625,6 +771,7 @@ async fn run_body(
625	771	</div>
626	772	}
627	773	</section>
	774	+ </div>
628	775	})
629	776	}
630	777
		@@ -657,7 +804,8 @@ async fn run_page(cx: &Cx) -> Result<impl View> {
657	804	.collect();
658	805	let title = title_of(&ctx);
659	806	let rows = job_rows(&title, &jobs, &runners, lang);
660		- let refresh = !run_finished(&run.state) && run.state != RunState::WaitingApproval;
	807	+ let live =
	808	+ (!run_finished(&run.state)).then(\|\| format!("{}/events", run_href(&title, run.number)));
661	809	let notice = notice_from_query(cx, lang);
662	810	let header = run_view(state, &ctx, &run, writer, lang).await?;
663	811	let head = ctx.head(state, "ci").await?;
		@@ -665,10 +813,32 @@ async fn run_page(cx: &Cx) -> Result<impl View> {
665	813	Ok(view! {
666	814	shell(title: page_title.as_str(), viewer: ctx.viewer.as_ref(),
667	815	run_body(head: head, lang: lang, notice: notice, run: header, jobs: rows,
668		- writer_csrf: writer_csrf, refresh: refresh))
	816	+ writer_csrf: writer_csrf, live: live))
669	817	})
670	818	}
671	819
	820	+/// The live `changed` stream of a run that is not finished. It is sent on every change to the run
	821	+/// or to one of its jobs, so the page can render its rows again.
	822	+#[route(GET "/{owner}/{repo}/ci/runs/{run_number}/events")]
	823	+async fn run_events(cx: &Cx) -> Result<Response> {
	824	+ let owner = path_param::<Owner>(cx);
	825	+ let name = path_param::<Repo>(cx);
	826	+ let number = path_param::<RunNumber>(cx).to_owned();
	827	+ let ctx = load_repo(cx, owner, name).await?;
	828	+ let state = app_context::<AppState>(cx);
	829	+ let run = run_of(state, &ctx, &number).await?;
	830	+ // Subscribed before the status is read, so a change in between is sent.
	831	+ let wake = state.ci_live.subscribe_run(&run.id);
	832	+ let store = state.store.clone();
	833	+ let run_id = run.id.clone();
	834	+ let events = changed_stream(wake, false, move \|\| {
	835	+ let store = store.clone();
	836	+ let run_id = run_id.clone();
	837	+ async move { run_status(&store, &run_id).await }
	838	+ });
	839	+ events_response(cx, events)
	840	+}
	841	+
672	842	/// Cancels a run that has not finished. Queued and blocked jobs are cancelled at once; running jobs
673	843	/// get a cancel request that their runner sees on its next heartbeat.
674	844	#[route(POST "/{owner}/{repo}/ci/runs/{run_number}/cancel")]

crates/forge-web/src/pages/ci_tests.rsmodified+421 -5

		@@ -1,7 +1,10 @@
1	1	//! Route tests for the CI run and job pages. The fixture: alice owns the private repo `proj`, bob
2	2	//! has Write, dave has Read, and carol is outside the repo.
3	3
	4	+use std::io::Write;
4	5	use std::path::{Path, PathBuf};
	6	+use std::sync::Arc;
	7	+use std::time::Duration;
5	8
6	9	use forge_core::ci::{CiJob, CiRun, JobState, NewJob, RunState};
7	10	use forge_core::{Role, Store, Visibility};
		@@ -11,7 +14,7 @@ use topcoat::router::{Body, HeaderMap, Router, to_bytes};
11	14
12	15	use crate::auth::SESSION_COOKIE;
13	16	use crate::blobs::BlobStore;
14		-use crate::ci::{artifact_key, log_path};
	17	+use crate::ci::{CiLive, artifact_key, log_path};
15	18	use crate::test_support::{app_router, create_session, test_state};
16	19
17	20	const CSRF: &str = "ci-test-csrf";
		@@ -22,6 +25,7 @@ const RUN_BASE: &str = "/alice/proj/ci";
22	25	struct Fixture {
23	26	store: Store,
24	27	app: Router,
	28	+ live: Arc<CiLive>,
25	29	repo: forge_core::Repo,
26	30	alice: String,
27	31	data_dir: PathBuf,
		@@ -69,9 +73,11 @@ async fn fixture(dir: &Path) -> Fixture {
69	73	create_session(&store, &carol.id, "carol-token", CSRF).await;
70	74	let data_dir = state.config.data_dir.clone();
71	75	let blobs = state.blobs.clone();
	76	+ let live = Arc::clone(&state.ci_live);
72	77	Fixture {
73	78	store,
74	79	app: app_router(state),
	80	+ live,
75	81	repo,
76	82	alice: alice.id,
77	83	data_dir,
		@@ -300,23 +306,300 @@ async fn run_list_pages_25_at_a_time() {
300	306	);
301	307	}
302	308
	309	+/// The pages no longer reload themselves. A page that is not finished carries its live marker, and
	310	+/// a finished one carries none.
303	311	#[tokio::test]
304		-async fn meta_refresh_only_while_the_job_is_not_terminal() {
	312	+async fn pages_are_live_while_not_finished_and_never_meta_refresh() {
305	313	let dir = tempfile::tempdir().expect("tempdir");
306	314	let fx = fixture(dir.path()).await;
307	315	let run = run_with(&fx, &["build"], false).await;
308	316	let job = job_of(&fx, &run).await;
309	317	let url = format!("/alice/proj/ci/jobs/{}", job.id);
	318	+ let run_url = format!("{RUN_BASE}/runs/{}", run.number);
	319	+ let job_marker = format!("data-live=\"{url}/events\"");
	320	+ let run_marker = format!("data-live=\"{run_url}/events\"");
	321	+
310	322	let queued = get(&fx.app, &url, Some(&cookie_for("alice-token"))).await;
	323	+ assert!(queued.text().contains(&job_marker), "job live while queued");
	324	+ let run_page = get(&fx.app, &run_url, Some(&cookie_for("alice-token"))).await;
311	325	assert!(
312		- queued.text().contains("http-equiv=\"refresh\""),
313		- "refresh while queued"
	326	+ run_page.text().contains(&run_marker),
	327	+ "run live while queued"
314	328	);
	329	+ for page in [&queued, &run_page] {
	330	+ assert!(
	331	+ !page.text().contains("http-equiv=\"refresh\""),
	332	+ "no meta refresh"
	333	+ );
	334	+ }
	335	+
315	336	finish_job(&fx).await;
316	337	let done = get(&fx.app, &url, Some(&cookie_for("alice-token"))).await;
	338	+ assert!(
	339	+ !done.text().contains("data-live="),
	340	+ "job not live when finished"
	341	+ );
317	342	assert!(
318	343	!done.text().contains("http-equiv=\"refresh\""),
319		- "no refresh when finished"
	344	+ "no meta refresh"
	345	+ );
	346	+}
	347	+
	348	+/// Opens an event stream without reading it. The response head carries the status.
	349	+async fn open(
	350	+ app: &Router,
	351	+ uri: &str,
	352	+ cookie: Option<&str>,
	353	+) -> topcoat::router::response::Response {
	354	+ let mut builder = Request::builder().method("GET").uri(uri);
	355	+ if let Some(cookie) = cookie {
	356	+ builder = builder.header("cookie", cookie);
	357	+ }
	358	+ app.handle(builder.body(Body::from(String::new())).expect("request"))
	359	+ .await
	360	+}
	361	+
	362	+/// The next frame of an event stream, or None once the stream has ended. Fails after five seconds.
	363	+async fn next_event<S, E>(frames: &mut S) -> Option<Vec<u8>>
	364	+where
	365	+ S: futures_util::Stream<Item = Result<bytes::Bytes, E>> + Unpin,
	366	+ E: std::fmt::Debug,
	367	+{
	368	+ use futures_util::StreamExt;
	369	+
	370	+ let frame = tokio::time::timeout(Duration::from_secs(5), frames.next())
	371	+ .await
	372	+ .expect("an event within five seconds")?;
	373	+ Some(frame.expect("data").to_vec())
	374	+}
	375	+
	376	+#[tokio::test]
	377	+async fn live_events_need_the_same_access_as_the_page() {
	378	+ let dir = tempfile::tempdir().expect("tempdir");
	379	+ let fx = fixture(dir.path()).await;
	380	+ let run = run_with(&fx, &["build"], false).await;
	381	+ let job = job_of(&fx, &run).await;
	382	+ let urls = [
	383	+ format!("{RUN_BASE}/runs/{}/events", run.number),
	384	+ format!("/alice/proj/ci/jobs/{}/events", job.id),
	385	+ ];
	386	+ for url in &urls {
	387	+ let anon = open(&fx.app, url, None).await;
	388	+ assert_eq!(anon.status().as_u16(), 404, "anonymous {url}");
	389	+ let outsider = open(&fx.app, url, Some(&cookie_for("carol-token"))).await;
	390	+ assert_eq!(outsider.status().as_u16(), 404, "outsider {url}");
	391	+ let reader = open(&fx.app, url, Some(&cookie_for("dave-token"))).await;
	392	+ assert_eq!(reader.status().as_u16(), 200, "reader {url}");
	393	+ }
	394	+}
	395	+
	396	+/// Each stream opens with its status. The run stream follows every bump; the job stream sends only
	397	+/// status changes, so a log append is silent, and it ends once the job is terminal.
	398	+#[tokio::test]
	399	+async fn live_events_open_with_the_status_and_follow_changes() {
	400	+ use futures_util::StreamExt;
	401	+
	402	+ let dir = tempfile::tempdir().expect("tempdir");
	403	+ let fx = fixture(dir.path()).await;
	404	+ let run = run_with(&fx, &["build"], false).await;
	405	+ let job = job_of(&fx, &run).await;
	406	+ let alice = cookie_for("alice-token");
	407	+
	408	+ let mut run_frames = open(
	409	+ &fx.app,
	410	+ &format!("{RUN_BASE}/runs/{}/events", run.number),
	411	+ Some(&alice),
	412	+ )
	413	+ .await
	414	+ .into_body()
	415	+ .into_data_stream();
	416	+ let first = next_event(&mut run_frames).await.expect("the run status");
	417	+ assert_eq!(
	418	+ first,
	419	+ format!("event: changed\ndata: {}\n\n", run.state.as_str()).into_bytes()
	420	+ );
	421	+
	422	+ let mut job_frames = open(
	423	+ &fx.app,
	424	+ &format!("/alice/proj/ci/jobs/{}/events", job.id),
	425	+ Some(&alice),
	426	+ )
	427	+ .await
	428	+ .into_body()
	429	+ .into_data_stream();
	430	+ let first = next_event(&mut job_frames).await.expect("the job status");
	431	+ assert_eq!(
	432	+ first,
	433	+ format!("event: changed\ndata: {}\n\n", job.state.as_str()).into_bytes()
	434	+ );
	435	+
	436	+ // A bump with no new log bytes and no status change sends nothing.
	437	+ fx.live.bump_job(&job.id);
	438	+ let quiet = tokio::time::timeout(Duration::from_millis(200), job_frames.next()).await;
	439	+ assert!(quiet.is_err(), "a log append sends no status event");
	440	+
	441	+ // The run stream sends its status on every bump of the run.
	442	+ fx.live.bump_run(&run.id);
	443	+ let bumped = next_event(&mut run_frames).await.expect("the run status");
	444	+ let now = fx
	445	+ .store
	446	+ .run_by_id(&run.id)
	447	+ .await
	448	+ .expect("run")
	449	+ .expect("run");
	450	+ assert_eq!(
	451	+ bumped,
	452	+ format!("event: changed\ndata: {}\n\n", now.state.as_str()).into_bytes()
	453	+ );
	454	+
	455	+ // The job finishes: its final status is sent, then the stream ends.
	456	+ finish_job(&fx).await;
	457	+ fx.live.close_job(&job.id);
	458	+ let last = next_event(&mut job_frames).await.expect("the final status");
	459	+ assert_eq!(
	460	+ last,
	461	+ format!("event: changed\ndata: {}\n\n", JobState::Success.as_str()).into_bytes()
	462	+ );
	463	+ assert!(next_event(&mut job_frames).await.is_none(), "stream ends");
	464	+}
	465	+
	466	+/// The job's log events: each append is sent as a `log` event with its bytes, the id is the offset
	467	+/// past them, and the bytes before `?from=` are not sent again.
	468	+#[tokio::test]
	469	+async fn job_events_send_each_log_append_as_a_log_event() {
	470	+ let dir = tempfile::tempdir().expect("tempdir");
	471	+ let fx = fixture(dir.path()).await;
	472	+ let run = run_with(&fx, &["build"], false).await;
	473	+ let job = job_of(&fx, &run).await;
	474	+ let path = log_path(&fx.data_dir, &job.id);
	475	+ std::fs::create_dir_all(path.parent().expect("log dir")).expect("log dir");
	476	+ std::fs::write(&path, b"one\n").expect("log");
	477	+
	478	+ let mut frames = open(
	479	+ &fx.app,
	480	+ &format!("/alice/proj/ci/jobs/{}/events?from=4", job.id),
	481	+ Some(&cookie_for("alice-token")),
	482	+ )
	483	+ .await
	484	+ .into_body()
	485	+ .into_data_stream();
	486	+ let first = next_event(&mut frames).await.expect("the status");
	487	+ assert_eq!(
	488	+ first,
	489	+ format!("event: changed\ndata: {}\n\n", job.state.as_str()).into_bytes()
	490	+ );
	491	+
	492	+ let mut file = std::fs::OpenOptions::new()
	493	+ .append(true)
	494	+ .open(&path)
	495	+ .expect("log");
	496	+ file.write_all(b"two\nthree\n").expect("append");
	497	+ fx.live.bump_job(&job.id);
	498	+ let log = next_event(&mut frames).await.expect("the log event");
	499	+ assert_eq!(
	500	+ log,
	501	+ b"event: log\ndata: two\ndata: three\ndata: \nid: 14\n\n".to_vec()
	502	+ );
	503	+}
	504	+
	505	+/// An append longer than 64 KiB is sent in several `log` events, in order.
	506	+#[tokio::test]
	507	+async fn job_events_split_a_long_append_into_64_kib_events() {
	508	+ let dir = tempfile::tempdir().expect("tempdir");
	509	+ let fx = fixture(dir.path()).await;
	510	+ let run = run_with(&fx, &["build"], false).await;
	511	+ let job = job_of(&fx, &run).await;
	512	+ let path = log_path(&fx.data_dir, &job.id);
	513	+ std::fs::create_dir_all(path.parent().expect("log dir")).expect("log dir");
	514	+ std::fs::write(&path, b"").expect("log");
	515	+
	516	+ let mut frames = open(
	517	+ &fx.app,
	518	+ &format!("/alice/proj/ci/jobs/{}/events", job.id),
	519	+ Some(&cookie_for("alice-token")),
	520	+ )
	521	+ .await
	522	+ .into_body()
	523	+ .into_data_stream();
	524	+ next_event(&mut frames).await.expect("the status");
	525	+
	526	+ let long = "a".repeat(150 * 1024);
	527	+ let mut file = std::fs::OpenOptions::new()
	528	+ .append(true)
	529	+ .open(&path)
	530	+ .expect("log");
	531	+ file.write_all(long.as_bytes()).expect("append");
	532	+ fx.live.bump_job(&job.id);
	533	+
	534	+ let mut sizes = Vec::new();
	535	+ for _ in 0..3 {
	536	+ let frame = next_event(&mut frames).await.expect("a log event");
	537	+ let text = String::from_utf8(frame).expect("text");
	538	+ assert!(text.starts_with("event: log\n"), "{text}");
	539	+ let data = text
	540	+ .lines()
	541	+ .find_map(\|line\| line.strip_prefix("data: "))
	542	+ .expect("data line");
	543	+ sizes.push(data.len());
	544	+ if sizes.len() == 3 {
	545	+ assert!(
	546	+ text.contains("id: 153600\n"),
	547	+ "the last id is the end offset"
	548	+ );
	549	+ }
	550	+ }
	551	+ assert_eq!(sizes, vec![64 * 1024, 64 * 1024, 150 * 1024 - 128 * 1024]);
	552	+}
	553	+
	554	+/// A reconnect sends `Last-Event-ID`, and the stream resumes after that offset, not at `?from=`.
	555	+#[tokio::test]
	556	+async fn job_events_resume_after_the_last_event_id() {
	557	+ let dir = tempfile::tempdir().expect("tempdir");
	558	+ let fx = fixture(dir.path()).await;
	559	+ let run = run_with(&fx, &["build"], false).await;
	560	+ let job = job_of(&fx, &run).await;
	561	+ let path = log_path(&fx.data_dir, &job.id);
	562	+ std::fs::create_dir_all(path.parent().expect("log dir")).expect("log dir");
	563	+ std::fs::write(&path, b"one\ntwo\n").expect("log");
	564	+
	565	+ let request = Request::builder()
	566	+ .method("GET")
	567	+ .uri(format!("/alice/proj/ci/jobs/{}/events?from=0", job.id))
	568	+ .header("cookie", cookie_for("alice-token"))
	569	+ .header("last-event-id", "4")
	570	+ .body(Body::from(String::new()))
	571	+ .expect("request");
	572	+ let mut frames = fx.app.handle(request).await.into_body().into_data_stream();
	573	+ next_event(&mut frames).await.expect("the status");
	574	+ let log = next_event(&mut frames).await.expect("the log event");
	575	+ assert_eq!(log, b"event: log\ndata: two\ndata: \nid: 8\n\n".to_vec());
	576	+}
	577	+
	578	+/// The job page's log tail says where its text ends, so the events stream starts from there.
	579	+#[tokio::test]
	580	+async fn job_page_tail_carries_its_offset() {
	581	+ let dir = tempfile::tempdir().expect("tempdir");
	582	+ let fx = fixture(dir.path()).await;
	583	+ let run = run_with(&fx, &["build"], false).await;
	584	+ let job = job_of(&fx, &run).await;
	585	+ let path = log_path(&fx.data_dir, &job.id);
	586	+ std::fs::create_dir_all(path.parent().expect("log dir")).expect("log dir");
	587	+ std::fs::write(&path, b"one\ntwo\n").expect("log");
	588	+
	589	+ let page = get(
	590	+ &fx.app,
	591	+ &format!("/alice/proj/ci/jobs/{}", job.id),
	592	+ Some(&cookie_for("alice-token")),
	593	+ )
	594	+ .await;
	595	+ let text = page.text();
	596	+ assert!(
	597	+ text.contains("data-live-log=\"true\" data-offset=\"8\""),
	598	+ "tail offset"
	599	+ );
	600	+ assert!(
	601	+ text.contains(&format!("data-live-state=\"{}\"", job.state.as_str())),
	602	+ "rendered status"
320	603	);
321	604	}
322	605
		@@ -384,6 +667,67 @@ async fn log_txt_is_the_full_file_for_a_finished_job() {
384	667	);
385	668	}
386	669
	670	+/// The next chunk a follower sends. Fails the test when none comes within five seconds.
	671	+async fn next_chunk(
	672	+ chunks: &mut tokio::sync::mpsc::UnboundedReceiver<bytes::Bytes>,
	673	+) -> Option<bytes::Bytes> {
	674	+ tokio::time::timeout(Duration::from_secs(5), chunks.recv())
	675	+ .await
	676	+ .expect("a chunk within five seconds")
	677	+}
	678	+
	679	+/// A follower sends each append as soon as the job's signal is bumped. The wait is far longer
	680	+/// than the timeouts here, so a chunk that arrives in time was woken by the bump, not by a timer.
	681	+#[tokio::test]
	682	+async fn log_follower_is_woken_by_bumps_and_ends_on_close() {
	683	+ use futures_util::StreamExt;
	684	+
	685	+ let dir = tempfile::tempdir().expect("tempdir");
	686	+ let fx = fixture(dir.path()).await;
	687	+ let run = run_with(&fx, &["build"], false).await;
	688	+ let job = job_of(&fx, &run).await;
	689	+ let path = log_path(&fx.data_dir, &job.id);
	690	+ std::fs::create_dir_all(path.parent().expect("log dir")).expect("log dir");
	691	+ std::fs::write(&path, b"one\n").expect("log");
	692	+ let live = Arc::new(CiLive::default());
	693	+ let mut stream = Box::pin(super::ci_job::follow_log(
	694	+ fx.store.clone(),
	695	+ Arc::clone(&live),
	696	+ job.id.clone(),
	697	+ path.clone(),
	698	+ ));
	699	+ let (tx, mut chunks) = tokio::sync::mpsc::unbounded_channel();
	700	+ tokio::spawn(async move {
	701	+ while let Some(Ok(chunk)) = stream.next().await {
	702	+ if tx.send(chunk).is_err() {
	703	+ break;
	704	+ }
	705	+ }
	706	+ });
	707	+ assert_eq!(
	708	+ next_chunk(&mut chunks).await.as_deref(),
	709	+ Some(&b"one\n"[..])
	710	+ );
	711	+
	712	+ let mut file = std::fs::OpenOptions::new()
	713	+ .append(true)
	714	+ .open(&path)
	715	+ .expect("log");
	716	+ file.write_all(b"two\n").expect("append");
	717	+ live.bump_job(&job.id);
	718	+ assert_eq!(
	719	+ next_chunk(&mut chunks).await.as_deref(),
	720	+ Some(&b"two\n"[..])
	721	+ );
	722	+
	723	+ finish_job(&fx).await;
	724	+ live.close_job(&job.id);
	725	+ let end = tokio::time::timeout(Duration::from_secs(5), chunks.recv())
	726	+ .await
	727	+ .expect("the stream ends once the job is terminal");
	728	+ assert!(end.is_none());
	729	+}
	730	+
387	731	#[tokio::test]
388	732	async fn artifact_download_sends_attachment_headers() {
389	733	let dir = tempfile::tempdir().expect("tempdir");
		@@ -579,3 +923,75 @@ async fn manual_run_refuses_an_unknown_branch() {
579	923	.await;
580	924	assert_eq!(reader.status, 403, "reader cannot start a manual run");
581	925	}
	926	+
	927	+#[tokio::test]
	928	+async fn badge_follows_read_access_and_reports_the_latest_run() {
	929	+ let dir = tempfile::tempdir().expect("tempdir");
	930	+ let fx = fixture(dir.path()).await;
	931	+ let open = fx
	932	+ .store
	933	+ .create_repo(&fx.alice, "open", "", Visibility::Public, "main")
	934	+ .await
	935	+ .expect("public repo");
	936	+ let svg_path = "/alice/proj/badge.svg";
	937	+
	938	+ // The private repo: a run exists, but only readers see its badge.
	939	+ run_with(&fx, &["build"], false).await;
	940	+ finish_job(&fx).await;
	941	+ assert_eq!(get(&fx.app, svg_path, None).await.status, 404, "anonymous");
	942	+ let outsider = get(&fx.app, svg_path, Some(&cookie_for("carol-token"))).await;
	943	+ assert_eq!(outsider.status, 404, "outsider");
	944	+ let reader = get(&fx.app, svg_path, Some(&cookie_for("dave-token"))).await;
	945	+ assert_eq!(reader.status, 200, "reader");
	946	+ assert!(reader.text().contains("passing"));
	947	+ assert_eq!(
	948	+ reader
	949	+ .headers
	950	+ .get("content-type")
	951	+ .and_then(\|v\| v.to_str().ok()),
	952	+ Some("image/svg+xml")
	953	+ );
	954	+ assert_eq!(
	955	+ reader
	956	+ .headers
	957	+ .get("cache-control")
	958	+ .and_then(\|v\| v.to_str().ok()),
	959	+ Some("no-cache, max-age=0")
	960	+ );
	961	+
	962	+ // The public repo: anyone sees the badge. Unknown branches and jobs have no status.
	963	+ let needs: &[String] = &[];
	964	+ let specs = [NewJob {
	965	+ name: "build",
	966	+ spec_json: SPEC,
	967	+ needs,
	968	+ }];
	969	+ fx.store
	970	+ .create_run(
	971	+ &open.id,
	972	+ "push",
	973	+ "refs/heads/main",
	974	+ SHA,
	975	+ None,
	976	+ &fx.alice,
	977	+ false,
	978	+ &specs,
	979	+ None,
	980	+ )
	981	+ .await
	982	+ .expect("run");
	983	+ finish_job(&fx).await;
	984	+ let anon = get(&fx.app, "/alice/open/badge.svg", None).await;
	985	+ assert_eq!(anon.status, 200);
	986	+ assert!(anon.text().contains(">passing</text>"));
	987	+ let unknown = get(&fx.app, "/alice/open/badge.svg?branch=nope", None).await;
	988	+ assert_eq!(unknown.status, 200);
	989	+ assert!(unknown.text().contains(">no status</text>"));
	990	+ let job = get(&fx.app, "/alice/open/badge.svg?job=build", None).await;
	991	+ assert!(job.text().contains(">build</text>"));
	992	+ assert!(job.text().contains(">passing</text>"));
	993	+ let missing_job = get(&fx.app, "/alice/open/badge.svg?job=lint", None).await;
	994	+ assert!(missing_job.text().contains(">no status</text>"));
	995	+ let unknown_repo = get(&fx.app, "/alice/nothere/badge.svg", None).await;
	996	+ assert_eq!(unknown_repo.status, 404);
	997	+}

crates/forge-web/src/pages/mod.rsmodified+1 -0

		@@ -10,6 +10,7 @@ pub mod assets;
10	10	pub mod audit_log;
11	11	pub mod avatars;
12	12	pub mod blob;
	13	+pub mod ci_badge;
13	14	pub mod ci_job;
14	15	pub mod ci_runs;
15	16	#[cfg(test)]

crates/forge-web/src/pages/notifications.rsmodified+164 -29

		@@ -5,23 +5,28 @@
5	5	//! The sidebar box and the watch form are shared with the issue, pull and repo header views.
6	6
7	7	use std::collections::HashMap;
	8	+use std::time::Duration;
8	9
9	10	use forge_core::authz::{Action, authorize};
10	11	use forge_core::discussions::{DiscussionNotification, DiscussionReason};
11	12	use forge_core::store::advisories::AdvisoryNotice;
12	13	use forge_core::store::transfers::PendingTransfer;
13	14	use forge_core::{Notification, Reason, Role, SubState, User, WatchLevel};
	15	+use futures_util::{StreamExt, future, stream};
14	16	use serde::Deserialize;
15	17	use topcoat::Result;
16	18	use topcoat::context::{Cx, app_context};
17	19	use topcoat::router::content::Form;
	20	+use topcoat::router::content::sse::{Event, KeepAlive, Sse};
18	21	use topcoat::router::error::{SeeOther, bad_request, not_found, see_other};
19		-use topcoat::router::{page, path_param, query_params, route};
	22	+use topcoat::router::response::{IntoResponse, Response};
	23	+use topcoat::router::{Body, HeaderValue, StatusCode, page, path_param, query_params, route};
20	24	use topcoat::view::{View, component, view};
21	25
22		-use crate::auth::{percent_encode, require_csrf, require_user};
	26	+use crate::auth::{current_user, percent_encode, require_csrf, require_user};
23	27	use crate::i18n::misc::{self as g, K};
24	28	use crate::i18n::{self, Key, Lang};
	29	+use crate::notify_hub::publish_unread;
25	30	use crate::pages::issue_parts::{parse_number, stamp_unix};
26	31	use crate::pages::repo_ctx::{RepoCtx, load_repo};
27	32	use crate::pages::repo_transfer::transfer_list_for;
		@@ -363,6 +368,9 @@ async fn inbox(cx: &Cx) -> Result<impl View> {
363	368	}
364	369	}
365	370
	371	+ if !hidden.is_empty() {
	372	+ publish_unread(state, &user.id).await;
	373	+ }
366	374	let unread_count = state.store.unread_count(&user.id).await?;
367	375	let transfers = state.store.pending_transfers_for_user(&user.id).await?;
368	376	let csrf = viewer.1.csrf_token.clone();
		@@ -384,6 +392,7 @@ async fn open_notification(cx: &Cx) -> Result<SeeOther> {
384	392	return Err(not_found().into());
385	393	};
386	394	state.store.mark_read(&user.id, id).await?;
	395	+ publish_unread(state, &user.id).await;
387	396	if !can_read(state, &user, &target.repo_id).await? {
388	397	return Err(not_found().into());
389	398	}
		@@ -411,6 +420,7 @@ async fn open_discussion_notification(cx: &Cx) -> Result<SeeOther> {
411	420	.store
412	421	.mark_discussion_notification_read(&user.id, id)
413	422	.await?;
	423	+ publish_unread(state, &user.id).await;
414	424	if !can_read(state, &user, &note.repo_id).await? {
415	425	return Err(not_found().into());
416	426	}
		@@ -433,6 +443,7 @@ async fn read_discussion_notification(cx: &Cx, Form(form): Form<CsrfForm>) -> Re
433	443	.store
434	444	.mark_discussion_notification_read(&user.id, id)
435	445	.await?;
	446	+ publish_unread(state, &user.id).await;
436	447	Ok(see_other(back(cx)))
437	448	}
438	449
		@@ -450,6 +461,7 @@ async fn open_advisory_notification(cx: &Cx) -> Result<SeeOther> {
450	461	.store
451	462	.mark_advisory_notification_read(&user.id, repo_id)
452	463	.await?;
	464	+ publish_unread(state, &user.id).await;
453	465	if !can_manage_alerts(state, &user, repo_id).await? {
454	466	return Err(not_found().into());
455	467	}
		@@ -471,6 +483,7 @@ async fn read_advisory_notification(cx: &Cx, Form(form): Form<CsrfForm>) -> Resu
471	483	.store
472	484	.mark_advisory_notification_read(&user.id, repo_id)
473	485	.await?;
	486	+ publish_unread(state, &user.id).await;
474	487	Ok(see_other(back(cx)))
475	488	}
476	489
		@@ -481,6 +494,7 @@ async fn read_notification(cx: &Cx, Form(form): Form<CsrfForm>) -> Result<SeeOth
481	494	let state = app_context::<AppState>(cx);
482	495	let id = path_param::<Note>(cx);
483	496	state.store.mark_read(&user.id, id).await?;
	497	+ publish_unread(state, &user.id).await;
484	498	Ok(see_other(back(cx)))
485	499	}
486	500
		@@ -500,9 +514,50 @@ async fn mark_all_read(cx: &Cx, Form(form): Form<CsrfForm>) -> Result<SeeOther>
500	514	.store
501	515	.mark_all_advisory_notifications_read(&user.id, repo_id)
502	516	.await?;
	517	+ publish_unread(state, &user.id).await;
503	518	Ok(see_other(back(cx)))
504	519	}
505	520
	521	+/// How often an idle stream sends a `: keepalive` comment, so a reverse proxy does not drop the
	522	+/// connection. This is the only timer in the feature: counts are pushed as they change.
	523	+const STREAM_KEEP_ALIVE: Duration = Duration::from_secs(30);
	524	+
	525	+/// One `unread` event carrying the count.
	526	+fn unread_event(count: u64) -> topcoat::Result<Event> {
	527	+ Ok(Event::new().event("unread").data(count.to_string()))
	528	+}
	529	+
	530	+/// The bell's live count. Sends an `unread` event with the viewer's unread count now, then one
	531	+/// after each change to the inbox, until the viewer disconnects or the server shuts down. An
	532	+/// anonymous viewer gets a bare 401: an event stream has no login page to redirect to.
	533	+#[route(GET "/notifications/stream")]
	534	+async fn notification_stream(cx: &Cx) -> Result<Response> {
	535	+ let Some((user, _)) = current_user(cx).await? else {
	536	+ let mut res = Response::new(Body::empty());
	537	+ *res.status_mut() = StatusCode::UNAUTHORIZED;
	538	+ return Ok(res);
	539	+ };
	540	+ let state = app_context::<AppState>(cx);
	541	+ let subscription = state.notify_hub.subscribe(&user.id);
	542	+ // Read after subscribing: a change that lands in between is pushed again, never missed.
	543	+ let count = state.store.unread_count(&user.id).await?;
	544	+ let first = stream::once(future::ready(unread_event(count)));
	545	+ let changes = stream::unfold(subscription, \|mut sub\| async move {
	546	+ let count = sub.next_count().await?;
	547	+ Some((unread_event(count), sub))
	548	+ });
	549	+ let events = Sse::new(first.chain(changes)).keep_alive(
	550	+ KeepAlive::new()
	551	+ .interval(STREAM_KEEP_ALIVE)
	552	+ .text("keepalive"),
	553	+ );
	554	+ let mut res = events.into_response(cx)?;
	555	+ // Nginx buffers a proxied response unless told not to, which would hold the events back.
	556	+ res.headers_mut()
	557	+ .insert("x-accel-buffering", HeaderValue::from_static("no"));
	558	+ Ok(res)
	559	+}
	560	+
506	561	/// Subscribes or unsubscribes the viewer from an issue.
507	562	#[route(POST "/{owner}/{repo}/issues/{number}/subscription")]
508	563	async fn issue_subscription(cx: &Cx, Form(form): Form<SubscriptionForm>) -> Result<SeeOther> {
		@@ -624,6 +679,7 @@ pub(crate) async fn notify_box_for(
624	679	return Ok(None);
625	680	};
626	681	state.store.mark_issue_read(&user.id, issue_id).await?;
	682	+ publish_unread(state, &user.id).await;
627	683	let level = state.store.watch_level(&user.id, &ctx.repo.id).await?;
628	684	let subscription = state.store.subscription(&user.id, issue_id).await?;
629	685	let (subscribed, reason) = if level == WatchLevel::Ignoring {
		@@ -805,7 +861,9 @@ pub(crate) async fn notify_box(
805	861	})
806	862	}
807	863
808		-/// The Watch control in the repo header: a level select and a submit button, plus the count.
	864	+/// The Watch control in the repo header: a dropdown whose summary shows the current level, one
	865	+/// POST form per level inside it (the current one marked), and the watcher count. The forms post
	866	+/// without JavaScript; nav.js submits them in place and closes the menu.
809	867	#[component]
810	868	pub(crate) async fn watch_form(
811	869	lang: Lang,
		@@ -815,38 +873,69 @@ pub(crate) async fn watch_form(
815	873	watchers: u64,
816	874	) -> Result<impl View> {
817	875	let count = g::watchers_count(lang, watchers);
	876	+ let current = match level {
	877	+ "watching" => g::t(lang, K::OptWatching),
	878	+ "ignoring" => g::t(lang, K::OptIgnoring),
	879	+ _ => g::t(lang, K::OptParticipating),
	880	+ };
	881	+ let class = \|value: &str\| -> &'static str {
	882	+ if level == value {
	883	+ "watch-option is-current"
	884	+ } else {
	885	+ "watch-option"
	886	+ }
	887	+ };
	888	+ let (cls_participating, cls_watching, cls_ignoring) =
	889	+ (class("participating"), class("watching"), class("ignoring"));
818	890	Ok(view! {
819		- <form method="post" action=(action.as_str()) class="watch-form">
820		- csrf_field(token: csrf.as_str())
821		- <label class="watch-label">
822		- <span class="sr-only">(g::t(lang, K::NotificationLevel))</span>
823		- <select name="level">
824		- if level == "participating" {
825		- <option value="participating" selected="selected">(g::t(lang, K::OptParticipating))</option>
826		- } else {
827		- <option value="participating">(g::t(lang, K::OptParticipating))</option>
828		- }
829		- if level == "watching" {
830		- <option value="watching" selected="selected">(g::t(lang, K::OptWatching))</option>
831		- } else {
832		- <option value="watching">(g::t(lang, K::OptWatching))</option>
833		- }
834		- if level == "ignoring" {
835		- <option value="ignoring" selected="selected">(g::t(lang, K::OptIgnoring))</option>
836		- } else {
837		- <option value="ignoring">(g::t(lang, K::OptIgnoring))</option>
838		- }
839		- </select>
840		- </label>
841		- <button type="submit" class="button">(i18n::t(lang, Key::Save))</button>
	891	+ <div class="watch-control">
	892	+ <details class="menu watch-menu">
	893	+ <summary>
	894	+ <svg class="watch-eye" viewBox="0 0 24 24" aria-hidden="true"><path d="M1 12s4-7 11-7 11 7 11 7-4 7-11 7S1 12 1 12z"/><circle cx="12" cy="12" r="3"/></svg>
	895	+ <span>(current)</span>
	896	+ </summary>
	897	+ <ul class="menu-list watch-list">
	898	+ <li>
	899	+ <form method="post" action=(action.as_str()) class="watch-option-form">
	900	+ csrf_field(token: csrf.as_str())
	901	+ <input type="hidden" name="level" value="participating">
	902	+ <button type="submit" class=(cls_participating)>
	903	+ <span class="watch-option-title">(g::t(lang, K::OptParticipating))</span>
	904	+ <span class="watch-option-desc">(g::t(lang, K::WatchDescParticipating))</span>
	905	+ </button>
	906	+ </form>
	907	+ </li>
	908	+ <li>
	909	+ <form method="post" action=(action.as_str()) class="watch-option-form">
	910	+ csrf_field(token: csrf.as_str())
	911	+ <input type="hidden" name="level" value="watching">
	912	+ <button type="submit" class=(cls_watching)>
	913	+ <span class="watch-option-title">(g::t(lang, K::OptWatching))</span>
	914	+ <span class="watch-option-desc">(g::t(lang, K::WatchDescWatching))</span>
	915	+ </button>
	916	+ </form>
	917	+ </li>
	918	+ <li>
	919	+ <form method="post" action=(action.as_str()) class="watch-option-form">
	920	+ csrf_field(token: csrf.as_str())
	921	+ <input type="hidden" name="level" value="ignoring">
	922	+ <button type="submit" class=(cls_ignoring)>
	923	+ <span class="watch-option-title">(g::t(lang, K::OptIgnoring))</span>
	924	+ <span class="watch-option-desc">(g::t(lang, K::WatchDescIgnoring))</span>
	925	+ </button>
	926	+ </form>
	927	+ </li>
	928	+ </ul>
	929	+ </details>
842	930	<span class="muted small">(count.as_str())</span>
843		- </form>
	931	+ </div>
844	932	})
845	933	}
846	934
847	935	#[cfg(test)]
848	936	mod tests {
849	937	use forge_core::{Issue, Reason, Role, Store, SubState, Visibility, WatchLevel};
	938	+ use futures_util::StreamExt;
850	939	use topcoat::router::request::Request;
851	940	use topcoat::router::{Body, Router, to_bytes};
852	941
		@@ -1083,6 +1172,45 @@ mod tests {
1083	1172	assert_eq!(site.store.unread_count(&w.bob.id).await.expect("count"), 0);
1084	1173	}
1085	1174
	1175	+ #[tokio::test]
	1176	+ async fn the_stream_needs_a_session() {
	1177	+ let site = site().await;
	1178	+ let res = get(&site, "/notifications/stream", None).await;
	1179	+ assert_eq!(res.status, 401);
	1180	+ }
	1181	+
	1182	+ #[tokio::test]
	1183	+ async fn the_stream_sends_the_count_then_each_change() {
	1184	+ let site = site().await;
	1185	+ let w = world(&site, Visibility::Public).await;
	1186	+ let cookie = sign_in(&site.store, &w.bob.id, "bob-token").await;
	1187	+ notify(&site, &w, Reason::Mention).await;
	1188	+
	1189	+ let request = Request::builder()
	1190	+ .method("GET")
	1191	+ .uri("/notifications/stream")
	1192	+ .header("cookie", cookie.as_str())
	1193	+ .body(Body::empty())
	1194	+ .expect("request");
	1195	+ let res = site.router.handle(request).await;
	1196	+ assert_eq!(res.status().as_u16(), 200);
	1197	+ assert_eq!(
	1198	+ res.headers().get("content-type").unwrap(),
	1199	+ "text/event-stream"
	1200	+ );
	1201	+ assert_eq!(res.headers().get("x-accel-buffering").unwrap(), "no");
	1202	+
	1203	+ let mut frames = res.into_body().into_data_stream();
	1204	+ let first = frames.next().await.expect("a frame").expect("data");
	1205	+ assert_eq!(&first[..], b"event: unread\ndata: 1\n\n");
	1206	+
	1207	+ // Marking the row read pushes the new count to the open stream.
	1208	+ let read = post(&site, "/notifications/mark_all_read", &cookie, "").await;
	1209	+ assert_eq!(read.status, 303);
	1210	+ let next = frames.next().await.expect("a frame").expect("data");
	1211	+ assert_eq!(&next[..], b"event: unread\ndata: 0\n\n");
	1212	+ }
	1213	+
1086	1214	#[tokio::test]
1087	1215	async fn header_bell_shows_the_unread_count() {
1088	1216	let site = site().await;
		@@ -1172,11 +1300,18 @@ mod tests {
1172	1300
1173	1301	let header = get(&site, "/alice/demo/issues/1", Some(&cookie)).await;
1174	1302	assert!(header.body.contains("action=\"/alice/demo/watch\""));
1175		- assert!(
	1303	+ assert!(header.body.contains("<details class=\"menu watch-menu\">"));
	1304	+ assert!(!header.body.contains("<select"));
	1305	+ // One form per level, and only the watched level is marked current.
	1306	+ assert_eq!(header.body.matches("name=\"level\"").count(), 3);
	1307	+ assert_eq!(
1176	1308	header
1177	1309	.body
1178		- .contains("<option value=\"watching\" selected=\"selected\">")
	1310	+ .matches("class=\"watch-option is-current\"")
	1311	+ .count(),
	1312	+ 1
1179	1313	);
	1314	+ assert_eq!(header.body.matches("class=\"watch-option\"").count(), 2);
1180	1315	assert!(header.body.contains("1 watching"));
1181	1316
1182	1317	let bad = post(&site, "/alice/demo/watch", &cookie, "&level=loud").await;

crates/forge-web/src/views/mod.rsmodified+34 -5

		@@ -262,7 +262,8 @@ pub async fn shell(
262	262	<img class="avatar avatar-sm" src=(avatar_src) alt="" width="28" height="28">
263	263	</summary>
264	264	<ul class="menu-list">
265		- <li><a href="/settings/profile">(t(lang, Key::Profile))</a></li>
	265	+ <li><a href=(user_href(username.as_str()))>(mi::t(lang, MK::YourProfile))</a></li>
	266	+ <li><a href="/settings/profile">(mi::t(lang, MK::EditProfile))</a></li>
266	267	<li><a href="/settings/keys">(t(lang, Key::SshKeys))</a></li>
267	268	<li><a href="/settings/gpg">(t(lang, Key::GpgKeys))</a></li>
268	269	<li><a href="/settings/tokens">(t(lang, Key::AccessTokens))</a></li>
		@@ -398,6 +399,22 @@ pub fn avatar_url(username: &str, hash: Option<&str>) -> String {
398	399	format!("/avatars/{}?v={}", enc(username), hash.unwrap_or("default"))
399	400	}
400	401
	402	+/// Public profile URL of a user: `/{username}`.
	403	+pub fn user_href(username: &str) -> String {
	404	+ format!("/{}", enc(username))
	405	+}
	406	+
	407	+/// Link to a user's public profile. `login` is the real username (the link target), `shown` is
	408	+/// the text to display (it differs from `login` only for the ghost account). Both are escaped
	409	+/// like any other interpolation.
	410	+#[component]
	411	+pub async fn user_link(shown: &str, login: &str) -> Result<impl View> {
	412	+ let href = user_href(login);
	413	+ Ok(view! {
	414	+ <a href=(href.as_str())>(shown)</a>
	415	+ })
	416	+}
	417	+
401	418	/// Hidden CSRF input for a POST form.
402	419	#[component]
403	420	pub async fn csrf_field(token: &str) -> Result<impl View> {
		@@ -888,7 +905,11 @@ mod shell_tests {
888	905	for (at, _) in html.match_indices("<script") {
889	906	let tag = &html[at..];
890	907	let end = tag.find('>').expect("script tag closes");
891		- assert!(tag[..end].contains(" src="), "inline script: {}", &tag[..=end]);
	908	+ assert!(
	909	+ tag[..end].contains(" src="),
	910	+ "inline script: {}",
	911	+ &tag[..=end]
	912	+ );
892	913	}
893	914	}
894	915
		@@ -904,11 +925,19 @@ mod shell_tests {
904	925	.unwrap();
905	926	let res = router.handle(req).await;
906	927	assert_eq!(res.status(), 200);
907		- let csp = res.headers()["content-security-policy"].to_str().unwrap().to_owned();
	928	+ let csp = res.headers()["content-security-policy"]
	929	+ .to_str()
	930	+ .unwrap()
	931	+ .to_owned();
908	932	assert!(csp.contains("default-src 'self'"), "{csp}");
909	933	assert!(!csp.contains("unsafe-inline"), "{csp}");
910		- let html = String::from_utf8(to_bytes(res.into_body(), usize::MAX).await.unwrap().to_vec())
911		- .unwrap();
	934	+ let html = String::from_utf8(
	935	+ to_bytes(res.into_body(), usize::MAX)
	936	+ .await
	937	+ .unwrap()
	938	+ .to_vec(),
	939	+ )
	940	+ .unwrap();
912	941
913	942	let tag_at = html
914	943	.find(&format!("<script src=\"{NAV_SCRIPT_HREF}\""))

✨ feat: live notifications and ci pages over server-sent events

11 changed files +1566 -89