Live pages L1: an event bus of topics, versions and a replay buffer 7e01a7665aa5

                app.events.live.publish(&crate::events::topic::diffusion(consumer.id), "bump");

                app.events.live.publish(&crate::events::topic::diffusion(upstream.id), "bump");

    /// Topics for live pages (oxygenhub#203).

    pub live: LiveBus,

        Events { deliveries: broadcast::channel(16).0, proposals: broadcast::channel(256).0, live: LiveBus::default() }

/// What changed, for live pages (docs/live.md, oxygenhub#203): a topic and

/// a kind, never the data. A page that hears its topic fetches the part

/// that changed from its normal route, which checks the viewer's access.

#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize)]

pub struct LiveEvent {

    /// Increasing across the hub: what `Last-Event-ID` resumes from.

    pub id: u64,

    pub topic: String,

    /// The topic's own version: how many events it has had.

    pub version: u64,

    /// What happened: `push`, `comment`, `review`, `check`, `jobs`, ...

    pub kind: String,

}

/// Topics (docs/live.md §4).

pub mod topic {

    pub fn repo(id: i64) -> String {

        format!("repo:{id}")

    }

    pub fn proposal(repo: i64, n: i64) -> String {

        format!("repo:{repo}/proposal:{n}")

    }

    pub fn issue(repo: i64, n: i64) -> String {

        format!("repo:{repo}/issue:{n}")

    }

    pub fn run(id: i64) -> String {

        format!("run:{id}")

    }

    pub fn user(id: i64) -> String {

        format!("user:{id}")

    }

    pub fn diffusion(repo: i64) -> String {

        format!("diffusion:{repo}")

    }

}

/// Events kept for replay: a reconnecting page catches up from these, or is

/// told to reload when it missed more.

pub const REPLAY: usize = 1024;

/// The live event bus: one broadcast that subscribers filter by topic, a

/// version per topic, and the last [`REPLAY`] events.

pub struct LiveBus {

    tx: broadcast::Sender<LiveEvent>,

    state: std::sync::Mutex<BusState>,

}

#[derive(Default)]

struct BusState {

    next_id: u64,

    versions: std::collections::HashMap<String, u64>,

    recent: std::collections::VecDeque<LiveEvent>,

}

impl Default for LiveBus {

    fn default() -> Self {

        LiveBus { tx: broadcast::channel(1024).0, state: std::sync::Mutex::new(BusState { next_id: 1, ..Default::default() }) }

    }

}

/// What a reconnecting subscriber gets: the events it missed, or `Reload`

/// when they are no longer kept.

#[derive(Debug, PartialEq, Eq)]

pub enum Replay {

    Events(Vec<LiveEvent>),

    Reload,

}

impl LiveBus {

    /// Publish an event on `topic`.

    pub fn publish(&self, topic: &str, kind: &str) -> LiveEvent {

        let event = {

            let mut s = self.state.lock().unwrap_or_else(|p| p.into_inner());

            let id = s.next_id;

            s.next_id += 1;

            let version = {

                let v = s.versions.entry(topic.to_string()).or_insert(0);

                *v += 1;

                *v

            };

            let e = LiveEvent { id, topic: topic.into(), version, kind: kind.into() };

            if s.recent.len() == REPLAY {

                s.recent.pop_front();

            }

            s.recent.push_back(e.clone());

            e

        };

        let _ = self.tx.send(event.clone());

        event

    }

    pub fn subscribe(&self) -> broadcast::Receiver<LiveEvent> {

        self.tx.subscribe()

    }

    /// The newest event id (0 before any).

    pub fn last_id(&self) -> u64 {

        self.state.lock().unwrap_or_else(|p| p.into_inner()).next_id - 1

    }

    /// The events on `topics` after `last_id`, or `Reload` when some after

    /// it are no longer kept.

    pub fn since(&self, last_id: u64, topics: &[String]) -> Replay {

        let s = self.state.lock().unwrap_or_else(|p| p.into_inner());

        let oldest = s.recent.front().map(|e| e.id).unwrap_or(s.next_id);

        if last_id + 1 < oldest {

            return Replay::Reload;

        }

        Replay::Events(s.recent.iter().filter(|e| e.id > last_id && topics.contains(&e.topic)).cloned().collect())

    }

    /// The topics a webhook event touches: the repository always, and the

    /// proposal or issue its data names.

