import { Signal } from "@char/aftercare"; import { applyImmutable, type Op } from "@earendil-works/chord/delta"; const TOKEN_KEY = "berry.token"; export const token = new Signal(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( method: string, path: string, body?: unknown, signal?: AbortSignal, ): Promise { 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(); 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(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 }); }; }