live_stream.rs 86.3%

82 of 95 lines run, 19 of 35 functions · run #483 on api-unknown-fields · run missed

//! `GET /live?topics=`: one stream of changes for a page (docs/live.md §4,
//! oxygenhub#204). Every topic must be one the viewer may read, or nothing
//! is streamed; an event names a topic, its version and a kind, never data,
//! and the page fetches what changed from its normal routes.

use crate::auth::Viewer;
use crate::db;
use crate::events::{LiveEvent, Replay};
use crate::Shared;
use axum::extract::{Query, State};
use axum::http::{HeaderMap, StatusCode};
use axum::response::sse::{Event, KeepAlive, Sse};
use axum::response::{IntoResponse, Response};
use serde::Deserialize;

/// The most topics one page asks for.
const MAX_TOPICS: usize = 16;

/// A topic, read (docs/live.md §4).
#[derive(Debug, PartialEq, Eq)]
pub enum Topic {
    Repo(i64),
    Proposal(i64, i64),
    Issue(i64, i64),
    Run(i64),
    User(i64),
    Diffusion(i64),
}

pub fn parse_topic(s: &str) -> Option<Topic> {
    let num = |v: &str| v.parse::<i64>().ok().filter(|n| *n > 0);
    if let Some(rest) = s.strip_prefix("repo:") {
        return match rest.split_once('/') {
            None => num(rest).map(Topic::Repo),
            Some((id, sub)) => {
                let id = num(id)?;
                if let Some(n) = sub.strip_prefix("proposal:") {
                    num(n).map(|n| Topic::Proposal(id, n))
                } else if let Some(n) = sub.strip_prefix("issue:") {
                    num(n).map(|n| Topic::Issue(id, n))
                } else {
                    None
                }
            }
        };
    }
    if let Some(id) = s.strip_prefix("run:") {
        return num(id).map(Topic::Run);
    }
    if let Some(id) = s.strip_prefix("user:") {
        return num(id).map(Topic::User);
    }
    s.strip_prefix("diffusion:").and_then(num).map(Topic::Diffusion)
}

/// Whether `viewer` may hear of changes to `topic`: read access to its
/// repository, or, for `user:`, being that user.
async fn may_hear(app: &Shared, viewer: &Viewer, topic: &Topic) -> bool {
    let repo_id = match topic {
        Topic::Repo(r) | Topic::Proposal(r, _) | Topic::Issue(r, _) | Topic::Diffusion(r) => *r,
        Topic::Run(run) => match db::flow::run_by_id(&app.db, *run).await {
            Ok(Some(r)) => r.repo_id,
            _ => return false,
        },
        Topic::User(id) => return viewer.account().is_some_and(|a| a.id == *id),
    };
    let Ok(Some(row)) = db::repos::get_by_id(&app.db, repo_id).await else { return false };
    matches!(crate::auth::roles::role_for(&app.db, viewer.account(), &row).await, Ok(Some(_)))
}

#[derive(Deserialize)]
pub struct LiveQuery {
    #[serde(default)]
    topics: String,
}

fn sse_event(e: &LiveEvent) -> Event {
    Event::default()
        .id(e.id.to_string())
        .event("change")
        .data(serde_json::json!({ "topic": e.topic, "version": e.version, "kind": e.kind }).to_string())
}

