Carry events between the hub's processes on an event bus 33d98147511b

//! The event bus (docs/roles.md, #273): publish and subscribe for the

//! events live pages and workers react to, in this process and in every

//! other process sharing the hub's database.

//!

//! An event says that something happened (a push moved a bookmark, a

//! proposal's check was recorded, a run's jobs changed, a webhook delivery

//! is waiting, a choice in a conflict-resolution room) and carries ids and

//! kinds, never the data: subscribers read the rows they need. Delivery is

//! at most once. Anything that must not be lost is a row in a table (a

//! delivery, a job), and its event only says "look now"; every subscriber

//! also looks by itself from time to time, or when it starts.

//!

//! In process, each [`Topic`] is a `tokio::sync::broadcast` channel: what

//! the single-node edition has always done, and all there is on SQLite (one

//! process by definition). On PostgreSQL the bus also carries each event to

//! the other processes: [`Bus::publish`] delivers it here at once and sends

//! it with `pg_notify` through the pool, on one channel per topic named for

//! the hub's schema (two hubs in two schemas of one database never hear

//! each other). Each process listens on those channels on one connection of

//! its own, outside the pool, and relays what it hears into its local

//! channels without publishing it again. Every event is sent with this

//! process's origin id, and a process skips its own: each subscriber hears

//! an event once, in every process. If the listening connection is lost, the

//! process listens again on a new one, says that events published meanwhile

//! were missed, and nudges its own webhook worker to look now.

use crate::events::{hook_topics, LiveBus};

use crate::live::{RoomEvent, RoomKey};

use crate::Shared;

use hub_api::ProposalEvent;

use tokio::sync::broadcast;

/// What the bus carries: one local channel per topic and, on PostgreSQL,

/// one notification channel per topic.

#[derive(Clone, Copy, Debug, PartialEq, Eq, Hash)]

pub enum Topic {

    /// A webhook delivery is waiting (the worker's nudge).

    Deliveries,

    /// A proposal's event: a check recorded on a projected state.

    Proposals,

    /// Live pages' topics (docs/live.md): pushes, checks, runs' jobs,

    /// comments, reviews, notifications, Diffusion's bumps.

    Live,

    /// Choices and landings in conflict-resolution rooms.

    Rooms,

}

impl Topic {

    pub const ALL: [Topic; 4] = [Topic::Deliveries, Topic::Proposals, Topic::Live, Topic::Rooms];

    pub fn name(self) -> &'static str {

        match self {

            Topic::Deliveries => "deliveries",

            Topic::Proposals => "proposals",

            Topic::Live => "live",

            Topic::Rooms => "rooms",

        }

    }

    /// How many events a slow local subscriber may fall behind by.

    fn capacity(self) -> usize {

        match self {

            Topic::Deliveries => 16,

            Topic::Proposals | Topic::Rooms => 256,

            Topic::Live => 1024,

        }

    }

}

/// An event: its topic is its kind, and it carries ids and kinds only.

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

#[serde(tag = "t", rename_all = "snake_case")]

pub enum Event {

    /// A webhook delivery is waiting: the worker should look now.

    DeliveryWaiting,

    /// Something about a proposal of repository `repo` (the proposal

    /// page's stream, hub spec §13.1): a check recorded on a state.

    Proposal { repo: i64, event: ProposalEvent },

    /// A live page's topic changed (`repo:1`, `run:5`, …) with a kind

    /// (`push`, `check`, `jobs`, `comment`, …).

    Live { topic: String, kind: String },

    /// A viewer's choice, or the landing, in a conflict-resolution room.

    Room { room: RoomKey, event: RoomEvent },

}

impl Event {

    pub fn topic(&self) -> Topic {

        match self {

            Event::DeliveryWaiting => Topic::Deliveries,

            Event::Proposal { .. } => Topic::Proposals,

            Event::Live { .. } => Topic::Live,

            Event::Room { .. } => Topic::Rooms,

        }

    }

}

/// This process's end of the bus.

pub struct Bus {

    /// Sent with every event this process publishes, so it skips its own

    /// when they come back from the database.

    #[cfg_attr(not(feature = "postgres"), allow(dead_code))]

    origin: String,

    /// One broadcast per topic, in [`Topic::ALL`]'s order.

    channels: Vec<broadcast::Sender<Event>>,

    /// Live pages' events as this process numbers them, with the last few

    /// kept for `Last-Event-ID` (docs/live.md).

    pub live: LiveBus,

    #[cfg(feature = "postgres")]

    remote: pg::Remote,

}

impl Default for Bus {

    fn default() -> Self {

        Bus {

            origin: crate::auth::random_hex(8),

            channels: Topic::ALL.iter().map(|t| broadcast::channel(t.capacity()).0).collect(),

            live: LiveBus::default(),

            #[cfg(feature = "postgres")]

            remote: pg::Remote::default(),

        }

    }

}

impl Bus {

    /// Publish `event`: this process's subscribers hear it now, and every

    /// other process's soon after (PostgreSQL), at most once.

    pub fn publish(&self, event: Event) {

        #[cfg(feature = "postgres")]

        self.remote.send(&self.origin, &event);

        self.deliver(event);

    }

