Add ChatRepository and chat factories

app/db/seed/factories/ChatMessageFactory.ts

@@ -0,0 +1,13 @@
+import * as ChatRepository from "~/features/chat/ChatRepository.server";
+import { defineFactory } from "../core/defineFactory";
+import { faker } from "../core/faker";
+
+/** Creates user-sent chat messages. System messages are inserted through the repository directly. */
+export const { create, createMany } = defineFactory({
+	defaults: ({ seq }) => ({
+		contents: faker.lorem.sentence(),
+		publicId: `seed-msg-${seq}`,
+	}),
+	insert: (args: Parameters<typeof ChatRepository.insertMessage>[0]) =>
+		ChatRepository.insertMessage(args),
+});

app/db/seed/factories/ChatRoomFactory.ts

@@ -0,0 +1,13 @@
+import { addHours } from "date-fns";
+import * as ChatRepository from "~/features/chat/ChatRepository.server";
+import { defineFactory } from "../core/defineFactory";
+
+/** Creates chat rooms. Owner rows point at these via their `chatRoomId`. */
+export const { create } = defineFactory({
+	defaults: () => ({
+		type: "SQ_MATCH" as const,
+		expiresAt: addHours(new Date(), 12),
+	}),
+	insert: (args: Parameters<typeof ChatRepository.insertRoom>[0]) =>
+		ChatRepository.insertRoom(args),
+});

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

