API reference

// @ts-check
/**
 * Follow a CueStride room from an API key's stream, with only the items that changed.
 *
 * This is the reference client the API docs show, and CueStride's tests run this file as it is,
 * against the server. Plain JavaScript with no dependencies, for Node 20 or later and current
 * browsers: copy it whole.
 *
 * It keeps a copy of the room as a moderator sees it: the room, the playback, the rundown with
 * timer notes blank, and every message. It applies each update. When an update does not follow on,
 * does not fit, or a heartbeat's digest says the copy drifted, it reopens the stream, whose first
 * event is a fresh snapshot. It never mends the copy from the list reads, which carry notes.
 */

/**
 * @typedef {{ id: string, position: number, updatedAt: number, [field: string]: unknown }} Item
 * @typedef {{ remove?: readonly string[], patch?: ReadonlyArray<{ id: string, [field: string]: unknown }>, add?: readonly Item[] }} ListChanges
 * @typedef {{ revision: number, room: unknown, playback: unknown, timers: Item[], messages: Item[] }} RoomCopy
 * @typedef {{ end: "kicked", reason: string } | { end: "reopen" } | { end: "dropped" }} Outcome
 */

/** No event of any kind for this long, and the connection is taken for dead. */
const SILENCE_MS = 45_000;
/** The wait before reopening a stream that dropped: doubling from the first to the last. */
const FIRST_WAIT_MS = 2_000;
const LAST_WAIT_MS = 10_000;
/** The longest wait a timer holds: past it, Node fires at once. */
const LONGEST_WAIT_MS = 2_147_483_647;

/**
 * One list's changes applied: remove, then patch, then add, then sort by position. A patch sets
 * each field it names, null included, and leaves the rest. Null when they do not fit: an id the
 * list does not hold removed or patched, one it holds added, or positions that are not 0 to n−1.
 * @param {readonly Item[]} list
 * @param {ListChanges} changes
 * @returns {Item[] | null}
 */
export function applyChanges(list, changes) {
  const items = new Map(list.map((item) => [item.id, item]));
  for (const id of changes.remove ?? []) {
    if (!items.delete(id)) {
      return null;
    }
  }
  for (const fields of changes.patch ?? []) {
    const item = items.get(fields.id);
    if (item === undefined) {
      return null;
    }
    items.set(fields.id, { ...item, ...fields });
  }
  for (const item of changes.add ?? []) {
    if (items.has(item.id)) {
      return null;
    }
    items.set(item.id, item);
  }
  const next = [...items.values()].sort((a, b) => a.position - b.position);
  return next.every((item, i) => item.position === i) ? next : null;
}

/**
 * An update applied to a copy: `room` and `playback` replace theirs, a list comes whole as an
 * array or as item changes, and a section that is absent keeps its own. Null when a list's
 * changes do not fit.
 * @param {RoomCopy} copy
 * @param {{ revision: number, changes: { room?: unknown, playback?: unknown, timers?: Item[] | ListChanges, messages?: Item[] | ListChanges } }} update
 * @returns {RoomCopy | null}
 */
export function applyUpdate(copy, update) {
  const { room, playback, timers, messages } = update.changes;
  const nextTimers = listAfter(copy.timers, timers);
  const nextMessages = listAfter(copy.messages, messages);
  if (nextTimers === null || nextMessages === null) {
    return null;
  }
  return {
    revision: update.revision,
    room: room === undefined ? copy.room : room,
    playback: playback === undefined ? copy.playback : playback,
    timers: nextTimers,
    messages: nextMessages,
  };
}

/**
 * @param {Item[]} list
 * @param {Item[] | ListChanges | undefined} section
 * @returns {Item[] | null}
 */
function listAfter(list, section) {
  if (section === undefined) {
    return list;
  }
  return Array.isArray(section) ? section : applyChanges(list, section);
}

/**
 * A list's digest, as a heartbeat sends it: each item's JSON, its keys sorted at every level and
 * its top-level `updatedAt` left out, hashed; each hash in base 36 followed by a comma; and that
 * text hashed again, in base 36.
 * @param {readonly Item[]} list
 * @returns {string}
 */