    /// Hear the events on `topic` published from now on, in this process

    /// and the others.

    pub fn subscribe(&self, topic: Topic) -> broadcast::Receiver<Event> {

        self.channels[Self::index(topic)].subscribe()

    }

    /// A live page's `topic` changed with `kind` (docs/live.md).

    pub fn live(&self, topic: &str, kind: &str) {

        self.publish(Event::Live { topic: topic.to_string(), kind: kind.to_string() });

    }

    /// Live pages hear of a webhook event: the repository's topic, and the

    /// proposal's or issue's its data names.

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

        for (topic, kind) in hook_topics(repo_id, event, data) {

            self.live(&topic, &kind);

        }

    }

    /// A webhook delivery is waiting.

    pub fn delivery_waiting(&self) {

        self.publish(Event::DeliveryWaiting);

    }

    /// Announce something about a proposal of a repository.

    pub fn proposal(&self, repo_id: i64, event: ProposalEvent) {

        self.publish(Event::Proposal { repo: repo_id, event });

    }

    /// Tell this process's subscribers only: [`Bus::publish`]'s local half,

    /// and what the relay does with another process's event.

    fn deliver(&self, event: Event) {

        if let Event::Live { topic, kind } = &event {

            self.live.publish(topic, kind);

        }

        let _ = self.channels[Self::index(event.topic())].send(event);

    }

    fn index(topic: Topic) -> usize {

        Topic::ALL.iter().position(|t| *t == topic).expect("every topic is in ALL")

    }

    /// The process id of the database session listening for other

    /// processes' events, while it listens (for tests that end it).

    #[cfg(feature = "postgres")]

    pub fn listener_pid(&self) -> Option<i32> {

        *self.remote.pid.lock().unwrap_or_else(|p| p.into_inner())

    }

}

/// Start carrying events between processes: on PostgreSQL, the sender and

/// the listener; on SQLite, nothing (one process).

pub async fn start(app: &Shared) -> anyhow::Result<()> {

    #[cfg(feature = "postgres")]

    pg::start(app).await?;

    #[cfg(not(feature = "postgres"))]

    let _ = app;

    Ok(())

}

/// The most a notification's payload may be: PostgreSQL takes fewer than

/// 8000 bytes.

pub const MAX_PAYLOAD: usize = 7999;

/// An event as sent between processes: who sent it, and the event.

#[derive(serde::Serialize, serde::Deserialize)]

struct Envelope {

    o: String,

    e: Event,

}

/// The payload for `event` from `origin`, within [`MAX_PAYLOAD`]: a check's

/// summary too long for it is cut (the proposal page reads the check for

/// the rest); anything else too long (a choice's own text) is not sent to

/// other processes (`None`).

#[cfg_attr(not(feature = "postgres"), allow(dead_code))]

fn encode(origin: &str, event: &Event) -> Option<String> {

    let mut envelope = Envelope { o: origin.to_string(), e: event.clone() };

    loop {

        let text = serde_json::to_string(&envelope).ok()?;

        if text.len() <= MAX_PAYLOAD {

            return Some(text);

        }

        let Event::Proposal { event: ProposalEvent { check: Some(check), .. }, .. } = &mut envelope.e else { return None };

        if check.summary.is_empty() {

            return None;

        }

        // Escaping may make the summary longer in JSON than as text: cut

        // by what is over, and again if still over.

        let over = text.len() - MAX_PAYLOAD;

        let mut keep = check.summary.trim_end_matches('…').len().saturating_sub(over + 3);

        while !check.summary.is_char_boundary(keep) {

            keep -= 1;

        }

        check.summary.truncate(keep);

        check.summary.push('…');

        if keep == 0 {

            check.summary.clear();

        }

    }

}

/// The event in `payload`, unless this process (`origin`) sent it or it

/// cannot be read.

#[cfg_attr(not(feature = "postgres"), allow(dead_code))]

fn decode(origin: &str, payload: &str) -> Option<Event> {

    match serde_json::from_str::<Envelope>(payload) {

        Ok(e) if e.o != origin => Some(e.e),

        Ok(_) => None,

        Err(err) => {

            tracing::warn!("bus: an event that could not be read: {err}");

            None

        }

    }

}

/// The notification channel for `topic` of the hub in `schema`: a stable

/// hash of the schema, so channel names are plain and short whatever the

/// schema is called.

#[cfg_attr(not(feature = "postgres"), allow(dead_code))]

fn channel(schema: &str, topic: Topic) -> String {

    let h = blake3::hash(format!("oxygenhub bus\0{schema}").as_bytes());

    format!("oxyhub_{}_{}", &h.to_hex()[..16], topic.name())

}

#[cfg(feature = "postgres")]

mod pg {

    use super::{channel, decode, encode, Event, Topic};

    use crate::{App, Shared};

    use sqlx::postgres::{PgListener, PgPool, PgPoolOptions};

    use std::sync::{Mutex, OnceLock};

    use std::time::Duration;

    use tokio::sync::mpsc;

    /// How long a quiet listening connection goes before it is asked to

    /// answer, and how long it has to answer.

    const CHECK: Duration = Duration::from_secs(5);

