Live pages L2: GET /live?topics= and the hub.js client 67be8ac00c24

//! `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.events.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.events.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.events.live.last_id()),

        },

        None => (vec![], app.events.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,

            }

        }

    });

    Sse::new(futures::stream::iter(replay).chain(live)).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}");

        }

    }

}

// Live pages (docs/live.md, oxygenhub#204): elements marked

// data-live="<topic>" data-fragment="<name>" are fetched again from this

// page's own route (?fragment=<name>) when the hub says their topic

// changed, and swapped in. Someone typing inside one is not interrupted:

// it is refreshed when they leave it. Without JavaScript or a stream the

// page is as it was; a reload shows everything.

(function () {

  var marked = document.querySelectorAll("[data-live][data-fragment]");

  if (!marked.length || !window.EventSource || !window.fetch) return;

  var topics = {};

  Array.prototype.forEach.call(marked, function (el) { topics[el.getAttribute("data-live")] = true; });

  var busy = {};

  var typingIn = function (el) {

    var a = document.activeElement;

    return a && el.contains(a) && (a.tagName === "TEXTAREA" || (a.tagName === "INPUT" && a.type !== "hidden")) && a.value;

  };

  var refresh = function (el) {

    var name = el.getAttribute("data-fragment");

    if (busy[name]) return;

    if (typingIn(el)) {

      el.addEventListener("focusout", function again() {

        el.removeEventListener("focusout", again);

        setTimeout(function () { refresh(el); }, 0);

      });

      return;

    }

    busy[name] = true;

    var url = new URL(location.href);

    url.searchParams.set("fragment", name);

    fetch(url.toString(), { credentials: "same-origin", headers: { "X-Fragment": name } })

      .then(function (r) { return r.ok ? r.text() : null; })

      .then(function (html) {

        busy[name] = false;

        if (html !== null && !typingIn(el)) el.innerHTML = html;

      })

      .catch(function () { busy[name] = false; });

  };

  var es = new EventSource("/live?topics=" + encodeURIComponent(Object.keys(topics).join(",")));

  es.addEventListener("change", function (e) {

    var d;

    try { d = JSON.parse(e.data); } catch (err) { return; }

    Array.prototype.forEach.call(marked, function (el) {

      if (el.getAttribute("data-live") === d.topic) refresh(el);

    });

  });

  // Missed more than the hub keeps: everything marked is fetched again.

  es.addEventListener("reload", function () {

    Array.prototype.forEach.call(marked, refresh);

  });

})();

mod live_stream;

        .route("/live", get(live_stream::stream))

#[tokio::test(flavor = "multi_thread")]

async fn live_streams_only_topics_the_viewer_may_hear_and_replays() {

    let h = hub().await;

    create(&h.app, "andy", "open").await;

    h.app.repos.create(&RepoName::parse("andy/shut").unwrap(), None).unwrap();

    db::repos::create(&h.app.db, "andy", "shut", "", hub_api::Visibility::Private).await.unwrap();

    let open = db::repos::get(&h.app.db, "andy", "open").await.unwrap().unwrap();

    let shut = db::repos::get(&h.app.db, "andy", "shut").await.unwrap().unwrap();

    let live = &h.app.events.live;

    let first = live.publish(&oxygenhub::events::topic::repo(open.id), "push");

    live.publish(&oxygenhub::events::topic::repo(shut.id), "push");

    live.publish(&oxygenhub::events::topic::proposal(open.id, 3), "review");

    // The first bytes of a stream, within a moment.

    let read = |path: String, last: Option<u64>| {

        let app = h.app.clone();

        async move {

            let mut req = Request::get(&path);

            if let Some(l) = last {

                req = req.header("last-event-id", l.to_string());

            }

            let res = oxygenhub::router(app).oneshot(req.body(Body::empty()).unwrap()).await.unwrap();

            let status = res.status();

            let mut body = res.into_body();

            let mut text = String::new();

            while let Ok(Some(Ok(frame))) = tokio::time::timeout(std::time::Duration::from_millis(300), body.frame()).await {

                if let Some(d) = frame.data_ref() {

                    text.push_str(&String::from_utf8_lossy(d));

                }

            }

            (status, text)

        }

    };

    let topics = format!("/live?topics=repo:{},repo:{}/proposal:3", open.id, open.id);

    // From the first event on: what came after it on these topics only.

    let (st, text) = read(topics.clone(), Some(first.id)).await;

    assert_eq!(st, StatusCode::OK);

    assert!(text.contains("event: change") && text.contains(&format!("\"topic\":\"repo:{}/proposal:3\"", open.id)), "{text}");

    assert!(!text.contains(&format!("\"topic\":\"repo:{}\"", shut.id)), "never another topic: {text}");

    assert!(!text.contains(&format!("id: {}\n", first.id)), "not what was already seen: {text}");

    // A private repository's topic, or someone else's, is refused whole.

    let (st, _) = read(format!("/live?topics=repo:{},repo:{}", open.id, shut.id), None).await;

    assert_eq!(st, StatusCode::FORBIDDEN);

    assert_eq!(read("/live?topics=user:1".into(), None).await.0, StatusCode::FORBIDDEN);

    assert_eq!(read("/live?topics=nonsense".into(), None).await.0, StatusCode::BAD_REQUEST);

    assert_eq!(read("/live?topics=".into(), None).await.0, StatusCode::BAD_REQUEST);

    // Too far behind: told to reload.

    for _ in 0..oxygenhub::events::REPLAY {

        live.publish("repo:999999", "push");

    }

    let (_, text) = read(topics, Some(first.id)).await;

    assert!(text.contains("event: reload"), "{text}");

}

- **L2 (done, #204):** `GET /live?topics=a,b` (1 to 16 topics; every one

  must be one the viewer may hear, read access to its repository or, for

  `user:<id>`, being that user, or the request is refused whole, saying

  nothing of which); SSE `change` events `{topic, version, kind}` with the

  bus id as the event id; `Last-Event-ID` replays the missed events on

  those topics or sends `reload`, as does falling behind the broadcast.

  The client in `hub.js`: `data-live` + `data-fragment` elements refetch

  `?fragment=<name>` from the page's own route and swap it in, waiting

  while someone types inside one.