export function listDigest(list) {
  let text = "";
  for (const item of list) {
    text += `${hash53(canonical(item, true)).toString(36)},`;
  }
  return hash53(text).toString(36);
}

/**
 * JSON with every object's keys sorted, and without `updatedAt` at an item's top level.
 * @param {unknown} value
 * @param {boolean} [item]
 * @returns {string}
 */
function canonical(value, item = false) {
  if (value === null || typeof value !== "object") {
    return JSON.stringify(value) ?? "null";
  }
  if (Array.isArray(value)) {
    return `[${value.map((v) => canonical(v)).join(",")}]`;
  }
  const record = /** @type {Record<string, unknown>} */ (value);
  const keys = Object.keys(record)
    .filter((k) => record[k] !== undefined && !(item && k === "updatedAt"))
    .sort();
  return `{${keys.map((k) => `${JSON.stringify(k)}:${canonical(record[k])}`).join(",")}}`;
}

/**
 * A 53-bit hash of a string's UTF-16 code units: two 32-bit multiply-xor lanes, mixed.
 * @param {string} text
 * @returns {number}
 */
function hash53(text) {
  let h1 = 0xdeadbeef;
  let h2 = 0x41c6ce57;
  for (let i = 0; i < text.length; i += 1) {
    const c = text.charCodeAt(i);
    h1 = Math.imul(h1 ^ c, 2654435761);
    h2 = Math.imul(h2 ^ c, 1597334677);
  }
  h1 = Math.imul(h1 ^ (h1 >>> 16), 2246822507);
  h1 ^= Math.imul(h2 ^ (h2 >>> 13), 3266489909);
  h2 = Math.imul(h2 ^ (h2 >>> 16), 2246822507);
  h2 ^= Math.imul(h1 ^ (h1 >>> 13), 3266489909);
  return 4294967296 * (2097151 & h2) + (h1 >>> 0);
}

/**
 * The events of a server-sent event stream as `{ event, data }`, whatever chunks they arrive in.
 * Ending early, by `return` or `break` in the loop that reads them, closes the stream.
 * @param {ReadableStream<Uint8Array>} body
 * @returns {AsyncGenerator<{ event: string, data: string }>}
 */
export async function* serverEvents(body) {
  const reader = body.getReader();
  const decoder = new TextDecoder();
  let buffer = "";
  let event = "";
  /** @type {string[]} */
  let data = [];
  try {
    for (;;) {
      const chunk = await reader.read();
      if (chunk.done) {
        return;
      }
      buffer += decoder.decode(chunk.value, { stream: true });
      for (;;) {
        const end = buffer.search(/[\r\n]/);
        // A CR at the very end may be half of a CRLF still on its way.
        if (end === -1 || (buffer[end] === "\r" && end === buffer.length - 1)) {
          break;
        }
        const line = buffer.slice(0, end);
        buffer = buffer.slice(buffer.startsWith("\r\n", end) ? end + 2 : end + 1);
        if (line === "") {
          if (data.length > 0) {
            yield { event: event || "message", data: data.join("\n") };
          }
          event = "";
          data = [];
          continue;
        }
        const colon = line.indexOf(":");
        const field = colon === -1 ? line : line.slice(0, colon);
        const value = colon === -1 ? "" : line.slice(colon + 1).replace(/^ /, "");
        if (field === "event") {
          event = value;
        } else if (field === "data") {
          data.push(value);
        }
      }
    }
  } finally {
    await reader.cancel().catch(() => {});
  }
}