    /// How long to wait before listening again after failing to connect.

    const RETRY: Duration = Duration::from_secs(2);

    /// Events waiting to be sent to other processes; past this, new ones

    /// are dropped (and said so): the bus is at most once.

    const QUEUE: usize = 4096;

    #[derive(Default)]

    pub(super) struct Remote {

        /// The hub's schema and the queue to the sender, once started.

        out: OnceLock<(String, mpsc::Sender<(String, String)>)>,

        /// The listening session's process id, while it listens.

        pub(super) pid: Mutex<Option<i32>>,

    }

    impl Remote {

        /// Queue `event` for the other processes. Before [`start`], or

        /// for an event too large to send, it stays in this process.

        pub(super) fn send(&self, origin: &str, event: &Event) {

            let Some((schema, tx)) = self.out.get() else { return };

            let Some(payload) = encode(origin, event) else {

                tracing::warn!("bus: a {} event too large to send to other processes stays in this one", event.topic().name());

                return;

            };

            if tx.try_send((channel(schema, event.topic()), payload)).is_err() {

                tracing::warn!("bus: too many events waiting to go to other processes; one was dropped");

            }

        }

    }

    pub(super) async fn start(app: &Shared) -> anyhow::Result<()> {

        let schema: String = sqlx::query_scalar("SELECT CAST(current_schema() AS TEXT)").fetch_one(&app.db).await?;

        let (tx, rx) = mpsc::channel(QUEUE);

        if app.bus.remote.out.set((schema.clone(), tx)).is_err() {

            return Ok(());

        }

        tokio::spawn(send_all(app.db.clone(), rx));

        tokio::spawn(listen(app.clone(), schema));

        Ok(())

    }

    /// Send queued events in order, through the pool.

    async fn send_all(db: crate::db::Pool, mut rx: mpsc::Receiver<(String, String)>) {

        while let Some((channel, payload)) = rx.recv().await {

            if let Err(e) = sqlx::query("SELECT pg_notify($1, $2)").bind(&channel).bind(&payload).execute(&db).await {

                tracing::warn!("bus: sending an event to other processes: {e}");

            }

        }

    }

    /// A connection of its own, outside the pool, listening on every topic.

    async fn connect(app: &App, channels: &[String]) -> anyhow::Result<(PgPool, PgListener, i32)> {

        let options = (*app.db.connect_options()).clone();

        let pool = PgPoolOptions::new().max_connections(1).max_lifetime(None).idle_timeout(None).connect_with(options).await?;

        let mut listener = PgListener::connect_with(&pool).await?;

        listener.ignore_pool_close_event(true);

        // A lost connection ends this listener; [`listen`] makes another.

        listener.eager_reconnect(false);

        listener.listen_all(channels.iter().map(String::as_str)).await?;

        let pid: i32 = sqlx::query_scalar("SELECT pg_backend_pid()").fetch_one(&mut listener).await?;

        Ok((pool, listener, pid))

    }

    /// Listen for other processes' events and relay them here, for as long

    /// as the process runs, on a new connection whenever one is lost.

    async fn listen(app: Shared, schema: String) {

        let channels: Vec<String> = Topic::ALL.iter().map(|t| channel(&schema, *t)).collect();

        let mut lost = false;

        loop {

            let (pool, mut listener, pid) = match tokio::time::timeout(CHECK * 2, connect(&app, &channels)).await {

                Ok(Ok(c)) => c,

                Ok(Err(e)) => {

                    tracing::warn!("bus: listening for other processes' events: {e}");

                    tokio::time::sleep(RETRY).await;

                    continue;

                }

                Err(_) => {

                    tracing::warn!("bus: listening for other processes' events: timed out");

                    tokio::time::sleep(RETRY).await;

                    continue;

                }

            };

            *app.bus.remote.pid.lock().unwrap_or_else(|p| p.into_inner()) = Some(pid);

            if lost {

                tracing::warn!("bus: listening again; events other processes published meanwhile were missed");

                // Look now for what a missed nudge would have said.

                app.bus.deliver(Event::DeliveryWaiting);

            }

            loop {

                tokio::select! {

                    heard = listener.try_recv() => match heard {

                        Ok(Some(n)) => relay(&app, n.payload()),

                        Ok(None) => {

                            tracing::warn!("bus: the connection listening for other processes' events was lost");

                            break;

                        }

                        Err(e) => {

                            tracing::warn!("bus: the connection listening for other processes' events failed: {e}");

                            break;

                        }

                    },

                    _ = tokio::time::sleep(CHECK) => {

                        let ping = sqlx::query_scalar::<_, i32>("SELECT 1").fetch_one(&mut listener);

                        if !matches!(tokio::time::timeout(CHECK, ping).await, Ok(Ok(_))) {

                            tracing::warn!("bus: the connection listening for other processes' events stopped answering");

                            break;

                        }

                    }

                }

            }

            lost = true;

            *app.bus.remote.pid.lock().unwrap_or_else(|p| p.into_inner()) = None;

            drop(listener);

            let _ = tokio::time::timeout(CHECK, pool.close()).await;

        }

    }

    /// Another process's event, told to this one's subscribers (and its

