Something went wrong. Try again.
Monorepo for the nibling.run platform
Something went wrong. Try again.
1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859import { EphemeralStore, VersionVector } from "loro-crdt";import type { Doc, Unsubscribe } from "./doc.ts";
/** Anything that carries structured-clone messages both ways: a MessagePort, a BroadcastChannel, a relay socket adapter. */export interface Port { postMessage(message: Message): void; addEventListener(type: "message", listener: (event: MessageEvent<Message>) => void): void; removeEventListener(type: "message", listener: (event: MessageEvent<Message>) => void): void;}
export type Message = | { readonly t: "hello"; readonly vv: Uint8Array } | { readonly t: "update"; readonly bytes: Uint8Array } | { readonly t: "presence"; readonly bytes: Uint8Array };
/** Cursor and selection state shared over the same port. Entries expire after `timeout` ms without an update. */export function createPresence(timeout = 30_000): EphemeralStore { return new EphemeralStore(timeout);}
/** * Keeps `doc` in sync with whoever is on the other end of `port`. * Each side announces its version vector, the other answers with what it is missing, * and from then on every local commit is forwarded. The relay can forward these * messages without understanding Loro (D16). */export function connect(doc: Doc, port: Port, presence?: EphemeralStore): Unsubscribe { const { loro } = doc;
const onMessage = ({ data }: MessageEvent<Message>) => { switch (data.t) { case "hello": port.postMessage({ t: "update", bytes: loro.export({ mode: "update", from: VersionVector.decode(data.vv) }) }); if (presence) port.postMessage({ t: "presence", bytes: presence.encodeAll() }); return; case "update": loro.import(data.bytes); return; case "presence": presence?.apply(data.bytes); return; } };
port.addEventListener("message", onMessage); if ("start" in port && typeof port.start === "function") port.start();
const stopUpdates = doc.onUpdate((bytes) => port.postMessage({ t: "update", bytes })); const stopPresence = presence?.subscribeLocalUpdates((bytes) => port.postMessage({ t: "presence", bytes }));
port.postMessage({ t: "hello", vv: loro.oplogVersion().encode() });
return () => { port.removeEventListener("message", onMessage); stopUpdates(); stopPresence?.(); };}