    pub fn publish_hook(&self, repo_id: i64, event: &str, data: &serde_json::Value) {

        let kind = event.split('.').next().unwrap_or(event);

        self.publish(&topic::repo(repo_id), kind);

        let number = data["number"].as_i64().or_else(|| data["issue"]["number"].as_i64());

        let is_issue = event.starts_with("issue") || event == "comment.issue";

        let is_proposal = event.starts_with("proposal") || matches!(event, "review" | "comment" | "comment.proposal");

        match number {

            Some(n) if is_issue => {

                self.publish(&topic::issue(repo_id, n), kind);

            }

            Some(n) if is_proposal => {

                self.publish(&topic::proposal(repo_id, n), kind);

            }

            _ => {}

        }

    }

}

#[cfg(test)]

mod tests {

    use super::*;

    #[test]

    fn events_have_ids_and_versions_and_replay_or_say_reload() {

        let bus = LiveBus::default();

        let mut rx = bus.subscribe();

        let a = bus.publish("repo:1", "push");

        let b = bus.publish("repo:2", "push");

        let c = bus.publish("repo:1", "check");

        assert_eq!((a.id, b.id, c.id), (1, 2, 3));

        assert_eq!((a.version, b.version, c.version), (1, 1, 2), "versions count per topic");

        assert_eq!(rx.try_recv().unwrap(), a, "subscribers hear every event");

        assert_eq!(bus.since(1, &["repo:1".into()]), Replay::Events(vec![c.clone()]), "only the topics asked for");

        assert_eq!(bus.since(3, &["repo:1".into()]), Replay::Events(vec![]));

        for _ in 0..REPLAY {

            bus.publish("repo:3", "push");

        }

        assert_eq!(bus.since(1, &["repo:1".into()]), Replay::Reload, "too old to replay");

        assert_eq!(bus.last_id(), 3 + REPLAY as u64);

    }

    #[test]

    fn a_webhook_event_names_its_repository_and_its_proposal_or_issue() {

        let bus = LiveBus::default();

        bus.publish_hook(7, "review", &serde_json::json!({ "number": 4, "verdict": "approve" }));

        bus.publish_hook(7, "issue.opened", &serde_json::json!({ "issue": { "number": 9 } }));

        bus.publish_hook(7, "comment.issue", &serde_json::json!({ "number": 9 }));

        bus.publish_hook(7, "push", &serde_json::json!({ "updates": [] }));

        let Replay::Events(all) = bus.since(0, &["repo:7".into(), "repo:7/proposal:4".into(), "repo:7/issue:9".into()]) else { panic!() };

        let seen: Vec<(String, String)> = all.into_iter().map(|e| (e.topic, e.kind)).collect();

        assert_eq!(

            seen,

            [

                ("repo:7", "review"),

                ("repo:7/proposal:4", "review"),

                ("repo:7", "issue"),

                ("repo:7/issue:9", "issue"),

                ("repo:7", "comment"),

                ("repo:7/issue:9", "comment"),

                ("repo:7", "push"),

            ]

            .map(|(t, k)| (t.to_string(), k.to_string()))

        );

    }

}

    if !changed.is_empty() {

        app.events.live.publish(&crate::events::topic::run(run_id), "jobs");

    }

    // Live pages hear of it whether or not any hook does (oxygenhub#203).

    app.events.live.publish_hook(repo.id, event, &data);

        app.events.live.publish(&crate::events::topic::user(id), "notification");

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

    let runs_before = db::flow::runs(&h.app.db, api_row.id, 100, None).await.unwrap().len();

    // The push's run is made in the background: wait for it, or pass_all

    // can finish before it exists and v5 never passes (a flake under load).

    for _ in 0..300 {

        if db::flow::runs(&h.app.db, api_row.id, 100, None).await.unwrap().len() > runs_before {

            break;

        }

        tokio::time::sleep(std::time::Duration::from_millis(20)).await;

    }

    // Live pages hear of the push (oxygenhub#203).

    let heard = |topic: String| match h.app.events.live.since(0, &[topic]) {

        oxygenhub::events::Replay::Events(e) => e.into_iter().map(|e| e.kind).collect::<Vec<_>>(),

        oxygenhub::events::Replay::Reload => vec!["reload".into()],

    };

    assert!(heard(oxygenhub::events::topic::repo(row.id)).contains(&"push".to_string()), "the push is on the repository's topic");

    assert!(heard(oxygenhub::events::topic::run(run.id)).contains(&"jobs".to_string()), "the run's topic hears its jobs");

## 5a. Status

- **L1 (done, #203):** `events::LiveBus` in the hub: one broadcast of

  `{id, topic, version, kind}` that subscribers filter by topic (a channel

  per topic buys nothing at this size), a version per topic, and the last

  1024 events for `Last-Event-ID` (older: "reload"). Published from

  `hooks::emit` (the repository's topic, and the proposal's or issue's the

  event names), `notify::tell` (`user:<id>`), Flow job changes

  (`run:<id>`) and Diffusion bumps (`diffusion:<repo id>`, both sides).