    /// rooms), never published again.

    fn relay(app: &App, payload: &str) {

        let Some(event) = decode(&app.bus.origin, payload) else { return };

        if let Event::Room { room, event } = &event {

            app.rooms.apply(room, event);

        }

        app.bus.deliver(event);

    }

}

#[cfg(test)]

mod tests {

    use super::*;

    use crate::events::Replay;

    fn check(summary: &str) -> ProposalEvent {

        ProposalEvent {

            number: 0,

            kind: "check".into(),

            check: Some(hub_api::Check {

                state: "s".into(),

                name: "build".into(),

                status: hub_api::CheckStatus::Passed,

                summary: summary.into(),

                url: String::new(),

                reporter: "ci".into(),

                seal: None,

                started_ms: 0,

                finished_ms: None,

            }),

        }

    }

    #[test]

    fn subscribers_hear_their_topic_in_order_and_nothing_else() {

        let bus = Bus::default();

        let mut deliveries = bus.subscribe(Topic::Deliveries);

        let mut proposals = bus.subscribe(Topic::Proposals);

        let mut live = bus.subscribe(Topic::Live);

        bus.delivery_waiting();

        bus.proposal(7, check("ok"));

        bus.live("repo:7", "push");

        bus.live("repo:7", "check");

        assert_eq!(deliveries.try_recv().unwrap(), Event::DeliveryWaiting);

        assert!(deliveries.try_recv().is_err(), "once");

        assert_eq!(proposals.try_recv().unwrap(), Event::Proposal { repo: 7, event: check("ok") });

        assert!(proposals.try_recv().is_err());

        assert_eq!(live.try_recv().unwrap(), Event::Live { topic: "repo:7".into(), kind: "push".into() });

        assert_eq!(live.try_recv().unwrap(), Event::Live { topic: "repo:7".into(), kind: "check".into() });

        // Live events are numbered for replay as they are published.

        let Replay::Events(kept) = bus.live.since(0, &["repo:7".into()]) else { panic!() };

        assert_eq!(kept.iter().map(|e| (e.id, e.version, e.kind.as_str())).collect::<Vec<_>>(), [(1, 1, "push"), (2, 2, "check")]);

    }

    #[test]

    fn a_webhook_event_is_heard_on_its_live_topics() {

        let bus = Bus::default();

        let mut live = bus.subscribe(Topic::Live);

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

        assert_eq!(live.try_recv().unwrap(), Event::Live { topic: "repo:7".into(), kind: "review".into() });

        assert_eq!(live.try_recv().unwrap(), Event::Live { topic: "repo:7/proposal:4".into(), kind: "review".into() });

    }

    #[test]

    fn events_travel_with_their_origin_and_a_process_skips_its_own() {

        let room = RoomKey { repo_id: 1, bookmark: "main".into(), path: "a.txt".into() };

        let events = [

            Event::DeliveryWaiting,

            Event::Proposal { repo: 7, event: check("42 tests") },

            Event::Live { topic: "repo:7".into(), kind: "push".into() },

            Event::Room {

                room,

                event: RoomEvent::Choice {

                    region: "1".into(),

                    choice: crate::live::Choice { choice: "side2".into(), text: String::new(), by: "andy".into() },

                    client: "tab-a".into(),

                },

            },

        ];

        for e in events {

            let payload = encode("aaaa", &e).unwrap();

            assert_eq!(decode("bbbb", &payload), Some(e.clone()), "another process hears it");

            assert_eq!(decode("aaaa", &payload), None, "its sender skips it");

        }

        assert_eq!(decode("bbbb", "not json"), None);

    }

    #[test]

    fn payloads_fit_a_notification() {

        // A long check summary is cut to fit, on a character boundary.

        let long = "é\"".repeat(6000);

        let payload = encode("aaaa", &Event::Proposal { repo: 7, event: check(&long) }).unwrap();

        assert!(payload.len() <= MAX_PAYLOAD, "{}", payload.len());

        let Some(Event::Proposal { event, .. }) = decode("bbbb", &payload) else { panic!() };

        let summary = event.check.unwrap().summary;

        assert!(summary.ends_with('…') && long.starts_with(summary.trim_end_matches('…')), "{}", summary.len());

        // A choice's own text that does not fit stays in its process.

        let big = Event::Room {

            room: RoomKey { repo_id: 1, bookmark: "main".into(), path: "a".into() },

            event: RoomEvent::Choice {

                region: "1".into(),

                choice: crate::live::Choice { choice: "text".into(), text: "x".repeat(9000), by: "andy".into() },

                client: "c".into(),

            },

        };

        assert_eq!(encode("aaaa", &big), None);

    }

    #[test]

    fn channels_are_per_schema_and_topic_and_plain() {

        assert_eq!(channel("hub_a", Topic::Live), channel("hub_a", Topic::Live));

        assert_ne!(channel("hub_a", Topic::Live), channel("hub_b", Topic::Live));

        assert_ne!(channel("hub_a", Topic::Live), channel("hub_a", Topic::Rooms));

        for t in Topic::ALL {

            let c = channel(&"x".repeat(200), t);

            assert!(c.len() < 64 && c.chars().all(|c| c.is_ascii_lowercase() || c.is_ascii_digit() || c == '_'), "{c}");

        }

    }

}