/**
 * Follows a room until `signal` aborts or the server ends the key's stream: it returns the reason
 * the server gave with `kicked`, such as "replaced", "revoked", "expired", "secret replaced",
 * "disabled", "api off", or "disconnected by the operator", or "aborted". It calls `onCopy` with
 * the copy after every change; the copy is frozen through and through, for reading only. A
 * refusal that waiting will not mend, a 4xx such as a revoked key's 401 or a full room's 403,
 * throws an Error with the
 * answer's `status` and `code`, and so does an answer that is not an event stream. An error
 * `onCopy` throws comes out as it is. A dropped stream, a network failure, or a 5xx, such as a
 * proxy's 502 while the server restarts, is waited out and the stream opened again.
 * @param {{
 *   url: string,
 *   key: string,
 *   onCopy: (copy: Readonly<RoomCopy>) => void,
 *   signal?: AbortSignal,
 *   request?: typeof fetch,
 *   wait?: (ms: number, signal?: AbortSignal) => Promise<void>,
 *   silenceMs?: number,
 * }} options `url` is the room's stream, such as
 *   https://app.cuestride.com/api/v1/rooms/K7QW3ND2XA/events, and `key` one that reads the room.
 * @returns {Promise<string>}
 */
export async function followRoom({
  url,
  key,
  onCopy,
  signal,
  request = fetch,
  wait = pauseFor,
  silenceMs = SILENCE_MS,
}) {
  let pause = FIRST_WAIT_MS;
  for (;;) {
    if (signal?.aborted) {
      return "aborted";
    }
    // One connection's own abort: the caller's signal, or a silence that outlasts the heartbeats.
    const connection = new AbortController();
    const stop = () => connection.abort();
    signal?.addEventListener("abort", stop);
    let silence = setTimeout(stop, silenceMs);
    const heard = () => {
      clearTimeout(silence);
      silence = setTimeout(stop, silenceMs);
    };
    const progress = { connected: false, applied: 0 };
    /** @type {Outcome} */
    let outcome = { end: "dropped" };
    /** How long a 429 says to wait, or null. */
    let told = null;
    try {
      const res = await request(`${url}?lists=items`, {
        headers: { authorization: `Bearer ${key}` },
        signal: connection.signal,
      });
      if (res.status === 429) {
        // Too many opens, or ten streams for this key already: wait as long as it says.
        await res.body?.cancel();
        told = retryAfterMs(res.headers.get("retry-after"), pause);
      } else if (res.status >= 500) {
        // The server or a proxy in front of it, restarting: as a stream that dropped.
        await res.body?.cancel();
      } else if (!res.ok || res.body === null) {
        throw await refusal(res);
      } else if (!(res.headers.get("content-type") ?? "").includes("text/event-stream")) {
        await res.body.cancel();
        throw Object.assign(new Error(`${url} is not a room's stream`), {
          status: res.status,
          code: "NOT_A_STREAM",
        });
      } else {
        outcome = await follow(res.body, onCopy, heard, progress);
      }
    } catch (error) {
      if (error !== null && typeof error === "object" && "caller" in error) {
        throw error.caller;
      }
      if (/** @type {{ status?: number }} */ (error).status !== undefined) {
        throw error;
      }
      // The network failed, or the silence ended the connection: as a stream that dropped.
    } finally {
      clearTimeout(silence);
      signal?.removeEventListener("abort", stop);
    }
    if (outcome.end === "kicked") {
      return outcome.reason;
    }
    if (signal?.aborted) {
      return "aborted";
    }
    if (told !== null) {
      await wait(told, signal);
      continue;
    }
    // After a stream that delivered its snapshot, the least wait again, however it ended.
    if (progress.connected) {
      pause = FIRST_WAIT_MS;
    }
    // A copy that needs a fresh snapshot after applying updates reopens at once; one that needs it
    // again before any update waits, so a change that never fits cannot spin the client.
    if (outcome.end === "reopen" && progress.applied > 0) {
      continue;
    }
    await wait(pause, signal);
    pause = Math.min(pause * 2, LAST_WAIT_MS);
  }
}

/**
 * The milliseconds a Retry-After asks for, in seconds or as a date; `fallback` when it says
 * neither; never less than a second, nor more than a timer holds.
 * @param {string | null} value
 * @param {number} fallback
 * @returns {number}
 */
