// 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, insertComment, insertDump, insertPlaylist, PASSWORD, PNG, post, put, resetDb, type TestUser, } from "./helpers.ts"; import { assert, assertEquals, assertExists, assertFalse } from "@std/assert"; // deno-lint-ignore no-explicit-any type Msg = Record; interface Client { ws: WebSocket; messages: Msg[]; /** Resolves with the next unread message of `type` (and matching `where`). */ next(type: string, where?: (m: Msg) => boolean): Promise; /** Every message of `type` received so far. */ all(type: string): Msg[]; send(data: unknown): void; /** Round-trips a ping, so anything the server sent before it has arrived. */ sync(): Promise; closed: Promise; } async function withServer( fn: ( connect: (user?: TestUser, origin?: string) => Promise, ) => Promise, ): Promise { 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((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 WebSocket( `ws://127.0.0.1:${port}/ws${query}`, { headers: { origin } } as unknown as string[], ); const messages: Msg[] = []; const read = new Set(); const waiters: (() => void)[] = []; ws.onmessage = (e) => { messages.push(JSON.parse(e.data)); waiters.splice(0).forEach((w) => w()); }; const closed = new Promise((resolve) => { ws.addEventListener("close", (e) => resolve(e)); }); const client: Client = { ws, messages, closed, async next(type, where = () => true) { const deadline = Date.now() + 3000; while (true) { const hit = messages.find((m) => !read.has(m) && m.type === type && where(m) ); 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((r) => waiters.push(r)), new Promise((r) => setTimeout(r, 50)), ]); } }, all: (type) => messages.filter((m) => 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((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[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: Msg) => 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", success: true, 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("/api/dumps", { auth: owner.auth, form: formWithFile(PNG), }); assertEquals((await x.next("dump_created")).dump.id, pub.body.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.body.data.id}`, { auth: owner.auth, json: { isPrivate: true }, }); assertEquals((await x.next("dump_deleted")).dumpId, pub.body.data.id); await put(`/api/dumps/${pub.body.data.id}`, { auth: owner.auth, json: { isPrivate: false }, }); await x.next("dump_created", (m) => m.dump.id === pub.body.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.body.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("/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.body.data.token); assertFalse(res.body.data.token === user.token); });