jevstrudel.git / website / src / jev / party.mjs
party.mjsannotatedparty.mjssource519 lines · 18.4 KB · raw

Listening parties: a host plays a song, guests who open its link hear the same song with the same Jev decisions, in time with the host.

One room per party, a Durable Object in the site's Worker (worker/src/party-room.ts has the protocol). The host's page is the only one that asks Jev: it announces each performance (run: the song), every decision its jev()s settle (decision, from jevCore's watchDecisions) and when its scheduler started (clock, in the room's clock). A guest's jev()s follow those decisions (jevCore's followDecisions) and its scheduler starts at the cycle the host is on, so a guest joining mid-song starts mid-song. 🔥/😴 from anyone go to the room, which counts them and tells everyone the totals, so the host's Jev hears every listener.

Time: each page estimates the room's clock from a few ping/pong exchanges (NTP-style: the round trip with the smallest delay, its midpoint). The host sends its start in the room's clock; a guest turns it back into its own. Both define "started" the same way, at the moment their scheduler starts, so the scheduler's own fixed latencies cancel.

The page calls in at three places (useReplContext): partyBeforeStart and partyStarted around each scheduler start, and partyToggled when it stops.

22import { followDecisions, setReactions, watchDecisions } from './jevCore.mjs';
23import { nowPlaying, playSong, readSongs, showSong, songPath } from './client.mjs';
24import { setOwnCode } from './sandbox.mjs';

26export const PARTY_PATH = '/jev/party/';

Live parties listed to join (the listening tab): GET gives them, and GET <entry id> the room to join, from a room that is listed.

29export const PARTIES_PATH = '/jev/parties';
30const PARTY_PARAM = 'party';
31const HOST_KEY = (room) => `jevstrudel.party.${room}`;
32const PINGS = 6; // at joining; the best of them sets the offset
33const REPING_MS = 60_000;
34const ROOM = /^[A-Za-z0-9_-]{16,43}$/;

128 random bits, base64url: the room id is the party's only secret from people it was not shared with.

