gerbeur/api/test/ws.test.ts

// The live side over a real socket: presence, votes, likes, chat, and the
// broadcasts REST writes trigger. Each test runs its own server.
import {
  app,
  count,
  createUser,
  db,
  formWithFile,
  get,
  insertComment,
  insertDump,
  insertPlaylist,
  PASSWORD,
  PNG,
  post,
  put,
  resetDb,
  type TestUser,
} from "./helpers.ts";
import { assert, assertEquals, assertExists, assertFalse } from "@std/assert";
import type {
  AuthTokens,
  Dump,
  ServerToClientMessage,
} from "../model/interfaces.ts";

// deno.json's "dom" lib types WebSocket's second argument as protocols only,
// hiding Deno's own overload, which takes handshake headers.
const DenoWebSocket = WebSocket as unknown as new (
  url: string,
  options: { headers: HeadersInit },
) => WebSocket;

type Msg = ServerToClientMessage;
type MsgOf<T extends Msg["type"]> = Extract<Msg, { type: T }>;

interface Client {
  ws: WebSocket;
  messages: Msg[];
  /** Resolves with the next unread message of `type` (and matching `where`). */
  next<T extends Msg["type"]>(
    type: T,
    where?: (m: MsgOf<T>) => boolean,
  ): Promise<MsgOf<T>>;
  /** Every message of `type` received so far. */
  all<T extends Msg["type"]>(type: T): MsgOf<T>[];
  send(data: unknown): void;
  /** Round-trips a ping, so anything the server sent before it has arrived. */
  sync(): Promise<void>;
  closed: Promise<CloseEvent>;
}

async function withServer(
  fn: (
    connect: (user?: TestUser, origin?: string) => Promise<Client>,
  ) => Promise<void>,
): Promise<void> {
  resetDb();
  // Oak's own server, not app.handle(): an upgraded request has no response
  // for handle() to return. Its listen() promise never settles after the
  // abort, hence the sanitizers being off for these tests.
  const controller = new AbortController();
  const port = await new Promise<number>((resolve) => {
    app.addEventListener("listen", (e) => resolve(e.port), { once: true });
    app.listen({ hostname: "127.0.0.1", port: 0, signal: controller.signal });
  });
  const clients: Client[] = [];

  const connect = (user?: TestUser, origin = "http://allowed.test") => {
    const query = user ? `?token=${user.token}` : "";
    const ws = new DenoWebSocket(
      `ws://127.0.0.1:${port}/ws${query}`,
      { headers: { origin } },
    );
    const messages: Msg[] = [];
    const read = new Set<Msg>();
    const waiters: (() => void)[] = [];
    ws.onmessage = (e) => {
      messages.push(JSON.parse(e.data));
      waiters.splice(0).forEach((w) => w());
    };
    const closed = new Promise<CloseEvent>((resolve) => {
      ws.addEventListener("close", (e) => resolve(e));
    });
    const client: Client = {
      ws,
      messages,
      closed,
      async next<T extends Msg["type"]>(
        type: T,
        where: (m: MsgOf<T>) => boolean = () => true,
      ) {
        const deadline = Date.now() + 3000;
        while (true) {
          const hit = messages.find((m): m is MsgOf<T> =>
            !read.has(m) && m.type === type && where(m as MsgOf<T>)
          );
          if (hit) {
            read.add(hit);
            return hit;
          }
          if (Date.now() > deadline) {
            throw new Error(
              `No ${type} message; got ${
                messages.map((m) => m.type).join(", ")
              }`,
            );
          }
          await Promise.race([
            new Promise<void>((r) => waiters.push(r)),
            new Promise((r) => setTimeout(r, 50)),
          ]);
        }
      },
      all: <T extends Msg["type"]>(type: T) =>
        messages.filter((m): m is MsgOf<T> => m.type === type),
      send: (data) =>
        ws.send(typeof data === "string" ? data : JSON.stringify(data)),
      async sync() {
        client.send({ type: "ping" });
        await client.next("pong");
      },
    };
    clients.push(client);
    return new Promise<Client>((resolve, reject) => {
      ws.onopen = async () => {
        await client.next("welcome");
        resolve(client);
      };
      ws.onerror = () => reject(new Error("socket error"));
    });
  };

  try {
    await fn(connect);
  } finally {
    for (const c of clients) {
      if (
        c.ws.readyState === WebSocket.OPEN ||
        c.ws.readyState === WebSocket.CONNECTING
      ) {
        c.ws.close();
      }
    }
    await Promise.all(clients.map((c) => c.closed));
    // Let the server see the closes and drop the clients from its registry.
    await new Promise((r) => setTimeout(r, 20));
    controller.abort();
  }
}