//! The event bus (docs/roles.md, #273): a hub hears its own events once;

//! on PostgreSQL, two hubs on one database hear each other's: a push made

//! through a web process reaches live pages and the webhook worker of

//! another process, and a process whose listening connection is ended

//! listens again on a new one.

use oxygenhub::bus::{Event, Topic};

use oxygenhub::config::Config;

use std::path::Path;

use std::time::Duration;

fn hermetic() {

    static ONCE: std::sync::Once = std::sync::Once::new();

    static DIR: std::sync::OnceLock<tempfile::TempDir> = std::sync::OnceLock::new();

    ONCE.call_once(|| {

        let d = tempfile::tempdir().unwrap();

        std::env::set_var("OXY_HOME", d.path());

        DIR.set(d).ok();

    });

}

fn config(dir: &Path, roles: &[&str]) -> Config {

    let mut config = Config::load(dir).unwrap();

    config.server.roles = roles.iter().map(|r| r.to_string()).collect();

    config.registration.setup_code = "test-setup".into();

    config

}

/// The next event on `rx` that `want` picks, within `limit`.

async fn heard<T: Clone>(rx: &mut tokio::sync::broadcast::Receiver<T>, limit: Duration, want: impl Fn(&T) -> bool) -> Option<T> {

    tokio::time::timeout(limit, async {

        loop {

            match rx.recv().await {

                Ok(e) if want(&e) => return Some(e),

                Ok(_) | Err(tokio::sync::broadcast::error::RecvError::Lagged(_)) => continue,

                Err(_) => return None,

            }

        }

    })

    .await

    .ok()

    .flatten()

}

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

async fn a_hub_hears_its_own_events_once() {

    hermetic();

    let dir = tempfile::tempdir().unwrap();

    let app = oxygenhub::build(config(dir.path(), &[])).await.unwrap();

    let mut live = app.bus.subscribe(Topic::Live);

    let mut deliveries = app.bus.subscribe(Topic::Deliveries);

    app.bus.live("repo:1", "push");

    app.bus.delivery_waiting();

    let second = Duration::from_secs(1);

    assert_eq!(heard(&mut live, second, |_| true).await, Some(Event::Live { topic: "repo:1".into(), kind: "push".into() }));

    assert_eq!(heard(&mut deliveries, second, |_| true).await, Some(Event::DeliveryWaiting));

    // On PostgreSQL the event also went out through the database and came

    // back to this process's listener, which skips its own.

    tokio::time::sleep(Duration::from_millis(500)).await;

    assert!(live.try_recv().is_err(), "heard once");

    assert!(deliveries.try_recv().is_err(), "heard once");

    assert_eq!(app.bus.live.last_id(), 1);

}

#[cfg(feature = "postgres")]

mod postgres {

    use super::*;

    use axum::body::Body;

    use axum::http::{header, Request, StatusCode};

    use axum::routing::post;

    use http_body_util::BodyExt;

    use oxy_core::repo::Repo;

    use oxygenhub::Shared;

    use std::sync::{Arc, Mutex};

    use tower::ServiceExt;

    fn pg(dir: &Path, schema: &str, roles: &[&str]) -> Config {

        let mut c = config(dir, roles);

        c.database.schema = schema.to_string();

        c

    }

    async fn call(

        app: &Shared,

        method: &str,

        path: &str,

        auth: (Option<&str>, Option<&str>),

        body: Option<(&str, String)>,

    ) -> (StatusCode, String, Option<String>) {

        let mut req = Request::builder().method(method).uri(path);

        if let Some(b) = auth.0 {

            req = req.header(header::AUTHORIZATION, format!("Bearer {b}"));

        }

        if let Some(c) = auth.1 {

            req = req.header(header::COOKIE, format!("oxyhub={c}"));

        }

        let body = match body {

            Some((kind, b)) => {

                req = req.header(header::CONTENT_TYPE, kind);

                Body::from(b)

            }

            None => Body::empty(),

        };

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

        let status = res.status();

        let cookie = res

            .headers()

            .get(header::SET_COOKIE)

            .and_then(|v| v.to_str().ok())

            .and_then(|v| v.split(';').next())

            .and_then(|kv| kv.split_once('='))

            .map(|(_, v)| v.to_string());

        let bytes = res.into_body().collect().await.unwrap().to_bytes();

        (status, String::from_utf8_lossy(&bytes).into_owned(), cookie)

    }

    /// A webhook receiver that records the events it gets.

    async fn receiver() -> (String, Arc<Mutex<Vec<String>>>) {

        let seen: Arc<Mutex<Vec<String>>> = Arc::default();

        let s2 = seen.clone();

        let router = axum::Router::new().route(

            "/hook",

            post(move |headers: header::HeaderMap| {

                let s2 = s2.clone();

                async move {

                    let event = headers.get("x-oxyhub-event").and_then(|v| v.to_str().ok()).unwrap_or("").to_string();

                    s2.lock().unwrap().push(event);

                    StatusCode::OK

                }

            }),

        );

        let listener = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();

        let port = listener.local_addr().unwrap().port();

        tokio::spawn(async move { axum::serve(listener, router).await.unwrap() });

        (format!("http://127.0.0.1:{port}/hook"), seen)

    }

