1// Listening parties: a host plays a song, guests who open its link hear the
2// same song with the same Jev decisions, in time with the host.
3//
4// One room per party, a Durable Object in the site's Worker
5// (worker/src/party-room.ts has the protocol). The host's page is the only
6// one that asks Jev: it announces each performance (`run`: the song), every
7// decision its jev()s settle (`decision`, from jevCore's watchDecisions) and
8// when its scheduler started (`clock`, in the room's clock). A guest's
9// jev()s follow those decisions (jevCore's followDecisions) and its
10// scheduler starts at the cycle the host is on, so a guest joining mid-song
11// starts mid-song. 🔥/😴 from anyone go to the room, which counts them and
12// tells everyone the totals, so the host's Jev hears every listener.
13//
14// Time: each page estimates the room's clock from a few ping/pong
15// exchanges (NTP-style: the round trip with the smallest delay, its
16// midpoint). The host sends its start in the room's clock; a guest turns
17// it back into its own. Both define "started" the same way, at the moment
18// their scheduler starts, so the scheduler's own fixed latencies cancel.
19//
20// The page calls in at three places (useReplContext): partyBeforeStart and
21// 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';
25
26export const PARTY_PATH = '/jev/party/';
27// Live parties listed to join (the listening tab): GET gives them, and
28// 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}$/;
35
36// 128 random bits, base64url: the room id is the party's only secret from
37// 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}
45
46// 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}
52
53// The room's clock from ping/pong samples: `c` sent and `r` received in the
54// page's clock, `s` the room's reading. The best sample is the fastest round
55// 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}
60
61// The cycle a guest starts at: where the host is now, at the host's cps.
62// 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}
66
67// ── 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};
92
93// ── 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;
103
104// 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 }
110
111// guest: the playing song's receivers, and whether the listener is in
112// (played, and has not stopped since)
113let receivers = null;
114let listening = false;
115let ourStop = false; // a stop this module made, not the listener
116
117// host: the decisions watcher, this page's last start, and messages that
118// wait for the room to number the run
119let watching = null;
120let started = null; // { at, cps } in this page's clock
121let queue = null;
122
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}
135
136// Whether a party this page hosts is listed for anyone to join: the
137// lobby's "show me here" (lobby.mjs sets it, which imports this module).
138// Sent as the host connects, and again the moment it changes
139// (`listedChanged`), so the party leaves or rejoins the live list at once;
140// 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}
149
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}
282
283// A guest's start waiting for the host's clock (partyStarted) ends when the
284// clock arrives, and also when the guest leaves or stops: evaluations run
285// one at a time (packages/core/repl.mjs), so a start left waiting would hold
286// 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}
292
293// ── 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}
299
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}
310
311// The page opened a party link (?party=<room>): join as a guest, or as the
312// 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}
325
326// Listening along from the lobby (lobby.mjs): join `room`, which the tab
327// listened to started for us, as a guest, and play the host's song as soon
328// as the room says which it is. The click on "listen along" is the gesture
329// 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}
339// 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}
345
346// "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}
360
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}
383
384// A reaction: through the room when in a party (it counts everyone's and
385// 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}
390
391// ── host ──────────────────────────────────────────
392const songOf = () => readSongs().find((s) => s.id === nowPlaying.get());
393
394// Announce the performance that is starting (or playing): the song, then
395// 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}
408// Watch the performance's decisions, and send those already settled and
409// 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}
416// a message that belongs to the current run, which the room may not have
417// numbered yet
418function sendRun(message) {
419  if (queue) queue.push(message);
420  else if (run) send({ ...message, run: run.run });
421}
422
423// ── the scheduler's hooks (useReplContext) ─────────
424// Before the song's opening is asked about: the host announces the new
425// 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}
438
439// The last step before the scheduler's first tick. The host notes its
440// start and tells the room; a guest waits for the host's start and begins
441// 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}
467
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}
484
485// ── 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}
492
493// Load the performance's song into the editor (the host may have moved on
494// to another), and, for a listener who is in, play it: it starts in sync
495// 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}
518
519export const isGuest = () => state.role === 'guest' && state.status !== 'off';