live.rs 100.0%

89 of 89 lines run, 20 of 23 functions · run #545 on main · run missed

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

`/`	Search
`g` `c`	Code
`g` `h`	History
`g` `b`	Bookmarks
`g` `t`	Trust
`y`	Permalink to this state (files and trees)
`?`	This sheet

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

Coverage / crates/hub/src/live.rs 100.0%

Keyboard