Collaborative editing

import * as Y from "yjs";
 
export type Shape = { x: number; y: number; color: string };
export const LOCAL = Symbol("local"); // transaction origin
 
export function createBoard(doc = new Y.Doc()) {
  const shapes = doc.getMap<Y.Map<unknown>>("shapes");
  const order = doc.getArray<string>("order"); // z-order
  const title = doc.getText("title");
 
  function add(id: string, s: Shape) {
    // one transaction = one update, one observer call
    doc.transact(() => {
      shapes.set(id, new Y.Map(Object.entries(s)));
      order.push([id]);
    }, LOCAL);
  }
 
  function move(id: string, x: number, y: number) {
    const s = shapes.get(id);
    if (!s) return;
    doc.transact(() => {
      s.set("x", x); // per-field: concurrent x and color
      s.set("y", y); // edits both survive
    }, LOCAL);
  }
 
  return { doc, shapes, order, title, add, move };
}

import * as Y from "yjs";
import { LOCAL, createBoard } from "./board";
 
const { doc, shapes } = createBoard();
 
// deep: fires for nested Y.Map changes too
shapes.observeDeep((events, tr) => {
  if (tr.origin === LOCAL) return; // already drawn
  for (const e of events) {
    // e.path: ["s1"], e.changes.keys: Map of key → action
    // redraw what e.target / e.path points at
  }
});
 
// every local or remote change, as a binary update
doc.on("update", (update: Uint8Array, origin: unknown) => {
  if (origin !== "remote") send(update); // → provider
});
 
declare function send(u: Uint8Array): void;
export function receive(update: Uint8Array) {
  Y.applyUpdate(doc, update, "remote"); // origin tag
}

import * as Y from "yjs";
 
// the two-step sync every provider runs on connect
export function syncFrom(from: Y.Doc, to: Y.Doc) {
  const sv = Y.encodeStateVector(to); // what `to` has
  const diff = Y.encodeStateAsUpdate(from, sv); // the rest
  Y.applyUpdate(to, diff);
  return diff.byteLength;
}

import * as Y from "yjs";
import {
  Awareness,
  applyAwarenessUpdate,
  encodeAwarenessUpdate,
  removeAwarenessStates,
} from "y-protocols/awareness";
 
export type Presence = {
  user: { name: string; color: string };
  cursor: { x: number; y: number } | null;
  selection: string[]; // selected shape ids
};
 
export function createPresence(
  doc: Y.Doc,
  send: (u: Uint8Array) => void,
) {
  const aw = new Awareness(doc);
  // "update" also fires on the 15 s keep-alive renewal
  aw.on("update", (ch: Changes, origin: unknown) => {
    if (origin === "remote") return;
    const ids = [...ch.added, ...ch.updated, ...ch.removed];
    send(encodeAwarenessUpdate(aw, ids));
  });
  return {
    aw,
    set: <K extends keyof Presence>(k: K, v: Presence[K]) =>
      aw.setLocalStateField(k, v),
    receive: (u: Uint8Array) =>
      applyAwarenessUpdate(aw, u, "remote"),
    // others only: Map<clientID, state>
    peers: () =>
      [...aw.getStates()].filter(
        ([id]) => id !== doc.clientID,
      ),
    leave: () =>
      removeAwarenessStates(aw, [doc.clientID], "leave"),
  };
}
 
type Changes = {
  added: number[];
  updated: number[];
  removed: number[];
};

import { WebsocketProvider } from "y-websocket";
import { IndexeddbPersistence } from "y-indexeddb";
import * as Y from "yjs";
 
const doc = new Y.Doc();
 
// offline first: load the local copy, then go online
const local = new IndexeddbPersistence("board:42", doc);
await local.whenSynced;
 
const ws = new WebsocketProvider(
  "wss://collab.example.com", // server URL
  "board:42", // room = document name
  doc,
  { params: { token: "…" } }, // → ?token=…
);
ws.on("sync", (synced: boolean) => {});
ws.awareness.setLocalStateField("user", { name: "Zach" });

import { HocuspocusProvider } from "@hocuspocus/provider";
import * as Y from "yjs";
 
declare function getToken(): Promise<string>;
 
const doc = new Y.Doc();
const provider = new HocuspocusProvider({
  url: "wss://collab.example.com",
  name: "board:42", // document name
  document: doc,
  token: getToken, // re-read on every reconnect
  onSynced: () => {
    // first full sync done: safe to render or seed
  },
  onAuthenticationFailed: ({ reason }) => {
    // show "no access", stop retrying
  },
});
 
provider.setAwarenessField("user", { name: "Zach" });
// later: provider.destroy()

HOST=localhost PORT=1234 \
  bunx --package @y/websocket-server@0.1.1 y-websocket

import { Hocuspocus } from "@hocuspocus/server";
import { Database } from "bun:sqlite";
import * as Y from "yjs";
 
type User = { id: string; canEdit: boolean };
declare function verify(token: string): Promise<User | null>;
 