function wsTest(name: string, fn: Parameters<typeof withServer>[0]) {
  Deno.test({
    name,
    sanitizeOps: false,
    sanitizeResources: false,
    fn: () => withServer(fn),
  });
}

wsTest("welcome: anonymous and signed-in state", async (connect) => {
  const user = await createUser({ createdAt: "2024-01-01T00:00:00.000Z" });
  const { id: dumpId } = insertDump({ userId: user.id });
  db.prepare(
    `INSERT INTO votes (dump_id, user_id, created_at) VALUES (?, ?, ?);`,
  )
    .run(dumpId, user.id, new Date().toISOString());

  const anon = await connect();
  const [welcome] = anon.all("welcome");
  assertEquals(welcome.users, []);
  assertEquals(welcome.myVotes, []);
  assertEquals(welcome.unreadChatCount, 0);
  assertEquals(welcome.chatReadAt, undefined);

  const me = await connect(user);
  const [mine] = me.all("welcome");
  assertEquals(mine.myVotes, [dumpId]);
  assertEquals(mine.users.map((u) => u.username), [user.username]);
  assertEquals(mine.chatReadAt, "2024-01-01T00:00:00.000Z");
  // The anonymous client hears about the arrival.
  const presence = await anon.next(
    "presence_update",
    (m) => m.users.length === 1,
  );
  assertEquals(presence.users[0].userId, user.id);
});

wsTest(
  "presence: one entry per user however many tabs, gone on close",
  async (connect) => {
    const user = await createUser();
    const watcher = await connect();
    const tab1 = await connect(user);
    const tab2 = await connect(user);
    assertEquals(tab2.all("welcome")[0].users.length, 1);
    tab1.ws.close();
    await tab1.closed;
    tab2.ws.close();
    await tab2.closed;
    await watcher.next("presence_update", (m) => m.users.length === 0);
  },
);

wsTest(
  "a refused origin never connects; a revoked token connects anonymously",
  async (connect) => {
    let refused = false;
    await connect(undefined, "http://evil.test").catch(() => (refused = true));
    assert(refused);

    const user = await createUser();
    db.prepare(`UPDATE users SET tokens_valid_after = ? WHERE id = ?;`)
      .run(new Date(Date.now() + 60_000).toISOString(), user.id);
    const client = await connect(user);
    assertEquals(client.all("welcome")[0].users, []);
    client.send({ type: "vote_cast", dumpId: "x" });
    assertEquals(
      (await client.next("error")).message,
      "Authentication required",
    );
  },
);

wsTest(
  "votes: ack to the voter, count to everyone, errors carry context",
  async (connect) => {
    const owner = await createUser();
    const voter = await createUser();
    const { id: dumpId } = insertDump({ userId: owner.id });
    const watcher = await connect();
    const v = await connect(voter);

    v.send({ type: "vote_cast", dumpId });
    assertEquals(await v.next("vote_ack"), {
      type: "vote_ack",
      dumpId,
      action: "cast",
      voteCount: 1,
    });
    const update = await watcher.next("votes_update");
    assertEquals([update.voteCount, update.voterId, update.action], [
      1,
      voter.id,
      "cast",
    ]);

    v.send({ type: "vote_cast", dumpId });
    const dup = await v.next("error");
    assertEquals([dup.message, dup.dumpId, dup.action], [
      "Already voted",
      dumpId,
      "cast",
    ]);

    v.send({ type: "vote_remove", dumpId });
    assertEquals((await v.next("vote_ack")).voteCount, 0);
    v.send({ type: "vote_remove", dumpId });
    assertEquals((await v.next("error")).message, "Vote not found");

    // The owner was notified of the upvote, live.
    assertEquals(
      count(
        `SELECT COUNT(*) as n FROM notifications WHERE user_id = ?;`,
        owner.id,
      ),
      1,
    );

    const anon = await connect();
    anon.send({ type: "vote_cast", dumpId });
    assertEquals((await anon.next("error")).message, "Authentication required");
  },
);

