colabbd/adapters/vim/bridge.test.ts

import { afterEach, beforeEach, describe, expect, test } from "bun:test";
import { spawn } from "bun";

describe("Bridge lifecycle", () => {
  let daemon: ReturnType<typeof Bun.serve>;
  const DAEMON_PORT = 4042;

  beforeEach(async () => {
    const { decode } = await import("../../src/protocol");
    const { getOrCreateSession, getSession, removeSession } = await import(
      "../../src/session"
    );

    // start daemon for bridge to connect to
    daemon = Bun.serve({
      port: DAEMON_PORT,
      fetch(req, server) {
        const url = new URL(req.url);
        if (url.pathname === "/ws") {
          const upgraded = server.upgrade(req, {
            data: { room: null, client: null },
          });
          if (!upgraded) {
            return new Response("websocket upgrade failed", { status: 400 });
          }
          return;
        }
        return new Response("test daemon");
      },
      websocket: {
        open(ws) {
          const client = { ws };
          ws.data.client = client;
        },
        message(ws, raw) {
          const msg = decode(raw.toString());
          if (!msg) return;

          const client = ws.data.client;
          if (!client) return;

          switch (msg.type) {
            case "join": {
              const session = getOrCreateSession(msg.room);
              ws.data.room = msg.room;
              session.join(client);
              break;
            }
            case "leave": {
              if (ws.data.room) {
                const session = getSession(ws.data.room);
                session?.leave(client);
                removeSession(ws.data.room);
                ws.data.room = null;
              }
              break;
            }
            case "update": {
              if (ws.data.room) {
                const session = getSession(ws.data.room);
                session?.applyUpdate(new Uint8Array(msg.data), client);
              }
              break;
            }
          }
        },
        close(ws) {
          if (ws.data.room && ws.data.client) {
            const session = getSession(ws.data.room);
            session?.leave(ws.data.client);
            removeSession(ws.data.room);
          }
        },
      },
    });
  });

  afterEach(() => {
    daemon.stop();
  });

  test("bridge starts and signals ready", async () => {
    const bridge = spawn({
      cmd: ["bun", "adapters/vim/bridge.ts"],
      env: {
        ...process.env,
        COLLABD_URL: `ws://localhost:${DAEMON_PORT}/ws`,
      },
      stdin: "pipe",
      stdout: "pipe",
      stderr: "pipe",
    });

    // read first line from stdout
    const reader = bridge.stdout.getReader();
    const decoder = new TextDecoder();

    const { value } = await reader.read();
    expect(value).toBeDefined();

    const output = decoder.decode(value);
    const firstLine = output.trim().split("\n")[0];
    const msg = JSON.parse(firstLine);

    expect(msg.type).toBe("ready");

    bridge.kill();
    await bridge.exited;
  });

  test("bridge connects to daemon and disconnects cleanly", async () => {
    const bridge = spawn({
      cmd: ["bun", "adapters/vim/bridge.ts"],
      env: {
        ...process.env,
        COLLABD_URL: `ws://localhost:${DAEMON_PORT}/ws`,
      },
      stdin: "pipe",
      stdout: "pipe",
      stderr: "pipe",
    });

    const reader = bridge.stdout.getReader();
    const decoder = new TextDecoder();

    // wait for ready
    let { value } = await reader.read();
    expect(value).toBeDefined();
    let output = decoder.decode(value);
    expect(output).toContain('"type":"ready"');

    // send connect message
    bridge.stdin.write(
      `${JSON.stringify({ type: "connect", room: "test" })}\n`,
    );

    // wait for connected message
    ({ value } = await reader.read());
    expect(value).toBeDefined();
    output = decoder.decode(value);
    expect(output).toContain('"type":"connected"');
    expect(output).toContain('"room":"test"');

    // send disconnect message
    bridge.stdin.write(`${JSON.stringify({ type: "disconnect" })}\n`);
    bridge.stdin.end();

    // bridge should exit cleanly
    const exitCode = await bridge.exited;
    expect(exitCode).toBe(0);
  });

  test("bridge handles content synchronization", async () => {
    const bridge = spawn({
      cmd: ["bun", "adapters/vim/bridge.ts"],
      env: {
        ...process.env,
        COLLABD_URL: `ws://localhost:${DAEMON_PORT}/ws`,
      },
      stdin: "pipe",
      stdout: "pipe",
      stderr: "pipe",
    });

    const reader = bridge.stdout.getReader();

    // wait for ready
    await reader.read();

    // connect to room
    bridge.stdin.write(
      `${JSON.stringify({ type: "connect", room: "content-test" })}\n`,
    );

    // wait for connected and peers messages
    await reader.read();
    await reader.read();

    // send content
    bridge.stdin.write(
      `${JSON.stringify({ type: "content", text: "hello world" })}\n`,
    );

    // give it time to process
    await new Promise((r) => setTimeout(r, 100));

    // disconnect
    bridge.stdin.write(`${JSON.stringify({ type: "disconnect" })}\n`);
    bridge.stdin.end();

    const exitCode = await bridge.exited;
    expect(exitCode).toBe(0);
  });
});

describe("awareness", () => {
  test("forwards awareness from daemon to vim", async () => {
    // Start daemon
    const server = Bun.serve({
      port: 4043,
      fetch(req, server) {
        if (new URL(req.url).pathname === "/ws") {
          server.upgrade(req);
          return;
        }
        return new Response("not found", { status: 404 });
      },
      websocket: {
        open(ws) {
          // Simulate sending awareness after join
          setTimeout(() => {
            ws.send(JSON.stringify({
              type: "awareness",
              data: { clientId: 99, cursor: { line: 3, col: 7 }, name: "peer" }
            }));
          }, 100);
        },
        message() {},
        close() {},
      },
    });

    const output: string[] = [];
    const bridge = spawn({
      cmd: ["bun", "adapters/vim/bridge.ts"],
      env: { ...process.env, COLLABD_URL: "ws://localhost:4043/ws" },
      stdin: "pipe",
      stdout: "pipe",
    });

    const reader = bridge.stdout.getReader();
    const decoder = new TextDecoder();

    // Collect output
    (async () => {
      while (true) {
        const { done, value } = await reader.read();
        if (done) break;
        output.push(decoder.decode(value));
      }
    })();

    await new Promise(r => setTimeout(r, 50));
    bridge.stdin.write(JSON.stringify({ type: "connect", room: "test" }) + "\n");
    await new Promise(r => setTimeout(r, 200));

    const awarenessMsg = output.join("").split("\n")
      .filter(Boolean)
      .map(l => JSON.parse(l))
      .find(m => m.type === "cursor");

    expect(awarenessMsg).toBeDefined();
    expect(awarenessMsg.data.line).toBe(3);
    expect(awarenessMsg.data.col).toBe(7);
    expect(awarenessMsg.data.name).toBe("peer");

    bridge.kill();
    server.stop();
  });
});