diff --git a/app/db/seed/factories/ChatMessageFactory.ts b/app/db/seed/factories/ChatMessageFactory.ts new file mode 100644 index 000000000..ee2ec8680 --- /dev/null +++ b/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[0]) => + ChatRepository.insertMessage(args), +}); diff --git a/app/db/seed/factories/ChatRoomFactory.ts b/app/db/seed/factories/ChatRoomFactory.ts new file mode 100644 index 000000000..9437769a3 --- /dev/null +++ b/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[0]) => + ChatRepository.insertRoom(args), +}); diff --git a/app/features/chat/ChatRepository.server.test.ts b/app/features/chat/ChatRepository.server.test.ts new file mode 100644 index 000000000..5911673a3 --- /dev/null +++ b/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().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(); diff --git a/app/features/chat/ChatRepository.server.ts b/app/features/chat/ChatRepository.server.ts new file mode 100644 index 000000000..3f1eb8b88 --- /dev/null +++ b/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, +) { + 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, +) { + 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); +} diff --git a/scripts/benchmark-db/cases.ts b/scripts/benchmark-db/cases.ts index 9ea677627..ae9c1afce 100644 --- a/scripts/benchmark-db/cases.ts +++ b/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), diff --git a/scripts/benchmark-db/fixtures.ts b/scripts/benchmark-db/fixtures.ts index 07eedb5bd..310c19173 100644 --- a/scripts/benchmark-db/fixtures.ts +++ b/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 { 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().as("count")]) + .groupBy("roomId") + .orderBy("count", "desc") + .limit(1) + .executeTakeFirst(); + + return row?.roomId ?? null; +} + async function resolveCalendarWindow() { const row = await db .selectFrom("CalendarEventDate")