@@ -0,0 +1,221 @@
+import { addDays, subDays } from "date-fns";
+import { beforeEach, describe, expect, test } from "vitest";
+import * as ChatMessageFactory from "~/db/seed/factories/ChatMessageFactory";
+import * as ChatRoomFactory from "~/db/seed/factories/ChatRoomFactory";
+import * as UserFactory from "~/db/seed/factories/UserFactory";
+import { db } from "~/db/sql";
+import * as ChatRepository from "./ChatRepository.server";
+
+const users = UserFactory.pool();
+
+beforeEach(async () => {
+	await users.create(2);
+});
+
+describe("ChatRepository.insertMessage", () => {
+	test("returns the inserted row", async () => {
+		const room = await ChatRoomFactory.create();
+
+		const message = await ChatRepository.insertMessage({
+			roomId: room.id,
+			authorUserId: users.id(1),
+			contents: "hello",
+			publicId: "abc123",
+		});
+
+		expect(message.roomId).toBe(room.id);
+		expect(message.authorUserId).toBe(users.id(1));
+		expect(message.contents).toBe("hello");
+		expect(message.type).toBeNull();
+	});
+
+	test("retried insert with the same publicId returns the existing row without double-inserting", async () => {
+		const room = await ChatRoomFactory.create();
+
+		const first = await ChatRepository.insertMessage({
+			roomId: room.id,
+			authorUserId: users.id(1),
+			contents: "hello",
+			publicId: "abc123",
+		});
+		const retried = await ChatRepository.insertMessage({
+			roomId: room.id,
+			authorUserId: users.id(1),
+			contents: "hello again",
+			publicId: "abc123",
+		});
+
+		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", "=", room.id)
+			.executeTakeFirstOrThrow();
+		expect(count.count).toBe(1);
+	});
+});
+
+describe("ChatRepository.insertSystemMessage", () => {
+	test("inserts a typed message with no contents", async () => {
+		const room = await ChatRoomFactory.create();
+
+		const message = await ChatRepository.insertSystemMessage({
+			roomId: room.id,
+			type: "SCORE_REPORTED",
+			authorUserId: users.id(1),
+		});
+
+		expect(message.type).toBe("SCORE_REPORTED");
+		expect(message.contents).toBeNull();
+		expect(message.authorUserId).toBe(users.id(1));
+		expect(message.publicId).not.toBe("");
+	});
+});
+
+describe("ChatRepository.findAllMessagesByRoomId", () => {
+	test("returns messages oldest first with authors resolved", async () => {
+		const room = await ChatRoomFactory.create();
+		await ChatMessageFactory.create({
+			roomId: room.id,
+			authorUserId: users.id(1),
+		});
+		await ChatMessageFactory.create({
+			roomId: room.id,
+			authorUserId: users.id(2),
+		});
+
+		const messages = await ChatRepository.findAllMessagesByRoomId(room.id);
+
+		expect(messages).toHaveLength(2);
+		expect(messages[0].id).toBeLessThan(messages[1].id);
+		expect(messages[0].author?.id).toBe(users.id(1));
+		expect(messages[1].author?.id).toBe(users.id(2));
+		expect(messages[0].author?.username).toBeTruthy();
+	});
+
+	test("keeps the latest messages when over the limit", async () => {
+		const room = await ChatRoomFactory.create();
+		const [, second, third] = await ChatMessageFactory.createMany(3, {
+			roomId: room.id,
+			authorUserId: users.id(1),
+		});
+
+		const messages = await ChatRepository.findAllMessagesByRoomId(room.id, {
+			limit: 2,
+		});
+
+		expect(messages.map((message) => message.id)).toEqual([
+			second.id,
+			third.id,
+		]);
+	});
+
+	test("does not return another room's messages", async () => {
+		const room = await ChatRoomFactory.create();
+		const otherRoom = await ChatRoomFactory.create();
+		await ChatMessageFactory.create({
+			roomId: room.id,
+			authorUserId: users.id(1),
+		});
+		await ChatMessageFactory.create({
+			roomId: otherRoom.id,
+			authorUserId: users.id(1),
+		});
+
+		const messages = await ChatRepository.findAllMessagesByRoomId(room.id);
+
+		expect(messages).toHaveLength(1);
+		expect(messages[0].roomId).toBe(room.id);
+	});
+});
+
+describe("ChatRepository.upsertReadIndicator", () => {
+	test("creates the indicator on first upsert", async () => {
+		const room = await ChatRoomFactory.create();
+
+		await ChatRepository.upsertReadIndicator({
+			userId: users.id(1),
+			roomId: room.id,
+			lastSeenMessageId: 5,
+		});
+
+		const indicator = await readIndicator(users.id(1), room.id);
+		expect(indicator.lastSeenMessageId).toBe(5);
+	});
+
+	test("never regresses to an older message", async () => {
+		const room = await ChatRoomFactory.create();
+
+		await ChatRepository.upsertReadIndicator({
+			userId: users.id(1),
+			roomId: room.id,
+			lastSeenMessageId: 5,
+		});
+		await ChatRepository.upsertReadIndicator({
+			userId: users.id(1),
+			roomId: room.id,
+			lastSeenMessageId: 3,
+		});
+
+		expect((await readIndicator(users.id(1), room.id)).lastSeenMessageId).toBe(
+			5,
+		);
+
+		await ChatRepository.upsertReadIndicator({
+			userId: users.id(1),
+			roomId: room.id,
+			lastSeenMessageId: 9,
+		});
+
+		expect((await readIndicator(users.id(1), room.id)).lastSeenMessageId).toBe(
+			9,
+		);
+	});
+});
+
+describe("ChatRepository.closeExpiredRooms", () => {
+	test("closes only rooms expired before the cutoff", async () => {
+		const expiredRoom = await ChatRoomFactory.create({
+			expiresAt: subDays(new Date(), 60),
+		});
+		const openRoom = await ChatRoomFactory.create({
+			expiresAt: addDays(new Date(), 1),
+		});
+
+		const closedCount = await ChatRepository.closeExpiredRooms(
+			subDays(new Date(), 30),
+		);
+
+		expect(closedCount).toBe(1);
+		expect((await roomById(expiredRoom.id)).closedAt).not.toBeNull();
+		expect((await roomById(openRoom.id)).closedAt).toBeNull();
+	});
+
+	test("leaves already-closed rooms untouched", async () => {
+		await ChatRoomFactory.create({ expiresAt: subDays(new Date(), 60) });
+		await ChatRepository.closeExpiredRooms(subDays(new Date(), 30));
+
+		const closedAgainCount = await ChatRepository.closeExpiredRooms(
+			subDays(new Date(), 30),
+		);
+
+		expect(closedAgainCount).toBe(0);
+	});
+});
+
+const readIndicator = (userId: number, roomId: number) =>
+	db
+		.selectFrom("ChatMessageReadIndicator")
+		.selectAll()
+		.where("userId", "=", userId)
+		.where("roomId", "=", roomId)
+		.executeTakeFirstOrThrow();
+
+const roomById = (id: number) =>
+	db
+		.selectFrom("ChatRoom")
+		.selectAll()
+		.where("id", "=", id)
+		.executeTakeFirstOrThrow();