38export function randomId(bytes = 16) {
39  const a = crypto.getRandomValues(new Uint8Array(bytes));
40  return btoa(String.fromCharCode(...a))
41    .replace(/\+/g, '-')
42    .replace(/\//g, '_')
43    .replace(/=+$/, '');
44}

A short hash of a song's code, so a guest can tell the host edited it.

47export function codeHash(code = '') {
48  let h = 0x811c9dc5;
49  for (let i = 0; i < code.length; i++) h = Math.imul(h ^ code.charCodeAt(i), 0x01000193);
50  return (h >>> 0).toString(16).padStart(8, '0');
51}

The room's clock from ping/pong samples: c sent and r received in the page's clock, s the room's reading. The best sample is the fastest round trip; the room's reading is taken to be at its midpoint.

56export function clockOffset(samples) {
57  const best = [...samples].sort((a, b) => a.r - a.c - (b.r - b.c))[0];
58  return best ? { offset: best.s - (best.c + best.r) / 2, rtt: best.r - best.c } : null;
59}

The cycle a guest starts at: where the host is now, at the host's cps. Negative when the host starts in the future (a guest faster to start).

63export function cycleNow(clock, roomNow) {
64  return ((roomNow - clock.startedAt) / 1000) * clock.cps;
65}

── state, for the UI ─────────────────────────────

68let state = {
69  room: null,
70  role: null, // 'host' | 'guest'
71  status: 'off', // 'connecting' | 'open' | 'reconnecting' | 'ended' | 'off'
72  people: 0,
73  hostHere: false,
74  song: null, // the performance's song id
75  note: null, // anything the listener should know (edited song, joined mid-song, …)
76  rtt: null,
77  joinedAt: null, // the cycle a guest's scheduler started at
78  waiting: false, // a guest waiting for the host to press play
79};
80const listeners = new Set();
81function set(patch) {
82  state = { ...state, ...patch };
83  for (const l of listeners) l();
84}
85export const partyStore = {
86  subscribe(l) {
87    listeners.add(l);
88    return () => listeners.delete(l);
89  },
90  get: () => state,
91};

── the connection ────────────────────────────────

94let editorRef = null;
95let socket = null;
96let key = null;
97let retry = null;
98let retryMs = 1000;
99let pinger = null;
100let samples = [];
101let offset = 0; // the room's clock minus this page's
102const roomNow = () => Date.now() + offset;

the current performance, as the room has it

105let run = null; // { run, song, code }
106let clock = null; // { startedAt, cps }
107let clockWaiters = [];
108const decisions = new Map(); // `${view}:${segment}` → { view, segment, d }
109const reactions = new Map(); // segment → { fire, sleep }

guest: the playing song's receivers, and whether the listener is in (played, and has not stopped since)

113let receivers = null;
114let listening = false;
115let ourStop = false; // a stop this module made, not the listener

host: the decisions watcher, this page's last start, and messages that wait for the room to number the run

119let watching = null;
120let started = null; // { at, cps } in this page's clock
121let queue = null;

123function send(message) {
124  if (socket?.readyState === 1) socket.send(JSON.stringify(message));
125}
126
127export function linkFor(room, songId) {
128  return `${location.origin}${songPath({ id: songId })}?${PARTY_PARAM}=${room}`;
129}
130
131function ping() {
132  samples = [];
133  for (let i = 0; i < PINGS; i++) setTimeout(() => send({ t: 'ping', c: Date.now() }), i * 150);
134}

Whether a party this page hosts is listed for anyone to join: the lobby's "show me here" (lobby.mjs sets it, which imports this module). Sent as the host connects, and again the moment it changes (listedChanged), so the party leaves or rejoins the live list at once; its guests stay in either way.

141let listed = () => false;
142export const setListedCheck = (fn) => void (listed = fn);
143export function listedChanged() {
144  if (state.role !== 'host') return;
145  const on = listed();
146  // the room checks the name is its own before keeping it (party-room.ts)
147  send({ t: 'listed', listed: on, ...(on ? { room: state.room } : {}) });
148}

150function connect() {
151  clearTimeout(retry);
152  const { room, role } = state;
153  const protocol = location.protocol === 'https:' ? 'wss:' : 'ws:';
154  const params = new URLSearchParams({ role, ...(role === 'host' ? { key, listed: listed() ? '1' : '0' } : {}) });
155  const ws = new WebSocket(`${protocol}//${location.host}${PARTY_PATH}${room}?${params}`);
156  socket = ws;
157  ws.onopen = () => {
158    retryMs = 1000;
159    ping();
160    clearInterval(pinger);
161    pinger = setInterval(ping, REPING_MS);
162  };
163  ws.onmessage = (e) => {
164    let m;
165    try {
166      m = JSON.parse(e.data);
167    } catch {
168      return;
169    }
170    if (socket === ws) onMessage(m);
171  };
172  ws.onclose = (e) => {
173    if (socket !== ws) return;
174    clearInterval(pinger);
175    if (e.code >= 4400 && e.code < 4600) {
176      set({ status: 'ended', note: e.reason || 'this party has ended' });
177      socket = null;
178      return;
179    }
180    if (state.status === 'off') return;
181    set({ status: 'reconnecting' });
182    retry = setTimeout(connect, retryMs);
183    retryMs = Math.min(retryMs * 2, 10_000);
184  };
185}
186
187function resetPerformance(next) {
188  run = next;
189  clock = null;
190  decisions.clear();
191  reactions.clear();
192  receivers = null;
193}
194
195function onMessage(m) {
196  switch (m.t) {
197    case 'pong': {
198      samples.push({ c: m.c, s: m.s, r: Date.now() });
199      const best = clockOffset(samples);
200      if (best) {
201        offset = best.offset;
202        set({ rtt: Math.round(best.rtt) });
203      }
204      return;
205    }
206    case 'welcome': {
207      set({ status: 'open', people: m.people, hostHere: m.host });
208      // the same performance as before a reconnect: carry on with it
209      const same = Boolean(run && m.run && run.run === m.run.run);
210      if (state.role === 'host') {
211        if (!started) return;
212        // a party started while a song plays joins it now; a reconnect
213        // re-sends what the room may have missed (it keeps one of each)
214        if (same) share();
215        else announce();
216        return;
217      }
218      if (!same) resetPerformance(m.run);
219      for (const x of m.decisions ?? []) {
220        decisions.set(`${x.view}:${x.segment}`, x);
221        if (same) receivers?.[x.view]?.(x.segment, x.d);
222      }
223      for (const [segment, counts] of Object.entries(m.reactions ?? {})) {
224        reactions.set(Number(segment), counts);
225        if (same) setReactions(Number(segment), counts);
226      }
227      if (m.clock) gotClock(m.clock);
228      set({ song: m.run?.song ?? null });
229      const joining = playOnJoin && Boolean(m.run);
230      if (joining) takePlayOnJoin();
231      followSong({ restart: joining || (!same && listening) });
232      return;
233    }
234    case 'people':
235      set({ people: m.n, hostHere: m.host });
236      return;
237    case 'run':
238      if (state.role === 'host') {
239        // the room numbered our run: what waited for it goes now
240        run = { run: m.run, song: m.song, code: m.code };
241        const waiting = queue ?? [];
242        queue = null;
243        for (const message of waiting) send({ ...message, run: m.run });
244        return;
245      }
246      resetPerformance({ run: m.run, song: m.song, code: m.code });
247      set({ song: m.song, joinedAt: null });
248      takePlayOnJoin();
249      followSong({ restart: true });
250      return;
251    case 'clock':
252      if (state.role === 'guest' && run && m.run === run.run) gotClock(m);
253      return;
254    case 'decision': {
255      if (state.role !== 'guest' || !run || m.run !== run.run) return;
256      decisions.set(`${m.view}:${m.segment}`, m);
257      receivers?.[m.view]?.(m.segment, m.d);
258      return;
259    }
260    case 'reactions':
261      reactions.set(m.segment, { fire: m.fire, sleep: m.sleep });
262      setReactions(m.segment, m);
263      return;
264    case 'end':
265      if (state.role === 'guest' && run && m.run === run.run) {
266        clock = null;
267        stopForHost();
268        set({ waiting: listening, joinedAt: null });
269      }
270      return;
271    case 'refused':
272    case 'error':
273      set({ note: m.why });
274      return;
275  }
276}
277
278function gotClock(c) {
279  clock = { startedAt: c.startedAt, cps: c.cps };
280  releaseClockWaiters();
281}

A guest's start waiting for the host's clock (partyStarted) ends when the clock arrives, and also when the guest leaves or stops: evaluations run one at a time (packages/core/repl.mjs), so a start left waiting would hold every later one. Without a clock partyStarted returns and sets nothing.

287function releaseClockWaiters() {
288  const waiting = clockWaiters;
289  clockWaiters = [];
290  for (const resolve of waiting) resolve();
291}

── joining and leaving ───────────────────────────

294function begin(room, role) {
295  resetPerformance(null);
296  set({ room, role, status: 'connecting', people: 0, hostHere: false, note: null, joinedAt: null, waiting: false });
297  connect();
298}

300function urlWithParty(room) {
301  try {
302    const url = new URL(location.href);
303    if (room) url.searchParams.set(PARTY_PARAM, room);
304    else url.searchParams.delete(PARTY_PARAM);
305    history.replaceState(history.state, '', url);
306  } catch {
307    // a sandboxed page: the party works, the address bar does not show it
308  }
309}

The page opened a party link (?party=<room>): join as a guest, or as the host when this tab started that party (its key is in sessionStorage).

313export function initParty(ref) {
314  editorRef = ref;
315  if (typeof location === 'undefined') return;
316  const room = new URLSearchParams(location.search).get(PARTY_PARAM);
317  if (!room || !ROOM.test(room)) return;
318  try {
319    key = sessionStorage.getItem(HOST_KEY(room));
320  } catch {
321    key = null;
322  }
323  begin(room, key ? 'host' : 'guest');
324}

Listening along from the lobby (lobby.mjs): join room, which the tab listened to started for us, as a guest, and play the host's song as soon as the room says which it is. The click on "listen along" is the gesture that lets it play.

330let playOnJoin = false;
331export function joinParty(room) {
332  if (!ROOM.test(room)) return;
333  if (state.status !== 'off' && state.status !== 'ended') leaveParty();
334  key = null;
335  playOnJoin = true;
336  urlWithParty(room);
337  begin(room, 'guest');
338}

A listener who joined from the lobby is in once the song is known.

340function takePlayOnJoin() {
341  if (!playOnJoin) return;
342  playOnJoin = false;
343  listening = true;
344}

"listen together": start a party on the song that is playing (or about to).

347export function startParty() {
348  if (state.status !== 'off' && state.status !== 'ended') return state.room;
349  const room = randomId();
350  key = randomId(24);
351  try {
352    sessionStorage.setItem(HOST_KEY(room), key);
353  } catch {
354    // a host that cannot keep its key cannot reclaim the party on reload
355  }
356  urlWithParty(room);
357  begin(room, 'host');
358  return room;
359}

361export function leaveParty() {
362  const { room, role } = state;
363  set({ status: 'off' });
364  clearTimeout(retry);
365  clearInterval(pinger);
366  socket?.close(1000, 'left');
367  socket = null;
368  watching?.stop();
369  watching = null;
370  if (role === 'host') {
371    try {
372      sessionStorage.removeItem(HOST_KEY(room));
373    } catch {}
374  }
375  if (role === 'guest' && listening) stopForHost();
376  listening = false;
377  playOnJoin = false;
378  resetPerformance(null);
379  urlWithParty(null);
380  set({ room: null, role: null, people: 0, song: null, note: null, joinedAt: null, waiting: false });
381  releaseClockWaiters();
382}

A reaction: through the room when in a party (it counts everyone's and tells everyone, the host's Jev included), else straight to the booth.

386export function react(booth, segment, kind) {
387  if (state.status === 'open' && run) send({ t: 'react', segment, kind });
388  else booth.react(segment, kind);
389}

── host ──────────────────────────────────────────

392const songOf = () => readSongs().find((s) => s.id === nowPlaying.get());

Announce the performance that is starting (or playing): the song, then every decision as its jev()s settle it, and the clock once it starts.

396function announce() {
397  const song = songOf();
398  if (!song) {
399    set({ note: 'only a song from the list can be shared: pick one, and the party plays it' });
400    return;
401  }
402  const code = editorRef?.current?.code ?? '';
403  queue = [];
404  send({ t: 'run', song: song.id, code: codeHash(code) });
405  set({ song: song.id, note: code === song.code ? null : 'you edited this song: guests hear it as published' });
406  share();
407}

Watch the performance's decisions, and send those already settled and the clock (again, after a reconnect: the room keeps one of each).

410function share() {
411  watching?.stop();
412  watching = watchDecisions((view, segment, d) => sendRun({ t: 'decision', view, segment, d }));
413  for (const x of watching.settled) sendRun({ t: 'decision', view: x.view, segment: x.segment, d: x.decision });
414  if (started) sendRun({ t: 'clock', startedAt: started.at + offset, cps: started.cps });
415}

a message that belongs to the current run, which the room may not have numbered yet

418function sendRun(message) {
419  if (queue) queue.push(message);
420  else if (run) send({ ...message, run: run.run });
421}

── the scheduler's hooks (useReplContext) ───────── Before the song's opening is asked about: the host announces the new performance and watches its decisions; a guest's jev()s start following.

426export function partyBeforeStart() {
427  started = null;
428  if (state.status === 'off' || state.status === 'ended') return;
429  if (state.role === 'host') {
430    if (state.status === 'open') announce();
431    return;
432  }
433  listening = true;
434  receivers = followDecisions();
435  for (const { view, segment, d } of decisions.values()) receivers[view]?.(segment, d);
436  for (const [segment, counts] of reactions) setReactions(segment, counts);
437}

The last step before the scheduler's first tick. The host notes its start and tells the room; a guest waits for the host's start and begins at the cycle the host is on now.

442export async function partyStarted(scheduler) {
443  if (state.role === 'guest' && state.status !== 'off' && state.status !== 'ended') {
444    if (!clock) {
445      set({ waiting: true });
446      await new Promise((resolve) => clockWaiters.push(resolve));
447      set({ waiting: false });
448    }
449    if (!clock) return; // left meanwhile
450    if (scheduler.cps !== clock.cps) scheduler.setCps(clock.cps);
451    let cycle = cycleNow(clock, roomNow());
452    if (cycle < 0) {
453      await new Promise((r) => setTimeout(r, (-cycle / clock.cps) * 1000));
454      cycle = cycleNow(clock, roomNow());
455    }
456    // the cyclist's first tick begins at lastEnd (packages/core/cyclist.mjs)
457    scheduler.lastEnd = Math.max(0, cycle);
458    set({ joinedAt: scheduler.lastEnd });
459    if (songOf()?.id !== run?.song) setTimeout(() => followSong({ restart: true }), 0);
460    return;
461  }
462  started = { at: Date.now(), cps: scheduler.cps };
463  if (state.role === 'host' && state.status === 'open') {
464    sendRun({ t: 'clock', startedAt: started.at + offset, cps: started.cps });
465  }
466}

468export function partyToggled(playing) {
469  if (playing) return;
470  releaseClockWaiters();
471  started = null;
472  if (state.role === 'host') {
473    watching?.stop();
474    watching = null;
475    sendRun({ t: 'end' });
476    return;
477  }
478  if (state.role === 'guest') {
479    // the listener's own stop: they are out until they press play again
480    if (ourStop) ourStop = false;
481    else listening = false;
482  }
483}

── guest ─────────────────────────────────────────

486function stopForHost() {
487  const editor = editorRef?.current;
488  if (!editor?.repl?.scheduler?.started) return;
489  ourStop = true; // partyToggled consumes it
490  editor.stop();
491}

Load the performance's song into the editor (the host may have moved on to another), and, for a listener who is in, play it: it starts in sync once the host's clock arrives.

496function followSong({ restart = false } = {}) {
497  const editor = editorRef?.current;
498  const song = readSongs().find((s) => s.id === run?.song);
499  if (!editor || !song) return;
500  if (song.code && codeHash(song.code) !== run.code) {
501    set({ note: 'the host edited this song: you hear it as published, with the host’s Jev' });
502  }
503  const differs = nowPlaying.get() !== song.id || editor.code !== song.code;
504  if (!listening) {
505    if (differs) {
506      setOwnCode(editor, song.code);
507      nowPlaying.set(song.id);
508      showSong(song);
509    }
510    return;
511  }
512  if (differs || restart) {
513    // playSong stops first; that stop is ours (partyToggled consumes it)
514    ourStop = Boolean(editor.repl?.scheduler?.started);
515    playSong(editor, song);
516  }
517}

519export const isGuest = () => state.role === 'guest' && state.status !== 'off';