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