app/features/chat/ChatRepository.server.ts

@@ -0,0 +1,136 @@
+import type { Transaction } from "kysely";
+import { sql } from "kysely";
+import { db } from "~/db/sql";
+import type { DB, Tables, TablesInsertable } from "~/db/tables";
+import { databaseTimestampNow, dateToDatabaseTimestamp } from "~/utils/dates";
+import { shortNanoid } from "~/utils/id";
+import {
+	commonUserSelect,
+	jsonObjectFrom,
+	userChatNameHue,
+} from "~/utils/kysely.server";
+import type { PersistedSystemMessageType } from "./chat-types";
+
+const MESSAGES_DEFAULT_LIMIT = 500;
+
+/** Returns the latest `limit` messages of a room, oldest first, authors resolved live. */
+export async function findAllMessagesByRoomId(
+	roomId: number,
+	{ limit = MESSAGES_DEFAULT_LIMIT }: { limit?: number } = {},
+) {
+	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"),
+		])
+		.where("ChatMessage.roomId", "=", roomId)
+		.orderBy("ChatMessage.id", "desc")
+		.limit(limit)
+		.execute();
+
+	return rows.reverse();
+}
+
+/** 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 },
+	trx?: Transaction<DB>,
+) {
+	const executor = trx ?? db;
+
+	return executor
+		.insertInto("ChatRoom")
+		.values({
+			type: args.type,
+			expiresAt: dateToDatabaseTimestamp(args.expiresAt),
+		})
+		.returningAll()
+		.executeTakeFirstOrThrow();
+}
+
+type InsertMessageArgs = Pick<
+	TablesInsertable["ChatMessage"],
+	"roomId" | "publicId"
+> & {
+	authorUserId: number;
+	contents: string;
+};
+
+/** Inserts a user message. A `publicId` conflict (retried send) returns the existing row instead. */
+export async function insertMessage(args: InsertMessageArgs) {
+	const inserted = await db
+		.insertInto("ChatMessage")
+		.values(args)
+		.onConflict((oc) => oc.column("publicId").doNothing())
+		.returningAll()
+		.executeTakeFirst();
+
+	if (inserted) return inserted;
+
+	return db
+		.selectFrom("ChatMessage")
+		.selectAll()
+		.where("ChatMessage.publicId", "=", args.publicId)
+		.executeTakeFirstOrThrow();
+}
+
+/** Inserts a system message rendered client-side from its `type`; `authorUserId` is the actor it describes. */
+export function insertSystemMessage(
+	args: {
+		roomId: number;
+		type: PersistedSystemMessageType;
+		authorUserId: number;
+	},
+	trx?: Transaction<DB>,
+) {
+	const executor = trx ?? db;
+
+	return executor
+		.insertInto("ChatMessage")
+		.values({ ...args, publicId: shortNanoid() })
+		.returningAll()
+		.executeTakeFirstOrThrow();
+}
+
+/** Marks the newest message a user has seen in a room. Never regresses: upserts keep the MAX. */
+export async function upsertReadIndicator(
+	args: TablesInsertable["ChatMessageReadIndicator"],
+) {
+	await db
+		.insertInto("ChatMessageReadIndicator")
+		.values(args)
+		.onConflict((oc) =>
+			oc.columns(["userId", "roomId"]).doUpdateSet({
+				lastSeenMessageId: sql`max("ChatMessageReadIndicator"."lastSeenMessageId", "excluded"."lastSeenMessageId")`,
+			}),
+		)
+		.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
+		.updateTable("ChatRoom")
+		.set({ closedAt: databaseTimestampNow() })
+		.where("ChatRoom.expiresAt", "<", dateToDatabaseTimestamp(expiredBefore))
+		.where("ChatRoom.closedAt", "is", null)
+		.executeTakeFirst();
+
+	return Number(result.numUpdatedRows);
+}

