//! Live collaboration on the conflict resolver (hub spec §6.10): everyone

//! looking at the same conflict sees who else is there, the choices each

//! makes region by region (the last choice for a region wins), and the

//! resolution when one of them commits it. Rooms live in memory: they are

//! for people working together now, not a record.

//!

//! 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).

use std::collections::{BTreeMap, HashMap};

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

use tokio::sync::broadcast;

/// One conflict being resolved: repository, bookmark, path.

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

pub struct RoomKey {

    pub repo_id: i64,

    pub bookmark: String,

    pub path: String,

}

/// A region's current choice: `side1`…, `all`, `none` or `text`, with the

/// text for `text`, and who made it.

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

pub struct Choice {

    pub choice: String,

    pub text: String,

    pub by: String,

}

/// What a room tells its viewers.

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

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

pub enum LiveEvent {

    /// Everything so far, sent to a viewer who joins.

    Snapshot { present: Vec<String>, choices: BTreeMap<String, Choice>, keep: Option<Choice> },

    /// Who is looking now.

    Presence { present: Vec<String> },

    /// A viewer chose for region `region` (`keep` for a candidate page).

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

    /// A resolution was committed; `url` is its change.

    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.

    Landed { url: String, by: String },

}

struct Room {

    tx: broadcast::Sender<LiveEvent>,

    /// Connected viewers by client id: their user names.

    present: HashMap<String, String>,

    choices: BTreeMap<String, Choice>,

}

impl Room {

    fn names(&self) -> Vec<String> {

        let mut v: Vec<String> = self.present.values().cloned().collect();

        v.sort();

        v.dedup();

        v

    }

}

#[derive(Default)]

pub struct Rooms {

    rooms: Mutex<HashMap<RoomKey, Room>>,

}

/// A viewer's place in a room; leaving happens when it is dropped (the

/// viewer's event stream closed).

pub struct Seat {

    rooms: Arc<Rooms>,

    key: RoomKey,

    client: String,

}

impl Drop for Seat {

    fn drop(&mut self) {

        self.rooms.leave(&self.key, &self.client);

    }

}

impl Rooms {

    /// Join `key` as `client` (one browser tab) of `user`: the events to

    /// follow, starting with a snapshot, and the seat that leaves on drop.

    pub fn join(&self, key: RoomKey, client: &str, user: &str) -> (broadcast::Receiver<LiveEvent>, LiveEvent) {

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

        let room =

            rooms.entry(key).or_insert_with(|| Room { tx: broadcast::channel(256).0, present: HashMap::new(), choices: BTreeMap::new() });

        room.present.insert(client.to_string(), user.to_string());

        let rx = room.tx.subscribe();

        let present = room.names();

        let _ = room.tx.send(LiveEvent::Presence { present: present.clone() });

        let mut choices = room.choices.clone();

        let keep = choices.remove("keep");

        (rx, LiveEvent::Snapshot { present, choices, keep })

    }

    /// A seat for `client` in `key`, to hold while its stream is open.

    pub fn seat(self: &Arc<Self>, key: RoomKey, client: &str) -> Seat {

        Seat { rooms: Arc::clone(self), key, client: client.to_string() }

    }

    fn leave(&self, key: &RoomKey, client: &str) {

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

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

        room.present.remove(client);

        if room.present.is_empty() {

            // Nobody is looking: forget the room and its choices.

            rooms.remove(key);

        } else {

            let _ = room.tx.send(LiveEvent::Presence { present: room.names() });

        }

    }

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

    pub fn choose(&self, key: &RoomKey, client: &str, region: &str, choice: Choice) -> bool {

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

    }

    /// Tell a room its conflict was resolved.

    pub fn landed(&self, key: &RoomKey, url: &str, by: &str) {

        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

    }

    /// How many rooms are open (for tests and the admin page).

    pub fn open(&self) -> usize {

        self.rooms.lock().unwrap().len()

    }

}

#[cfg(test)]

mod tests {

    use super::*;

    fn key() -> RoomKey {

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

    }

    fn pick(choice: &str, by: &str) -> Choice {

        Choice { choice: choice.into(), text: String::new(), by: by.into() }

    }

    #[test]

    fn viewers_see_each_other_and_each_others_choices() {

        let rooms = Arc::new(Rooms::default());

        let (mut a_rx, a_snap) = rooms.join(key(), "tab-a", "andy");

        assert_eq!(a_snap, LiveEvent::Snapshot { present: vec!["andy".into()], choices: BTreeMap::new(), keep: None });

        let seat_a = rooms.seat(key(), "tab-a");

        assert!(matches!(a_rx.try_recv(), Ok(LiveEvent::Presence { .. })), "a's own arrival");

        assert!(rooms.choose(&key(), "tab-a", "1", pick("side2", "andy")));

        assert!(matches!(a_rx.try_recv(), Ok(LiveEvent::Choice { region, .. }) if region == "1"));

        // A second viewer gets the choices so far and is announced.

        let (mut b_rx, b_snap) = rooms.join(key(), "tab-b", "bea");

        let seat_b = rooms.seat(key(), "tab-b");

        let LiveEvent::Snapshot { present, choices, .. } = b_snap else { panic!() };

        assert_eq!(present, vec!["andy", "bea"]);

        assert_eq!(choices["1"].choice, "side2");

        assert_eq!(a_rx.try_recv().unwrap(), LiveEvent::Presence { present: vec!["andy".into(), "bea".into()] });

        // The last choice for a region wins; keep is apart for candidate pages.

        rooms.choose(&key(), "tab-b", "1", pick("all", "bea"));

        rooms.choose(&key(), "tab-b", "keep", pick("1", "bea"));

        let _ = b_rx.try_recv();

        drop(seat_b);

        assert_eq!(a_rx.try_recv().unwrap(), LiveEvent::Choice { region: "1".into(), choice: pick("all", "bea"), client: "tab-b".into() });

        let _ = a_rx.try_recv();

        assert_eq!(a_rx.try_recv().unwrap(), LiveEvent::Presence { present: vec!["andy".into()] }, "bea left");

        rooms.landed(&key(), "/andy/demo/change/abc", "andy");

        assert!(matches!(a_rx.try_recv(), Ok(LiveEvent::Landed { .. })));

        // The last viewer leaving closes the room.

        drop(seat_a);

        assert_eq!(rooms.open(), 0);

        assert!(!rooms.choose(&key(), "tab-a", "1", pick("none", "andy")));

    }

}