char-slop/berry

git clone https://git.t4t.associates/char-slop/berry

Charlotte Somspeed up conversation switchingd1e4a762026-10-08

import { Signal } from "@char/aftercare";
import { applyImmutable, type Op } from "@earendil-works/chord/delta";

const TOKEN_KEY = "berry.token";

export const token = new Signal<string | null>(localStorage.getItem(TOKEN_KEY));
export const connected = new Signal(false);

export function setToken(value: string | null) {
  if (value === null) localStorage.removeItem(TOKEN_KEY);
  else localStorage.setItem(TOKEN_KEY, value);
  token.set(value);
}

export class ApiError extends Error {
  constructor(readonly status: number, message: string) {
    super(message);
  }
}

export async function api<T = unknown>(
  method: string,
  path: string,
  body?: unknown,
  signal?: AbortSignal,
): Promise<T> {
  const response = await fetch(path, {
    method,
    signal,
    headers: {
      authorization: `Bearer ${token.get()}`,
      ...(body === undefined ? {} : { "content-type": "application/json" }),
    },
    ...(body === undefined ? {} : { body: JSON.stringify(body) }),
  });
  const json = await response.json().catch(() => null);
  if (response.status === 401) setToken(null);
  if (!response.ok) throw new ApiError(response.status, json?.error ?? response.statusText);
  return json as T;
}

type Topic = {
  value?: { current: unknown };
  listeners: Set<(value: unknown) => void>;
  /** A snapshot still taking its older entries, and whether its listeners see it meanwhile. */
  filling?: { value: { entries: unknown[] }; older: number; shown: boolean };
};

const topics = new Map<string, Topic>();
let socket: WebSocket | null = null;
let retry = 0;

const send = (message: object) => {
  if (socket?.readyState === WebSocket.OPEN) socket.send(JSON.stringify(message));
};

function connect() {
  const current = new WebSocket(
    `${location.protocol === "https:" ? "wss" : "ws"}://${location.host}/ws`,
  );
  socket = current;
  current.onopen = () => current.send(JSON.stringify({ t: "auth", token: token.get() }));
  current.onmessage = (event) => {
    const message = JSON.parse(event.data);
    if (message.t === "ready") {
      retry = 0;
      connected.set(true);
      for (const name of topics.keys()) send({ t: "sub", topic: name });
    } else if (message.t === "snapshot") {
      const topic = topics.get(message.topic);
      if (topic === undefined) return;
      // a first snapshot shows as it arrives, newest first; a later one, such as a reconnect's, once it is whole
      topic.filling = message.older
        ? { value: message.value, older: message.older, shown: topic.value === undefined }
        : undefined;
      if (topic.filling?.shown !== false) publish(topic, message.value);
    } else if (message.t === "older") {
      const topic = topics.get(message.topic);
      const filling = topic?.filling;
      if (filling === undefined) return;
      const { value } = filling;
      filling.value = { ...value, entries: [...message.entries, ...value.entries] };
      filling.older -= message.entries.length;
      if (filling.older <= 0) topic!.filling = undefined;
      if (filling.shown || filling.older <= 0) publish(topic, filling.value);
    } else if (message.t === "ops") {
      const topic = topics.get(message.topic);
      if (topic?.value) publish(topic, applyImmutable(topic.value.current, message.ops as Op[]));
    } else if (message.t === "error") {
      console.error(`[berry] ${message.topic ?? "socket"}: ${message.message}`);
    }
  };
  current.onclose = (event) => {
    if (socket !== current) return;
    socket = null;
    connected.set(false);
    if (event.code === 4001) setToken(null);
    else if (token.get() !== null) setTimeout(connect, Math.min(10_000, 250 * 2 ** retry++));
  };
}

token.subscribeImmediate((value) => {
  if (value !== null && socket === null) connect();
  if (value === null) socket?.close();
});

function publish(topic: Topic | undefined, value: unknown) {
  if (topic === undefined) return;
  topic.value = { current: value };
  for (const listener of topic.listeners) listener(value);
}

/**
 * Follow a replicated json value: `listener` gets its snapshot once there is one, and then each revision, whose
 * unchanged parts keep their identity. Returns the release, for when its view goes away.
 */
export function watch<T>(name: string, listener: (value: T) => void): () => void {
  if (!topics.has(name)) {
    topics.set(name, { listeners: new Set() });
    send({ t: "sub", topic: name });
  }
  const topic = topics.get(name)!;
  const own = (value: unknown) => listener(value as T);
  topic.listeners.add(own);
  if (topic.value) own(topic.value.current);
  return () => {
    if (!topic.listeners.delete(own) || topic.listeners.size > 0) return;
    topics.delete(name);
    send({ t: "unsub", topic: name });
  };
}