OXYGENHUB
# e8e7158bbc73 oxygenhub / crates / hub / src / live.rs
6.9 KiB · Rust · last changed Resolve conflicts together, live 1 month ago source blame history raw
1
//! Live collaboration on the conflict resolver (hub spec §6.10): everyone
2
//! looking at the same conflict sees who else is there, the choices each
3
//! makes region by region (the last choice for a region wins), and the
4
//! resolution when one of them commits it. Rooms live in memory: they are
5
//! for people working together now, not a record.
6
7
use std::collections::{BTreeMap, HashMap};
8
use std::sync::{Arc, Mutex};
9
use tokio::sync::broadcast;
10
11
/// One conflict being resolved: repository, bookmark, path.
12
#[derive(Clone, Debug, PartialEq, Eq, Hash)]
13
pub struct RoomKey {
14
    pub repo_id: i64,
15
    pub bookmark: String,
16
    pub path: String,
17
}
18
19
/// A region's current choice: `side1`…, `all`, `none` or `text`, with the
20
/// text for `text`, and who made it.
21
#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize)]
22
pub struct Choice {
23
    pub choice: String,
24
    pub text: String,
25
    pub by: String,
26
}
27
28
/// What a room tells its viewers.
29
#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize)]
30
#[serde(tag = "kind", rename_all = "lowercase")]
31
pub enum LiveEvent {
32
    /// Everything so far, sent to a viewer who joins.
33
    Snapshot { present: Vec<String>, choices: BTreeMap<String, Choice>, keep: Option<Choice> },
34
    /// Who is looking now.
35
    Presence { present: Vec<String> },
36
    /// A viewer chose for region `region` (`keep` for a candidate page).
37
    Choice { region: String, choice: Choice, client: String },
38
    /// A resolution was committed; `url` is its change.
39
    Landed { url: String, by: String },
40
}
41
42
struct Room {
43
    tx: broadcast::Sender<LiveEvent>,
44
    /// Connected viewers by client id: their user names.
45
    present: HashMap<String, String>,
46
    choices: BTreeMap<String, Choice>,
47
}
48
49
impl Room {
50
    fn names(&self) -> Vec<String> {
51
        let mut v: Vec<String> = self.present.values().cloned().collect();
52
        v.sort();
53
        v.dedup();
54
        v
55
    }
56
}
57
58
#[derive(Default)]
59
pub struct Rooms {
60
    rooms: Mutex<HashMap<RoomKey, Room>>,
61
}
62
63
/// A viewer's place in a room; leaving happens when it is dropped (the
64
/// viewer's event stream closed).
65
pub struct Seat {
66
    rooms: Arc<Rooms>,
67
    key: RoomKey,
68
    client: String,
69
}
70
71
impl Drop for Seat {
72
    fn drop(&mut self) {
73
        self.rooms.leave(&self.key, &self.client);
74
    }
75
}
76
77
impl Rooms {
78
    /// Join `key` as `client` (one browser tab) of `user`: the events to
79
    /// follow, starting with a snapshot, and the seat that leaves on drop.
80
    pub fn join(&self, key: RoomKey, client: &str, user: &str) -> (broadcast::Receiver<LiveEvent>, LiveEvent) {
81
        let mut rooms = self.rooms.lock().unwrap();
82
        let room =
83
            rooms.entry(key).or_insert_with(|| Room { tx: broadcast::channel(256).0, present: HashMap::new(), choices: BTreeMap::new() });
84
        room.present.insert(client.to_string(), user.to_string());
85
        let rx = room.tx.subscribe();
86
        let present = room.names();
87
        let _ = room.tx.send(LiveEvent::Presence { present: present.clone() });
88
        let mut choices = room.choices.clone();
89
        let keep = choices.remove("keep");
90
        (rx, LiveEvent::Snapshot { present, choices, keep })
91
    }
92
93
    /// A seat for `client` in `key`, to hold while its stream is open.
94
    pub fn seat(self: &Arc<Self>, key: RoomKey, client: &str) -> Seat {
95
        Seat { rooms: Arc::clone(self), key, client: client.to_string() }
96
    }
97
98
    fn leave(&self, key: &RoomKey, client: &str) {
99
        let mut rooms = self.rooms.lock().unwrap();
100
        let Some(room) = rooms.get_mut(key) else { return };
101
        room.present.remove(client);
102
        if room.present.is_empty() {
103
            // Nobody is looking: forget the room and its choices.
104
            rooms.remove(key);
105
        } else {
106
            let _ = room.tx.send(LiveEvent::Presence { present: room.names() });
107
        }
108
    }
109
110
    /// Record and share a choice; `false` when nobody is in the room.
111
    pub fn choose(&self, key: &RoomKey, client: &str, region: &str, choice: Choice) -> bool {
112
        let mut rooms = self.rooms.lock().unwrap();
113
        let Some(room) = rooms.get_mut(key) else { return false };
114
        room.choices.insert(region.to_string(), choice.clone());
115
        let _ = room.tx.send(LiveEvent::Choice { region: region.to_string(), choice, client: client.to_string() });
116
        true
117
    }
118
119
    /// Tell a room its conflict was resolved.
120
    pub fn landed(&self, key: &RoomKey, url: &str, by: &str) {
121
        let rooms = self.rooms.lock().unwrap();
122
        if let Some(room) = rooms.get(key) {
123
            let _ = room.tx.send(LiveEvent::Landed { url: url.to_string(), by: by.to_string() });
124
        }
125
    }
126
127
    /// How many rooms are open (for tests and the admin page).
128
    pub fn open(&self) -> usize {
129
        self.rooms.lock().unwrap().len()
130
    }
131
}
132
133
#[cfg(test)]
134
mod tests {
135
    use super::*;
136
137
    fn key() -> RoomKey {
138
        RoomKey { repo_id: 1, bookmark: "main".into(), path: "a.txt".into() }
139
    }
140
141
    fn pick(choice: &str, by: &str) -> Choice {
142
        Choice { choice: choice.into(), text: String::new(), by: by.into() }
143
    }
144
145
    #[test]
146
    fn viewers_see_each_other_and_each_others_choices() {
147
        let rooms = Arc::new(Rooms::default());
148
        let (mut a_rx, a_snap) = rooms.join(key(), "tab-a", "andy");
149
        assert_eq!(a_snap, LiveEvent::Snapshot { present: vec!["andy".into()], choices: BTreeMap::new(), keep: None });
150
        let seat_a = rooms.seat(key(), "tab-a");
151
        assert!(matches!(a_rx.try_recv(), Ok(LiveEvent::Presence { .. })), "a's own arrival");
152
153
        assert!(rooms.choose(&key(), "tab-a", "1", pick("side2", "andy")));
154
        assert!(matches!(a_rx.try_recv(), Ok(LiveEvent::Choice { region, .. }) if region == "1"));
155
156
        // A second viewer gets the choices so far and is announced.
157
        let (mut b_rx, b_snap) = rooms.join(key(), "tab-b", "bea");
158
        let seat_b = rooms.seat(key(), "tab-b");
159
        let LiveEvent::Snapshot { present, choices, .. } = b_snap else { panic!() };
160
        assert_eq!(present, vec!["andy", "bea"]);
161
        assert_eq!(choices["1"].choice, "side2");
162
        assert_eq!(a_rx.try_recv().unwrap(), LiveEvent::Presence { present: vec!["andy".into(), "bea".into()] });
163
164
        // The last choice for a region wins; keep is apart for candidate pages.
165
        rooms.choose(&key(), "tab-b", "1", pick("all", "bea"));
166
        rooms.choose(&key(), "tab-b", "keep", pick("1", "bea"));
167
        let _ = b_rx.try_recv();
168
        drop(seat_b);
169
        assert_eq!(a_rx.try_recv().unwrap(), LiveEvent::Choice { region: "1".into(), choice: pick("all", "bea"), client: "tab-b".into() });
170
        let _ = a_rx.try_recv();
171
        assert_eq!(a_rx.try_recv().unwrap(), LiveEvent::Presence { present: vec!["andy".into()] }, "bea left");
172
173
        rooms.landed(&key(), "/andy/demo/change/abc", "andy");
174
        assert!(matches!(a_rx.try_recv(), Ok(LiveEvent::Landed { .. })));
175
176
        // The last viewer leaving closes the room.
177
        drop(seat_a);
178
        assert_eq!(rooms.open(), 0);
179
        assert!(!rooms.choose(&key(), "tab-a", "1", pick("none", "andy")));
180
    }
181
}