scripts/benchmark-db/cases.ts

@@ -7,6 +7,7 @@ import * as LogInLinkRepository from "~/features/auth/LogInLinkRepository.server
 import * as BadgeRepository from "~/features/badges/BadgeRepository.server";
 import * as BuildRepository from "~/features/builds/BuildRepository.server";
 import * as CalendarRepository from "~/features/calendar/CalendarRepository.server";
+import * as ChatRepository from "~/features/chat/ChatRepository.server";
 import * as FriendRepository from "~/features/friends/FriendRepository.server";
 import * as ImageRepository from "~/features/img-upload/ImageRepository.server";
 import * as LeaderboardRepository from "~/features/leaderboards/LeaderboardRepository.server";
@@ -232,6 +233,11 @@ export function buildCases(fx: Fixtures): {
 		(eventId) => CalendarRepository.findTopThreeResultsByEventIds([eventId]),
 	);
 
+	// ChatRepository
+	add("ChatRepository.findAllMessagesByRoomId", fx.heavyChatRoomId, (roomId) =>
+		ChatRepository.findAllMessagesByRoomId(roomId),
+	);
+
 	// FriendRepository
 	add("FriendRepository.findByUserIdWithActivity", fx.heavyFriendPair, (pair) =>
 		FriendRepository.findByUserIdWithActivity(pair.userId),

scripts/benchmark-db/fixtures.ts

@@ -41,6 +41,8 @@ export interface Fixtures {
 	recentTournamentIds: number[] | null;
 	heavyTeam: { id: number; customUrl: string; memberUserId: number } | null;
 	heavyCalendarEventId: number | null;
+	/** Chat room with the most messages. Null until the prod copy has post-migration chat data. */
+	heavyChatRoomId: number | null;
 	resultsEventId: number | null;
 	calendarAuthorId: number | null;
 	calendarWindow: { startTime: Date; endTime: Date } | null;
@@ -150,6 +152,7 @@ export async function resolveFixtures(): Promise<Fixtures> {
 		recentTournamentIds: await resolveRecentTournamentIds(),
 		heavyTeam: await resolveHeavyTeam(),
 		heavyCalendarEventId: await resolveHeavyCalendarEventId(),
+		heavyChatRoomId: await resolveHeavyChatRoomId(),
 		resultsEventId: await resolveResultsEventId(),
 		calendarAuthorId: await resolveCalendarAuthorId(),
 		calendarWindow: await resolveCalendarWindow(),
@@ -595,6 +598,18 @@ async function resolveCalendarAuthorId() {
 	return row?.authorId ?? null;
 }
 
+async function resolveHeavyChatRoomId() {
+	const row = await db
+		.selectFrom("ChatMessage")
+		.select(({ fn }) => ["roomId", fn.countAll<number>().as("count")])
+		.groupBy("roomId")
+		.orderBy("count", "desc")
+		.limit(1)
+		.executeTakeFirst();
+
+	return row?.roomId ?? null;
+}
+
 async function resolveCalendarWindow() {
 	const row = await db
 		.selectFrom("CalendarEventDate")