wsTest(
  "votes on a private dump: others get a 404, the count stays with the owner",
  async (connect) => {
    const owner = await createUser();
    const other = await createUser();
    const { id: dumpId } = insertDump({ userId: owner.id, isPrivate: true });
    const o = await connect(owner);
    const x = await connect(other);
    x.send({ type: "vote_cast", dumpId });
    assertEquals((await x.next("error")).message, "Dump not found");
    o.send({ type: "vote_cast", dumpId });
    await o.next("vote_ack");
    await o.next("votes_update");
    await x.sync();
    assertEquals(x.all("votes_update"), []);
  },
);

wsTest("comment likes over the socket", async (connect) => {
  const author = await createUser();
  const liker = await createUser();
  const { id: dumpId } = insertDump({ userId: author.id });
  const commentId = insertComment({ dumpId, userId: author.id });
  const a = await connect(author);
  const l = await connect(liker);
  l.send({ type: "comment_like_cast", commentId });
  assertEquals((await l.next("comment_like_ack")).likeCount, 1);
  assertEquals((await a.next("comment_likes_update")).likerId, liker.id);
  // The like notification reaches the author live.
  const note = await a.next("notification_created");
  assertEquals(note.notification.type, "comment_liked");
  l.send({ type: "comment_like_remove", commentId });
  assertEquals((await l.next("comment_like_ack")).likeCount, 0);
  l.send({ type: "comment_like_cast", commentId: crypto.randomUUID() });
  assertEquals((await l.next("error")).message, "Comment not found");
});

wsTest(
  "chat: members hear messages, anonymous sockets don't",
  async (connect) => {
    const alice = await createUser();
    const bob = await createUser();
    const a = await connect(alice);
    const b = await connect(bob);
    const anon = await connect();

    a.send({ type: "chat_send", body: "  hello  ", nonce: "n1" });
    const mine = await a.next("chat_message");
    assertEquals(mine.nonce, "n1");
    assertEquals(mine.message.body, "hello");
    const theirs = await b.next("chat_message");
    assertEquals(theirs.message.id, mine.message.id);

    b.send({ type: "chat_send", body: "reply", replyToId: mine.message.id });
    const reply = await a.next(
      "chat_message",
      (m) => m.message.body === "reply",
    );
    assertEquals(reply.message.replyToBody, "hello");

    await anon.sync();
    assertEquals(anon.all("chat_message"), []);
    anon.send({ type: "chat_send", body: "hi", nonce: "anon" });
    const err = await anon.next("error");
    assertEquals([err.scope, err.nonce, err.message], [
      "chat",
      "anon",
      "Authentication required",
    ]);
  },
);

wsTest("chat: bad sends are reported with their nonce", async (connect) => {
  const user = await createUser();
  const c = await connect(user);
  c.send({ type: "chat_send", body: "   ", nonce: "blank" });
  assertEquals((await c.next("error")).nonce, "blank");
  c.send({ type: "chat_send", body: 42, nonce: "number" });
  assertEquals((await c.next("error")).message, "Message cannot be empty");
  c.send({ type: "chat_send", body: "x".repeat(2001), nonce: "long" });
  assertEquals(
    (await c.next("error", (m) => m.nonce === "long")).scope,
    "chat",
  );
  // A nonce over 64 characters isn't echoed.
  c.send({ type: "chat_send", body: " ", nonce: "n".repeat(65) });
  assertEquals((await c.next("error")).nonce, undefined);
});

wsTest(
  "chat: edits and deletes by the author or a moderator only",
  async (connect) => {
    const author = await createUser();
    const other = await createUser();
    const mod = await createUser({ role: "moderator" });
    const a = await connect(author);
    const o = await connect(other);
    const m = await connect(mod);

    a.send({ type: "chat_send", body: "original" });
    const { message } = await a.next("chat_message");

    o.send({ type: "chat_edit", id: message.id, body: "hacked" });
    assertEquals(
      (await o.next("error")).message,
      "Not authorized to edit this message",
    );
    o.send({ type: "chat_delete", id: message.id });
    assertEquals(
      (await o.next("error")).message,
      "Not authorized to delete this message",
    );

    a.send({ type: "chat_edit", id: message.id, body: "edited" });
    assertEquals((await o.next("chat_message_updated")).message.body, "edited");

    // A moderator demoted since connecting loses the power at once.
    db.prepare(`UPDATE users SET role = 'user' WHERE id = ?;`).run(mod.id);
    m.send({ type: "chat_delete", id: message.id });
    assertEquals((await m.next("error")).scope, "chat");
    db.prepare(`UPDATE users SET role = 'moderator' WHERE id = ?;`).run(mod.id);
    m.send({ type: "chat_delete", id: message.id });
    assertEquals((await a.next("chat_message_deleted")).id, message.id);

    a.send({ type: "chat_edit", id: 5, body: "x" });
    assertEquals((await a.next("error")).message, "Invalid message");
  },
);