function retryAfterMs(value, fallback) {
  if (value === null || value.trim() === "") {
    return fallback;
  }
  const clamp = (/** @type {number} */ ms) => Math.min(Math.max(ms, 1000), LONGEST_WAIT_MS);
  const seconds = Number(value);
  if (Number.isFinite(seconds)) {
    // Seconds as they are: a product too large for a number is past the cap anyway.
    return clamp(seconds * 1000);
  }
  const at = Date.parse(value);
  return Number.isFinite(at) ? clamp(at - Date.now()) : fallback;
}

/**
 * A refusal as an Error with its status and code.
 * @param {Response} res
 * @returns {Promise<Error>}
 */
async function refusal(res) {
  const answer = await res.json().catch(() => ({}));
  return Object.assign(new Error(answer.error?.message ?? `HTTP ${res.status}`), {
    status: res.status,
    code: answer.error?.code,
  });
}

/**
 * A pause that ends early when `signal` aborts.
 * @param {number} ms
 * @param {AbortSignal} [signal]
 * @returns {Promise<void>}
 */
function pauseFor(ms, signal) {
  return new Promise((resolve) => {
    const done = () => {
      clearTimeout(timer);
      signal?.removeEventListener("abort", done);
      resolve();
    };
    const timer = setTimeout(done, ms);
    signal?.addEventListener("abort", done);
  });
}

/**
 * A value frozen with everything in it, so no write to the copy can change what the next update
 * applies to. What is frozen already, an item no update touched, is left as it is.
 * @template T
 * @param {T} value
 * @returns {T}
 */
function freeze(value) {
  if (value !== null && typeof value === "object" && !Object.isFrozen(value)) {
    for (const inner of Object.values(value)) {
      freeze(inner);
    }
    Object.freeze(value);
  }
  return value;
}

/**
 * One connection's events, until the copy needs a fresh snapshot, the server ends the stream for
 * this key, or the stream drops. `progress` says whether a snapshot came and how many updates were
 * applied, whichever way the connection ends.
 * @param {ReadableStream<Uint8Array>} body
 * @param {(copy: Readonly<RoomCopy>) => void} onCopy
 * @param {() => void} heard
 * @param {{ connected: boolean, applied: number }} progress
 * @returns {Promise<Outcome>}
 */
async function follow(body, onCopy, heard, progress) {
  /** @type {RoomCopy | null} */
  let copy = null;
  /**
   * The copy, frozen through and through, handed to `onCopy`; an error it throws is the caller's.
   * @param {RoomCopy} next
   * @returns {RoomCopy}
   */
  const deliver = (next) => {
    const frozen = freeze(next);
    try {
      onCopy(frozen);
    } catch (error) {
      throw { caller: error };
    }
    return frozen;
  };
  for await (const { event, data } of serverEvents(body)) {
    heard();
    if (event === "snapshot") {
      const { revision, room, playback, timers, messages } = JSON.parse(data);
      progress.connected = true;
      copy = deliver({ revision, room, playback, timers, messages });
    } else if (event === "update") {
      const update = JSON.parse(data);
      // Not the next revision: one was lost, or this one came twice.
      /** @type {RoomCopy | null} */
      const next =
        copy !== null && update.revision === copy.revision + 1 ? applyUpdate(copy, update) : null;
      if (next === null) {
        return { end: "reopen" };
      }
      progress.applied += 1;
      copy = deliver(next);
    } else if (event === "heartbeat" && copy !== null) {
      const beat = JSON.parse(data);
      if (beat.revision > copy.revision) {
        // The last update before a quiet spell was lost.
        return { end: "reopen" };
      }
      const drifted =
        beat.revision === copy.revision &&
        beat.digest !== undefined &&
        (beat.digest.timers !== listDigest(copy.timers) ||
          beat.digest.messages !== listDigest(copy.messages));
      if (drifted) {
        return { end: "reopen" };
      }
    } else if (event === "kicked") {
      return { end: "kicked", reason: JSON.parse(data).reason };
    }
  }
  return { end: "dropped" };
}