const db = new Database("docs.sqlite");
db.run(`create table if not exists docs
  (name text primary key, data blob)`);
 
export const hocuspocus = new Hocuspocus({
  debounce: 2000, // store 2 s after the last change…
  maxDebounce: 10_000, // …but at least every 10 s
  async onAuthenticate({ token, connectionConfig }) {
    const user = await verify(token);
    if (!user) throw new Error("unauthorized"); // rejected
    connectionConfig.readOnly = !user.canEdit;
    return { user }; // becomes `context` in other hooks
  },
  async onLoadDocument({ documentName, document }) {
    const row = db
      .query<{ data: Uint8Array }, [string]>(
        "select data from docs where name = ?",
      )
      .get(documentName);
    if (row) Y.applyUpdate(document, row.data);
  },
  async onStoreDocument({ documentName, document }) {
    db.query("insert or replace into docs values (?, ?)")
      .run(documentName, Y.encodeStateAsUpdate(document));
  },
});

import { hocuspocus } from "./hocus";
 
type Conn = ReturnType<typeof hocuspocus.handleConnection>;
type Data = { req: Request; conn?: Conn };
 
// Bun: `new Server()` is Node-only, attach to Bun.serve
Bun.serve({
  port: 1234,
  fetch(req, server) {
    if (server.upgrade(req, { data: { req } })) return;
    return new Response("Upgrade required", { status: 426 });
  },
  websocket: {
    data: {} as Data,
    open(ws) {
      ws.data.conn = hocuspocus.handleConnection(
        ws,
        ws.data.req,
      );
    },
    message(ws, msg) {
      if (typeof msg === "string") return; // binary only
      ws.data.conn?.handleMessage(new Uint8Array(msg));
    },
    close(ws, code, reason) {
      const ev = { code, reason } as CloseEvent;
      ws.data.conn?.handleClose(ev);
    },
  },
});

import { YServer } from "y-partyserver";
import {
  routePartykitRequest,
  type Connection,
  type ConnectionContext,
} from "partyserver";
import * as Y from "yjs";
 
type Role = { editor: boolean };
declare function verify(
  token: string | null,
): Promise<Role | null>;
 
export class Board extends YServer {
  static override callbackOptions = {
    debounceWait: 2000, // onSave 2 s after edits stop
    debounceMaxWait: 10_000,
  };
 
  override async onLoad() {
    const { storage } = this.ctx;
    const saved = await storage.get<Uint8Array>("doc");
    if (saved) Y.applyUpdate(this.document, saved);
  }
 
  override async onSave() {
    const state = Y.encodeStateAsUpdate(this.document);
    await this.ctx.storage.put("doc", state);
  }
 
  override onConnect(
    conn: Connection,
    ctx: ConnectionContext,
  ) {
    const role = ctx.request.headers.get("x-role");
    conn.setState({ editor: role === "editor" });
    return super.onConnect(conn, ctx);
  }
 
  override isReadOnly(conn: Connection) {
    return !(conn.state as Role | null)?.editor;
  }
}
 
// runs in the Worker, before the DO sees the socket
async function auth(req: Request) {
  const token = new URL(req.url).searchParams.get("token");
  const role = await verify(token);
  if (!role) return new Response(null, { status: 401 });
  const headers = new Headers(req.headers); // overwrite
  headers.set("x-role", role.editor ? "editor" : "viewer");
  return new Request(req, { headers });
}
 
export default {
  async fetch(req: Request, env: Cloudflare.Env) {
    const res = await routePartykitRequest(req, env, {
      onBeforeConnect: auth,
    });
    return res ?? new Response("Not found", { status: 404 });
  },
};

import { Database } from "bun:sqlite";
import * as Y from "yjs";
 
export function updateLog(db: Database) {
  db.run(`create table if not exists updates (
    doc text, id integer primary key, data blob)`);
  const rows = db.query<{ data: Uint8Array }, [string]>(
    "select data from updates where doc = ? order by id",
  );
  const add = db.query(
    "insert into updates (doc, data) values (?, ?)",
  );
  const clear = db.query(
    "delete from updates where doc = ?",
  );
  const load = (doc: string) =>
    Y.mergeUpdates(rows.all(doc).map((r) => r.data));
 
  return {
    load, // → Y.applyUpdate(ydoc, load(name))
    append: (doc: string, u: Uint8Array) => add.run(doc, u),
    // replace N rows with one snapshot (same content)
    compact: db.transaction((doc: string) => {
      const ydoc = new Y.Doc();
      Y.applyUpdate(ydoc, load(doc));
      clear.run(doc);
      add.run(doc, Y.encodeStateAsUpdate(ydoc)); // GC'd
      ydoc.destroy();
    }),
    count: (doc: string) => rows.all(doc).length,
  };
}

import * as Y from "yjs";
 