/// `GET /live?topics=repo:1,repo:1/proposal:4`.
pub async fn stream(State(app): State<Shared>, viewer: Viewer, headers: HeaderMap, Query(q): Query<LiveQuery>) -> Response {
    use futures::StreamExt;
    let mut topics: Vec<String> = q.topics.split(',').map(str::trim).filter(|t| !t.is_empty()).map(String::from).collect();
    topics.sort();
    topics.dedup();
    if topics.is_empty() || topics.len() > MAX_TOPICS {
        return (StatusCode::BAD_REQUEST, format!("ask for 1 to {MAX_TOPICS} topics, as /live?topics=repo:1")).into_response();
    }
    for t in &topics {
        let Some(topic) = parse_topic(t) else {
            return (StatusCode::BAD_REQUEST, format!("{t:?} is not a topic")).into_response();
        };
        if !may_hear(&app, &viewer, &topic).await {
            // The same answer for every topic one may not hear: whether it
            // exists is not said either.
            return (StatusCode::FORBIDDEN, "not every topic asked for may be heard").into_response();
        }
    }
    // Subscribed before the replay is read, so nothing falls between them.
    let rx = app.bus.live.subscribe();
    let last: Option<u64> = headers.get("last-event-id").and_then(|v| v.to_str().ok()).and_then(|v| v.parse().ok());
    let (replay, after) = match last {
        Some(id) => match app.bus.live.since(id, &topics) {
            Replay::Events(e) => {
                let after = e.last().map(|e| e.id).unwrap_or(id);
                (e.iter().map(|e| Ok(sse_event(e))).collect::<Vec<_>>(), after)
            }
            Replay::Reload => (vec![Ok(Event::default().event("reload").data("{}"))], app.bus.live.last_id()),
        },
        None => (vec![], app.bus.live.last_id()),
    };
    let live = tokio_stream::wrappers::BroadcastStream::new(rx).filter_map(move |item| {
        let topics = topics.clone();
        async move {
            match item {
                Ok(e) if e.id > after && topics.contains(&e.topic) => Some(Ok::<_, std::convert::Infallible>(sse_event(&e))),
                // A subscriber that fell behind is told to reload.
                Err(_) => Some(Ok(Event::default().event("reload").data("{}"))),
                _ => None,
            }
        }
    });
    // A first comment at once, so a proxy that waits for body bytes (the
    // tunnel does) sends the headers now rather than at the first keep-alive.
    let hello = std::iter::once(Ok(Event::default().comment("live")));
    Sse::new(futures::StreamExt::take_until(futures::stream::iter(hello.chain(replay)).chain(live), crate::shutdown::stopping()))
        .keep_alive(KeepAlive::default())
        .into_response()
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn topics_are_read_strictly() {
        assert_eq!(parse_topic("repo:7"), Some(Topic::Repo(7)));
        assert_eq!(parse_topic("repo:7/proposal:4"), Some(Topic::Proposal(7, 4)));
        assert_eq!(parse_topic("repo:7/issue:9"), Some(Topic::Issue(7, 9)));
        assert_eq!(parse_topic("run:12"), Some(Topic::Run(12)));
        assert_eq!(parse_topic("user:3"), Some(Topic::User(3)));
        assert_eq!(parse_topic("diffusion:7"), Some(Topic::Diffusion(7)));
        for bad in ["repo:", "repo:x", "repo:-1", "repo:7/wiki:1", "repo:7/proposal:", "team:1", "", "run:0"] {
            assert_eq!(parse_topic(bad), None, "{bad}");
        }
    }
}

