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

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)]
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)]
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 },
}

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.
    pub fn choose(&self, key: &RoomKey, client: &str, region: &str, choice: Choice) -> bool {
        let mut rooms = self.rooms.lock().unwrap();
        let Some(room) = rooms.get_mut(key) else { return false };
        room.choices.insert(region.to_string(), choice.clone());
        let _ = room.tx.send(LiveEvent::Choice { region: region.to_string(), choice, client: client.to_string() });
        true
    }

    /// Tell a room its conflict was resolved.
    pub fn landed(&self, key: &RoomKey, url: &str, by: &str) {
        let rooms = self.rooms.lock().unwrap();
        if let Some(room) = rooms.get(key) {
            let _ = room.tx.send(LiveEvent::Landed { url: url.to_string(), by: by.to_string() });
        }
    }

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