    /// A push through a web-only hub: a live page's subscriber on a

    /// worker-only hub hears it, and that hub's webhook worker, nudged over

    /// the bus, delivers its webhook at once rather than on its

    /// once-a-minute pass. The web hub hears its own push once.

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

    async fn a_push_through_one_hub_reaches_the_other_hubs_pages_and_worker() {

        hermetic();

        let (da, dbb) = (tempfile::tempdir().unwrap(), tempfile::tempdir().unwrap());

        let schema = format!("hub_bus_{}", da.path().file_name().unwrap().to_string_lossy());

        let mut ca = pg(da.path(), &schema, &["web"]);

        ca.auth.push_tokens = vec!["site".into()];

        ca.registration.mode = "open".into();

        ca.outbound.allow = vec!["http://127.0.0.1".into()];

        let tcp = tokio::net::TcpListener::bind("127.0.0.1:0").await.unwrap();

        let base = format!("http://127.0.0.1:{}", tcp.local_addr().unwrap().port());

        ca.server.base_url = base.clone();

        let a = oxygenhub::build(ca).await.unwrap();

        let router = oxygenhub::router(a.clone());

        tokio::spawn(async move { axum::serve(tcp, router).await.unwrap() });

        let mut cb = pg(dbb.path(), &schema, &["worker"]);

        cb.outbound.allow = vec!["http://127.0.0.1".into()];

        let b = oxygenhub::build(cb).await.unwrap();

        // An account, an admin token, a repository and its webhook, on A.

        let form = "application/x-www-form-urlencoded";

        let (_, _, cookie) = call(

            &a,

            "POST",

            "/register",

            (None, None),

            Some((form, "username=andy&email=a%40example.com&password=correct-horse-battery&setup=test-setup".into())),

        )

        .await;

        let andy = cookie.expect("signed in");

        let (_, page, _) = call(&a, "GET", "/settings/tokens", (None, Some(&andy)), None).await;

        let at = page.find("name=\"_csrf\" value=\"").unwrap() + "name=\"_csrf\" value=\"".len();

        let csrf = page[at..].split('"').next().unwrap().to_string();

        let (_, page, _) =

            call(&a, "POST", "/settings/tokens", (None, Some(&andy)), Some((form, format!("_csrf={csrf}&name=t&scopes=repo%3Aadmin"))))

                .await;

        let token = format!("oxyhub_{}", page.split("oxyhub_").nth(1).unwrap().split('<').next().unwrap().trim());

        let json = "application/json";

        let (st, body, _) = call(

            &a,

            "POST",

            "/api/v1/repos",

            (Some(&token), None),

            Some((json, r#"{"owner":"andy","name":"demo","visibility":"public"}"#.into())),

        )

        .await;

        assert_eq!(st, StatusCode::CREATED, "{body}");

        let (hook, seen) = receiver().await;

        let (st, body, _) = call(

            &a,

            "POST",

            "/api/v1/repos/andy/demo/hooks",

            (Some(&token), None),

            Some((json, format!(r#"{{"url":"{hook}","secret":"s","events":"push"}}"#))),

        )

        .await;

        assert_eq!(st, StatusCode::CREATED, "{body}");

        let repo = oxygenhub::db::repos::get(&a.db, "andy", "demo").await.unwrap().unwrap();

        let topic = oxygenhub::events::topic::repo(repo.id);

        // B's worker leads the webhook worker and has made its first pass.

        let start = std::time::Instant::now();

        while !b.leader.leading().contains(&"hooks.worker") {

            assert!(start.elapsed() < Duration::from_secs(10), "{:?}", b.leader.leading());

            tokio::time::sleep(Duration::from_millis(100)).await;

        }

        tokio::time::sleep(Duration::from_millis(500)).await;

        let mut b_live = b.bus.live.subscribe();

        let work = tempfile::tempdir().unwrap();

        let mut client = Repo::init(work.path()).unwrap();

        client.config.set("remote.origin.url", &format!("{base}/andy/demo")).unwrap();

        client.config.set("remote.origin.token", "site").unwrap();

        client.save_config().unwrap();

        std::fs::write(work.path().join("a.txt"), b"a\n").unwrap();

        client.snapshot("t").unwrap();

        client.commit("base", &[]).unwrap();

        assert!(client.push("origin", false).unwrap().iter().all(|r| r.ok));

        let push = heard(&mut b_live, Duration::from_secs(10), |e| e.topic == topic).await.expect("B hears A's push");

        assert_eq!(push.kind, "push");

        let start = std::time::Instant::now();

        while seen.lock().unwrap().is_empty() {

            assert!(start.elapsed() < Duration::from_secs(10), "B's worker was not nudged");

            tokio::time::sleep(Duration::from_millis(100)).await;

        }

        assert_eq!(*seen.lock().unwrap(), ["push"]);

        eprintln!("delivered by the other process in {:.1} s", start.elapsed().as_secs_f64());

        // A heard its own push once, not again from the database.

        tokio::time::sleep(Duration::from_millis(500)).await;

        let oxygenhub::events::Replay::Events(on_a) = a.bus.live.since(0, std::slice::from_ref(&topic)) else { panic!() };

        assert_eq!(on_a.iter().filter(|e| e.kind == "push").count(), 1, "{on_a:?}");

    }

    /// Proposal events and resolver rooms' choices cross over; a process

    /// whose listening connection is ended listens again, looks for

    /// deliveries it may have missed, and hears the other again.

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

    async fn two_hubs_hear_each_other_and_listen_again_after_losing_the_connection() {

        hermetic();

        let (da, dbb) = (tempfile::tempdir().unwrap(), tempfile::tempdir().unwrap());

        let schema = format!("hub_bus_{}", da.path().file_name().unwrap().to_string_lossy());

        let a = oxygenhub::build(pg(da.path(), &schema, &[])).await.unwrap();

        let b = oxygenhub::build(pg(dbb.path(), &schema, &[])).await.unwrap();

        // A hub in another schema of the same database hears neither.

        let dc = tempfile::tempdir().unwrap();

        let c = oxygenhub::build(pg(dc.path(), &format!("{schema}_other"), &[])).await.unwrap();

        let mut c_live = c.bus.subscribe(Topic::Live);

        let mut b_proposals = b.bus.subscribe(Topic::Proposals);

        let check = hub_api::ProposalEvent { number: 0, kind: "check".into(), check: None };

        a.bus.proposal(7, check.clone());

        let got = heard(&mut b_proposals, Duration::from_secs(5), |_| true).await;

        assert_eq!(got, Some(Event::Proposal { repo: 7, event: check }));

        // A choice made on A reaches the same room's viewers on B.

        let key = oxygenhub::live::RoomKey { repo_id: 1, bookmark: "main".into(), path: "a.txt".into() };

        let (mut b_room, _) = b.rooms.join(key.clone(), "tab-b", "bea");

        let choice = oxygenhub::live::Choice { choice: "side2".into(), text: String::new(), by: "andy".into() };

        let event = oxygenhub::live::RoomEvent::Choice { region: "1".into(), choice: choice.clone(), client: "tab-a".into() };

        a.bus.publish(Event::Room { room: key.clone(), event });

        let got = heard(&mut b_room, Duration::from_secs(5), |e| matches!(e, oxygenhub::live::LiveEvent::Choice { .. })).await;

        assert_eq!(got, Some(oxygenhub::live::LiveEvent::Choice { region: "1".into(), choice, client: "tab-a".into() }));

        // B's listening connection is ended, as when the network drops.

        let pid = b.bus.listener_pid().expect("b listens");

        let mut b_deliveries = b.bus.subscribe(Topic::Deliveries);

        let ended: bool = sqlx::query_scalar("SELECT pg_terminate_backend($1)").bind(pid).fetch_one(&a.db).await.unwrap();

        assert!(ended);

        let start = std::time::Instant::now();

        while b.bus.listener_pid().is_none_or(|p| p == pid) {

            assert!(start.elapsed() < Duration::from_secs(15), "b did not listen again");

            tokio::time::sleep(Duration::from_millis(100)).await;

        }

        eprintln!("listening again in {:.1} s", start.elapsed().as_secs_f64());

        // It looks now for deliveries a missed nudge would have announced.

        assert_eq!(heard(&mut b_deliveries, Duration::from_secs(5), |_| true).await, Some(Event::DeliveryWaiting));

        let mut b_live = b.bus.subscribe(Topic::Live);

        a.bus.live("repo:9", "push");

        let got = heard(&mut b_live, Duration::from_secs(5), |_| true).await;

        assert_eq!(got, Some(Event::Live { topic: "repo:9".into(), kind: "push".into() }));

        tokio::time::sleep(Duration::from_millis(300)).await;

        assert!(c_live.try_recv().is_err(), "another schema's hub hears nothing");

    }

}

- **An event bus between the hub's processes (#273):** on PostgreSQL, what one process publishes (a push, a check, a run's jobs, a comment, a waiting webhook delivery, a choice in the conflict resolver) reaches every other process over `LISTEN/NOTIFY`, so live pages on any web process see a push made through another, and a separate worker delivers a push's webhooks at once instead of on its once-a-minute pass. Nothing to configure; on SQLite the bus stays in process. Events say "look now" and are delivered at most once; a process that loses its listening connection listens again on a new one and logs that events may have been missed. A live page that reconnects with an event id its process never gave (another web process's, or one from before a restart) is now told to reload instead of waiting for newer ids. In the conflict resolver, choices and landings cross processes; who is present stays per process. See docs/roles.md.

    app.bus.proposal(row.id, ProposalEvent { number: 0, kind: "check".into(), check: Some(api.clone()) });

    app.bus.live(&crate::events::topic::proposal(row.id, p.number), "assistant");

                app.bus.live(&crate::events::topic::proposal(row.id, p.number), "assistant");

    app.bus.live(&crate::events::topic::proposal(row.id, p.number), "assistant");

                app.bus.live(&crate::events::topic::proposal(row.id, p.number), "comment");

                app.bus.live(&crate::events::topic::proposal(row.id, p.number), "assistant");

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

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

//! Live pages' events (docs/live.md, oxygenhub#203): what a topic is, and

//! the replay buffer that numbers each process's events for `Last-Event-ID`.

//! They travel on the event bus ([`crate::bus`]), which carries them between

//! processes too.

    /// Increasing in this process: what `Last-Event-ID` resumes from. Each

    /// process numbers the events it hears, its own and other processes'.

/// The live pages' buffer: one broadcast that subscribers filter by topic, a

    /// Record an event on `topic` and tell this process's subscribers.

    /// Producers publish through the bus ([`crate::bus::Bus::live`]), which

    /// records it here and in every other process.

        // An id this process never gave: from before it restarted, or from

        // another process's numbering (a page that reconnected through

        // another web node, docs/roles.md). What it missed is unknown.

        if last_id >= s.next_id {

            return Replay::Reload;

        }

}

/// The live topics a webhook event touches, with the kind for each: the

/// repository always, and the proposal or issue its data names.

pub fn hook_topics(repo_id: i64, event: &str, data: &serde_json::Value) -> Vec<(String, String)> {

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

    let mut topics = vec![(topic::repo(repo_id), kind.clone())];

    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 => topics.push((topic::issue(repo_id, n), kind)),

        Some(n) if is_proposal => topics.push((topic::proposal(repo_id, n), kind)),

        _ => {}

    topics

        assert_eq!(bus.since(4, &["repo:1".into()]), Replay::Reload, "an id never given here: another process's, or before a restart");

        let mut seen = vec![];

        seen.extend(hook_topics(7, "review", &serde_json::json!({ "number": 4, "verdict": "approve" })));

        seen.extend(hook_topics(7, "issue.opened", &serde_json::json!({ "issue": { "number": 9 } })));

        seen.extend(hook_topics(7, "comment.issue", &serde_json::json!({ "number": 9 })));

        seen.extend(hook_topics(7, "push", &serde_json::json!({ "updates": [] })));

        app.bus.proposal(row.id, hub_api::ProposalEvent { number: 0, kind: "check".into(), check: Some(api.clone()) });

        app.bus.proposal(row.id, hub_api::ProposalEvent { number: 0, kind: "check".into(), check: Some(api.clone()) });

            app.bus.proposal(row.id, hub_api::ProposalEvent { number: 0, kind: "check".into(), check: Some(api.clone()) });

        app.bus.proposal(row.id, hub_api::ProposalEvent { number: 0, kind: "check".into(), check: Some(api.clone()) });

        app.bus.proposal(row.id, hub_api::ProposalEvent { number: 0, kind: "check".into(), check: Some(api.clone()) });

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

    app.bus.hook(repo.id, event, &data);

    app.bus.delivery_waiting();

    let mut rx = app.bus.subscribe(crate::bus::Topic::Deliveries);

pub mod bus;

    /// Events for live pages and workers, here and in the other processes

    /// sharing the database (docs/roles.md, #273).

    pub bus: bus::Bus,

        bus: bus::Bus::default(),

    bus::start(&app).await?;

//!

//! With several web processes (docs/roles.md), choices and landings travel

//! on the event bus ([`crate::bus`]) to every process's room for the same

//! conflict; who is present, and a newcomer's snapshot, are this process's

//! own (a room's viewers on other processes are not counted).

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

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

    Landed { url: String, by: String },

}

/// What one viewer does that every viewer of the room hears, in this

/// process and (over the bus) in the others.

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

#[serde(tag = "kind", rename_all = "lowercase")]

pub enum RoomEvent {

    /// `client` chose for `region`.

    Choice { region: String, choice: Choice, client: String },

    /// The conflict was resolved; `url` is the change.

    /// Record and share a choice; `false` when nobody is in the room here.

        self.apply(key, &RoomEvent::Choice { region: region.to_string(), choice, client: client.to_string() })

        self.apply(key, &RoomEvent::Landed { url: url.to_string(), by: by.to_string() });

    }

    /// Apply a viewer's event to the room for `key` in this process, if

    /// anyone here is in it: record a choice, and tell the viewers. `false`

    /// when nobody here is.

    pub fn apply(&self, key: &RoomKey, event: &RoomEvent) -> bool {

        let mut rooms = self.rooms.lock().unwrap();

        let Some(room) = rooms.get_mut(key) else { return false };

        let told = match event {

            RoomEvent::Choice { region, choice, client } => {

                room.choices.insert(region.clone(), choice.clone());

                LiveEvent::Choice { region: region.clone(), choice: choice.clone(), client: client.clone() }

            }

            RoomEvent::Landed { url, by } => LiveEvent::Landed { url: url.clone(), by: by.clone() },

        };

        let _ = room.tx.send(told);

        true

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

    let rx = app.bus.live.subscribe();

        Some(id) => match app.bus.live.since(id, &topics) {

            Replay::Reload => (vec![Ok(Event::default().event("reload").data("{}"))], app.bus.live.last_id()),

        None => (vec![], app.bus.live.last_id()),

    app.bus.live(&crate::events::topic::proposal(row.id, number), "comment");