From 08cf44991db9944b6982631db4f801225cd731ac Mon Sep 17 00:00:00 2001 From: Devin Zuczek Date: Tue, 21 Jul 2026 01:17:20 -0400 Subject: [PATCH] chat, little buggy --- apps/chat/migrations/0001_message.sql | 22 + apps/chat/migrations/0002_thread.sql | 37 + apps/chat/package.json | 2 + apps/chat/src/chat.app.ts | 445 ++++++- apps/chat/src/context.ts | 12 +- apps/chat/src/message-db.ts | 150 +++ apps/chat/src/test/integration/api.test.ts | 1244 +++++++++++++++++++- apps/chat/src/thread-db.ts | 462 ++++++++ apps/chat/vitest.config.ts | 34 + apps/chat/wrangler.jsonc | 35 + 10 files changed, 2435 insertions(+), 8 deletions(-) create mode 100644 apps/chat/migrations/0001_message.sql create mode 100644 apps/chat/migrations/0002_thread.sql create mode 100644 apps/chat/src/message-db.ts create mode 100644 apps/chat/src/thread-db.ts diff --git a/apps/chat/migrations/0001_message.sql b/apps/chat/migrations/0001_message.sql new file mode 100644 index 0000000..1893f5b --- /dev/null +++ b/apps/chat/migrations/0001_message.sql @@ -0,0 +1,22 @@ +-- Chat messages, owned by the `chat` worker. One row per message posted to a thread; +-- columns rather than a JSON blob (mirroring the reference model), since every field is a +-- scalar the server reads: threads are listed newest-first by (chat_thread_id, +-- chat_message_id), and `time_sent` is carried for display only. +-- +-- `chat_message_id` is server-assigned (AUTOINCREMENT) so ids are unique across every +-- thread, matching the client's expectation of a global message id. `contents` is the +-- client's envelope — `{"Type":0,"Version":1,"Data":"..."}` — stored verbatim as an +-- opaque string and served back untouched, so new message types need no schema change. +-- `moderation_state` is the ChatModerationState enum (0 = none). Kept in sync with +-- SCHEMA_DDL in src/message-db.ts. + +CREATE TABLE IF NOT EXISTS message ( + chat_message_id INTEGER PRIMARY KEY AUTOINCREMENT, + chat_thread_id INTEGER NOT NULL, + sender_player_id INTEGER NOT NULL, + time_sent TEXT NOT NULL, + contents TEXT NOT NULL, + moderation_state INTEGER NOT NULL DEFAULT 0 + ); +CREATE INDEX IF NOT EXISTS idx_message_thread ON message (chat_thread_id, chat_message_id); +CREATE INDEX IF NOT EXISTS idx_message_sender ON message (sender_player_id); diff --git a/apps/chat/migrations/0002_thread.sql b/apps/chat/migrations/0002_thread.sql new file mode 100644 index 0000000..f3e7337 --- /dev/null +++ b/apps/chat/migrations/0002_thread.sql @@ -0,0 +1,37 @@ +-- Chat threads and their membership, owned by the `chat` worker. +-- +-- A thread is a conversation — a DM pair, a group chat, or a system thread. Membership +-- in `thread_member` is the authorization gate: a player may read or post to a thread +-- only if they hold a row here, and the same rows render the `playerIds` array the +-- client shows. There is deliberately no FK to accounts, here or on +-- `message.sender_player_id`: that table belongs to the `auth` worker, and a thread +-- outlives the accounts in it. +-- +-- `latest_message_id` is denormalized onto the thread so the thread list renders from +-- one indexed row per thread instead of a per-thread MAX() over `message`; it also +-- orders that list (message ids being monotonic, newest thread = highest id). Kept in +-- sync on every insert — see touchThread in src/thread-db.ts. +-- +-- The per-viewer fields live on the membership row, not the thread: two players in one +-- DM have independent read positions, snoozes, and favorites. Kept in sync with +-- THREAD_SCHEMA_DDL in src/thread-db.ts. + +CREATE TABLE IF NOT EXISTS message_thread ( + chat_thread_id INTEGER PRIMARY KEY AUTOINCREMENT, + -- Null for DMs and unnamed groups; the client falls back to rendering the members. + chat_thread_name TEXT, + latest_message_id INTEGER, + created_at TEXT NOT NULL + ); +CREATE INDEX IF NOT EXISTS idx_message_thread_latest ON message_thread (latest_message_id); + +CREATE TABLE IF NOT EXISTS thread_member ( + chat_thread_id INTEGER NOT NULL, + player_id INTEGER NOT NULL, + last_read_message_id INTEGER, + snoozed_until TEXT, + is_favorited INTEGER NOT NULL DEFAULT 0, + PRIMARY KEY (chat_thread_id, player_id) + ); +-- The thread-list query is "every thread this player is in", so player_id leads. +CREATE INDEX IF NOT EXISTS idx_thread_member_player ON thread_member (player_id); diff --git a/apps/chat/package.json b/apps/chat/package.json index cf5089e..2ec0acb 100644 --- a/apps/chat/package.json +++ b/apps/chat/package.json @@ -12,10 +12,12 @@ "deploy": "run-wrangler-deploy", "dev": "run-wrangler-dev", "fix:workers-types": "run-wrangler-types", + "migrate": "run-wrangler-migrate", "test": "run-vitest" }, "dependencies": { "@repo/hono-helpers": "workspace:*", + "@repo/jwt": "workspace:*", "hono": "4.12.27", "workers-tagged-logger": "1.0.1" }, diff --git a/apps/chat/src/chat.app.ts b/apps/chat/src/chat.app.ts index af1ca37..b5fb63a 100644 --- a/apps/chat/src/chat.app.ts +++ b/apps/chat/src/chat.app.ts @@ -1,9 +1,217 @@ import { Hono } from 'hono' import { useWorkersLogger } from 'workers-tagged-logger' -import { withNotFound, withOnError } from '@repo/hono-helpers' +import { logger, withNotFound, withOnError } from '@repo/hono-helpers' +import { validateAndGetAccountId } from '@repo/jwt' +import { NotificationType } from '../../notify/src/notification-types' +import { getThreadMessages } from './message-db' +import { + addThreadMember, + getOrCreateThreadWithMembers, + getThreadForPlayer, + getThreadMemberIds, + getThreadsForPlayer, + isThreadMember, + leftChatContents, + markThreadRead, + postMessage, + removeThreadMember, + setThreadFavorited, + setThreadName, + setThreadSnoozed, + SYSTEM_SENDER_ID, +} from './thread-db' + +import type { Context } from 'hono' import type { App } from './context' +import type { ChatMessage } from './message-db' + +/** + * Resolve the account id from a Bearer token. Returns `null` when the header is + * missing, the token is invalid, or the `sub` claim isn't an integer. + */ +async function authedId(c: Context): Promise { + return validateAndGetAccountId(c.req.raw, await c.env.JWT_SECRET.get()) +} + +/** + * How many items a `MessageCount` query param asks for. The client sends 16; anything + * missing, unparseable, or out of range falls back to the default rather than 400ing, + * and the cap keeps a hand-written request from pulling a whole thread history. + */ +const DEFAULT_MESSAGE_COUNT = 16 +const MAX_MESSAGE_COUNT = 100 + +/** What the client asks for when opening a thread (`messageCount=50`). */ +const DEFAULT_THREAD_MESSAGE_COUNT = 50 + +function messageCount(c: Context, fallback = DEFAULT_MESSAGE_COUNT): number { + // The GET routes spell it `MessageCount` in the query; the POST forms spell it + // `messageCount` in the body. Accept either, wherever it turns up. + const raw = Number.parseInt(c.req.query('MessageCount') ?? c.req.query('messageCount') ?? '', 10) + if (Number.isNaN(raw) || raw <= 0) return fallback + return Math.min(raw, MAX_MESSAGE_COUNT) +} + +/** The page size a POST form asks for, which may also arrive in the body. */ +async function formMessageCount(c: Context, fallback: number): Promise { + const raw = Number.parseInt((await formField(c, 'messageCount')) ?? '', 10) + if (Number.isNaN(raw) || raw <= 0) return messageCount(c, fallback) + return Math.min(raw, MAX_MESSAGE_COUNT) +} + +/** + * What a chat action reports back to the client alongside its payload — the reference's + * ChatResult. Only success and "bad arguments" are reachable here. + */ +const CHAT_SUCCESS = 0 +const CHAT_INVALID_ARGUMENTS = 1 +const CHAT_MEMBERSHIP_NOT_FOUND = 3 +const CHAT_PLAYER_ALREADY_ON_THREAD = 4 + +/** The hub is a single global Durable Object instance, as every worker addresses it. */ +const HUB_INSTANCE = 'global' + +/** + * Push ChatMessageReceived to everyone in the thread once a message lands, so the + * conversation updates live instead of on the next poll. + * + * The sender is notified too, deliberately: the client doesn't fold the HTTP response + * into its local thread cache, so without a self-targeted push its own outgoing message + * doesn't appear until the thread is refetched. + * + * Best-effort — a hub failure is logged and swallowed, since the message has already + * committed and the client will still see it on the next fetch. + */ +async function pushChatMessage(c: Context, message: ChatMessage): Promise { + try { + const hub = c.env.RECFLARE_NOTIFICATIONS_HUB.getByName(HUB_INSTANCE) + const members = await getThreadMemberIds(c.env.DB, message.chatThreadId) + await Promise.all( + members.map((playerId) => + hub.notifyPlayer(playerId, NotificationType.ChatMessageReceived, { ...message }) + ) + ) + } catch (err) { + logger.error('failed to push ChatMessageReceived notification', { + chatThreadId: message.chatThreadId, + chatMessageId: message.chatMessageId, + error: err instanceof Error ? err.message : String(err), + }) + } +} + +/** + * Send a message to a thread that already exists — every message after the one that + * opened the conversation. `/thread/18` is what the client posts; `/thread/18/message` is + * the same call under the reference's other spelling, so both routes land here. + * + * Answers `{chatResult, chatThread}` — the whole thread with its messages, not just the + * message that was sent, so the client re-renders the conversation from one response. + * Blank or missing contents stores nothing and reports invalid-arguments, still with the + * thread attached, rather than an error status. + */ +async function sendToThread(c: Context) { + const id = await authedId(c) + if (id === null) return c.body(null, 401) + + const chatThreadId = Number.parseInt(c.req.param('id') ?? '', 10) + if (!(await isThreadMember(c.env.DB, chatThreadId, id))) return c.notFound() + + // Stored exactly as sent: the envelope carries its own Type/Version and may hold + // fields we know nothing about (the client sends Version 2 with a `<=>` prefix in + // Data, and a `Blocks` array alongside it), so nothing here parses or rewrites it. + const contents = (await formField(c, 'messageContents'))?.trim() + const posted = + contents === undefined || contents === '' + ? null + : await postMessage(c.env.DB, { chatThreadId, senderPlayerId: id, contents }) + if (posted !== null) await pushChatMessage(c, posted) + + const thread = await threadWithMessages(c, chatThreadId, id, DEFAULT_THREAD_MESSAGE_COUNT) + return c.json({ + chatResult: posted === null ? CHAT_INVALID_ARGUMENTS : CHAT_SUCCESS, + chatThread: thread, + }) +} + +/** + * Move the caller's read pointer on a thread, to `chatMessageId` or (undefined) to the + * thread's latest message. Answers the bare ChatResult integer the reference sends. + */ +async function markRead(c: Context, chatMessageId?: number) { + const id = await authedId(c) + if (id === null) return c.body(null, 401) + + // Every route reaching here constrains `:id` to digits, so the parse can't fail. + const chatThreadId = Number.parseInt(c.req.param('id') ?? '', 10) + if (!(await isThreadMember(c.env.DB, chatThreadId, id))) return c.notFound() + + await markThreadRead(c.env.DB, chatThreadId, id, chatMessageId) + return c.json(CHAT_SUCCESS) +} + +/** + * A thread rendered for opening a conversation: the thread's own fields plus a page of + * its messages, newest first. `latestMessage` gives way to the full page — the client is + * sent one or the other, never both — and `messages` is always present, empty for a + * thread with nothing in it yet. + * + * Null when the caller isn't a member (or the thread doesn't exist); membership is the + * gate, so the two cases are indistinguishable from outside. + */ +async function threadWithMessages( + c: Context, + chatThreadId: number, + playerId: number, + limit: number +) { + const thread = await getThreadForPlayer(c.env.DB, chatThreadId, playerId) + if (thread === null) return null + + const messages = await getThreadMessages(c.env.DB, chatThreadId, { limit }) + const { latestMessage: _latest, ...rest } = thread + return { ...rest, messages } +} + +/** Ceiling on a new thread's roster, counting the caller. */ +const MAX_THREAD_MEMBERS = 50 + +/** Longest a thread name may be; anything beyond is truncated, not rejected. */ +const MAX_THREAD_NAME_LENGTH = 128 + +/** + * What `snooze=True` stores in `snoozedUntil`. The client sends a boolean but reads back + * an instant, so "snoozed" is expressed as a time far enough out to mean indefinitely. + */ +const SNOOZED_INDEFINITELY = '9999-12-31T23:59:59Z' + +/** + * The repeated `ids` fields naming a new thread's members (`ids=2&ids=155`). The client + * sends them as a urlencoded body, but they're read from the query string too, since + * the same call is easy to hand-write that way. Values that aren't integers are dropped. + */ +async function memberIds(c: Context): Promise { + const raw = [...(c.req.queries('ids') ?? [])] + const form = await c.req.formData().catch(() => null) + if (form !== null) raw.push(...form.getAll('ids').map(String)) + return raw.map((value) => Number.parseInt(value, 10)).filter((id) => Number.isInteger(id)) +} + +/** A form boolean as the client spells it (`True`/`False`), tolerant of the variants. */ +async function formBool(c: Context, name: string): Promise { + const value = (await formField(c, name))?.trim().toLowerCase() + return value === 'true' || value === '1' || value === 'yes' +} + +/** A single form field, or the query param of the same name. Hono caches the body, so + * this is safe to call alongside `memberIds`. */ +async function formField(c: Context, name: string): Promise { + const form = await c.req.formData().catch(() => null) + const value = form?.get(name) + return typeof value === 'string' ? value : c.req.query(name) +} const app = new Hono() .use( @@ -21,7 +229,238 @@ const app = new Hono() .get('/', (c) => c.json({ service: 'chat', status: 'ok' })) - // Chat threads. No DB binding yet — returns `[]`. - .get('/thread', (c) => c.json([])) + // The player's own thread list, newest conversation first — each thread carrying its + // latest message and the caller's own read/snooze/favorite state. `MessageCount` is + // the page size (of threads, despite the name). Membership scopes the query, so a + // player only ever sees their own threads. + .get('/thread', async (c) => { + const id = await authedId(c) + if (id === null) return c.body(null, 401) + return c.json(await getThreadsForPlayer(c.env.DB, id, { limit: messageCount(c) })) + }) + + // Send to a set of players (`ids=155&ids=2&messageContents=…`) — the client's + // create-thread-and-post-first-message call, in one. Resolves to the thread those + // players already share rather than opening a second one. + // + // `messageContents` is the same envelope a message carries + // (`{"Type":0,"Version":1,"Data":"…"}`) and is stored verbatim, unparsed. The client + // also sends it blank, right after /thread/withmembers: that opens the thread without + // posting an empty message, and reports invalid-arguments the way the reference does. + .post('/thread', async (c) => { + const id = await authedId(c) + if (id === null) return c.body(null, 401) + + const members = [...new Set([id, ...(await memberIds(c))])] + if (members.length < 2 || members.length > MAX_THREAD_MEMBERS) return c.body(null, 400) + + const chatThreadId = await getOrCreateThreadWithMembers(c.env.DB, members, id) + + const contents = (await formField(c, 'messageContents'))?.trim() + const posted = + contents === undefined || contents === '' + ? null + : await postMessage(c.env.DB, { chatThreadId, senderPlayerId: id, contents }) + if (posted !== null) await pushChatMessage(c, posted) + + const thread = await getThreadForPlayer(c.env.DB, chatThreadId, id) + if (thread === null) throw new Error(`thread ${chatThreadId} vanished after creation`) + // The reference answers a wrapper here, not a bare thread. + return c.json({ + chatThread: thread, + chatResult: posted === null ? CHAT_INVALID_ARGUMENTS : CHAT_SUCCESS, + }) + }) + + // "Open the chat with these people" — the client's GetChatBetweenPlayers. Fetch or + // create: the thread whose membership is exactly `ids` plus the caller, opened only + // if they don't already share one. Returning a fresh empty thread each call would + // bury the real conversation and hand the client a thread with no messages. + // + // Answers the thread with a `messages` array (what `messageCount` sizes) rather than + // the list's single `latestMessage`, so the client can open straight into the + // conversation. The array is always present, empty for a brand-new thread. + .post('/thread/withmembers', async (c) => { + const id = await authedId(c) + if (id === null) return c.body(null, 401) + + const members = [...new Set([id, ...(await memberIds(c))])] + // A thread needs someone else in it; naming only yourself is a bad request + // rather than a lonely thread. + if (members.length < 2 || members.length > MAX_THREAD_MEMBERS) return c.body(null, 400) + + const chatThreadId = await getOrCreateThreadWithMembers(c.env.DB, members, id) + const limit = await formMessageCount(c, DEFAULT_THREAD_MESSAGE_COUNT) + const thread = await threadWithMessages(c, chatThreadId, id, limit) + if (thread === null) throw new Error(`thread ${chatThreadId} vanished after creation`) + return c.json(thread) + }) + + // A page of one thread's messages, newest first — a bare array, not a thread object. + // The client reads a conversation through either spelling: `/thread/2?messageCount=50` + // and `/thread/2/message?MessageCount=16` answer the same thing, so they share a + // handler; only the default page size differs, matching what each caller sends. + // + // 404 rather than 403 for a thread the caller isn't in: whether a thread exists is + // itself private, so a non-member gets the same answer as for a thread that's gone. + // An empty thread is still a 200 with `[]` — a conversation just opened with someone + // has no messages yet and still has to open. + // One thread with its recent messages — what the client opens a conversation with + // (`/thread/13?messageCount=50`). An OBJECT, the same shape /thread/withmembers + // answers: the client parses this one as a thread and rejects a bare array + // ("expected '{', actual '['"). Only /thread/:id/message below serves an array. + // + // 404s only for a thread the caller isn't in, not for one that's simply empty: a + // thread just opened with someone has no messages yet and still has to open. + .get('/thread/:id{[0-9]+}', async (c) => { + const id = await authedId(c) + if (id === null) return c.body(null, 401) + + const chatThreadId = Number.parseInt(c.req.param('id'), 10) + const limit = messageCount(c, DEFAULT_THREAD_MESSAGE_COUNT) + const thread = await threadWithMessages(c, chatThreadId, id, limit) + return thread === null ? c.notFound() : c.json(thread) + }) + + // Send a message to a thread that already exists — every message after the one that + // opened the conversation. `/thread/18` is what the client posts; `/thread/18/message` + // is the same call under the reference's other spelling. + // + // Answers the SendMessageResponse wrapper (`{chatMessage, chatResult}`), not a bare + // message. Blank or missing contents is invalid-arguments with no message attached, + // rather than an error status. + .post('/thread/:id{[0-9]+}', (c) => sendToThread(c)) + .post('/thread/:id{[0-9]+}/message', (c) => sendToThread(c)) + + // Rename a thread (`name=my chat`). Any member may rename — there's no owner — and an + // empty name clears it back to unnamed, which renders as the member list. Answers a + // bare ChatResult: 3 when the caller isn't on the thread, 0 on success. + .on(['POST', 'PUT'], '/thread/:id{[0-9]+}/rename', async (c) => { + const id = await authedId(c) + if (id === null) return c.body(null, 401) + + const chatThreadId = Number.parseInt(c.req.param('id'), 10) + if (!(await isThreadMember(c.env.DB, chatThreadId, id))) { + return c.json(CHAT_MEMBERSHIP_NOT_FOUND) + } + + const name = ((await formField(c, 'name')) ?? '').trim().slice(0, MAX_THREAD_NAME_LENGTH) + await setThreadName(c.env.DB, chatThreadId, name) + return c.json(CHAT_SUCCESS) + }) + + // Leave a thread. The thread and its history survive — only the caller's membership + // goes, so they stop seeing it and the remaining members keep the conversation. + // + // A "Player <@U…> left" notice is posted first, so the others see why the roster + // changed; the leaver is still a member at that moment and gets the push too, which + // is what tells their client the thread is gone. + .on(['POST', 'DELETE'], '/thread/:id{[0-9]+}/leave', async (c) => { + const id = await authedId(c) + if (id === null) return c.body(null, 401) + + const chatThreadId = Number.parseInt(c.req.param('id'), 10) + if (!(await isThreadMember(c.env.DB, chatThreadId, id))) { + return c.json(CHAT_MEMBERSHIP_NOT_FOUND) + } + + const notice = await postMessage(c.env.DB, { + chatThreadId, + senderPlayerId: SYSTEM_SENDER_ID, + contents: leftChatContents(id), + }) + await pushChatMessage(c, notice) + + await removeThreadMember(c.env.DB, chatThreadId, id) + return c.json(CHAT_SUCCESS) + }) + + // Snooze or unsnooze a thread (`snooze=True`), for the caller alone — snoozing is a + // per-member setting, so it never affects what anyone else sees. + // + // The client sends a boolean while the field it reads back is `snoozedUntil`, a time. + // `True` is therefore stored as a far-future instant meaning "muted indefinitely", and + // `False` clears it. If the real server instead snoozes for a fixed window, this is + // the one line to change. + .on(['POST', 'PUT'], '/thread/:id{[0-9]+}/snooze', async (c) => { + const id = await authedId(c) + if (id === null) return c.body(null, 401) + + const chatThreadId = Number.parseInt(c.req.param('id'), 10) + if (!(await isThreadMember(c.env.DB, chatThreadId, id))) { + return c.json(CHAT_MEMBERSHIP_NOT_FOUND) + } + + const on = await formBool(c, 'snooze') + await setThreadSnoozed(c.env.DB, chatThreadId, id, on ? SNOOZED_INDEFINITELY : null) + return c.json(CHAT_SUCCESS) + }) + + // Favorite or unfavorite a thread (`favorite=True`), for the caller alone — like + // snoozing, it's a per-member flag that pins the thread in their own inbox. + .on(['PUT', 'POST'], '/thread/:id{[0-9]+}/favorite', async (c) => { + const id = await authedId(c) + if (id === null) return c.body(null, 401) + + const chatThreadId = Number.parseInt(c.req.param('id'), 10) + if (!(await isThreadMember(c.env.DB, chatThreadId, id))) { + return c.json(CHAT_MEMBERSHIP_NOT_FOUND) + } + + await setThreadFavorited(c.env.DB, chatThreadId, id, await formBool(c, 'favorite')) + return c.json(CHAT_SUCCESS) + }) + + // Add a player to a thread (`/thread/20/member/2`). Gated on the caller already being + // in it — you can only pull someone into a conversation you're part of. + // + // Answers a bare ChatResult rather than an HTTP status, as the reference does: 3 when + // the caller isn't a member (which doubles as "no such thread", keeping a thread's + // existence private), 4 when the target is already on it, 0 on success. Idempotent — + // re-adding an existing member changes nothing. + .post('/thread/:id{[0-9]+}/member/:playerId{[0-9]+}', async (c) => { + const id = await authedId(c) + if (id === null) return c.body(null, 401) + + const chatThreadId = Number.parseInt(c.req.param('id'), 10) + if (!(await isThreadMember(c.env.DB, chatThreadId, id))) { + return c.json(CHAT_MEMBERSHIP_NOT_FOUND) + } + + const playerId = Number.parseInt(c.req.param('playerId'), 10) + if (await isThreadMember(c.env.DB, chatThreadId, playerId)) { + return c.json(CHAT_PLAYER_ALREADY_ON_THREAD) + } + + await addThreadMember(c.env.DB, chatThreadId, playerId) + return c.json(CHAT_SUCCESS) + }) + + // Move the caller's read pointer — `/thread/15/read` for the whole thread, or + // `/thread/15/message/:messageId/read` for a specific message, which the client uses + // when the view sits on a message rather than the bottom. Both verbs, as the client + // sends either. Answers the bare ChatResult integer the reference does. + // + // The pointer only moves forward, and never past the thread's real latest message: an + // id the client made up (or one it read from a synthetic message) can't strand the + // thread as permanently read. + .on(['PUT', 'POST'], '/thread/:id{[0-9]+}/read', (c) => markRead(c)) + .on(['PUT', 'POST'], '/thread/:id{[0-9]+}/message/:messageId{[0-9]+}/read', (c) => + markRead(c, Number.parseInt(c.req.param('messageId'), 10)) + ) + + // A page of one thread's messages, newest first — a bare array, unlike /thread/:id. + // `MessageCount` is the page size. 404 rather than 403 for a thread the caller isn't + // in: whether a thread exists is itself private, so a non-member gets the same answer + // as for a thread that's gone. + .get('/thread/:id{[0-9]+}/message', async (c) => { + const id = await authedId(c) + if (id === null) return c.body(null, 401) + + const chatThreadId = Number.parseInt(c.req.param('id'), 10) + if (!(await isThreadMember(c.env.DB, chatThreadId, id))) return c.notFound() + + return c.json(await getThreadMessages(c.env.DB, chatThreadId, { limit: messageCount(c) })) + }) export default app diff --git a/apps/chat/src/context.ts b/apps/chat/src/context.ts index 329ac40..0f0da3b 100644 --- a/apps/chat/src/context.ts +++ b/apps/chat/src/context.ts @@ -1,8 +1,18 @@ import type { HonoApp } from '@repo/hono-helpers' import type { SharedHonoEnv, SharedHonoVariables } from '@repo/hono-helpers/src/types' +// Type-only import (erased at build) of the DO class owned by the `notify` worker, +// so this worker can push websocket notifications through its RPC surface. +import type { NotificationsHub } from '../../notify/src/notifications-hub' export type Env = SharedHonoEnv & { - // add additional Bindings here + // Shared Secrets Store binding for the HS256 JWT signing key. Resolve the value + // with `await env.JWT_SECRET.get()`; all workers bind the same store so tokens + // signed by `auth` verify here. + JWT_SECRET: SecretsStoreSecret + /** Shared `recflare` D1 — this worker owns the `message` and thread tables. */ + DB: D1Database + /** The `notify` worker's NotificationsHub DO — pushes ChatMessageReceived to members. */ + RECFLARE_NOTIFICATIONS_HUB: DurableObjectNamespace } /** Variables can be extended */ diff --git a/apps/chat/src/message-db.ts b/apps/chat/src/message-db.ts new file mode 100644 index 0000000..0682193 --- /dev/null +++ b/apps/chat/src/message-db.ts @@ -0,0 +1,150 @@ +/** + * Chat messages on the shared `recflare` D1 database — the individual messages posted + * to a chat thread (a DM pair or a group). Stored as columns rather than a JSON blob + * (mirroring the reference model): every field is a scalar the server itself reads, and the + * one client-shaped part — `contents` — is already an opaque string. + * + * `contents` is the client's envelope, e.g. + * `{"Type":0,"Version":1,"Data":"This is jordanparki7 from your Oculus friends."}`, + * where `Type` selects how the client renders `Data` (plain text, an invite, an image + * …) and `Version` versions that encoding. It is stored verbatim and served back + * untouched, so new message types need no schema change here. + * + * `chatMessageId` is server-assigned and unique across all threads (AUTOINCREMENT), the + * way the client expects to be able to reference a message by id alone. + * + * The `chat` worker owns this schema/migration (migrations/0001_message.sql, applied + * under its own `migrations_table` so it doesn't clash with the other workers' + * migrations that share the database). `SCHEMA_DDL` mirrors that migration so tests can + * build the table directly. + */ + +/** Schema DDL (mirror of migrations/0001_message.sql). */ +export const SCHEMA_DDL: string[] = [ + `CREATE TABLE IF NOT EXISTS message ( + chat_message_id INTEGER PRIMARY KEY AUTOINCREMENT, + chat_thread_id INTEGER NOT NULL, + sender_player_id INTEGER NOT NULL, + time_sent TEXT NOT NULL, + contents TEXT NOT NULL, + moderation_state INTEGER NOT NULL DEFAULT 0 + )`, + // Thread listing is always (thread, id) — newest-first pages walk this index. + `CREATE INDEX IF NOT EXISTS idx_message_thread ON message (chat_thread_id, chat_message_id)`, + `CREATE INDEX IF NOT EXISTS idx_message_sender ON message (sender_player_id)`, +] + +/** + * Whether a message has been touched by moderation. `None` is the overwhelmingly common + * case and the column default; the others let a message be withheld from the thread + * without deleting the row. + */ +export enum ChatModerationState { + None = 0, + Flagged = 1, + Hidden = 2, +} + +/** A chat message, in the shape the client sends and receives it. */ +export interface ChatMessage { + chatMessageId: number + chatThreadId: number + senderPlayerId: number + /** ISO-8601 UTC instant, as .NET serializes `DateTime` (e.g. `2022-05-22T12:47:03.6536656`). */ + timeSent: string + /** The raw message envelope, e.g. `{"Type":0,"Version":1,"Data":"hello"}`. */ + contents: string + moderationState: ChatModerationState +} + +/** A new message, before the server assigns its id and (by default) its timestamp. */ +export interface NewChatMessage { + chatThreadId: number + senderPlayerId: number + contents: string + /** Defaults to now. Pass only when replaying a message with its original timestamp. */ + timeSent?: string + moderationState?: ChatModerationState +} + +/** The stored row, before it's mapped back to the client's camelCase shape. */ +interface MessageRow { + chat_message_id: number + chat_thread_id: number + sender_player_id: number + time_sent: string + contents: string + moderation_state: number +} + +function toMessage(row: MessageRow): ChatMessage { + return { + chatMessageId: row.chat_message_id, + chatThreadId: row.chat_thread_id, + senderPlayerId: row.sender_player_id, + timeSent: row.time_sent, + contents: row.contents, + moderationState: row.moderation_state, + } +} + +/** Post a message to a thread, returning it with its server-assigned id. */ +export async function insertMessage(db: D1Database, message: NewChatMessage): Promise { + const row = await db + .prepare( + `INSERT INTO message (chat_thread_id, sender_player_id, time_sent, contents, moderation_state) + VALUES (?1, ?2, ?3, ?4, ?5) + RETURNING *` + ) + .bind( + message.chatThreadId, + message.senderPlayerId, + message.timeSent ?? new Date().toISOString(), + message.contents, + message.moderationState ?? ChatModerationState.None + ) + .first() + // RETURNING on an INSERT that ran always yields the row; a null here means the + // insert itself failed, which D1 would already have thrown for. + if (row === null) throw new Error('failed to insert chat message') + return toMessage(row) +} + +/** + * A page of a thread's messages, newest first. `before` pages backwards through the + * history: pass the `chatMessageId` of the oldest message you already have. + */ +export async function getThreadMessages( + db: D1Database, + chatThreadId: number, + { limit = 50, before }: { limit?: number; before?: number } = {} +): Promise { + const { results } = before + ? await db + .prepare( + `SELECT * FROM message WHERE chat_thread_id = ?1 AND chat_message_id < ?2 + ORDER BY chat_message_id DESC LIMIT ?3` + ) + .bind(chatThreadId, before, limit) + .all() + : await db + .prepare( + `SELECT * FROM message WHERE chat_thread_id = ?1 + ORDER BY chat_message_id DESC LIMIT ?2` + ) + .bind(chatThreadId, limit) + .all() + return results.map(toMessage) +} + +/** A single message by id, or null if there's no such message. */ +export async function getMessage( + db: D1Database, + chatMessageId: number +): Promise { + const row = await db + .prepare('SELECT * FROM message WHERE chat_message_id = ?1') + .bind(chatMessageId) + .first() + return row === null ? null : toMessage(row) +} diff --git a/apps/chat/src/test/integration/api.test.ts b/apps/chat/src/test/integration/api.test.ts index 16ccd71..34ea7c6 100644 --- a/apps/chat/src/test/integration/api.test.ts +++ b/apps/chat/src/test/integration/api.test.ts @@ -1,10 +1,74 @@ -import { SELF } from 'cloudflare:test' -import { describe, expect, it } from 'vitest' +import { adminSecretsStore, env, SELF } from 'cloudflare:test' +import { beforeAll, beforeEach, describe, expect, it } from 'vitest' import '../../chat.app' +import { + ChatModerationState, + getMessage, + getThreadMessages, + insertMessage, + SCHEMA_DDL, +} from '../../message-db' +import { + createThread, + findThreadWithMembers, + getThreadForPlayer, + getThreadsForPlayer, + isThreadMember, + leftChatContents, + markThreadRead, + postMessage, + removeThreadMember, + setThreadFavorited, + startedChatContents, + SYSTEM_SENDER_ID, + THREAD_SCHEMA_DDL, +} from '../../thread-db' + +import type { Env } from '../../context' +import type { ChatMessage } from '../../message-db' + +declare module 'cloudflare:test' { + interface ProvidedEnv extends Env {} +} + const ORIGIN = 'https://example.com' +// Mint a token the way the `auth` worker does, signing with the shared test key seeded +// into the JWT_SECRET store. +const TEST_SECRET = 'test-signing-key' + +function b64url(input: ArrayBuffer | string): string { + const bytes = typeof input === 'string' ? new TextEncoder().encode(input) : new Uint8Array(input) + let binary = '' + for (const byte of bytes) binary += String.fromCharCode(byte) + return btoa(binary).replace(/\+/g, '-').replace(/\//g, '_').replace(/=+$/, '') +} + +async function bearer(sub: number): Promise> { + const now = Math.floor(Date.now() / 1000) + const signingInput = `${b64url(JSON.stringify({ alg: 'HS256', typ: 'JWT' }))}.${b64url( + JSON.stringify({ sub: String(sub), exp: now + 3600 }) + )}` + const key = await crypto.subtle.importKey( + 'raw', + new TextEncoder().encode(TEST_SECRET), + { name: 'HMAC', hash: 'SHA-256' }, + false, + ['sign'] + ) + const sig = await crypto.subtle.sign('HMAC', key, new TextEncoder().encode(signingInput)) + return { Authorization: `Bearer ${signingInput}.${b64url(sig)}` } +} + +beforeAll(async () => { + // Seed the shared JWT signing key into the local Secrets Store so .get() resolves. + await adminSecretsStore(env.JWT_SECRET).create(TEST_SECRET) + for (const stmt of SCHEMA_DDL) await env.DB.prepare(stmt).run() + for (const stmt of THREAD_SCHEMA_DDL) await env.DB.prepare(stmt).run() +}) + describe('chat endpoints', () => { it('GET / reports service status', async () => { const res = await SELF.fetch(`${ORIGIN}/`) @@ -12,9 +76,1181 @@ describe('chat endpoints', () => { expect(await res.json()).toEqual({ service: 'chat', status: 'ok' }) }) - it('GET /thread returns an empty array', async () => { + it('GET /thread 401s without a token', async () => { const res = await SELF.fetch(`${ORIGIN}/thread`) + expect(res.status).toBe(401) + }) + + it('GET /thread serves the caller their own threads', async () => { + const player = 881001 + const thread = await createThread(env.DB, [player, 881002]) + const latest = await postMessage(env.DB, { + chatThreadId: thread, + senderPlayerId: 881002, + timeSent: '2022-02-21T18:08:56.0362822', + contents: '{"Type":0,"Version":1,"Data":"hi"}', + }) + + const res = await SELF.fetch(`${ORIGIN}/thread?MessageCount=16&Mode=0`, { + headers: await bearer(player), + }) expect(res.status).toBe(200) - expect(await res.json()).toEqual([]) + expect(await res.json()).toEqual([ + { + latestMessage: latest, + chatThreadId: thread, + playerIds: [player, 881002], + lastReadMessageId: 0, + chatThreadName: '', + snoozedUntil: null, + isFavorited: false, + }, + ]) + }) + + it('GET /thread/:id/message serves the thread newest first', async () => { + const player = 881003 + const thread = await createThread(env.DB, [player, 881004]) + const older = await postMessage(env.DB, { + chatThreadId: thread, + senderPlayerId: 881004, + timeSent: '2022-02-19T22:13:56.7224503', + contents: '{"Type":0,"Version":1,"Data":"on discord?"}', + }) + const newer = await postMessage(env.DB, { + chatThreadId: thread, + senderPlayerId: 881004, + timeSent: '2022-02-21T18:08:56.0362822', + contents: '{"Type":0,"Version":1,"Data":"hi"}', + }) + + const res = await SELF.fetch(`${ORIGIN}/thread/${thread}/message?MessageCount=16&Mode=0`, { + headers: await bearer(player), + }) + expect(res.status).toBe(200) + expect(await res.json()).toEqual([newer, older]) + }) + + it('GET /thread/:id/message honours MessageCount', async () => { + const player = 881005 + const thread = await createThread(env.DB, [player]) + for (const data of ['one', 'two', 'three']) { + await postMessage(env.DB, { + chatThreadId: thread, + senderPlayerId: player, + contents: JSON.stringify({ Type: 0, Version: 1, Data: data }), + }) + } + + const res = await SELF.fetch(`${ORIGIN}/thread/${thread}/message?MessageCount=2`, { + headers: await bearer(player), + }) + expect(await res.json()).toHaveLength(2) + }) + + // A thread's existence is itself private, so a non-member gets the same 404 as for a + // thread that never existed — not a 403 that confirms it's there. + it('GET /thread/:id/message hides threads the caller is not in', async () => { + const thread = await createThread(env.DB, [881006, 881007]) + const res = await SELF.fetch(`${ORIGIN}/thread/${thread}/message`, { + headers: await bearer(881008), + }) + expect(res.status).toBe(404) + + const missing = await SELF.fetch(`${ORIGIN}/thread/999999/message`, { + headers: await bearer(881008), + }) + expect(missing.status).toBe(404) + }) +}) + +describe('message storage', () => { + // A real message as the client sends it, kept verbatim (including the JSON-in-a-string + // `contents` envelope) so the round-trip is tested against the actual payload shape. + const CONTENTS = + '{"Type":0,"Version":1,"Data":"This is jordanparki7 from your Oculus friends. We\'re friends in Rec Room now!"}' + + it('round-trips a message, assigning an id', async () => { + const stored = await insertMessage(env.DB, { + chatThreadId: 116181128, + senderPlayerId: 10441985, + timeSent: '2022-05-22T12:47:03.6536656', + contents: CONTENTS, + }) + expect(stored.chatMessageId).toBeGreaterThan(0) + expect(stored).toEqual({ + chatMessageId: stored.chatMessageId, + chatThreadId: 116181128, + senderPlayerId: 10441985, + timeSent: '2022-05-22T12:47:03.6536656', + contents: CONTENTS, + moderationState: ChatModerationState.None, + }) + expect(await getMessage(env.DB, stored.chatMessageId)).toEqual(stored) + }) + + it('defaults timeSent to now', async () => { + const stored = await insertMessage(env.DB, { + chatThreadId: 999, + senderPlayerId: 42, + contents: CONTENTS, + }) + expect(Date.parse(stored.timeSent)).toBeGreaterThan(Date.now() - 60_000) + }) + + it('lists a thread newest first and pages backwards', async () => { + const thread = 116181129 + const first = await insertMessage(env.DB, { + chatThreadId: thread, + senderPlayerId: 1, + contents: CONTENTS, + }) + const second = await insertMessage(env.DB, { + chatThreadId: thread, + senderPlayerId: 2, + contents: CONTENTS, + }) + + const page = await getThreadMessages(env.DB, thread) + expect(page.map((m) => m.chatMessageId)).toEqual([second.chatMessageId, first.chatMessageId]) + + const older = await getThreadMessages(env.DB, thread, { before: second.chatMessageId }) + expect(older.map((m) => m.chatMessageId)).toEqual([first.chatMessageId]) + + // Messages from other threads never leak into a thread's listing. + expect(await getThreadMessages(env.DB, 404)).toEqual([]) + }) +}) + +describe('thread storage', () => { + // The viewing player from the captured thread-list response. + const VIEWER = 10441985 + + function contents(data: string): string { + return JSON.stringify({ Type: 0, Version: 1, Data: data }) + } + + it('renders the thread list in the shape the client expects', async () => { + const dm = await createThread(env.DB, [9489959, VIEWER]) + const latest = await postMessage(env.DB, { + chatThreadId: dm, + senderPlayerId: VIEWER, + timeSent: '2022-05-22T12:47:03.6536656', + contents: contents( + "This is jordanparki7 from your Oculus friends. We're friends in Rec Room now!" + ), + }) + await markThreadRead(env.DB, dm, VIEWER, latest.chatMessageId) + + const [thread] = await getThreadsForPlayer(env.DB, VIEWER) + expect(thread).toEqual({ + latestMessage: latest, + chatThreadId: dm, + playerIds: [9489959, VIEWER], + lastReadMessageId: latest.chatMessageId, + chatThreadName: '', + snoozedUntil: null, + isFavorited: false, + }) + }) + + it('keeps a named group thread with all its members', async () => { + const members = [VIEWER, 10452682, 12534039, 12535328, 12631702] + const group = await createThread(env.DB, members, 'Group Chat =]') + await postMessage(env.DB, { + chatThreadId: group, + senderPlayerId: VIEWER, + contents: contents('sussy baka'), + }) + + const thread = await getThreadForPlayer(env.DB, group, VIEWER) + expect(thread?.chatThreadName).toBe('Group Chat =]') + expect(thread?.playerIds).toEqual(members.sort((a, b) => a - b)) + }) + + // System notices ("Player <@U…> started a chat") and player messages both carry + // markup the server must not touch — the mention token, and HTML entities the + // client escaped itself. Stored and served back byte-for-byte. + it('stores message contents verbatim, markup and all', async () => { + const thread = await createThread(env.DB, [VIEWER, 29565301]) + const notice = await postMessage(env.DB, { + chatThreadId: thread, + senderPlayerId: 29565301, + contents: contents('Player <@U29565301> started a chat'), + }) + const escaped = await postMessage(env.DB, { + chatThreadId: thread, + senderPlayerId: 29563053, + contents: contents('Ly2 bae <<<333'), + }) + + expect(await getMessage(env.DB, notice.chatMessageId)).toEqual(notice) + expect((await getMessage(env.DB, escaped.chatMessageId))?.contents).toBe(escaped.contents) + }) + + it('orders threads newest first and honours the page size', async () => { + const viewer = 777001 + const older = await createThread(env.DB, [viewer, 1]) + const newer = await createThread(env.DB, [viewer, 2]) + await postMessage(env.DB, { + chatThreadId: older, + senderPlayerId: viewer, + contents: contents('first'), + }) + await postMessage(env.DB, { + chatThreadId: newer, + senderPlayerId: viewer, + contents: contents('second'), + }) + + const threads = await getThreadsForPlayer(env.DB, viewer) + expect(threads.map((t) => t.chatThreadId)).toEqual([newer, older]) + expect(await getThreadsForPlayer(env.DB, viewer, { limit: 1 })).toHaveLength(1) + }) + + it('gates reads on membership', async () => { + const thread = await createThread(env.DB, [777002, 777003]) + expect(await isThreadMember(env.DB, thread, 777002)).toBe(true) + expect(await isThreadMember(env.DB, thread, 777004)).toBe(false) + // A non-member sees neither the thread nor its place in their own list. + expect(await getThreadForPlayer(env.DB, thread, 777004)).toBeNull() + expect(await getThreadsForPlayer(env.DB, 777004)).toEqual([]) + }) + + it('keeps read state, favorites, and snoozes per viewer', async () => { + const a = 777005 + const b = 777006 + const thread = await createThread(env.DB, [a, b]) + const first = await postMessage(env.DB, { + chatThreadId: thread, + senderPlayerId: a, + contents: contents('one'), + }) + const second = await postMessage(env.DB, { + chatThreadId: thread, + senderPlayerId: b, + contents: contents('two'), + }) + + await markThreadRead(env.DB, thread, a, second.chatMessageId) + await setThreadFavorited(env.DB, thread, a, true) + await markThreadRead(env.DB, thread, b, first.chatMessageId) + + const forA = await getThreadForPlayer(env.DB, thread, a) + const forB = await getThreadForPlayer(env.DB, thread, b) + expect(forA?.lastReadMessageId).toBe(second.chatMessageId) + expect(forA?.isFavorited).toBe(true) + expect(forB?.lastReadMessageId).toBe(first.chatMessageId) + expect(forB?.isFavorited).toBe(false) + + // A late ack from a second client can't walk the thread back to unread. + await markThreadRead(env.DB, thread, a, first.chatMessageId) + expect((await getThreadForPlayer(env.DB, thread, a))?.lastReadMessageId).toBe( + second.chatMessageId + ) + }) + + it('leaves an empty thread with no latest message', async () => { + const thread = await createThread(env.DB, [777007]) + expect(await getThreadForPlayer(env.DB, thread, 777007)).toMatchObject({ + latestMessage: null, + lastReadMessageId: 0, + playerIds: [777007], + }) + }) + + it('drops a removed member from the roster but keeps the thread', async () => { + const thread = await createThread(env.DB, [777008, 777009]) + await postMessage(env.DB, { + chatThreadId: thread, + senderPlayerId: 777008, + contents: contents('mellon'), + }) + await removeThreadMember(env.DB, thread, 777009) + + expect(await getThreadForPlayer(env.DB, thread, 777009)).toBeNull() + expect((await getThreadForPlayer(env.DB, thread, 777008))?.playerIds).toEqual([777008]) + }) +}) + +describe('POST /thread/withmembers', () => { + async function withMembers(caller: number, body: string) { + return SELF.fetch(`${ORIGIN}/thread/withmembers`, { + method: 'POST', + headers: { + ...(await bearer(caller)), + 'Content-Type': 'application/x-www-form-urlencoded', + }, + body, + }) + } + + it('opens a thread with the named players plus the caller', async () => { + const caller = 882001 + const res = await withMembers(caller, 'ids=2&ids=155&messageCount=50') + expect(res.status).toBe(200) + + const thread = (await res.json()) as { + chatThreadId: number + playerIds: number[] + messages: unknown[] + } + // A page of messages, not the list's single latestMessage. A brand-new thread + // isn't empty: it opens with the system "started a chat" notice. + expect(thread.messages).toEqual([ + expect.objectContaining({ + senderPlayerId: SYSTEM_SENDER_ID, + contents: startedChatContents(caller), + }), + ]) + expect(thread).toMatchObject({ + playerIds: [2, 155, caller], + lastReadMessageId: 0, + chatThreadName: '', + snoozedUntil: null, + isFavorited: false, + }) + expect(thread).not.toHaveProperty('latestMessage') + + // The thread is real: it shows up in the caller's list, and its members can read it. + expect((await getThreadsForPlayer(env.DB, caller)).map((t) => t.chatThreadId)).toContain( + thread.chatThreadId + ) + expect(await isThreadMember(env.DB, thread.chatThreadId, 155)).toBe(true) + }) + + // Two nulls GetChatBetweenPlayers can't survive: lastReadMessageId deserializes into + // a non-nullable int ("expected 'Number Token', actual 'null'"), and chatThreadName + // is dereferenced unchecked (NullReferenceException). Unread is 0, unnamed is ''. + it('never serializes lastReadMessageId or chatThreadName as null', async () => { + const res = await withMembers(882008, 'ids=153') + const body = await res.text() + expect(body).not.toContain('"lastReadMessageId":null') + expect(body).not.toContain('"chatThreadName":null') + + const thread = JSON.parse(body) as { lastReadMessageId: number; chatThreadName: string } + expect(thread.lastReadMessageId).toBe(0) + expect(thread.chatThreadName).toBe('') + }) + + it('collapses duplicate ids and the caller naming themselves', async () => { + const caller = 882002 + const res = await withMembers(caller, `ids=${caller}&ids=882003&ids=882003`) + expect(((await res.json()) as { playerIds: number[] }).playerIds).toEqual([caller, 882003]) + }) + + // Fetch-or-create: reopening a chat with the same people must land back in the + // conversation that already has the history, not a fresh empty one. + it('returns the existing thread rather than opening a second', async () => { + const caller = 882004 + const first = (await (await withMembers(caller, 'ids=882005')).json()) as { + chatThreadId: number + } + await postMessage(env.DB, { + chatThreadId: first.chatThreadId, + senderPlayerId: 882005, + contents: '{"Type":0,"Version":1,"Data":"hi"}', + }) + + const second = (await (await withMembers(caller, 'ids=882005')).json()) as { + chatThreadId: number + messages: unknown[] + } + expect(second.chatThreadId).toBe(first.chatThreadId) + // The opening notice plus the real message — reopening adds neither a thread nor + // a second notice. + expect(second.messages).toHaveLength(2) + }) + + // Membership is matched as a whole set, so a DM isn't mistaken for a group that + // happens to contain the same two people. + it('does not confuse a subset or superset for the same thread', async () => { + const caller = 882009 + const pair = (await (await withMembers(caller, 'ids=882010')).json()) as { + chatThreadId: number + } + const trio = (await (await withMembers(caller, 'ids=882010&ids=882011')).json()) as { + chatThreadId: number + } + expect(trio.chatThreadId).not.toBe(pair.chatThreadId) + }) + + it('honours messageCount when paging the thread', async () => { + const caller = 882012 + const opened = (await (await withMembers(caller, 'ids=882013')).json()) as { + chatThreadId: number + } + for (const data of ['one', 'two', 'three']) { + await postMessage(env.DB, { + chatThreadId: opened.chatThreadId, + senderPlayerId: caller, + contents: JSON.stringify({ Type: 0, Version: 1, Data: data }), + }) + } + + const paged = (await (await withMembers(caller, 'ids=882013&messageCount=2')).json()) as { + messages: unknown[] + } + expect(paged.messages).toHaveLength(2) + }) + + it('rejects a thread with nobody else in it', async () => { + const caller = 882006 + expect((await withMembers(caller, '')).status).toBe(400) + expect((await withMembers(caller, `ids=${caller}`)).status).toBe(400) + expect((await withMembers(caller, 'ids=notanumber')).status).toBe(400) + }) + + it('rejects an oversized roster', async () => { + const ids = Array.from({ length: 60 }, (_, i) => `ids=${883000 + i}`).join('&') + expect((await withMembers(882007, ids)).status).toBe(400) + }) + + it('401s without a token', async () => { + const res = await SELF.fetch(`${ORIGIN}/thread/withmembers`, { + method: 'POST', + headers: { 'Content-Type': 'application/x-www-form-urlencoded' }, + body: 'ids=2', + }) + expect(res.status).toBe(401) + }) +}) + +describe('POST /thread', () => { + async function createViaPost(caller: number, body: string) { + return SELF.fetch(`${ORIGIN}/thread`, { + method: 'POST', + headers: { + ...(await bearer(caller)), + 'Content-Type': 'application/x-www-form-urlencoded', + }, + body, + }) + } + + // The call the client actually makes after /thread/withmembers: members, blank + // contents. A blank field must not post an empty message, and reports + // invalid-arguments rather than success. + it('opens a thread with no message when messageContents is blank', async () => { + const caller = 884001 + const res = await createViaPost(caller, 'ids=155&ids=2&messageContents=') + expect(res.status).toBe(200) + + const body = (await res.json()) as { + chatThread: { chatThreadId: number; latestMessage: { senderPlayerId: number } | null } + chatResult: number + } + expect(body.chatResult).toBe(1) + expect(body.chatThread).toMatchObject({ playerIds: [2, 155, caller] }) + + // Nothing of the caller's was posted — but the thread still isn't empty: it opens + // with the system notice, which is what the client needs to render it at all. + const messages = await getThreadMessages(env.DB, body.chatThread.chatThreadId) + expect(messages).toEqual([ + expect.objectContaining({ + senderPlayerId: SYSTEM_SENDER_ID, + contents: startedChatContents(caller), + }), + ]) + expect(body.chatThread.latestMessage?.senderPlayerId).toBe(SYSTEM_SENDER_ID) + }) + + it('posts the first message when messageContents is given', async () => { + const caller = 884002 + const contents = '{"Type":0,"Version":1,"Data":"hi"}' + const res = await createViaPost( + caller, + `ids=884003&messageContents=${encodeURIComponent(contents)}` + ) + + const body = (await res.json()) as { + chatThread: { + chatThreadId: number + latestMessage: { contents: string; senderPlayerId: number } | null + } + chatResult: number + } + expect(body.chatResult).toBe(0) + // Stored verbatim, attributed to the caller, and already the thread's latest. + expect(body.chatThread.latestMessage).toMatchObject({ contents, senderPlayerId: caller }) + // The opening notice, then the caller's message. + expect(await getThreadMessages(env.DB, body.chatThread.chatThreadId)).toHaveLength(2) + }) + + // Sending to people you already have a thread with appends to it, rather than + // stranding the message in a second conversation. + it('appends to the existing thread with the same members', async () => { + const caller = 884005 + const first = (await ( + await createViaPost(caller, 'ids=884006&messageContents=%7B%22Data%22%3A%22one%22%7D') + ).json()) as { chatThread: { chatThreadId: number } } + const second = (await ( + await createViaPost(caller, 'ids=884006&messageContents=%7B%22Data%22%3A%22two%22%7D') + ).json()) as { chatThread: { chatThreadId: number } } + + expect(second.chatThread.chatThreadId).toBe(first.chatThread.chatThreadId) + // The opening notice, then both messages. + expect(await getThreadMessages(env.DB, first.chatThread.chatThreadId)).toHaveLength(3) + }) + + it('rejects a thread with nobody else in it', async () => { + expect((await createViaPost(884004, 'messageContents=')).status).toBe(400) + }) + + it('401s without a token', async () => { + const res = await SELF.fetch(`${ORIGIN}/thread`, { + method: 'POST', + headers: { 'Content-Type': 'application/x-www-form-urlencoded' }, + body: 'ids=2&messageContents=', + }) + expect(res.status).toBe(401) + }) +}) + +describe('GET /thread/:id', () => { + async function openThread(caller: number, chatThreadId: number, query = '?messageCount=50') { + return SELF.fetch(`${ORIGIN}/thread/${chatThreadId}${query}`, { + headers: await bearer(caller), + }) + } + + it('opens a thread with its recent messages, newest first', async () => { + const caller = 885001 + const chatThreadId = await createThread(env.DB, [caller, 885002]) + const older = await postMessage(env.DB, { + chatThreadId, + senderPlayerId: 885002, + timeSent: '2022-02-19T22:13:56.7224503', + contents: '{"Type":0,"Version":1,"Data":"on discord?"}', + }) + const newer = await postMessage(env.DB, { + chatThreadId, + senderPlayerId: 885002, + timeSent: '2022-02-21T18:08:56.0362822', + contents: '{"Type":0,"Version":1,"Data":"hi"}', + }) + + const res = await openThread(caller, chatThreadId) + expect(res.status).toBe(200) + + // An object, not a bare array: the client parses this one as a thread and rejects + // an array outright ("expected '{', actual '['"). + const body = await res.text() + expect(body.startsWith('{')).toBe(true) + + const thread = JSON.parse(body) as { messages: unknown[] } + expect(thread).toMatchObject({ + chatThreadId, + playerIds: [caller, 885002], + lastReadMessageId: 0, + chatThreadName: '', + snoozedUntil: null, + isFavorited: false, + }) + expect(thread.messages).toEqual([newer, older]) + expect(thread).not.toHaveProperty('latestMessage') + }) + + // The sibling route serves the same messages as a bare array — the two shapes are + // deliberately different, and the client depends on which is which. + it('carries the same messages /thread/:id/message serves as an array', async () => { + const caller = 885010 + const chatThreadId = await createThread(env.DB, [caller, 885011]) + await postMessage(env.DB, { + chatThreadId, + senderPlayerId: 885011, + contents: '{"Type":0,"Version":1,"Data":"hi"}', + }) + + const thread = (await (await openThread(caller, chatThreadId)).json()) as { + messages: unknown[] + } + const messages = await ( + await SELF.fetch(`${ORIGIN}/thread/${chatThreadId}/message?MessageCount=50`, { + headers: await bearer(caller), + }) + ).json() + expect(Array.isArray(messages)).toBe(true) + expect(thread.messages).toEqual(messages) + }) + + it('honours messageCount', async () => { + const caller = 885003 + const chatThreadId = await createThread(env.DB, [caller, 885004]) + for (const data of ['one', 'two', 'three']) { + await postMessage(env.DB, { + chatThreadId, + senderPlayerId: caller, + contents: JSON.stringify({ Type: 0, Version: 1, Data: data }), + }) + } + + const res = await openThread(caller, chatThreadId, '?messageCount=2') + expect(((await res.json()) as { messages: unknown[] }).messages).toHaveLength(2) + }) + + // A thread opened moments ago has nothing in it and still has to open — an empty + // messages array, not a 404. + it('opens an empty thread with an empty messages array', async () => { + const caller = 885005 + const chatThreadId = await createThread(env.DB, [caller, 885006]) + + const res = await openThread(caller, chatThreadId) + expect(res.status).toBe(200) + + const { messages } = (await res.json()) as { messages: unknown[] } + // Built directly by createThread with no starter, so genuinely empty. + expect(messages).toEqual([]) + }) + + it('hides threads the caller is not in', async () => { + const chatThreadId = await createThread(env.DB, [885007, 885008]) + expect((await openThread(885009, chatThreadId)).status).toBe(404) + expect((await openThread(885009, 999999)).status).toBe(404) + }) + + it('401s without a token', async () => { + const res = await SELF.fetch(`${ORIGIN}/thread/1?messageCount=50`) + expect(res.status).toBe(401) + }) +}) + +describe('ChatMessageReceived push', () => { + /** The stubbed hub (see vitest.config.ts) records what it was sent. */ + interface SentNotification { + playerId: number + notificationType: number + data: Record + } + const hub = env.RECFLARE_NOTIFICATIONS_HUB as unknown as { + getByName(name: string): { takeSent(): Promise } + } + + async function send(caller: number, body: string) { + return SELF.fetch(`${ORIGIN}/thread`, { + method: 'POST', + headers: { + ...(await bearer(caller)), + 'Content-Type': 'application/x-www-form-urlencoded', + }, + body, + }) + } + + beforeEach(async () => { + await hub.getByName('global').takeSent() + }) + + // Every member is notified, the sender included: the client doesn't fold the HTTP + // response into its thread cache, so without a self-push its own message doesn't + // show until the next refetch. + it('notifies every member of the thread, sender included', async () => { + const caller = 886001 + const contents = '{"Type":0,"Version":1,"Data":"hi"}' + const res = await send( + caller, + `ids=886002&ids=886003&messageContents=${encodeURIComponent(contents)}` + ) + const { chatThread } = (await res.json()) as { + chatThread: { chatThreadId: number; latestMessage: { chatMessageId: number } } + } + + const sent = await hub.getByName('global').takeSent() + expect(sent.map((n) => n.playerId).sort((a, b) => a - b)).toEqual([caller, 886002, 886003]) + // NotificationType.ChatMessageReceived + expect(sent.every((n) => n.notificationType === 90)).toBe(true) + expect(sent[0]!.data).toEqual({ + chatMessageId: chatThread.latestMessage.chatMessageId, + chatThreadId: chatThread.chatThreadId, + senderPlayerId: caller, + timeSent: expect.any(String), + contents, + moderationState: 0, + }) + }) + + it('pushes nothing when there is no message to push', async () => { + await send(886004, 'ids=886005&messageContents=') + expect(await hub.getByName('global').takeSent()).toEqual([]) + }) +}) + +// A membership row whose thread row is gone must never be resolved to: it would hand +// back an id nothing can render ("thread N vanished after creation"), and because the +// oldest match wins it would keep winning on every later call. +describe('orphaned membership rows', () => { + it('ignores members of a thread whose message_thread row is gone', async () => { + const caller = 887001 + const other = 887002 + const orphaned = await createThread(env.DB, [caller, other]) + await env.DB.prepare('DELETE FROM message_thread WHERE chat_thread_id = ?1') + .bind(orphaned) + .run() + + expect(await findThreadWithMembers(env.DB, [caller, other])).toBeNull() + + // Opening the chat recovers: a usable thread comes back, and it isn't the orphan. + const res = await SELF.fetch(`${ORIGIN}/thread/withmembers`, { + method: 'POST', + headers: { + ...(await bearer(caller)), + 'Content-Type': 'application/x-www-form-urlencoded', + }, + body: `ids=${other}&messageCount=50`, + }) + expect(res.status).toBe(200) + + const thread = (await res.json()) as { chatThreadId: number; playerIds: number[] } + expect(thread.chatThreadId).not.toBe(orphaned) + expect(thread.playerIds).toEqual([caller, other]) + + // And it stays stable — the orphan never wins a later lookup. + expect(await findThreadWithMembers(env.DB, [caller, other])).toBe(thread.chatThreadId) + }) +}) + +describe('marking a thread read', () => { + async function read(caller: number, path: string, method = 'POST') { + return SELF.fetch(`${ORIGIN}${path}`, { method, headers: await bearer(caller) }) + } + + it('moves the pointer to a specific message, on both verbs', async () => { + const caller = 888001 + const chatThreadId = await createThread(env.DB, [caller, 888002]) + const first = await postMessage(env.DB, { + chatThreadId, + senderPlayerId: 888002, + contents: '{"Type":0,"Version":1,"Data":"one"}', + }) + const second = await postMessage(env.DB, { + chatThreadId, + senderPlayerId: 888002, + contents: '{"Type":0,"Version":1,"Data":"two"}', + }) + + const res = await read(caller, `/thread/${chatThreadId}/message/${first.chatMessageId}/read`) + expect(res.status).toBe(200) + // The bare ChatResult integer, not an envelope. + expect(await res.json()).toBe(0) + expect((await getThreadForPlayer(env.DB, chatThreadId, caller))?.lastReadMessageId).toBe( + first.chatMessageId + ) + + await read(caller, `/thread/${chatThreadId}/message/${second.chatMessageId}/read`, 'PUT') + expect((await getThreadForPlayer(env.DB, chatThreadId, caller))?.lastReadMessageId).toBe( + second.chatMessageId + ) + }) + + it('marks the whole thread read without a message id', async () => { + const caller = 888003 + const chatThreadId = await createThread(env.DB, [caller, 888004]) + const latest = await postMessage(env.DB, { + chatThreadId, + senderPlayerId: 888004, + contents: '{"Type":0,"Version":1,"Data":"hi"}', + }) + + expect((await read(caller, `/thread/${chatThreadId}/read`)).status).toBe(200) + expect((await getThreadForPlayer(env.DB, chatThreadId, caller))?.lastReadMessageId).toBe( + latest.chatMessageId + ) + }) + + // The client acks whatever id it was shown — including the synthetic message's + // 9007199254740976. Clamping keeps that from stranding the thread as read forever. + it('clamps an id beyond the thread to the real latest message', async () => { + const caller = 888005 + const chatThreadId = await createThread(env.DB, [caller, 888006]) + const real = await postMessage(env.DB, { + chatThreadId, + senderPlayerId: 888006, + contents: '{"Type":0,"Version":1,"Data":"hi"}', + }) + + await read(caller, `/thread/${chatThreadId}/message/9007199254740976/read`) + expect((await getThreadForPlayer(env.DB, chatThreadId, caller))?.lastReadMessageId).toBe( + real.chatMessageId + ) + + // A later real message is still unread, rather than swallowed by the bogus ack. + const next = await postMessage(env.DB, { + chatThreadId, + senderPlayerId: 888006, + contents: '{"Type":0,"Version":1,"Data":"later"}', + }) + expect( + (await getThreadForPlayer(env.DB, chatThreadId, caller))?.lastReadMessageId + ).toBeLessThan(next.chatMessageId) + }) + + it('never moves the pointer backwards', async () => { + const caller = 888007 + const chatThreadId = await createThread(env.DB, [caller, 888008]) + const first = await postMessage(env.DB, { + chatThreadId, + senderPlayerId: 888008, + contents: '{"Type":0,"Version":1,"Data":"one"}', + }) + const second = await postMessage(env.DB, { + chatThreadId, + senderPlayerId: 888008, + contents: '{"Type":0,"Version":1,"Data":"two"}', + }) + + await read(caller, `/thread/${chatThreadId}/message/${second.chatMessageId}/read`) + await read(caller, `/thread/${chatThreadId}/message/${first.chatMessageId}/read`) + expect((await getThreadForPlayer(env.DB, chatThreadId, caller))?.lastReadMessageId).toBe( + second.chatMessageId + ) + }) + + it('is gated on membership and auth', async () => { + const chatThreadId = await createThread(env.DB, [888009, 888010]) + expect((await read(888011, `/thread/${chatThreadId}/read`)).status).toBe(404) + + const anon = await SELF.fetch(`${ORIGIN}/thread/${chatThreadId}/read`, { method: 'POST' }) + expect(anon.status).toBe(401) + }) +}) + +describe('POST /thread/:id', () => { + // The exact body the client sends: a Version 2 envelope whose Data carries a `<=>` + // prefix. Nothing in the worker parses it, so it must survive byte-for-byte. + const CONTENTS = '{"Type":0,"Version":2,"Data":"<=>hey"}' + + async function send(caller: number, path: string, contents = CONTENTS) { + return SELF.fetch(`${ORIGIN}${path}`, { + method: 'POST', + headers: { + ...(await bearer(caller)), + 'Content-Type': 'application/x-www-form-urlencoded', + }, + body: `messageContents=${encodeURIComponent(contents)}`, + }) + } + + it('appends to an existing thread and answers the send wrapper', async () => { + const caller = 889001 + const chatThreadId = await createThread(env.DB, [caller, 889002], null, caller) + + const res = await send(caller, `/thread/${chatThreadId}`) + expect(res.status).toBe(200) + + const body = (await res.json()) as { + chatResult: number + chatThread: { chatThreadId: number; playerIds: number[]; messages: ChatMessage[] } + } + expect(body.chatResult).toBe(0) + + // The whole thread comes back, newest message first, with the opening notice + // beneath it — the shape the client re-renders the conversation from. + expect(body.chatThread).toMatchObject({ chatThreadId, playerIds: [caller, 889002] }) + expect(body.chatThread.messages).toHaveLength(2) + expect(body.chatThread.messages[0]).toMatchObject({ + chatThreadId, + senderPlayerId: caller, + contents: CONTENTS, + moderationState: 0, + }) + expect(body.chatThread.messages[1]).toMatchObject({ senderPlayerId: SYSTEM_SENDER_ID }) + + // And it's stored, not just echoed. + expect(await getThreadMessages(env.DB, chatThreadId)).toEqual(body.chatThread.messages) + }) + + it('accepts the /thread/:id/message spelling too', async () => { + const caller = 889003 + const chatThreadId = await createThread(env.DB, [caller, 889004], null, caller) + + const res = await send(caller, `/thread/${chatThreadId}/message`) + expect(((await res.json()) as { chatResult: number }).chatResult).toBe(0) + expect(await getThreadMessages(env.DB, chatThreadId)).toHaveLength(2) + }) + + it('pushes ChatMessageReceived to every member', async () => { + const hub = env.RECFLARE_NOTIFICATIONS_HUB as unknown as { + getByName(name: string): { + takeSent(): Promise> + } + } + const caller = 889005 + const chatThreadId = await createThread(env.DB, [caller, 889006], null, caller) + await hub.getByName('global').takeSent() + + await send(caller, `/thread/${chatThreadId}`) + const sent = await hub.getByName('global').takeSent() + expect(sent.map((n) => n.playerId).sort((a, b) => a - b)).toEqual([caller, 889006]) + expect(sent.every((n) => n.notificationType === 90)).toBe(true) + }) + + it('reports invalid arguments for blank contents without storing anything', async () => { + const caller = 889007 + const chatThreadId = await createThread(env.DB, [caller, 889008], null, caller) + + const res = await send(caller, `/thread/${chatThreadId}`, ' ') + expect(res.status).toBe(200) + + const body = (await res.json()) as { chatResult: number; chatThread: { messages: unknown[] } } + expect(body.chatResult).toBe(1) + // The thread still comes back — only the opening notice is in it. + expect(body.chatThread.messages).toHaveLength(1) + expect(await getThreadMessages(env.DB, chatThreadId)).toHaveLength(1) + }) + + it('is gated on membership and auth', async () => { + const chatThreadId = await createThread(env.DB, [889009, 889010], null, 889009) + expect((await send(889011, `/thread/${chatThreadId}`)).status).toBe(404) + + const anon = await SELF.fetch(`${ORIGIN}/thread/${chatThreadId}`, { + method: 'POST', + headers: { 'Content-Type': 'application/x-www-form-urlencoded' }, + body: 'messageContents=hi', + }) + expect(anon.status).toBe(401) + }) +}) + +describe('POST /thread/:id/member/:playerId', () => { + async function addMember(caller: number, chatThreadId: number, playerId: number) { + return SELF.fetch(`${ORIGIN}/thread/${chatThreadId}/member/${playerId}`, { + method: 'POST', + headers: await bearer(caller), + }) + } + + it('adds a player to a thread the caller is in', async () => { + const caller = 890001 + const chatThreadId = await createThread(env.DB, [caller, 890002], null, caller) + + const res = await addMember(caller, chatThreadId, 890003) + expect(res.status).toBe(200) + expect(await res.json()).toBe(0) + + expect(await isThreadMember(env.DB, chatThreadId, 890003)).toBe(true) + // The new member sees the thread, its history included. + const thread = await getThreadForPlayer(env.DB, chatThreadId, 890003) + expect(thread?.playerIds).toEqual([caller, 890002, 890003]) + }) + + it('reports the player is already on the thread', async () => { + const caller = 890004 + const chatThreadId = await createThread(env.DB, [caller, 890005], null, caller) + expect(await (await addMember(caller, chatThreadId, 890005)).json()).toBe(4) + }) + + // A non-member gets the same answer as for a thread that doesn't exist, so the + // endpoint can't be used to probe for threads. + it('refuses a caller who is not on the thread', async () => { + const chatThreadId = await createThread(env.DB, [890006, 890007], null, 890006) + expect(await (await addMember(890008, chatThreadId, 890009)).json()).toBe(3) + expect(await isThreadMember(env.DB, chatThreadId, 890009)).toBe(false) + + expect(await (await addMember(890008, 999999, 890009)).json()).toBe(3) + }) + + it('401s without a token', async () => { + const res = await SELF.fetch(`${ORIGIN}/thread/1/member/2`, { method: 'POST' }) + expect(res.status).toBe(401) + }) +}) + +describe('renaming and leaving a thread', () => { + async function post(caller: number, path: string, body?: string, method = 'POST') { + return SELF.fetch(`${ORIGIN}${path}`, { + method, + headers: { + ...(await bearer(caller)), + ...(body === undefined ? {} : { 'Content-Type': 'application/x-www-form-urlencoded' }), + }, + body, + }) + } + + it('renames a thread for everyone on it', async () => { + const caller = 891001 + const chatThreadId = await createThread(env.DB, [caller, 891002], null, caller) + + const res = await post(caller, `/thread/${chatThreadId}/rename`, 'name=my%20chat') + expect(res.status).toBe(200) + expect(await res.json()).toBe(0) + + // Both members see the new name. + expect((await getThreadForPlayer(env.DB, chatThreadId, caller))?.chatThreadName).toBe('my chat') + expect((await getThreadForPlayer(env.DB, chatThreadId, 891002))?.chatThreadName).toBe('my chat') + }) + + it('clears the name back to unnamed, never null', async () => { + const caller = 891003 + const chatThreadId = await createThread(env.DB, [caller, 891004], 'Group Chat =]', caller) + + await post(caller, `/thread/${chatThreadId}/rename`, 'name=') + expect((await getThreadForPlayer(env.DB, chatThreadId, caller))?.chatThreadName).toBe('') + }) + + it('truncates an overlong name rather than rejecting it', async () => { + const caller = 891005 + const chatThreadId = await createThread(env.DB, [caller, 891006], null, caller) + + await post(caller, `/thread/${chatThreadId}/rename`, `name=${'x'.repeat(200)}`) + expect((await getThreadForPlayer(env.DB, chatThreadId, caller))?.chatThreadName).toHaveLength( + 128 + ) + }) + + it('refuses to rename a thread the caller is not on', async () => { + const chatThreadId = await createThread(env.DB, [891007, 891008], null, 891007) + expect(await (await post(891009, `/thread/${chatThreadId}/rename`, 'name=nope')).json()).toBe(3) + expect((await getThreadForPlayer(env.DB, chatThreadId, 891007))?.chatThreadName).toBe('') + }) + + it('leaves a thread, posting the notice and keeping the history', async () => { + const caller = 891010 + const stayer = 891011 + const chatThreadId = await createThread(env.DB, [caller, stayer], null, caller) + await postMessage(env.DB, { + chatThreadId, + senderPlayerId: caller, + contents: '{"Type":0,"Version":1,"Data":"bye"}', + }) + + const res = await post(caller, `/thread/${chatThreadId}/leave`) + expect(res.status).toBe(200) + expect(await res.json()).toBe(0) + + // Gone for the leaver, intact for everyone else. + expect(await isThreadMember(env.DB, chatThreadId, caller)).toBe(false) + expect(await getThreadForPlayer(env.DB, chatThreadId, caller)).toBeNull() + + const remaining = await getThreadForPlayer(env.DB, chatThreadId, stayer) + expect(remaining?.playerIds).toEqual([stayer]) + expect(remaining?.latestMessage).toMatchObject({ + senderPlayerId: SYSTEM_SENDER_ID, + contents: leftChatContents(caller), + }) + // Opening notice, the message, and the leave notice. + expect(await getThreadMessages(env.DB, chatThreadId)).toHaveLength(3) + }) + + it('accepts DELETE for leave as well as POST', async () => { + const caller = 891012 + const chatThreadId = await createThread(env.DB, [caller, 891013], null, caller) + + expect( + await (await post(caller, `/thread/${chatThreadId}/leave`, undefined, 'DELETE')).json() + ).toBe(0) + expect(await isThreadMember(env.DB, chatThreadId, caller)).toBe(false) + }) + + it('reports membership-not-found when leaving a thread you are not on', async () => { + const chatThreadId = await createThread(env.DB, [891014, 891015], null, 891014) + expect(await (await post(891016, `/thread/${chatThreadId}/leave`)).json()).toBe(3) + // Nothing was posted to a thread the caller has no business touching. + expect(await getThreadMessages(env.DB, chatThreadId)).toHaveLength(1) + }) + + it('401s without a token', async () => { + const rename = await SELF.fetch(`${ORIGIN}/thread/1/rename`, { method: 'POST' }) + expect(rename.status).toBe(401) + const leave = await SELF.fetch(`${ORIGIN}/thread/1/leave`, { method: 'POST' }) + expect(leave.status).toBe(401) + }) +}) + +describe('POST /thread/:id/snooze', () => { + async function snooze(caller: number, chatThreadId: number, body: string) { + return SELF.fetch(`${ORIGIN}/thread/${chatThreadId}/snooze`, { + method: 'POST', + headers: { + ...(await bearer(caller)), + 'Content-Type': 'application/x-www-form-urlencoded', + }, + body, + }) + } + + it('snoozes and unsnoozes for the caller alone', async () => { + const caller = 892001 + const other = 892002 + const chatThreadId = await createThread(env.DB, [caller, other], null, caller) + + const res = await snooze(caller, chatThreadId, 'snooze=True') + expect(res.status).toBe(200) + expect(await res.json()).toBe(0) + + expect((await getThreadForPlayer(env.DB, chatThreadId, caller))?.snoozedUntil).toBe( + '9999-12-31T23:59:59Z' + ) + // Snoozing is per-member: the other player is untouched. + expect((await getThreadForPlayer(env.DB, chatThreadId, other))?.snoozedUntil).toBeNull() + + await snooze(caller, chatThreadId, 'snooze=False') + expect((await getThreadForPlayer(env.DB, chatThreadId, caller))?.snoozedUntil).toBeNull() + }) + + it('refuses a thread the caller is not on', async () => { + const chatThreadId = await createThread(env.DB, [892003, 892004], null, 892003) + expect(await (await snooze(892005, chatThreadId, 'snooze=True')).json()).toBe(3) + }) + + it('401s without a token', async () => { + const res = await SELF.fetch(`${ORIGIN}/thread/1/snooze`, { + method: 'POST', + headers: { 'Content-Type': 'application/x-www-form-urlencoded' }, + body: 'snooze=True', + }) + expect(res.status).toBe(401) + }) +}) + +describe('PUT /thread/:id/favorite', () => { + async function favorite(caller: number, chatThreadId: number, body: string, method = 'PUT') { + return SELF.fetch(`${ORIGIN}/thread/${chatThreadId}/favorite`, { + method, + headers: { + ...(await bearer(caller)), + 'Content-Type': 'application/x-www-form-urlencoded', + }, + body, + }) + } + + it('favorites and unfavorites for the caller alone', async () => { + const caller = 893001 + const other = 893002 + const chatThreadId = await createThread(env.DB, [caller, other], null, caller) + + const res = await favorite(caller, chatThreadId, 'favorite=True') + expect(res.status).toBe(200) + expect(await res.json()).toBe(0) + + expect((await getThreadForPlayer(env.DB, chatThreadId, caller))?.isFavorited).toBe(true) + // Per-member, like snoozing: the other player's inbox is untouched. + expect((await getThreadForPlayer(env.DB, chatThreadId, other))?.isFavorited).toBe(false) + + await favorite(caller, chatThreadId, 'favorite=False') + expect((await getThreadForPlayer(env.DB, chatThreadId, caller))?.isFavorited).toBe(false) + }) + + it('accepts POST as well as PUT', async () => { + const caller = 893003 + const chatThreadId = await createThread(env.DB, [caller, 893004], null, caller) + + expect(await (await favorite(caller, chatThreadId, 'favorite=True', 'POST')).json()).toBe(0) + expect((await getThreadForPlayer(env.DB, chatThreadId, caller))?.isFavorited).toBe(true) + }) + + it('refuses a thread the caller is not on', async () => { + const chatThreadId = await createThread(env.DB, [893005, 893006], null, 893005) + expect(await (await favorite(893007, chatThreadId, 'favorite=True')).json()).toBe(3) + }) + + it('401s without a token', async () => { + const res = await SELF.fetch(`${ORIGIN}/thread/1/favorite`, { + method: 'PUT', + headers: { 'Content-Type': 'application/x-www-form-urlencoded' }, + body: 'favorite=True', + }) + expect(res.status).toBe(401) }) }) diff --git a/apps/chat/src/thread-db.ts b/apps/chat/src/thread-db.ts new file mode 100644 index 0000000..84e49ee --- /dev/null +++ b/apps/chat/src/thread-db.ts @@ -0,0 +1,462 @@ +/** + * Chat threads and their membership on the shared `recflare` D1 database. A thread is a + * conversation — a DM pair, a named group chat, or a system thread; the messages in it + * live in `message` (see message-db.ts). + * + * Membership (`thread_member`) does double duty: it is the authorization gate — a + * player may read or post to a thread only if they hold a row — and it is what renders + * the `playerIds` array the client shows. Nothing here has a foreign key to accounts, + * here or on a message's sender: that table belongs to the `auth` worker, and a thread + * outlives the accounts in it. + * + * The thread denormalizes `latest_message_id` so the thread list renders from one + * indexed row per thread rather than a per-thread MAX() over `message`, and so it can + * be ordered by recency without a join — message ids are monotonic, so the highest id + * is the newest thread. `postMessage` keeps it in sync. + * + * The per-viewer fields — `lastReadMessageId`, `snoozedUntil`, `isFavorited` — live on + * the membership row, not the thread: two players in one DM have independent read + * positions, snoozes, and favorites. + * + * The `chat` worker owns this schema/migration (migrations/0002_thread.sql). + * `THREAD_SCHEMA_DDL` mirrors it so tests can build the tables directly. + */ + +import { insertMessage } from './message-db' + +import type { ChatMessage, NewChatMessage } from './message-db' + +/** Schema DDL (mirror of migrations/0002_thread.sql). */ +export const THREAD_SCHEMA_DDL: string[] = [ + `CREATE TABLE IF NOT EXISTS message_thread ( + chat_thread_id INTEGER PRIMARY KEY AUTOINCREMENT, + chat_thread_name TEXT, + latest_message_id INTEGER, + created_at TEXT NOT NULL + )`, + `CREATE INDEX IF NOT EXISTS idx_message_thread_latest ON message_thread (latest_message_id)`, + `CREATE TABLE IF NOT EXISTS thread_member ( + chat_thread_id INTEGER NOT NULL, + player_id INTEGER NOT NULL, + last_read_message_id INTEGER, + snoozed_until TEXT, + is_favorited INTEGER NOT NULL DEFAULT 0, + PRIMARY KEY (chat_thread_id, player_id) + )`, + // The thread-list query is "every thread this player is in", so player_id leads. + `CREATE INDEX IF NOT EXISTS idx_thread_member_player ON thread_member (player_id)`, +] + +/** + * A thread as the client receives it: the thread, its members, its most recent message, + * and the viewing player's own read/snooze/favorite state. This is the element shape of + * the thread-list response. + */ +export interface ChatThread { + /** Null only for a thread with no messages yet. */ + latestMessage: ChatMessage | null + chatThreadId: number + playerIds: number[] + /** + * 0 when the player has never read the thread — never null. The client deserializes + * this into a non-nullable int and fails the whole response on a null ("expected + * 'Number Token', actual 'null'"), unlike `latestMessage`, which it accepts as null. + */ + lastReadMessageId: number + /** + * Empty for DMs and unnamed groups — never null. The client dereferences this name + * without a null check (a null NullReferenceExceptions its way out of + * GetChatBetweenPlayers) and falls back to naming the members when it's blank. + */ + chatThreadName: string + snoozedUntil: string | null + isFavorited: boolean +} + +/** The joined row backing a rendered thread, before it's shaped for the client. */ +interface ThreadRow { + chat_thread_id: number + chat_thread_name: string | null + player_ids: string | null + last_read_message_id: number | null + snoozed_until: string | null + is_favorited: number + msg_chat_message_id: number | null + msg_chat_thread_id: number | null + msg_sender_player_id: number | null + msg_time_sent: string | null + msg_contents: string | null + msg_moderation_state: number | null +} + +function toThread(row: ThreadRow): ChatThread { + return { + latestMessage: + row.msg_chat_message_id === null + ? null + : { + chatMessageId: row.msg_chat_message_id, + chatThreadId: row.msg_chat_thread_id!, + senderPlayerId: row.msg_sender_player_id!, + timeSent: row.msg_time_sent!, + contents: row.msg_contents!, + moderationState: row.msg_moderation_state!, + }, + chatThreadId: row.chat_thread_id, + // group_concat of the membership rows, already ordered by player id. + playerIds: row.player_ids === null ? [] : row.player_ids.split(',').map(Number), + // Null in the column means "never read"; the client insists on a number. + lastReadMessageId: row.last_read_message_id ?? 0, + // Null in the column means "unnamed"; the client dereferences it unchecked. + chatThreadName: row.chat_thread_name ?? '', + snoozedUntil: row.snoozed_until, + isFavorited: row.is_favorited !== 0, + } +} + +/** + * The thread list as it renders for one player, newest conversation first — the + * `?MessageCount=N` page of the thread endpoint. + * + * Reads only threads the player is a member of, so the membership join is the + * authorization check as well as the query. The inner ordered subquery around + * group_concat is what makes `playerIds` come back sorted rather than in row order. + */ +export async function getThreadsForPlayer( + db: D1Database, + playerId: number, + { limit = 50 }: { limit?: number } = {} +): Promise { + const { results } = await db + .prepare( + `SELECT + t.chat_thread_id, + t.chat_thread_name, + (SELECT group_concat(player_id) FROM + (SELECT player_id FROM thread_member WHERE chat_thread_id = t.chat_thread_id + ORDER BY player_id)) AS player_ids, + me.last_read_message_id, + me.snoozed_until, + me.is_favorited, + msg.chat_message_id AS msg_chat_message_id, + msg.chat_thread_id AS msg_chat_thread_id, + msg.sender_player_id AS msg_sender_player_id, + msg.time_sent AS msg_time_sent, + msg.contents AS msg_contents, + msg.moderation_state AS msg_moderation_state + FROM thread_member me + JOIN message_thread t ON t.chat_thread_id = me.chat_thread_id + LEFT JOIN message msg ON msg.chat_message_id = t.latest_message_id + WHERE me.player_id = ?1 + ORDER BY t.latest_message_id DESC + LIMIT ?2` + ) + .bind(playerId, limit) + .all() + return results.map(toThread) +} + +/** One thread as it renders for one player, or null if they aren't a member of it. */ +export async function getThreadForPlayer( + db: D1Database, + chatThreadId: number, + playerId: number +): Promise { + const row = await db + .prepare( + `SELECT + t.chat_thread_id, + t.chat_thread_name, + (SELECT group_concat(player_id) FROM + (SELECT player_id FROM thread_member WHERE chat_thread_id = t.chat_thread_id + ORDER BY player_id)) AS player_ids, + me.last_read_message_id, + me.snoozed_until, + me.is_favorited, + msg.chat_message_id AS msg_chat_message_id, + msg.chat_thread_id AS msg_chat_thread_id, + msg.sender_player_id AS msg_sender_player_id, + msg.time_sent AS msg_time_sent, + msg.contents AS msg_contents, + msg.moderation_state AS msg_moderation_state + FROM thread_member me + JOIN message_thread t ON t.chat_thread_id = me.chat_thread_id + LEFT JOIN message msg ON msg.chat_message_id = t.latest_message_id + WHERE me.chat_thread_id = ?1 AND me.player_id = ?2` + ) + .bind(chatThreadId, playerId) + .first() + return row === null ? null : toThread(row) +} + +/** + * Whether a player may read or post to a thread. Every thread-scoped route gates on + * this before touching messages. + */ +export async function isThreadMember( + db: D1Database, + chatThreadId: number, + playerId: number +): Promise { + const row = await db + .prepare('SELECT 1 AS ok FROM thread_member WHERE chat_thread_id = ?1 AND player_id = ?2') + .bind(chatThreadId, playerId) + .first<{ ok: number }>() + return row !== null +} + +/** + * The pseudo-player system messages are sent as. Not a real account — the client renders + * a message from this sender as a notice rather than as someone speaking, which is why + * `message.sender_player_id` carries no foreign key and permits negative ids. + */ +export const SYSTEM_SENDER_ID = -5 + +/** + * The notice a thread opens with: `Player <@U10441985> started a chat`. The `<@U…>` token + * is a mention the client resolves to a display name, so the id goes in raw. + */ +export function startedChatContents(playerId: number): string { + return JSON.stringify({ + Type: 0, + Version: 1, + Data: `Player <@U${playerId}> started a chat`, + }) +} + +/** + * The notice left behind when someone walks out of a group: `Player <@U14922080> left`. + * Same `<@U…>` mention token the opening notice uses. + */ +export function leftChatContents(playerId: number): string { + return JSON.stringify({ Type: 0, Version: 1, Data: `Player <@U${playerId}> left` }) +} + +/** + * Rename a thread. An empty name clears it back to unnamed, which renders as the member + * list rather than a blank title. + */ +export async function setThreadName( + db: D1Database, + chatThreadId: number, + name: string +): Promise { + await db + .prepare('UPDATE message_thread SET chat_thread_name = ?2 WHERE chat_thread_id = ?1') + .bind(chatThreadId, name === '' ? null : name) + .run() +} + +/** + * Open a thread between a set of players, returning its new id. `name` is null for DMs + * and unnamed groups. Duplicate player ids collapse, so a caller need not dedupe. + * + * Pass `startedBy` to open the thread the way the real server does — with a system + * "started a chat" notice as its first message. A thread with no messages at all is one + * the client won't display, so every thread born from a request gets one; the parameter + * is optional only so tests can build a bare thread directly. + * + * Every call opens a *distinct* thread, even for a member set that already has one — + * threads are not keyed by their membership, and the same pair may hold several. + */ +export async function createThread( + db: D1Database, + playerIds: number[], + name: string | null = null, + startedBy?: number +): Promise { + const row = await db + .prepare( + `INSERT INTO message_thread (chat_thread_name, created_at) VALUES (?1, ?2) + RETURNING chat_thread_id` + ) + .bind(name, new Date().toISOString()) + .first<{ chat_thread_id: number }>() + if (row === null) throw new Error('failed to create chat thread') + + const members = [...new Set(playerIds)] + if (members.length > 0) { + await db.batch( + members.map((playerId) => + db + .prepare( + `INSERT OR IGNORE INTO thread_member (chat_thread_id, player_id) + VALUES (?1, ?2)` + ) + .bind(row.chat_thread_id, playerId) + ) + ) + } + + if (startedBy !== undefined) { + await postMessage(db, { + chatThreadId: row.chat_thread_id, + senderPlayerId: SYSTEM_SENDER_ID, + contents: startedChatContents(startedBy), + }) + } + return row.chat_thread_id +} + +/** + * The existing thread whose membership is *exactly* this set of players, or null. The + * oldest match wins, so a set that somehow accumulated duplicates keeps resolving to the + * conversation with the history in it. + * + * This is what makes "open a chat with these people" reuse the conversation you already + * have with them rather than starting an empty one each time. Matching is on the whole + * set: a DM and a group that happens to contain those two people are different threads. + * + * Only threads that still have a `message_thread` row can match. Membership rows whose + * thread is gone are ignored rather than resolved to: matching one would hand back an id + * that nothing else in the worker can render, and — since the oldest match wins — it + * would keep winning on every subsequent call. + */ +export async function findThreadWithMembers( + db: D1Database, + playerIds: number[] +): Promise { + const members = [...new Set(playerIds)] + if (members.length === 0) return null + + // ?1 is the member count; ?2… are the ids themselves. + const placeholders = members.map((_, i) => `?${i + 2}`).join(', ') + const row = await db + .prepare( + `SELECT m.chat_thread_id FROM thread_member m + JOIN message_thread t ON t.chat_thread_id = m.chat_thread_id + GROUP BY m.chat_thread_id + HAVING COUNT(*) = ?1 + AND COUNT(CASE WHEN m.player_id IN (${placeholders}) THEN 1 END) = ?1 + ORDER BY m.chat_thread_id + LIMIT 1` + ) + .bind(members.length, ...members) + .first<{ chat_thread_id: number }>() + return row?.chat_thread_id ?? null +} + +/** + * The thread with exactly these members, opening one if it doesn't exist yet. Two + * simultaneous first-messages to the same set can still race into two threads; the + * oldest-match rule in `findThreadWithMembers` means both parties converge on one of + * them afterwards. + */ +export async function getOrCreateThreadWithMembers( + db: D1Database, + playerIds: number[], + startedBy: number +): Promise { + return ( + (await findThreadWithMembers(db, playerIds)) ?? + (await createThread(db, playerIds, null, startedBy)) + ) +} + +/** Everyone in a thread, ordered by id — the fan-out list for a push notification. */ +export async function getThreadMemberIds(db: D1Database, chatThreadId: number): Promise { + const { results } = await db + .prepare('SELECT player_id FROM thread_member WHERE chat_thread_id = ?1 ORDER BY player_id') + .bind(chatThreadId) + .all<{ player_id: number }>() + return results.map((r) => r.player_id) +} + +/** Add a player to an existing thread. A no-op if they're already in it. */ +export async function addThreadMember( + db: D1Database, + chatThreadId: number, + playerId: number +): Promise { + await db + .prepare('INSERT OR IGNORE INTO thread_member (chat_thread_id, player_id) VALUES (?1, ?2)') + .bind(chatThreadId, playerId) + .run() +} + +/** Remove a player from a thread. The thread and its messages outlive the membership. */ +export async function removeThreadMember( + db: D1Database, + chatThreadId: number, + playerId: number +): Promise { + await db + .prepare('DELETE FROM thread_member WHERE chat_thread_id = ?1 AND player_id = ?2') + .bind(chatThreadId, playerId) + .run() +} + +/** + * Post a message and advance the thread's denormalized `latest_message_id` — the only + * way messages should be written, so the thread list never goes stale. Two statements + * rather than a batch, because the update needs the id the insert assigns. + */ +export async function postMessage(db: D1Database, message: NewChatMessage): Promise { + const stored = await insertMessage(db, message) + await db + .prepare('UPDATE message_thread SET latest_message_id = ?2 WHERE chat_thread_id = ?1') + .bind(stored.chatThreadId, stored.chatMessageId) + .run() + return stored +} + +/** + * Advance a player's read position, to a specific message or (with no id) to the whole + * thread. Only ever moves forward: an out-of-order ack from a second client can't walk + * the thread back to unread. + * + * The id is also clamped to the thread's real latest message, so a client acking an id + * that was never stored can't strand the pointer beyond every future message and leave + * the thread permanently "read". + */ +export async function markThreadRead( + db: D1Database, + chatThreadId: number, + playerId: number, + chatMessageId?: number +): Promise { + await db + .prepare( + `UPDATE thread_member + SET last_read_message_id = MAX( + COALESCE(last_read_message_id, 0), + MIN( + COALESCE(?3, (SELECT latest_message_id FROM message_thread WHERE chat_thread_id = ?1), 0), + COALESCE((SELECT latest_message_id FROM message_thread WHERE chat_thread_id = ?1), 0) + ) + ) + WHERE chat_thread_id = ?1 AND player_id = ?2` + ) + .bind(chatThreadId, playerId, chatMessageId ?? null) + .run() +} + +/** Favorite or unfavorite a thread, for one player only. */ +export async function setThreadFavorited( + db: D1Database, + chatThreadId: number, + playerId: number, + isFavorited: boolean +): Promise { + await db + .prepare( + 'UPDATE thread_member SET is_favorited = ?3 WHERE chat_thread_id = ?1 AND player_id = ?2' + ) + .bind(chatThreadId, playerId, isFavorited ? 1 : 0) + .run() +} + +/** Snooze a thread's notifications until an instant, or clear the snooze with null. */ +export async function setThreadSnoozed( + db: D1Database, + chatThreadId: number, + playerId: number, + snoozedUntil: string | null +): Promise { + await db + .prepare( + 'UPDATE thread_member SET snoozed_until = ?3 WHERE chat_thread_id = ?1 AND player_id = ?2' + ) + .bind(chatThreadId, playerId, snoozedUntil) + .run() +} diff --git a/apps/chat/vitest.config.ts b/apps/chat/vitest.config.ts index de0d903..fff0591 100644 --- a/apps/chat/vitest.config.ts +++ b/apps/chat/vitest.config.ts @@ -9,6 +9,40 @@ export default defineConfig({ bindings: { ENVIRONMENT: 'VITEST', }, + // The worker's RECFLARE_NOTIFICATIONS_HUB binding points at the `notify` + // worker's DO (script_name: "notify"). That worker isn't part of this + // isolated test, so provide a minimal stub exposing the same NotificationsHub + // RPC surface. This one also records what it was sent and hands it back via + // `takeSent`, so tests can assert on the ChatMessageReceived fan-out. + workers: [ + { + name: 'notify', + modules: true, + compatibilityDate: '2026-06-16', + compatibilityFlags: ['nodejs_compat'], + durableObjects: { RECFLARE_NOTIFICATIONS_HUB: 'NotificationsHub' }, + script: ` + import { DurableObject } from 'cloudflare:workers' + export class NotificationsHub extends DurableObject { + constructor(ctx, env) { + super(ctx, env) + this.sent = [] + } + async notifyPlayer(playerId, notificationType, data) { + this.sent.push({ playerId, notificationType, data }) + return { delivered: 1, queued: false } + } + async broadcast() { return { delivered: 0 } } + async takeSent() { + const sent = this.sent + this.sent = [] + return sent + } + } + export default { fetch() { return new Response('ok') } } + `, + }, + ], }, }), ], diff --git a/apps/chat/wrangler.jsonc b/apps/chat/wrangler.jsonc index d5a8af2..b4b2a00 100644 --- a/apps/chat/wrangler.jsonc +++ b/apps/chat/wrangler.jsonc @@ -4,6 +4,41 @@ "main": "src/chat.app.ts", "compatibility_date": "2026-06-16", "compatibility_flags": ["nodejs_compat"], + // Shared `recflare` DB (created manually with `wrangler d1 create recflare`; the + // "local" placeholder is spliced out at deploy time). The `chat` worker owns the + // `message` table (schema/migration here); its own migrations_table keeps history + // separate from the other workers' migrations on the shared database. + "d1_databases": [ + { + "binding": "DB", + "database_name": "recflare", + "database_id": "local", + "migrations_dir": "migrations", + "migrations_table": "d1_migrations_chat" + } + ], + "logpush": false, + // Shared Secrets Store holding the HS256 JWT signing key. Every worker binds the + // same store as JWT_SECRET so tokens signed by `auth` verify here. The "local" + // store_id placeholder is replaced with RECFLARE_SECRETS_STORE at deploy time. + "secrets_store_secrets": [ + { + "binding": "JWT_SECRET", + "store_id": "local", + "secret_name": "JWT_SECRET" + } + ], + // The `notify` worker's NotificationsHub DO — used to push ChatMessageReceived to + // every member of a thread when a message lands. + "durable_objects": { + "bindings": [ + { + "name": "RECFLARE_NOTIFICATIONS_HUB", + "class_name": "NotificationsHub", + "script_name": "notify" + } + ] + }, "upload_source_maps": true, "observability": { "logs": {