1//! Browser Live Share: event-driven relay connections carrying the same sealed streams.
2
3pub use ::relay::code;
4#[path = "group.rs"]
5mod group;
6#[path = "model.rs"]
7mod model;
8#[path = "share.rs"]
9pub mod share;
10#[path = "wire.rs"]
11pub mod wire;
12use model::hex;
13pub use model::{Event, Peer, Reach, Relayed, Room, Trouble};
14pub use wire::{Caret, Guid, Hello, Presence, Spot};
15
16use minicbor::Encode;
17use std::{
18 collections::BTreeMap,
19 io,
20 sync::{
21 Arc, Mutex, Weak,
22 atomic::{AtomicBool, Ordering},
23 },
24 time::Duration,
25};
26use wasm_bindgen::prelude::*;
27use wire::{Side, kind};
28
29#[wasm_bindgen(module = "/src/live/web.js")]
30extern "C" {
31 #[wasm_bindgen(catch, js_name = liveConnect)]
32 fn connect(url: &str, message: &JsValue, closed: &JsValue) -> Result<u32, JsValue>;
33 #[wasm_bindgen(catch, js_name = liveSend)]
34 fn send(id: u32, bytes: &[u8]) -> Result<(), JsValue>;
35 #[wasm_bindgen(js_name = liveClose)]
36 fn close(id: u32);
37}
38
39pub struct Live(Arc<Shared>);
40
41struct Shared {
42 me: Hello,
43 room: Room,
44 url: String,
45 events: Box<dyn Fn(Event) + Send + Sync>,
46 stopped: AtomicBool,
47 state: Mutex<State>,
48}
49
50struct State {
51 socket: u32,
52 slot: u32,
53 relayed: Relayed,
54 failed: u32,
55 outdated: Option<u16>,
56 burned: bool,
57 streams: BTreeMap<u32, Stream>,
58 members: BTreeMap<u32, group::Member>,
59 sealer: Option<group::Sealer>,
60 presence: Presence,
61 scheduled: bool,
62}
63
64struct Stream {
65 bytes: Vec<u8>,
66 opening: Option<wire::Opening>,
67 receive: Option<wire::Sealer>,
68 line: Line,
69 peer: Option<Peer>,
70}
71
72#[derive(Clone)]
73pub struct Line {
74 shared: Weak<Shared>,
75 slot: u32,
76 sealer: Arc<Mutex<Option<wire::Sealer>>>,
77}
78
79impl Line {
80 pub fn send(&self, kind: u16, body: &impl Encode<()>) -> io::Result<()> {
81 let shared = self.shared.upgrade().ok_or(io::ErrorKind::NotConnected)?;
82 let mut bytes = Vec::new();
83 let sealed = self
84 .sealer
85 .lock()
86 .unwrap()
87 .as_mut()
88 .ok_or(io::ErrorKind::NotConnected)?
89 .send(&mut bytes, kind, body);
90 if let Err(error) = sealed {
91 shared.disconnected();
92 return Err(error);
93 }
94 shared.stream(self.slot, &bytes)
95 }
96
97 pub fn hang_up(&self, reason: &str) {
98 let _ = self.send(
99 kind::BYE,
100 &wire::Bye {
101 reason: reason.into(),
102 },
103 );
104 }
105}
106
107#[derive(Clone)]
108pub struct Sender(Weak<Shared>);
109
110impl Sender {
111 pub fn send(&self, kind: u16, body: &impl Encode<()>, to: Option<&[[u8; 16]]>) {
112 if let Some(shared) = self.0.upgrade() {
113 match to {
114 None => {
115 let _ = shared.group(kind, body, None);
116 }
117 Some(to) => {
118 let slots: Vec<_> = shared
119 .state
120 .lock()
121 .unwrap()
122 .members
123 .iter()
124 .filter(|(_, m)| {
125 m.peer.as_ref().is_some_and(|p| to.contains(&p.hello.peer))
126 })
127 .map(|(slot, _)| *slot)
128 .collect();
129 for slot in slots {
130 let _ = shared.group(kind, body, Some(slot));
131 }
132 }
133 }
134 }
135 }
136}
137
138impl Live {
139 pub fn start(
140 me: Hello,
141 room: &Room,
142 _: Option<Reach>,
143 relay: Option<&str>,
144 events: impl Fn(Event) + Send + Sync + 'static,
145 ) -> io::Result<Self> {
146 let tag = room.tag().ok_or(io::ErrorKind::InvalidInput)?;
147 let relay = relay
148 .ok_or(io::ErrorKind::NotConnected)?
149 .trim_end_matches('/');
150 if !relay.starts_with("wss://") && !relay.starts_with("ws://") {
151 return Err(io::ErrorKind::InvalidInput.into());
152 }
153 let shared = Arc::new(Shared {
154 me,
155 room: room.clone(),
156 url: format!("{relay}/v1/room/{tag}"),
157 events: Box::new(events),
158 stopped: AtomicBool::new(false),
159 state: Mutex::new(State {
160 socket: 0,
161 slot: 0,
162 relayed: Relayed::Unknown,
163 failed: 0,
164 outdated: None,
165 burned: false,
166 streams: BTreeMap::new(),
167 members: BTreeMap::new(),
168 sealer: None,
169 presence: Presence::default(),
170 scheduled: false,
171 }),
172 });
173 shared.connect()?;
174 let weak = Arc::downgrade(&shared);
175 crate::task::spawn("live ping", move || async move {
176 let (_, wait) = crate::task::channel();
177 loop {
178 crate::task::wait(&wait, Some(Duration::from_secs(15))).await;
179 let Some(shared) = weak
180 .upgrade()
181 .filter(|s| !s.stopped.load(Ordering::Acquire))
182 else {
183 break;
184 };
185 let lines: Vec<_> = shared
186 .state
187 .lock()
188 .unwrap()
189 .streams
190 .values()
191 .map(|s| s.line.clone())
192 .collect();
193 for line in lines {
194 let _ = line.send(kind::PING, &());
195 }
196 if matches!(shared.room, Room::Notebook(_)) {
197 let _ = shared.group(kind::PING, &(), None);
198 }
199 }
200 })?;
201 Ok(Self(shared))
202 }
203
204 pub fn code(&self) -> Option<String> {
205 match &self.0.room {
206 Room::Code { code, .. } => Some(code.clone()),
207 _ => None,
208 }
209 }
210 pub fn relayed(&self) -> Relayed {
211 self.0.state.lock().unwrap().relayed.clone()
212 }
213 pub fn failed(&self) -> u32 {
214 self.0.state.lock().unwrap().failed
215 }
216 pub fn other_version(&self) -> Option<u16> {
217 self.0.state.lock().unwrap().outdated
218 }
219 pub fn burned(&self) -> bool {
220 self.0.state.lock().unwrap().burned
221 }
222 pub fn sender(&self) -> Sender {
223 Sender(Arc::downgrade(&self.0))
224 }
225 pub fn peers(&self) -> Vec<Peer> {
226 let state = self.0.state.lock().unwrap();
227 let mut peers = BTreeMap::new();
228 for peer in state
229 .members
230 .values()
231 .filter_map(|m| m.peer.as_ref())
232 .chain(state.streams.values().filter_map(|s| s.peer.as_ref()))
233 {
234 peers.insert(peer.hello.peer, peer.clone());
235 }
236 peers.into_values().collect()
237 }
238 pub fn line(&self, peer: &[u8; 16]) -> Option<Line> {
239 self.0
240 .state
241 .lock()
242 .unwrap()
243 .streams
244 .values()
245 .find(|s| s.peer.as_ref().is_some_and(|p| &p.hello.peer == peer))
246 .map(|s| s.line.clone())
247 }
248 pub fn set_presence(&self, presence: Presence) {
249 let mut state = self.0.state.lock().unwrap();
250 if state.presence == presence {
251 return;
252 }
253 state.presence = presence;
254 if std::mem::replace(&mut state.scheduled, true) {
255 return;
256 }
257 drop(state);
258 let weak = Arc::downgrade(&self.0);
259 let _ = crate::task::spawn("live presence", move || async move {
260 let (_, wait) = crate::task::channel();
261 crate::task::wait(&wait, Some(Duration::from_millis(100))).await;
262 if let Some(shared) = weak.upgrade() {
263 let presence = {
264 let mut state = shared.state.lock().unwrap();
265 state.scheduled = false;
266 state.presence.clone()
267 };
268 let _ = shared.group(kind::PRESENCE, &presence, None);
269 }
270 });
271 }
272 pub fn leave(self, reason: &str) {
273 let lines: Vec<_> = self
274 .0
275 .state
276 .lock()
277 .unwrap()
278 .streams
279 .values()
280 .map(|s| s.line.clone())
281 .collect();
282 for line in lines {
283 line.hang_up(reason);
284 }
285 }
286}
287
288impl Drop for Live {
289 fn drop(&mut self) {
290 self.0.stopped.store(true, Ordering::Release);
291 close(self.0.state.lock().unwrap().socket);
292 }
293}
294
295impl Shared {
296 fn connect(self: &Arc<Self>) -> io::Result<()> {
297 let weak = Arc::downgrade(self);
298 let message = Closure::<dyn FnMut(JsValue)>::new(move |data: JsValue| {
299 if let Some(shared) = weak.upgrade()
300 && let Err(error) = shared.heard(data)
301 {
302 if let Some(version) = error
303 .get_ref()
304 .and_then(|e| e.downcast_ref::<wire::Version>())
305 {
306 shared.state.lock().unwrap().outdated = Some(version.0);
307 } else if error.kind() == io::ErrorKind::InvalidData {
308 shared.state.lock().unwrap().failed += 1;
309 }
310 shared.disconnected();
311 }
312 })
313 .into_js_value();
314 let weak = Arc::downgrade(self);
315 let closed = Closure::<dyn FnMut()>::new(move || {
316 if let Some(shared) = weak.upgrade() {
317 shared.disconnected();
318 }
319 })
320 .into_js_value();
321 let socket = connect(&self.url, &message, &closed).map_err(js_error)?;
322 self.state.lock().unwrap().socket = socket;
323 Ok(())
324 }
325
326 fn disconnected(self: &Arc<Self>) {
327 let peers = {
328 let mut state = self.state.lock().unwrap();
329 let socket = std::mem::replace(&mut state.socket, 0);
330 if socket == 0 {
331 return;
332 }
333 close(socket);
334 let peers: Vec<_> = state
335 .streams
336 .values()
337 .filter_map(|s| {
338 *s.line.sealer.lock().unwrap() = None;
339 s.peer.as_ref().map(|p| p.hello.clone())
340 })
341 .collect();
342 state.streams.clear();
343 state.members.clear();
344 state.sealer = None;
345 state.relayed = Relayed::Unreachable(Trouble::Other);
346 peers
347 };
348 for hello in peers {
349 (self.events)(Event::Left(&hello));
350 }
351 (self.events)(Event::Changed);
352 let weak = Arc::downgrade(self);
353 let _ = crate::task::spawn("live reconnect", move || async move {
354 let (_, wait) = crate::task::channel();
355 crate::task::wait(&wait, Some(Duration::from_secs(2))).await;
356 if let Some(shared) = weak
357 .upgrade()
358 .filter(|s| !s.stopped.load(Ordering::Acquire))
359 {
360 let _ = shared.connect();
361 }
362 });
363 }
364
365 fn stream(&self, slot: u32, bytes: &[u8]) -> io::Result<()> {
366 let socket = self.state.lock().unwrap().socket;
367 for chunk in bytes.chunks(64 << 10) {
368 send(socket, &[&slot.to_be_bytes()[..], chunk].concat()).map_err(js_error)?;
369 }
370 Ok(())
371 }
372
373 fn group(
374 self: &Arc<Self>,
375 kind: u16,
376 body: &impl Encode<()>,
377 to: Option<u32>,
378 ) -> io::Result<()> {
379 let body = minicbor::to_vec(body).map_err(io::Error::other)?;
380 let (socket, sealed) = {
381 let mut state = self.state.lock().unwrap();
382 let sealed = state
383 .sealer
384 .as_mut()
385 .ok_or(io::ErrorKind::NotConnected)?
386 .seal(kind, &body, to.is_none());
387 (state.socket, sealed)
388 };
389 let sealed = match sealed {
390 Ok(sealed) => sealed,
391 Err(error) => {
392 self.disconnected();
393 return Err(error);
394 }
395 };
396 let mut bytes = match to {
397 Some(slot) => [&(::relay::GROUP | 1).to_be_bytes()[..], &slot.to_be_bytes()].concat(),
398 None => ::relay::BROADCAST.to_be_bytes().to_vec(),
399 };
400 bytes.extend(sealed);
401 send(socket, &bytes).map_err(js_error)
402 }
403
404 fn meet(self: &Arc<Self>, slot: u32) -> io::Result<()> {
405 if self.state.lock().unwrap().streams.contains_key(&slot) {
406 return Ok(());
407 }
408 let tag = self.room.tag().ok_or(io::ErrorKind::InvalidInput)?;
409 let (opening, bytes) = wire::Opening::new(Side::Initiator, &tag, &self.room.secret())?;
410 let line = Line {
411 shared: Arc::downgrade(self),
412 slot,
413 sealer: Arc::new(Mutex::new(None)),
414 };
415 self.state.lock().unwrap().streams.insert(
416 slot,
417 Stream {
418 bytes: Vec::new(),
419 opening: Some(opening),
420 receive: None,
421 line,
422 peer: None,
423 },
424 );
425 self.stream(
426 slot,
427 &[&(bytes.len() as u32).to_be_bytes()[..], &bytes].concat(),
428 )
429 }
430
431 fn heard(self: &Arc<Self>, data: JsValue) -> io::Result<()> {
432 if let Some(text) = data.as_string() {
433 let notice = text
434 .parse::<::relay::Notice>()
435 .map_err(|_| io::ErrorKind::InvalidData)?;
436 match notice {
437 ::relay::Notice::Welcome { you, members } => {
438 {
439 let mut state = self.state.lock().unwrap();
440 state.slot = you;
441 state.relayed = Relayed::Joined;
442 if matches!(self.room, Room::Notebook(_)) {
443 state.sealer =
444 Some(group::Sealer::new(&group::Keys::new(&self.room.secret()))?);
445 }
446 }
447 if matches!(self.room, Room::Code { .. }) {
448 for slot in members.into_iter().filter(|slot| *slot < you) {
449 self.meet(slot)?;
450 }
451 } else {
452 self.group(kind::HELLO, &self.me, None)?;
453 let presence = self.state.lock().unwrap().presence.clone();
454 self.group(kind::PRESENCE, &presence, None)?;
455 }
456 }
457 ::relay::Notice::Left(slot) => {
458 let hello = {
459 let mut state = self.state.lock().unwrap();
460 state.members.remove(&slot);
461 state.streams.remove(&slot).and_then(|s| {
462 *s.line.sealer.lock().unwrap() = None;
463 s.peer.map(|p| p.hello)
464 })
465 };
466 if let Some(hello) = hello {
467 (self.events)(Event::Left(&hello));
468 }
469 }
470 ::relay::Notice::Burned => self.state.lock().unwrap().burned = true,
471 _ => {}
472 }
473 (self.events)(Event::Changed);
474 return Ok(());
475 }
476 let bytes = js_sys::Uint8Array::new(&data).to_vec();
477 let (slot, bytes) = bytes
478 .split_first_chunk::<4>()
479 .ok_or(io::ErrorKind::InvalidData)?;
480 let slot = u32::from_be_bytes(*slot);
481 if slot & ::relay::GROUP != 0 {
482 return self.heard_group(slot & !::relay::GROUP, bytes);
483 }
484 self.heard_stream(slot, bytes)
485 }
486
487 fn heard_group(self: &Arc<Self>, slot: u32, frame: &[u8]) -> io::Result<()> {
488 let (kind, body, hello) = {
489 let mut state = self.state.lock().unwrap();
490 let member = match state.members.entry(slot) {
491 std::collections::btree_map::Entry::Occupied(e) => e.into_mut(),
492 std::collections::btree_map::Entry::Vacant(e) => e.insert(group::Member::new(
493 &group::Keys::new(&self.room.secret()),
494 frame,
495 )?),
496 };
497 let (kind, body) = member.open(frame)?;
498 (kind, body, member.hello().cloned())
499 };
500 match kind {
501 kind::HELLO | kind::HELLO_BACK => {
502 let hello: Hello =
503 minicbor::decode(&body).map_err(|_| io::ErrorKind::InvalidData)?;
504 if hello.peer == self.me.peer {
505 return Ok(());
506 }
507 let serves = hello.serves.is_some() && self.me.serves.is_none();
508 self.state
509 .lock()
510 .unwrap()
511 .members
512 .get_mut(&slot)
513 .unwrap()
514 .peer = Some(Peer {
515 hello: Arc::new(hello),
516 presence: None,
517 });
518 if kind == kind::HELLO {
519 self.group(kind::HELLO_BACK, &self.me, Some(slot))?;
520 let presence = self.state.lock().unwrap().presence.clone();
521 self.group(kind::PRESENCE, &presence, Some(slot))?;
522 }
523 if serves {
524 self.meet(slot)?;
525 }
526 (self.events)(Event::Changed);
527 }
528 kind::PRESENCE => {
529 let presence = minicbor::decode(&body).map_err(|_| io::ErrorKind::InvalidData)?;
530 if let Some(peer) = self
531 .state
532 .lock()
533 .unwrap()
534 .members
535 .get_mut(&slot)
536 .and_then(|m| m.peer.as_mut())
537 {
538 peer.presence = Some(presence);
539 }
540 (self.events)(Event::Changed);
541 }
542 kind if kind < 256 => {
543 if let Some(hello) = hello {
544 (self.events)(Event::Frame {
545 from: &hello,
546 kind,
547 body: &body,
548 });
549 }
550 }
551 _ => {}
552 }
553 Ok(())
554 }
555
556 fn heard_stream(self: &Arc<Self>, slot: u32, bytes: &[u8]) -> io::Result<()> {
557 {
558 let mut state = self.state.lock().unwrap();
559 let stream = state
560 .streams
561 .get_mut(&slot)
562 .ok_or(io::ErrorKind::InvalidData)?;
563 if stream.bytes.len() + bytes.len() > (16 << 20) + 4 {
564 return Err(io::ErrorKind::InvalidData.into());
565 }
566 stream.bytes.extend_from_slice(bytes);
567 }
568 loop {
569 let event = {
570 let mut state = self.state.lock().unwrap();
571 let stream = state.streams.get_mut(&slot).unwrap();
572 let Some(length) = stream.bytes.get(..4) else {
573 break;
574 };
575 let length = u32::from_be_bytes(length.try_into().unwrap()) as usize;
576 if length > 16 << 20 {
577 return Err(io::ErrorKind::InvalidData.into());
578 }
579 if stream.bytes.len() < length + 4 {
580 break;
581 }
582 let bytes: Vec<_> = stream.bytes.drain(..length + 4).collect();
583 if let Some(opening) = stream.opening.take() {
584 let (send, receive) = opening.finish(&bytes[4..])?;
585 *stream.line.sealer.lock().unwrap() = Some(send);
586 stream.receive = Some(receive);
587 Some((stream.line.clone(), None, kind::HELLO, Vec::new()))
588 } else {
589 let (kind, body) = stream
590 .receive
591 .as_mut()
592 .ok_or(io::ErrorKind::InvalidData)?
593 .receive(&mut io::Cursor::new(bytes))?;
594 if stream.peer.is_none() {
595 if kind != kind::HELLO {
596 return Err(io::ErrorKind::InvalidData.into());
597 }
598 let hello = minicbor::decode::<Hello>(&body)
599 .map_err(|_| io::ErrorKind::InvalidData)?;
600 stream.peer = Some(Peer {
601 hello: Arc::new(hello),
602 presence: None,
603 });
604 }
605 Some((
606 stream.line.clone(),
607 stream.peer.as_ref().map(|p| p.hello.clone()),
608 kind,
609 body,
610 ))
611 }
612 };
613 if let Some((line, hello, kind, body)) = event {
614 match hello {
615 None => line.send(kind::HELLO, &self.me)?,
616 Some(hello) if kind == kind::HELLO => (self.events)(Event::Met(&hello, &line)),
617 Some(hello) => (self.events)(Event::Frame {
618 from: &hello,
619 kind,
620 body: &body,
621 }),
622 }
623 }
624 }
625 Ok(())
626 }
627}
628
629fn js_error(error: JsValue) -> io::Error {
630 io::Error::new(
631 io::ErrorKind::NotConnected,
632 error
633 .as_string()
634 .unwrap_or_else(|| "Relay disconnected".into()),
635 )
636}