//! 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}");
        }
    }
}