wsTest("chat: a burst of writes is rate limited", async (connect) => {
  const user = await createUser();
  const c = await connect(user);
  for (let i = 0; i < 9; i++) {
    c.send({ type: "chat_send", body: `m${i}`, nonce: `${i}` });
  }
  const err = await c.next("error");
  assertEquals(err.nonce, "8");
  assertEquals(err.message, "You're sending messages too fast — wait a moment");
  await c.sync();
  assertEquals(c.all("chat_message").length, 8);
});

wsTest(
  "chat: mentions notify, unless the chat is already open",
  async (connect) => {
    const alice = await createUser();
    const bob = await createUser();
    const carol = await createUser();
    const a = await connect(alice);
    const b = await connect(bob);
    b.send({ type: "chat_focus", open: true });
    await b.sync();
    a.send({
      type: "chat_send",
      body: `hey @${bob.username} @${carol.username}`,
    });
    await a.next("chat_message");
    const notified = db.prepare(
      `SELECT user_id FROM notifications WHERE type = 'user_mentioned';`,
    ).all().map((r) => r.user_id);
    assertEquals(notified, [carol.id]);
  },
);

wsTest("chat: typing and read markers", async (connect) => {
  const alice = await createUser();
  const bob = await createUser();
  const a = await connect(alice);
  const b1 = await connect(bob);
  const b2 = await connect(bob);

  a.send({ type: "chat_typing", typing: true });
  const typing = await b1.next(
    "chat_typing_update",
    (m) => m.users.length === 1,
  );
  assertEquals(typing.users[0].userId, alice.id);
  // Sending clears it.
  a.send({ type: "chat_send", body: "done" });
  await b1.next("chat_typing_update", (m) => m.users.length === 0);
  const { message } = await b1.next("chat_message");

  b1.send({ type: "chat_read", at: message.createdAt });
  // Both of Bob's sockets drop the badge together.
  for (const b of [b1, b2]) {
    const update = await b.next("chat_read_update");
    assertEquals([update.readAt, update.unread], [message.createdAt, 0]);
  }
  await a.sync();
  assertEquals(a.all("chat_read_update"), []);
});

wsTest(
  "garbage frames are ignored and the socket stays up",
  async (connect) => {
    const user = await createUser();
    const c = await connect(user);
    for (
      const frame of [
        "not json",
        "null",
        "42",
        "[]",
        `{"no":"type"}`,
        `{"type":7}`,
        `{"type":"unknown"}`,
        `{"type":"vote_cast"}`,
        `{"type":"chat_read","at":"nonsense"}`,
      ]
    ) {
      c.send(frame);
    }
    c.ws.send(new Uint8Array([1, 2, 3]));
    await c.sync();
    assertEquals(c.ws.readyState, WebSocket.OPEN);
  },
);

wsTest("REST writes broadcast to the right audience", async (connect) => {
  const owner = await createUser();
  const other = await createUser();
  const o = await connect(owner);
  const x = await connect(other);

  // A public file dump arrives for everyone; a private one for no one.
  const pub = await post<Dump>("/api/dumps", {
    auth: owner.auth,
    form: formWithFile(PNG),
  });
  assertEquals((await x.next("dump_created")).dump.id, pub.data.id);
  await post("/api/dumps", {
    auth: owner.auth,
    form: formWithFile(PNG, "p.png", "image/png", { isPrivate: "true" }),
  });
  await x.sync();
  assertEquals(x.all("dump_created").length, 1);

  // Going private is a deletion for everyone else; going public an arrival.
  await put(`/api/dumps/${pub.data.id}`, {
    auth: owner.auth,
    json: { isPrivate: true },
  });
  assertEquals((await x.next("dump_deleted")).dumpId, pub.data.id);
  await put(`/api/dumps/${pub.data.id}`, {
    auth: owner.auth,
    json: { isPrivate: false },
  });
  await x.next("dump_created", (m) => m.dump.id === pub.data.id);

  // Comments on a private dump stay quiet.
  const priv = insertDump({ userId: owner.id, isPrivate: true });
  await post(`/api/dumps/${priv.id}/comments`, {
    auth: owner.auth,
    json: { body: "shh" },
  });
  await post(`/api/dumps/${pub.data.id}/comments`, {
    auth: owner.auth,
    json: { body: "hi" },
  });
  const created = await x.next("comment_created");
  assertEquals(created.comment.body, "hi");
  await x.sync();
  assertEquals(x.all("comment_created").length, 1);

  // A private playlist's events go to its owner alone.
  const pl = insertPlaylist({ userId: owner.id, isPublic: false });
  await post(`/api/playlists/${pl.id}/dumps/${priv.id}`, { auth: owner.auth });
  await o.next("playlist_dumps_updated");
  await x.sync();
  assertEquals(x.all("playlist_dumps_updated"), []);
});