1		//! `GET /live?topics=`: one stream of changes for a page (docs/live.md §4,
2		//! oxygenhub#204). Every topic must be one the viewer may read, or nothing
3		//! is streamed; an event names a topic, its version and a kind, never data,
4		//! and the page fetches what changed from its normal routes.
5
6		use crate::auth::Viewer;
7		use crate::db;
8		use crate::events::{LiveEvent, Replay};
9		use crate::Shared;
10		use axum::extract::{Query, State};
11		use axum::http::{HeaderMap, StatusCode};
12		use axum::response::sse::{Event, KeepAlive, Sse};
13		use axum::response::{IntoResponse, Response};
14		use serde::Deserialize;
15
16		/// The most topics one page asks for.
17		const MAX_TOPICS: usize = 16;
18
19		/// A topic, read (docs/live.md §4).
20		#[derive(Debug, PartialEq, Eq)]
21		pub enum Topic {
22		Repo(i64),
23		Proposal(i64, i64),
24		Issue(i64, i64),
25		Run(i64),
26		User(i64),
27		Diffusion(i64),
28		}
29
30	23×	pub fn parse_topic(s: &str) -> Option<Topic> {
31	25×	let num = \|v: &str\| v.parse::<i64>().ok().filter(\|n\| *n > 0);
32	23×	if let Some(rest) = s.strip_prefix("repo:") {
33	15×	return match rest.split_once('/') {
34	9×	None => num(rest).map(Topic::Repo),
35	6×	Some((id, sub)) => {
36	6×	let id = num(id)?;
37	6×	if let Some(n) = sub.strip_prefix("proposal:") {
38	4×	num(n).map(\|n\| Topic::Proposal(id, n))
39	2×	} else if let Some(n) = sub.strip_prefix("issue:") {
40	1×	num(n).map(\|n\| Topic::Issue(id, n))
41		} else {
42	1×	None
43		}
44		}
45		};
46	8×	}
47	8×	if let Some(id) = s.strip_prefix("run:") {
48	2×	return num(id).map(Topic::Run);
49	6×	}
50	6×	if let Some(id) = s.strip_prefix("user:") {
51	2×	return num(id).map(Topic::User);
52	4×	}
53	4×	s.strip_prefix("diffusion:").and_then(num).map(Topic::Diffusion)
54	23×	}
55
56		/// Whether `viewer` may hear of changes to `topic`: read access to its
57		/// repository, or, for `user:`, being that user.
58	8×	async fn may_hear(app: &Shared, viewer: &Viewer, topic: &Topic) -> bool {
59	8×	let repo_id = match topic {
60	7×	Topic::Repo(r) \| Topic::Proposal(r, _) \| Topic::Issue(r, _) \| Topic::Diffusion(r) => *r,
61	0×	Topic::Run(run) => match db::flow::run_by_id(&app.db, *run).await {
62	0×	Ok(Some(r)) => r.repo_id,
63	0×	_ => return false,
64		},
65	1×	Topic::User(id) => return viewer.account().is_some_and(\|a\| a.id == *id),
66		};
67	7×	let Ok(Some(row)) = db::repos::get_by_id(&app.db, repo_id).await else { return false };
68	7×	matches!(crate::auth::roles::role_for(&app.db, viewer.account(), &row).await, Ok(Some(_)))
69	8×	}
70
71		#[derive(Deserialize)]
72		pub struct LiveQuery {
73		#[serde(default)]
74		topics: String,
75		}
76
77	1×	fn sse_event(e: &LiveEvent) -> Event {
78	1×	Event::default()
79	1×	.id(e.id.to_string())
80	1×	.event("change")
81	1×	.data(serde_json::json!({ "topic": e.topic, "version": e.version, "kind": e.kind }).to_string())
82	1×	}
83
84		/// `GET /live?topics=repo:1,repo:1/proposal:4`.
85	7×	pub async fn stream(State(app): State<Shared>, viewer: Viewer, headers: HeaderMap, Query(q): Query<LiveQuery>) -> Response {
86		use futures::StreamExt;
87	10×	let mut topics: Vec<String> = q.topics.split(',').map(str::trim).filter(\|t\| !t.is_empty()).map(String::from).collect();
88	7×	topics.sort();
89	7×	topics.dedup();
90	7×	if topics.is_empty() \|\| topics.len() > MAX_TOPICS {
91	1×	return (StatusCode::BAD_REQUEST, format!("ask for 1 to {MAX_TOPICS} topics, as /live?topics=repo:1")).into_response();
92	6×	}
93	9×	for t in &topics {
94	9×	let Some(topic) = parse_topic(t) else {
95	1×	return (StatusCode::BAD_REQUEST, format!("{t:?} is not a topic")).into_response();
96		};
97	8×	if !may_hear(&app, &viewer, &topic).await {
98		// The same answer for every topic one may not hear: whether it
99		// exists is not said either.
100	2×	return (StatusCode::FORBIDDEN, "not every topic asked for may be heard").into_response();
101	6×	}
102		}
103		// Subscribed before the replay is read, so nothing falls between them.
104	3×	let rx = app.bus.live.subscribe();
105	3×	let last: Option<u64> = headers.get("last-event-id").and_then(\|v\| v.to_str().ok()).and_then(\|v\| v.parse().ok());
106	3×	let (replay, after) = match last {
107	2×	Some(id) => match app.bus.live.since(id, &topics) {
108	1×	Replay::Events(e) => {
109	1×	let after = e.last().map(\|e\| e.id).unwrap_or(id);
110	1×	(e.iter().map(\|e\| Ok(sse_event(e))).collect::<Vec<_>>(), after)
111		}
112	1×	Replay::Reload => (vec![Ok(Event::default().event("reload").data("{}"))], app.bus.live.last_id()),
113		},
114	1×	None => (vec![], app.bus.live.last_id()),
115		};
116	3×	let live = tokio_stream::wrappers::BroadcastStream::new(rx).filter_map(move \|item\| {
117	0×	let topics = topics.clone();
118	0×	async move {
119	0×	match item {
120	0×	Ok(e) if e.id > after && topics.contains(&e.topic) => Some(Ok::<_, std::convert::Infallible>(sse_event(&e))),
121		// A subscriber that fell behind is told to reload.
122	0×	Err(_) => Some(Ok(Event::default().event("reload").data("{}"))),
123	0×	_ => None,
124		}
125	0×	}
126	0×	});
127		// A first comment at once, so a proxy that waits for body bytes (the
128		// tunnel does) sends the headers now rather than at the first keep-alive.
129	3×	let hello = std::iter::once(Ok(Event::default().comment("live")));
130	3×	Sse::new(futures::StreamExt::take_until(futures::stream::iter(hello.chain(replay)).chain(live), crate::shutdown::stopping()))
131	3×	.keep_alive(KeepAlive::default())
132	3×	.into_response()
133	7×	}
134
135		#[cfg(test)]
136		mod tests {
137		use super::*;
138
139		#[test]
140	1×	fn topics_are_read_strictly() {
141	1×	assert_eq!(parse_topic("repo:7"), Some(Topic::Repo(7)));
142	1×	assert_eq!(parse_topic("repo:7/proposal:4"), Some(Topic::Proposal(7, 4)));
143	1×	assert_eq!(parse_topic("repo:7/issue:9"), Some(Topic::Issue(7, 9)));
144	1×	assert_eq!(parse_topic("run:12"), Some(Topic::Run(12)));
145	1×	assert_eq!(parse_topic("user:3"), Some(Topic::User(3)));
146	1×	assert_eq!(parse_topic("diffusion:7"), Some(Topic::Diffusion(7)));
147	8×	for bad in ["repo:", "repo:x", "repo:-1", "repo:7/wiki:1", "repo:7/proposal:", "team:1", "", "run:0"] {
148	8×	assert_eq!(parse_topic(bad), None, "{bad}");
149		}
150	1×	}
151		}

`/`	Search
`g` `c`	Code
`g` `h`	History
`g` `b`	Bookmarks
`g` `t`	Trust
`y`	Permalink to this state (files and trees)
`?`	This sheet

Coverage / crates/hub/src/web/live_stream.rs 86.3%

Keyboard