// an external store over a Y.Map for useSyncExternalStore
export function yMapStore<T>(map: Y.Map<T>) {
  let snap = map.toJSON() as Record<string, unknown>;
  const listeners = new Set<() => void>();
  const onChange = () => {
    snap = map.toJSON(); // new reference only on change
    listeners.forEach((l) => l());
  };
  map.observeDeep(onChange);
  return {
    subscribe(l: () => void) {
      listeners.add(l);
      return () => void listeners.delete(l);
    },
    getSnapshot: () => snap,
    destroy: () => map.unobserveDeep(onChange),
  };
}

"use client";
import { useSyncExternalStore } from "react";
import type { yMapStore } from "./store";
 
type Store = ReturnType<typeof yMapStore>;
 
export function useYMap(store: Store) {
  // 3rd arg: the snapshot used during server rendering
  return useSyncExternalStore(
    store.subscribe,
    store.getSnapshot,
    store.getSnapshot,
  );
}
 
type Shape = { x: number; y: number; color: string };
 
export function Board({ store }: { store: Store }) {
  const shapes = useYMap(store) as Record<string, Shape>;
  return (
    <svg viewBox="0 0 800 600">
      {Object.entries(shapes).map(([id, s]) => (
        <circle key={id} cx={s.x} cy={s.y} r={20}
          fill={s.color} />
      ))}
    </svg>
  );
}

import * as Y from "yjs";
import { LOCAL, createBoard } from "./board";
 
const board = createBoard();
// only transactions tagged LOCAL are undoable: never
// another user's edits
export const undo = new Y.UndoManager(
  [board.shapes, board.order, board.title],
  { trackedOrigins: new Set([LOCAL]), captureTimeout: 500 },
);
 
// keep selection with the stack item for a nice undo
undo.on("stack-item-added", (e) => {
  e.stackItem.meta.set("selection", ["s1"]);
});
undo.on("stack-item-popped", (e) => {
  const sel = e.stackItem.meta.get("selection");
  // restore selection
});
 
// ⌘Z / ⇧⌘Z
undo.undo();
undo.redo();
undo.stopCapturing(); // next edit starts a new step

import * as A from "@automerge/automerge";
 
type Board = { cards: { title: string; done: boolean }[] };
 
let a = A.from<Board>({ cards: [] });
let b = A.clone(a); // a second replica
 
a = A.change(a, (d) => {
  d.cards.push({ title: "Write spec", done: false });
});
b = A.change(b, (d) => {
  d.cards.push({ title: "Draw board", done: false });
});
 
const merged = A.merge(a, b); // both cards, same order
const bytes = A.save(merged); // compact binary
export const history = A.getHistory(merged).length;

import * as Y from "yjs";
 
// build once (a script), store the bytes as a constant
export function makeTemplate(): Uint8Array {
  const doc = new Y.Doc();
  doc.clientID = 1; // stable: same bytes on every build
  doc.transact(() => {
    doc.getText("title").insert(0, "Untitled board");
    doc.getArray<string>("order").push(["welcome"]);
  });
  return Y.encodeStateAsUpdate(doc);
}
 
// anywhere, any number of times: idempotent, never doubles
export function seed(doc: Y.Doc, template: Uint8Array) {
  Y.applyUpdate(doc, template);
}

import type { Awareness } from "y-protocols/awareness";
 
type Point = { x: number; y: number };
 
// pointermove fires 60–120×/s: send at most every 50 ms,
// always ending on the latest position
export function cursorSender(aw: Awareness, everyMs = 50) {
  let latest: Point | null = null;
  let timer: ReturnType<typeof setTimeout> | undefined;
  return (p: Point | null) => {
    latest = p;
    if (timer) return;
    timer = setTimeout(() => {
      timer = undefined;
      aw.setLocalStateField("cursor", latest);
    }, everyMs);
  };
}

import * as Y from "yjs";
 
// versions need deleted content: turn GC off
const doc = new Y.Doc({ gc: false });
 
export function saveVersion(): Uint8Array {
  return Y.encodeSnapshot(Y.snapshot(doc)); // small
}
 
// a read-only doc as it was at that version
export function openVersion(bytes: Uint8Array): Y.Doc {
  const snap = Y.decodeSnapshot(bytes);
  return Y.createDocFromSnapshot(doc, snap);
}
 
export { doc };

"use client";
import { useEffect, useState } from "react";
import { HocuspocusProvider } from "@hocuspocus/provider";
import * as Y from "yjs";
 
type Collab = { doc: Y.Doc; provider: HocuspocusProvider };
 
// one doc + provider per document name, cleaned up on
// unmount (and on StrictMode's dev double-mount)
export function useCollab(name: string, token: string) {
  const [collab, setCollab] = useState<Collab | null>(null);
  useEffect(() => {
    const doc = new Y.Doc();
    const provider = new HocuspocusProvider({
      url: process.env.NEXT_PUBLIC_COLLAB_URL ?? "",
      name,
      token,
      document: doc,
    });
    setCollab({ doc, provider });
    return () => {
      provider.destroy();
      doc.destroy();
    };
  }, [name, token]);
  return collab; // null until mounted: render a skeleton
}