wsTest("a password change closes the user's sockets", async (connect) => {
  const user = await createUser();
  const c = await connect(user);
  await new Promise((r) => setTimeout(r, 1100)); // see auth.test.ts
  const res = await post<AuthTokens>("/api/users/me/change-password", {
    auth: user.auth,
    json: { currentPassword: PASSWORD, newPassword: "a-new-password" },
  });
  assertEquals(res.status, 200);
  const closed = await c.closed;
  assertEquals(closed.reason, "Password changed");
  // Reconnecting with the old token is anonymous now.
  const again = await connect(user);
  assertEquals(again.all("welcome")[0].users, []);
  assertExists(res.data.token);
  assertFalse(res.data.token === user.token);
});

wsTest(
  "playlist_dumps_updated lists only what each viewer can see",
  async (connect) => {
    const owner = await createUser();
    const author = await createUser();
    const other = await createUser();
    const pl = insertPlaylist({ userId: owner.id });
    const pub = insertDump({ userId: owner.id });
    const mine = insertDump({ userId: owner.id, isPrivate: true });
    const theirs = insertDump({ userId: author.id, isPrivate: true });
    // Rows straight in: the dumps are private now, as if they went private after
    // being added.
    for (const [i, d] of [pub, mine, theirs].entries()) {
      db.prepare(
        `INSERT INTO playlist_dumps (playlist_id, dump_id, position, added_at)
       VALUES (?, ?, ?, ?);`,
      ).run(pl.id, d.id, i, new Date().toISOString());
    }
    const o = await connect(owner);
    const a = await connect(author);
    const x = await connect(other);
    const anon = await connect();

    const added = insertDump({ userId: owner.id });
    await post(`/api/playlists/${pl.id}/dumps/${added.id}`, {
      auth: owner.auth,
    });
    // The owner sees their own private dump but not the author's; the author,
    // not being the owner, sees neither private one; nor does anyone else.
    assertEquals((await o.next("playlist_dumps_updated")).dumpIds, [
      added.id,
      pub.id,
      mine.id,
    ]);
    for (const c of [a, x, anon]) {
      const msg = await c.next("playlist_dumps_updated");
      assertEquals(msg.dumpIds, [added.id, pub.id]);
      // The count is the playlist's, as its dumpCount gives it to everyone.
      assertEquals(msg.dumpCount, 4);
    }
  },
);

wsTest(
  "a new playlist cover is on disk by the time it's announced",
  async (connect) => {
    const owner = await createUser();
    const pl = insertPlaylist({ userId: owner.id });
    const c = await connect();
    const cover = new Uint8Array([...PNG, 9, 9, 9]);
    const upload = post(`/api/playlists/${pl.id}/image`, {
      auth: owner.auth,
      form: formWithFile(cover),
    });
    // Fetch the image the moment the socket says it changed, as a client would.
    const msg = await c.next("playlist_updated");
    const served = await get(
      `/api/playlists/${pl.id}/image?v=${msg.playlist.updatedAt}`,
    );
    assertEquals(served.status, 200);
    assertEquals(served.bytes, cover);
    assertEquals((await upload).status, 200);
  },
);

wsTest(
  "replacing a dump's file is broadcast like any edit",
  async (connect) => {
    const owner = await createUser();
    const other = await createUser();
    const { id } = insertDump({ userId: owner.id, fileName: "a.txt" });
    const x = await connect(other);
    const replace = (fields: Record<string, string>) =>
      put(`/api/dumps/${id}/file`, {
        auth: owner.auth,
        form: formWithFile(PNG, "b.png", "image/png", fields),
      });
    await replace({ title: "New title" });
    assertEquals((await x.next("dump_updated")).dump.title, "New title");
    await replace({ isPrivate: "true" });
    assertEquals((await x.next("dump_deleted")).dumpId, id);
    await replace({ isPrivate: "false" });
    assertEquals((await x.next("dump_created")).dump.id, id);
  },
);