OXYGENHUB
# 05f1a85a2842 oxygenhub / crates / hub / src / web / live_stream.rs
6.1 KiB · Rust · last changed Restarts drain: SIGTERM stops taking connections and lets requests finish 19 days ago source blame history raw
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 
pub fn parse_topic(s: &str) -> Option<Topic> {
31 
    let num = |v: &str| v.parse::<i64>().ok().filter(|n| *n > 0);
32 
    if let Some(rest) = s.strip_prefix("repo:") {
33 
        return match rest.split_once('/') {
34 
            None => num(rest).map(Topic::Repo),
35 
            Some((id, sub)) => {
36 
                let id = num(id)?;
37 
                if let Some(n) = sub.strip_prefix("proposal:") {
38 
                    num(n).map(|n| Topic::Proposal(id, n))
39 
                } else if let Some(n) = sub.strip_prefix("issue:") {
40 
                    num(n).map(|n| Topic::Issue(id, n))
41 
                } else {
42 
                    None
43 
                }
44 
            }
45 
        };
46 
    }
47 
    if let Some(id) = s.strip_prefix("run:") {
48 
        return num(id).map(Topic::Run);
49 
    }
50 
    if let Some(id) = s.strip_prefix("user:") {
51 
        return num(id).map(Topic::User);
52 
    }
53 
    s.strip_prefix("diffusion:").and_then(num).map(Topic::Diffusion)
54 
}
55
56 
/// Whether `viewer` may hear of changes to `topic`: read access to its
57 
/// repository, or, for `user:`, being that user.
58 
async fn may_hear(app: &Shared, viewer: &Viewer, topic: &Topic) -> bool {
59 
    let repo_id = match topic {
60 
        Topic::Repo(r) | Topic::Proposal(r, _) | Topic::Issue(r, _) | Topic::Diffusion(r) => *r,
61 
        Topic::Run(run) => match db::flow::run_by_id(&app.db, *run).await {
62 
            Ok(Some(r)) => r.repo_id,
63 
            _ => return false,
64 
        },
65 
        Topic::User(id) => return viewer.account().is_some_and(|a| a.id == *id),
66 
    };
67 
    let Ok(Some(row)) = db::repos::get_by_id(&app.db, repo_id).await else { return false };
68 
    matches!(crate::auth::roles::role_for(&app.db, viewer.account(), &row).await, Ok(Some(_)))
69 
}
70
71 
#[derive(Deserialize)]
72 
pub struct LiveQuery {
73 
    #[serde(default)]
74 
    topics: String,
75 
}
76
77 
fn sse_event(e: &LiveEvent) -> Event {
78 
    Event::default()
79 
        .id(e.id.to_string())
80 
        .event("change")
81 
        .data(serde_json::json!({ "topic": e.topic, "version": e.version, "kind": e.kind }).to_string())
82 
}
83
84 
/// `GET /live?topics=repo:1,repo:1/proposal:4`.
85 
pub async fn stream(State(app): State<Shared>, viewer: Viewer, headers: HeaderMap, Query(q): Query<LiveQuery>) -> Response {
86 
    use futures::StreamExt;
87 
    let mut topics: Vec<String> = q.topics.split(',').map(str::trim).filter(|t| !t.is_empty()).map(String::from).collect();
88 
    topics.sort();
89 
    topics.dedup();
90 
    if topics.is_empty() || topics.len() > MAX_TOPICS {
91 
        return (StatusCode::BAD_REQUEST, format!("ask for 1 to {MAX_TOPICS} topics, as /live?topics=repo:1")).into_response();
92 
    }
93 
    for t in &topics {
94 
        let Some(topic) = parse_topic(t) else {
95 
            return (StatusCode::BAD_REQUEST, format!("{t:?} is not a topic")).into_response();
96 
        };
97 
        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 
            return (StatusCode::FORBIDDEN, "not every topic asked for may be heard").into_response();
101 
        }
102 
    }
103 
    // Subscribed before the replay is read, so nothing falls between them.
104 
    let rx = app.events.live.subscribe();
105 
    let last: Option<u64> = headers.get("last-event-id").and_then(|v| v.to_str().ok()).and_then(|v| v.parse().ok());
106 
    let (replay, after) = match last {
107 
        Some(id) => match app.events.live.since(id, &topics) {
108 
            Replay::Events(e) => {
109 
                let after = e.last().map(|e| e.id).unwrap_or(id);
110 
                (e.iter().map(|e| Ok(sse_event(e))).collect::<Vec<_>>(), after)
111 
            }
112 
            Replay::Reload => (vec![Ok(Event::default().event("reload").data("{}"))], app.events.live.last_id()),
113 
        },
114 
        None => (vec![], app.events.live.last_id()),
115 
    };
116 
    let live = tokio_stream::wrappers::BroadcastStream::new(rx).filter_map(move |item| {
117 
        let topics = topics.clone();
118 
        async move {
119 
            match item {
120 
                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 
                Err(_) => Some(Ok(Event::default().event("reload").data("{}"))),
123 
                _ => None,
124 
            }
125 
        }
126 
    });
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 
    let hello = std::iter::once(Ok(Event::default().comment("live")));
130 
    Sse::new(futures::StreamExt::take_until(futures::stream::iter(hello.chain(replay)).chain(live), crate::shutdown::stopping()))
131 
        .keep_alive(KeepAlive::default())
132 
        .into_response()
133 
}
134
135 
#[cfg(test)]
136 
mod tests {
137 
    use super::*;
138
139 
    #[test]
140 
    fn topics_are_read_strictly() {
141 
        assert_eq!(parse_topic("repo:7"), Some(Topic::Repo(7)));
142 
        assert_eq!(parse_topic("repo:7/proposal:4"), Some(Topic::Proposal(7, 4)));
143 
        assert_eq!(parse_topic("repo:7/issue:9"), Some(Topic::Issue(7, 9)));
144 
        assert_eq!(parse_topic("run:12"), Some(Topic::Run(12)));
145 
        assert_eq!(parse_topic("user:3"), Some(Topic::User(3)));
146 
        assert_eq!(parse_topic("diffusion:7"), Some(Topic::Diffusion(7)));
147 
        for bad in ["repo:", "repo:x", "repo:-1", "repo:7/wiki:1", "repo:7/proposal:", "team:1", "", "run:0"] {
148 
            assert_eq!(parse_topic(bad), None, "{bad}");
149 
        }
150 
    }
151 
}