Add chat HTTP endpoints

app/features/chat/ChatRepository.server.test.ts

@@ -131,6 +131,79 @@ describe("ChatRepository.findAllMessagesByRoomId", () => {
 	});
 });
 
+describe("ChatRepository.findMessageById", () => {
+	test("returns the message with its author resolved", async () => {
+		const room = await ChatRoomFactory.create();
+		const inserted = await ChatMessageFactory.create({
+			roomId: room.id,
+			authorUserId: users.id(1),
+		});
+
+		const message = await ChatRepository.findMessageById(inserted.id);
+
+		expect(message?.roomId).toBe(room.id);
+		expect(message?.author?.id).toBe(users.id(1));
+		expect(message?.author?.username).toBeTruthy();
+	});
+
+	test("returns undefined for an unknown id", async () => {
+		expect(await ChatRepository.findMessageById(424242)).toBeUndefined();
+	});
+});
+
+describe("ChatRepository.findUnreadCountsByRoomIds", () => {
+	test("counts messages newer than the user's read indicator per room", async () => {
+		const room = await ChatRoomFactory.create();
+		const otherRoom = await ChatRoomFactory.create();
+		const [first] = await ChatMessageFactory.createMany(3, {
+			roomId: room.id,
+			authorUserId: users.id(2),
+		});
+		await ChatMessageFactory.create({
+			roomId: otherRoom.id,
+			authorUserId: users.id(2),
+		});
+		await ChatRepository.upsertReadIndicator({
+			userId: users.id(1),
+			roomId: room.id,
+			lastSeenMessageId: first.id,
+		});
+
+		const counts = await ChatRepository.findUnreadCountsByRoomIds(users.id(1), [
+			room.id,
+			otherRoom.id,
+		]);
+
+		expect(counts.sort((a, b) => a.roomId - b.roomId)).toEqual([
+			{ roomId: room.id, unreadCount: 2 },
+			{ roomId: otherRoom.id, unreadCount: 1 },
+		]);
+	});
+
+	test("leaves out rooms with nothing unread", async () => {
+		const room = await ChatRoomFactory.create();
+		const message = await ChatMessageFactory.create({
+			roomId: room.id,
+			authorUserId: users.id(2),
+		});
+		await ChatRepository.upsertReadIndicator({
+			userId: users.id(1),
+			roomId: room.id,
+			lastSeenMessageId: message.id,
+		});
+
+		expect(
+			await ChatRepository.findUnreadCountsByRoomIds(users.id(1), [room.id]),
+		).toEqual([]);
+	});
+
+	test("returns an empty array for no room ids", async () => {
+		expect(
+			await ChatRepository.findUnreadCountsByRoomIds(users.id(1), []),
+		).toEqual([]);
+	});
+});
+
 describe("ChatRepository.upsertReadIndicator", () => {
 	test("creates the indicator on first upsert", async () => {
 		const room = await ChatRoomFactory.create();

app/features/chat/ChatRepository.server.ts

@@ -1,4 +1,4 @@
-import type { Transaction } from "kysely";
+import type { ExpressionBuilder, Transaction } from "kysely";
 import { sql } from "kysely";
 import { db } from "~/db/sql";
 import type { DB, Tables, TablesInsertable } from "~/db/tables";
@@ -20,25 +20,7 @@ export async function findAllMessagesByRoomId(
 ) {
 	const rows = await db
 		.selectFrom("ChatMessage")
-		.select((eb) => [
-			"ChatMessage.id",
-			"ChatMessage.roomId",
-			"ChatMessage.authorUserId",
-			"ChatMessage.type",
-			"ChatMessage.contents",
-			"ChatMessage.publicId",
-			"ChatMessage.createdAt",
-			jsonObjectFrom(
-				eb
-					.selectFrom("User")
-					.select((userEb) => [
-						...commonUserSelect(userEb),
-						"User.pronouns",
-						userChatNameHue,
-					])
-					.whereRef("User.id", "=", "ChatMessage.authorUserId"),
-			).as("author"),
-		])
+		.select(messageWithAuthorSelect)
 		.where("ChatMessage.roomId", "=", roomId)
 		.orderBy("ChatMessage.id", "desc")
 		.limit(limit)
@@ -47,6 +29,15 @@ export async function findAllMessagesByRoomId(
 	return rows.reverse();
 }
 
+/** Returns a message with its author resolved live. */
+export function findMessageById(messageId: number) {
+	return db
+		.selectFrom("ChatMessage")
+		.select(messageWithAuthorSelect)
+		.where("ChatMessage.id", "=", messageId)
+		.executeTakeFirst();
+}
+
 /** Inserts a chat room, returning the row. Called in the owning entity's insert transaction. */
 export function insertRoom(
 	args: { type: Tables["ChatRoom"]["type"]; expiresAt: Date },
@@ -153,6 +144,36 @@ export async function upsertReadIndicator(
 		.execute();
 }
 
+/** Unread message counts per room: how many messages are newer than the user's read indicator. Rooms with nothing unread are left out. */
+export async function findUnreadCountsByRoomIds(
+	userId: number,
+	roomIds: number[],
+): Promise<Array<{ roomId: number; unreadCount: number }>> {
+	if (roomIds.length === 0) return [];
+
+	return db
+		.selectFrom("ChatMessage")
+		.leftJoin("ChatMessageReadIndicator", (join) =>
+			join
+				.onRef("ChatMessageReadIndicator.roomId", "=", "ChatMessage.roomId")
+				.on("ChatMessageReadIndicator.userId", "=", userId),
+		)
+		.select(({ fn }) => [
+			"ChatMessage.roomId",
+			fn.countAll<number>().as("unreadCount"),
+		])
+		.where("ChatMessage.roomId", "in", roomIds)
+		.where(({ eb, fn, val }) =>
+			eb(
+				"ChatMessage.id",
+				">",
+				fn.coalesce("ChatMessageReadIndicator.lastSeenMessageId", val(0)),
+			),
+		)
+		.groupBy("ChatMessage.roomId")
+		.execute();
+}
+
 /** Closes rooms whose expiry is before `expiredBefore`, returning how many. Messages are kept; access narrows. */
 export async function closeExpiredRooms(expiredBefore: Date) {
 	const result = await db
@@ -164,3 +185,25 @@ export async function closeExpiredRooms(expiredBefore: Date) {
 
 	return Number(result.numUpdatedRows);
 }
+
+function messageWithAuthorSelect(eb: ExpressionBuilder<DB, "ChatMessage">) {
+	return [
+		"ChatMessage.id",
+		"ChatMessage.roomId",
+		"ChatMessage.authorUserId",
+		"ChatMessage.type",
+		"ChatMessage.contents",
+		"ChatMessage.publicId",
+		"ChatMessage.createdAt",
+		jsonObjectFrom(
+			eb
+				.selectFrom("User")
+				.select((userEb) => [
+					...commonUserSelect(userEb),
+					"User.pronouns",
+					userChatNameHue,
+				])
+				.whereRef("User.id", "=", "ChatMessage.authorUserId"),
+		).as("author"),
+	] as const;
+}

app/features/chat/ChatRoomResolver.server.test.ts

@@ -2,7 +2,6 @@ import { addHours } from "date-fns";
 import { beforeEach, describe, expect, test } from "vitest";
 import * as ScrimPostFactory from "~/db/seed/factories/ScrimPostFactory";
 import * as SQGroupFactory from "~/db/seed/factories/SQGroupFactory";
-import * as SQMatchFactory from "~/db/seed/factories/SQMatchFactory";
 import * as TeamFactory from "~/db/seed/factories/TeamFactory";
 import * as TournamentFactory from "~/db/seed/factories/TournamentFactory";
 import * as TournamentTeamFactory from "~/db/seed/factories/TournamentTeamFactory";
@@ -13,6 +12,7 @@ import * as TournamentRepository from "~/features/tournament/TournamentRepositor
 import { dateToDatabaseTimestamp } from "~/utils/dates";
 import * as ChatRepository from "./ChatRepository.server";
 import * as ChatRoomResolver from "./ChatRoomResolver.server";
+import { setupSqMatch } from "./tests/fixtures";
 
 const users = UserFactory.pool();
 
@@ -24,15 +24,6 @@ beforeEach(async () => {
 	await users.create(12);
 });
 
-const setupSqMatch = async () => {
-	const alphaUserIds = [users.id(2), users.id(3), users.id(4), users.id(5)];
-	const bravoUserIds = [users.id(6), users.id(7), users.id(8), users.id(9)];
-
-	const match = await SQMatchFactory.create({ alphaUserIds, bravoUserIds });
-
-	return { match, alphaUserIds, bravoUserIds };
-};
-
 const setupStartedTournamentMatch = async () => {
 	const authorId = users.id(2);
 	const teamAlphaUserIds = [users.id(3), users.id(4), users.id(5), users.id(6)];
@@ -126,7 +117,7 @@ describe("ChatRoomResolver.resolve", () => {
 	});
 
 	test("resolves an SQ_MATCH room to both groups' members", async () => {
-		const { match, alphaUserIds, bravoUserIds } = await setupSqMatch();
+		const { match, alphaUserIds, bravoUserIds } = await setupSqMatch(users);
 
 		const [room] = await ChatRoomResolver.resolve([match.chatRoomId!]);
 
@@ -220,7 +211,7 @@ describe("ChatRoomResolver.resolve", () => {
 
 describe("ChatRoomResolver.findAllByUserId", () => {
 	test("returns the member's own group and match rooms of an SQ match", async () => {
-		const { match, alphaUserIds } = await setupSqMatch();
+		const { match, alphaUserIds } = await setupSqMatch(users);
 
 		const rooms = await ChatRoomResolver.findAllByUserId(alphaUserIds[0]);
 
@@ -256,7 +247,7 @@ describe("ChatRoomResolver.findAllByUserId", () => {
 	});
 
 	test("returns nothing for a non-participant", async () => {
-		await setupSqMatch();
+		await setupSqMatch(users);
 		await setupAcceptedScrim();
 
 		expect(await ChatRoomResolver.findAllByUserId(outsiderId())).toEqual([]);
@@ -274,7 +265,7 @@ describe("ChatRoomResolver.findAllByUserId", () => {
 
 describe("ChatRoomResolver.canObserve", () => {
 	test("site staff can observe both group chats of an SQ match", async () => {
-		const { match } = await setupSqMatch();
+		const { match } = await setupSqMatch(users);
 
 		const rooms = await ChatRoomResolver.resolve([
 			match.chatRoomId!,
@@ -290,7 +281,7 @@ describe("ChatRoomResolver.canObserve", () => {
 	});
 
 	test("a participant of one group cannot view the other group's chat", async () => {
-		const { match, alphaUserIds } = await setupSqMatch();
+		const { match, alphaUserIds } = await setupSqMatch(users);
 
 		const [bravoRoom] = await ChatRoomResolver.resolve([
 			await groupChatRoomId(match.bravoGroup.id),

app/features/chat/ChatSystemMessage.server.test.ts

@@ -1,5 +1,6 @@
 import { afterEach, beforeEach, describe, expect, test, vi } from "vitest";
 import * as EventBus from "~/features/events/core/EventBus.server";
+import { abortSubscriptions, flushEvents, subscribeTo } from "./tests/fixtures";
 
 process.env.SKALOP_SYSTEM_MESSAGE_URL = "http://skalop.test";
 process.env.SKALOP_TOKEN = "test-token";
@@ -10,8 +11,6 @@ const fetchMock = vi.fn(
 	async (_input: unknown, _init?: { body?: string }) => new Response(null),
 );
 
-const abortControllers: AbortController[] = [];
-
 beforeEach(() => {
 	vi.stubGlobal("fetch", fetchMock);
 });
@@ -19,30 +18,9 @@ beforeEach(() => {
 afterEach(() => {
 	vi.unstubAllGlobals();
 	fetchMock.mockClear();
-	for (const controller of abortControllers) {
-		controller.abort();
-	}
-	abortControllers.length = 0;
+	abortSubscriptions();
 });
 
-function subscribeTo(channel: string) {
-	const controller = new AbortController();
-	abortControllers.push(controller);
-
-	const received: EventBus.ServerEvent[] = [];
-	void (async () => {
-		for await (const event of EventBus.subscribe(
-			[channel],
-			controller.signal,
-		)) {
-			received.push(event);
-		}
-	})();
-	return received;
-}
-
-const flush = () => new Promise<void>((resolve) => setTimeout(resolve));
-
 describe("ChatSystemMessage.send", () => {
 	test("publishes a revalidate broadcast to its topic channel", async () => {
 		const received = subscribeTo("tournament__101");
@@ -53,7 +31,7 @@ describe("ChatSystemMessage.send", () => {
 			revalidateScope: "MATCH_RESULTS",
 			authorUserId: 5,
 		});
-		await flush();
+		await flushEvents();
 
 		expect(received).toEqual([
 			{ kind: "revalidate", scope: "MATCH_RESULTS", authorUserId: 5 },
@@ -76,7 +54,7 @@ describe("ChatSystemMessage.send", () => {
 			revalidateOnly: true,
 			authorUserId: 5,
 		});
-		await flush();
+		await flushEvents();
 
 		expect(received).toEqual([
 			{ kind: "revalidate", authorUserId: 5, type: "READY_CHECK_STARTED" },
@@ -92,7 +70,7 @@ describe("ChatSystemMessage.send", () => {
 			type: "TOURNAMENT_UPDATED",
 			revalidateOnly: true,
 		});
-		await flush();
+		await flushEvents();
 
 		expect(received).toEqual([{ kind: "revalidate" }]);
 	});
@@ -102,7 +80,7 @@ describe("ChatSystemMessage.send", () => {
 
 		ChatSystemMessage.send({ room: "tournament__104", revalidateOnly: true });
 		ChatSystemMessage.send({ room: "tournament__104", revalidateOnly: true });
-		await flush();
+		await flushEvents();
 
 		expect(received).toHaveLength(1);
 	});
@@ -115,7 +93,7 @@ describe("ChatSystemMessage.send", () => {
 			type: "USER_LEFT",
 			context: { name: "Sendou" },
 		});
-		await flush();
+		await flushEvents();
 
 		expect(received).toEqual([]);
 		expect(fetchMock).toHaveBeenCalledTimes(1);
@@ -131,7 +109,7 @@ describe("ChatSystemMessage.notifyNotificationsChanged", () => {
 		const bravo = subscribeTo(EventBus.userChannel(2));
 
 		ChatSystemMessage.notifyNotificationsChanged([1, 2]);
-		await flush();
+		await flushEvents();
 
 		expect(alpha).toEqual([{ kind: "notificationsChanged" }]);
 		expect(bravo).toEqual([{ kind: "notificationsChanged" }]);

app/features/chat/chat-schemas.ts

@@ -0,0 +1,9 @@
+import * as v from "valibot";
+import { hidden, textField } from "~/form/fields";
+import { SHORT_NANOID_LENGTH } from "~/utils/id";
+import { MESSAGE_MAX_LENGTH } from "./chat-constants";
+
+export const sendChatMessageSchema = v.object({
+	publicId: hidden(v.pipe(v.string(), v.length(SHORT_NANOID_LENGTH))),
+	contents: textField({ maxLength: MESSAGE_MAX_LENGTH }),
+});

app/features/chat/routes/api.chat.rooms.$id.messages.ts

@@ -0,0 +1,22 @@
+import type { LoaderFunctionArgs } from "react-router";
+import { requireUser } from "~/features/auth/core/user.server";
+import {
+	forbidden,
+	notFoundIfNullish,
+	parseParams,
+} from "~/utils/remix.server";
+import { idObject } from "~/utils/schema";
+import * as ChatRepository from "../ChatRepository.server";
+import * as ChatRoomResolver from "../ChatRoomResolver.server";
+
+export const loader = async ({ params }: LoaderFunctionArgs) => {
+	const user = requireUser();
+	const { id: roomId } = parseParams({ params, schema: idObject });
+
+	const room = notFoundIfNullish((await ChatRoomResolver.resolve([roomId]))[0]);
+	if (!ChatRoomResolver.canView(room, user.id)) {
+		forbidden();
+	}
+
+	return { messages: await ChatRepository.findAllMessagesByRoomId(roomId) };
+};

app/features/chat/routes/api.chat.rooms.ts

@@ -0,0 +1,34 @@
+import { requireUser } from "~/features/auth/core/user.server";
+import * as ChatRepository from "../ChatRepository.server";
+import * as ChatRoomResolver from "../ChatRoomResolver.server";
+
+/**
+ * The user's open chat rooms with server-computed unread counts. A background
+ * resource like the notifications peek: fetched after mount and refetched on
+ * `chatMessage` / `roomsChanged` events instead of riding any page loader.
+ */
+export const loader = async () => {
+	const user = requireUser();
+
+	const rooms = await ChatRoomResolver.findAllByUserId(user.id);
+	const unreadCounts = await ChatRepository.findUnreadCountsByRoomIds(
+		user.id,
+		rooms.map((room) => room.roomId),
+	);
+	const unreadCountByRoomId = new Map(
+		unreadCounts.map((row) => [row.roomId, row.unreadCount]),
+	);
+
+	return {
+		rooms: rooms.map((room) => ({
+			id: room.roomId,
+			type: room.type,
+			titleParams: room.titleParams,
+			url: room.url,
+			imageUrl: room.imageUrl,
+			participantUserIds: room.participantUserIds,
+			expiresAt: room.expiresAt,
+			unreadCount: unreadCountByRoomId.get(room.roomId) ?? 0,
+		})),
+	};
+};

app/features/chat/routes/chat-routes.test.ts

@@ -0,0 +1,350 @@
+import { subHours } from "date-fns";
+import type { ActionFunctionArgs, LoaderFunctionArgs } from "react-router";
+import { afterEach, beforeEach, describe, expect, test } from "vitest";
+import * as UserFactory from "~/db/seed/factories/UserFactory";
+import { db } from "~/db/sql";
+import * as EventBus from "~/features/events/core/EventBus.server";
+import { withUserId } from "~/utils/Test";
+import * as ChatRepository from "../ChatRepository.server";
+import {
+	abortSubscriptions,
+	flushEvents,
+	setupSqMatch,
+	subscribeTo,
+} from "../tests/fixtures";
+import { loader as roomsLoader } from "./api.chat.rooms";
+import { loader as messagesLoader } from "./api.chat.rooms.$id.messages";
+import { action as sendAction } from "./chat.$roomId.messages";
+import { action as readAction } from "./chat.$roomId.read";
+
+const users = UserFactory.pool();
+
+// ADMIN_ID is 1 under NODE_ENV=test, so the first pool user is site staff
+const adminId = () => users.id(1);
+const outsiderId = () => users.id(11);
+
+beforeEach(async () => {
+	await users.create(11);
+});
+
+afterEach(() => {
+	abortSubscriptions();
+});
+
+describe("chat messages action", () => {
+	test("inserts the message and returns it with the author resolved", async () => {
+		const { match, alphaUserIds } = await setupSqMatch(users);
+
+		const message = await sendMessageOk(alphaUserIds[0], match.chatRoomId!, {
+			publicId: "aaaaaaaaaa",
+			contents: "hello",
+		});
+
+		expect(message.roomId).toBe(match.chatRoomId);
+		expect(message.contents).toBe("hello");
+		expect(message.authorUserId).toBe(alphaUserIds[0]);
+		expect(message.author?.username).toEqual(expect.any(String));
+	});
+
+	test("publishes the message to participant user channels and the room channel", async () => {
+		const { match, alphaUserIds, bravoUserIds } = await setupSqMatch(users);
+		const bravoReceived = subscribeTo(EventBus.userChannel(bravoUserIds[0]));
+		const roomReceived = subscribeTo(
+			EventBus.chatRoomChannel(match.chatRoomId!),
+		);
+
+		const message = await sendMessageOk(alphaUserIds[0], match.chatRoomId!, {
+			publicId: "bbbbbbbbbb",
+			contents: "hello",
+		});
+		await flushEvents();
+
+		expect(bravoReceived).toEqual([
+			{ kind: "chatMessage", roomId: match.chatRoomId, message },
+		]);
+		expect(roomReceived).toEqual([
+			{ kind: "chatMessage", roomId: match.chatRoomId, message },
+		]);
+	});
+
+	test("retried send with the same publicId returns the existing message without publishing a duplicate row", async () => {
+		const { match, alphaUserIds } = await setupSqMatch(users);
+
+		const first = await sendMessageOk(alphaUserIds[0], match.chatRoomId!, {
+			publicId: "cccccccccc",
+			contents: "hello",
+		});
+		const retried = await sendMessageOk(alphaUserIds[0], match.chatRoomId!, {
+			publicId: "cccccccccc",
+			contents: "hello again",
+		});
+
+		expect(retried.id).toBe(first.id);
+		expect(retried.contents).toBe("hello");
+
+		const count = await db
+			.selectFrom("ChatMessage")
+			.select(({ fn }) => fn.countAll<number>().as("count"))
+			.where("ChatMessage.roomId", "=", match.chatRoomId!)
+			.executeTakeFirstOrThrow();
+		expect(count.count).toBe(1);
+	});
+
+	test("returns field errors for empty contents", async () => {
+		const { match, alphaUserIds } = await setupSqMatch(users);
+
+		const result = await sendMessage(alphaUserIds[0], match.chatRoomId!, {
+			publicId: "dddddddddd",
+			contents: "",
+		});
+
+		expect(result).toHaveProperty("fieldErrors");
+	});
+
+	test.each([
+		{ why: "a non-participant", userId: () => outsiderId() },
+		{ why: "a site staff observer", userId: () => adminId() },
+	])("403s $why", async ({ userId }) => {
+		const { match } = await setupSqMatch(users);
+
+		expect(
+			await statusOf(
+				sendMessage(userId(), match.chatRoomId!, {
+					publicId: "eeeeeeeeee",
+					contents: "hello",
+				}),
+			),
+		).toBe(403);
+	});
+
+	test("403s a participant once the room has expired", async () => {
+		const { match, alphaUserIds } = await setupSqMatch(users);
+		await ChatRepository.updateRoomExpiresAt({
+			roomId: match.chatRoomId!,
+			expiresAt: subHours(new Date(), 1),
+		});
+
+		expect(
+			await statusOf(
+				sendMessage(alphaUserIds[0], match.chatRoomId!, {
+					publicId: "ffffffffff",
+					contents: "hello",
+				}),
+			),
+		).toBe(403);
+	});
+
+	test("400s a publicId already claimed by another user's message", async () => {
+		const { match, alphaUserIds, bravoUserIds } = await setupSqMatch(users);
+		await sendMessageOk(alphaUserIds[0], match.chatRoomId!, {
+			publicId: "gggggggggg",
+			contents: "hello",
+		});
+
+		expect(
+			await statusOf(
+				sendMessage(bravoUserIds[0], match.chatRoomId!, {
+					publicId: "gggggggggg",
+					contents: "not mine",
+				}),
+			),
+		).toBe(400);
+	});
+
+	test("404s a room id with no owner", async () => {
+		expect(
+			await statusOf(
+				sendMessage(outsiderId(), 424242, {
+					publicId: "hhhhhhhhhh",
+					contents: "hello",
+				}),
+			),
+		).toBe(404);
+	});
+});
+
+describe("chat rooms loader", () => {
+	test("returns the user's rooms with server-computed unread counts", async () => {
+		const { match, alphaUserIds, bravoUserIds } = await setupSqMatch(users);
+		await sendMessageOk(alphaUserIds[0], match.chatRoomId!, {
+			publicId: "iiiiiiiiii",
+			contents: "hello",
+		});
+
+		const data = await loadRooms(bravoUserIds[0]);
+		const matchRoom = data.rooms.find((room) => room.id === match.chatRoomId);
+
+		expect(matchRoom).toMatchObject({
+			type: "SQ_MATCH",
+			unreadCount: 1,
+			url: expect.stringContaining(String(match.id)),
+		});
+		expect(matchRoom?.participantUserIds).toHaveLength(8);
+	});
+
+	test("does not count the sender's own message as unread on their other devices", async () => {
+		const { match, alphaUserIds } = await setupSqMatch(users);
+		await sendMessageOk(alphaUserIds[0], match.chatRoomId!, {
+			publicId: "jjjjjjjjjj",
+			contents: "hello",
+		});
+
+		const data = await loadRooms(alphaUserIds[0]);
+		const matchRoom = data.rooms.find((room) => room.id === match.chatRoomId);
+
+		expect(matchRoom?.unreadCount).toBe(0);
+	});
+});
+
+describe("chat read action", () => {
+	test("marking read clears the room's unread count", async () => {
+		const { match, alphaUserIds, bravoUserIds } = await setupSqMatch(users);
+		const message = await sendMessageOk(alphaUserIds[0], match.chatRoomId!, {
+			publicId: "kkkkkkkkkk",
+			contents: "hello",
+		});
+
+		await markRead(bravoUserIds[0], match.chatRoomId!, message.id);
+
+		const data = await loadRooms(bravoUserIds[0]);
+		const matchRoom = data.rooms.find((room) => room.id === match.chatRoomId);
+		expect(matchRoom?.unreadCount).toBe(0);
+	});
+
+	test("403s a non-participant", async () => {
+		const { match } = await setupSqMatch(users);
+
+		expect(await statusOf(markRead(outsiderId(), match.chatRoomId!, 1))).toBe(
+			403,
+		);
+	});
+});
+
+describe("chat room messages loader", () => {
+	test("returns the room's messages oldest first for a participant", async () => {
+		const { match, alphaUserIds, bravoUserIds } = await setupSqMatch(users);
+		await sendMessageOk(alphaUserIds[0], match.chatRoomId!, {
+			publicId: "llllllllll",
+			contents: "first",
+		});
+		await sendMessageOk(bravoUserIds[0], match.chatRoomId!, {
+			publicId: "mmmmmmmmmm",
+			contents: "second",
+		});
+
+		const data = await loadMessages(alphaUserIds[1], match.chatRoomId!);
+
+		expect(data.messages.map((message) => message.contents)).toEqual([
+			"first",
+			"second",
+		]);
+		expect(data.messages[0].author?.username).toEqual(expect.any(String));
+	});
+
+	test("site staff observer can read the room", async () => {
+		const { match, alphaUserIds } = await setupSqMatch(users);
+		await sendMessageOk(alphaUserIds[0], match.chatRoomId!, {
+			publicId: "nnnnnnnnnn",
+			contents: "hello",
+		});
+
+		const data = await loadMessages(adminId(), match.chatRoomId!);
+
+		expect(data.messages).toHaveLength(1);
+	});
+
+	test("403s a non-participant", async () => {
+		const { match } = await setupSqMatch(users);
+
+		expect(await statusOf(loadMessages(outsiderId(), match.chatRoomId!))).toBe(
+			403,
+		);
+	});
+
+	test("404s an unknown room", async () => {
+		expect(await statusOf(loadMessages(outsiderId(), 424242))).toBe(404);
+	});
+});
+
+function sendMessage(
+	userId: number,
+	roomId: number,
+	body: Record<string, unknown>,
+) {
+	const request = new Request(`http://app.com/chat/${roomId}/messages`, {
+		method: "POST",
+		body: JSON.stringify(body),
+		headers: [["Content-Type", "application/json"]],
+	});
+
+	return withUserId(userId, () =>
+		sendAction({
+			request,
+			params: { roomId: String(roomId) },
+			context: {} as any,
+			pattern: "",
+			url: new URL(request.url),
+		} as ActionFunctionArgs),
+	);
+}
+
+async function sendMessageOk(
+	userId: number,
+	roomId: number,
+	body: Record<string, unknown>,
+) {
+	const result = await sendMessage(userId, roomId, body);
+	if ("fieldErrors" in result) {
+		throw new Error(`send failed: ${JSON.stringify(result.fieldErrors)}`);
+	}
+
+	return result.message;
+}
+
+function markRead(userId: number, roomId: number, lastSeenMessageId: number) {
+	const request = new Request(`http://app.com/chat/${roomId}/read`, {
+		method: "POST",
+		body: JSON.stringify({ lastSeenMessageId }),
+		headers: [["Content-Type", "application/json"]],
+	});
+
+	return withUserId(userId, () =>
+		readAction({
+			request,
+			params: { roomId: String(roomId) },
+			context: {} as any,
+			pattern: "",
+			url: new URL(request.url),
+		} as ActionFunctionArgs),
+	);
+}
+
+function loadRooms(userId: number) {
+	return withUserId(userId, () => roomsLoader());
+}
+
+function loadMessages(userId: number, roomId: number) {
+	const request = new Request(
+		`http://app.com/api/chat/rooms/${roomId}/messages`,
+	);
+
+	return withUserId(userId, () =>
+		messagesLoader({
+			request,
+			params: { id: String(roomId) },
+			context: {} as any,
+			pattern: "",
+			url: new URL(request.url),
+		} as LoaderFunctionArgs),
+	);
+}
+
+async function statusOf(promise: Promise<unknown>) {
+	try {
+		await promise;
+		return 200;
+	} catch (thrown) {
+		if (thrown instanceof Response) return thrown.status;
+		throw thrown;
+	}
+}

app/features/chat/routes/chat.$roomId.messages.ts

@@ -0,0 +1,71 @@
+import type { ActionFunctionArgs } from "react-router";
+import * as v from "valibot";
+import { requireUser } from "~/features/auth/core/user.server";
+import * as EventBus from "~/features/events/core/EventBus.server";
+import { parseFormData } from "~/form/parse.server";
+import invariant from "~/utils/invariant";
+import {
+	badRequestIfFalsy,
+	forbidden,
+	notFoundIfNullish,
+	parseParams,
+} from "~/utils/remix.server";
+import { id } from "~/utils/schema";
+import * as ChatRepository from "../ChatRepository.server";
+import * as ChatRoomResolver from "../ChatRoomResolver.server";
+import { sendChatMessageSchema } from "../chat-schemas";
+
+const paramsSchema = v.object({ roomId: id });
+
+export const action = async ({ request, params }: ActionFunctionArgs) => {
+	const user = requireUser();
+	const { roomId } = parseParams({ params, schema: paramsSchema });
+
+	const room = notFoundIfNullish((await ChatRoomResolver.resolve([roomId]))[0]);
+	if (!ChatRoomResolver.canPost(room, user.id)) {
+		forbidden();
+	}
+
+	const result = await parseFormData({
+		request,
+		schema: sendChatMessageSchema,
+	});
+	if (!result.success) {
+		return { fieldErrors: result.fieldErrors };
+	}
+
+	const inserted = await ChatRepository.insertMessage({
+		roomId,
+		authorUserId: user.id,
+		contents: result.data.contents,
+		publicId: result.data.publicId,
+	});
+	// a publicId clash with another room's or user's message is not a retry of
+	// this send, and echoing the clashing row would leak it
+	badRequestIfFalsy(
+		inserted.roomId === roomId && inserted.authorUserId === user.id
+			? inserted
+			: null,
+	);
+
+	// the sender has the room open, so their own message never counts as unread
+	// on their other devices
+	await ChatRepository.upsertReadIndicator({
+		userId: user.id,
+		roomId,
+		lastSeenMessageId: inserted.id,
+	});
+
+	const message = await ChatRepository.findMessageById(inserted.id);
+	invariant(message, "inserted chat message not found");
+
+	EventBus.publish(
+		[
+			...room.participantUserIds.map(EventBus.userChannel),
+			EventBus.chatRoomChannel(roomId),
+		],
+		{ kind: "chatMessage", roomId, message },
+	);
+
+	return { message };
+};

app/features/chat/routes/chat.$roomId.read.ts

@@ -0,0 +1,34 @@
+import type { ActionFunctionArgs } from "react-router";
+import * as v from "valibot";
+import { requireUser } from "~/features/auth/core/user.server";
+import {
+	forbidden,
+	notFoundIfNullish,
+	parseBody,
+	parseParams,
+} from "~/utils/remix.server";
+import { id } from "~/utils/schema";
+import * as ChatRepository from "../ChatRepository.server";
+import * as ChatRoomResolver from "../ChatRoomResolver.server";
+
+const paramsSchema = v.object({ roomId: id });
+const bodySchema = v.object({ lastSeenMessageId: id });
+
+export const action = async ({ request, params }: ActionFunctionArgs) => {
+	const user = requireUser();
+	const { roomId } = parseParams({ params, schema: paramsSchema });
+	const data = await parseBody({ request, schema: bodySchema });
+
+	const room = notFoundIfNullish((await ChatRoomResolver.resolve([roomId]))[0]);
+	if (!ChatRoomResolver.canView(room, user.id)) {
+		forbidden();
+	}
+
+	await ChatRepository.upsertReadIndicator({
+		userId: user.id,
+		roomId,
+		lastSeenMessageId: data.lastSeenMessageId,
+	});
+
+	return null;
+};

app/features/chat/tests/fixtures.ts

@@ -0,0 +1,45 @@
+import * as SQMatchFactory from "~/db/seed/factories/SQMatchFactory";
+import type * as UserFactory from "~/db/seed/factories/UserFactory";
+import * as EventBus from "~/features/events/core/EventBus.server";
+
+/** SendouQ match between pool users 2-5 (alpha) and 6-9 (bravo), the owner of the chat rooms the tests exercise. */
+export async function setupSqMatch(users: ReturnType<typeof UserFactory.pool>) {
+	const alphaUserIds = [users.id(2), users.id(3), users.id(4), users.id(5)];
+	const bravoUserIds = [users.id(6), users.id(7), users.id(8), users.id(9)];
+
+	const match = await SQMatchFactory.create({ alphaUserIds, bravoUserIds });
+
+	return { match, alphaUserIds, bravoUserIds };
+}
+
+const abortControllers: AbortController[] = [];
+
+/** Collects the channel's published events into the returned array; release with {@link abortSubscriptions} in `afterEach`. */
+export function subscribeTo(channel: string) {
+	const controller = new AbortController();
+	abortControllers.push(controller);
+
+	const received: EventBus.ServerEvent[] = [];
+	void (async () => {
+		for await (const event of EventBus.subscribe(
+			[channel],
+			controller.signal,
+		)) {
+			received.push(event);
+		}
+	})();
+	return received;
+}
+
+/** Aborts every subscription opened via {@link subscribeTo}. */
+export function abortSubscriptions() {
+	for (const controller of abortControllers) {
+		controller.abort();
+	}
+	abortControllers.length = 0;
+}
+
+/** Lets queued EventBus deliveries drain. */
+export function flushEvents() {
+	return new Promise<void>((resolve) => setTimeout(resolve));
+}

app/routes.ts

@@ -49,6 +49,12 @@ export default [
 		"features/events/routes/sse.$connectionId.topics.ts",
 	),
 
+	route(
+		"/chat/:roomId/messages",
+		"features/chat/routes/chat.$roomId.messages.ts",
+	),
+	route("/chat/:roomId/read", "features/chat/routes/chat.$roomId.read.ts"),
+
 	route("/notifications", "features/notifications/routes/notifications.tsx"),
 	route(
 		"/notifications/seen",
@@ -323,6 +329,11 @@ export default [
 	route("/admin", "features/admin/routes/admin.tsx"),
 	route("/admin/streams", "features/admin/routes/admin.streams.tsx"),
 	route("/api/chat-users", "features/chat/routes/api.chat-users.ts"),
+	route("/api/chat/rooms", "features/chat/routes/api.chat.rooms.ts"),
+	route(
+		"/api/chat/rooms/:id/messages",
+		"features/chat/routes/api.chat.rooms.$id.messages.ts",
+	),
 	route("/api/layout", "features/layout/routes/api.layout.ts"),
 	route(
 		"/api/notifications",

scripts/benchmark-db/cases.ts

@@ -238,6 +238,15 @@ export function buildCases(fx: Fixtures): {
 	add("ChatRepository.findAllMessagesByRoomId", fx.heavyChatRoomId, (roomId) =>
 		ChatRepository.findAllMessagesByRoomId(roomId),
 	);
+	add("ChatRepository.findMessageById", fx.heavyChatMessageId, (messageId) =>
+		ChatRepository.findMessageById(messageId),
+	);
+	add(
+		"ChatRepository.findUnreadCountsByRoomIds",
+		both(fx.heavyUser, fx.heavyChatRoomId),
+		([user, roomId]) =>
+			ChatRepository.findUnreadCountsByRoomIds(user.id, [roomId]),
+	);
 
 	// ChatRoomResolver
 	add("ChatRoomResolver.resolve", fx.heavyChatRoomId, (roomId) =>

scripts/benchmark-db/fixtures.ts

@@ -43,6 +43,8 @@ export interface Fixtures {
 	heavyCalendarEventId: number | null;
 	/** Chat room with the most messages. Null until the prod copy has post-migration chat data. */
 	heavyChatRoomId: number | null;
+	/** Newest chat message. Null until the prod copy has post-migration chat data. */
+	heavyChatMessageId: number | null;
 	resultsEventId: number | null;
 	calendarAuthorId: number | null;
 	calendarWindow: { startTime: Date; endTime: Date } | null;
@@ -153,6 +155,7 @@ export async function resolveFixtures(): Promise<Fixtures> {
 		heavyTeam: await resolveHeavyTeam(),
 		heavyCalendarEventId: await resolveHeavyCalendarEventId(),
 		heavyChatRoomId: await resolveHeavyChatRoomId(),
+		heavyChatMessageId: await resolveHeavyChatMessageId(),
 		resultsEventId: await resolveResultsEventId(),
 		calendarAuthorId: await resolveCalendarAuthorId(),
 		calendarWindow: await resolveCalendarWindow(),
@@ -610,6 +613,15 @@ async function resolveHeavyChatRoomId() {
 	return row?.roomId ?? null;
 }
 
+async function resolveHeavyChatMessageId() {
+	const row = await db
+		.selectFrom("ChatMessage")
+		.select(({ fn }) => fn.max("ChatMessage.id").as("id"))
+		.executeTakeFirst();
+
+	return row?.id ?? null;
+}
+
 async function resolveCalendarWindow() {
 	const row = await db
 		.selectFrom("CalendarEventDate")