From 13b8b9ded84be991f91f41bf889f14feed85152b Mon Sep 17 00:00:00 2001 From: root Date: Wed, 16 Sep 2026 09:55:03 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=AE=9E=E7=8E=B0=E5=AE=9E=E6=97=B6?= =?UTF-8?q?=E5=8C=BF=E5=90=8D=E4=BC=9A=E8=AF=9D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- apps/api/package.json | 8 +- apps/api/src/app.module.ts | 10 +- apps/api/src/auth/auth.guard.ts | 17 +- .../api/src/conversation/chat-rate-limiter.ts | 22 + apps/api/src/conversation/chat.gateway.ts | 243 +++++++ .../conversation/conversation.controller.ts | 119 ++++ .../src/conversation/conversation.e2e-spec.ts | 638 +++++++++++++++++ .../src/conversation/conversation.module.ts | 16 + .../src/conversation/conversation.service.ts | 648 ++++++++++++++++++ apps/api/src/conversation/dto.ts | 42 ++ pnpm-lock.yaml | 250 ++++++- .../0010_realtime_conversation/migration.sql | 24 + prisma/schema.prisma | 30 +- tests/integration/database.spec.ts | 6 + 14 files changed, 2042 insertions(+), 31 deletions(-) create mode 100644 apps/api/src/conversation/chat-rate-limiter.ts create mode 100644 apps/api/src/conversation/chat.gateway.ts create mode 100644 apps/api/src/conversation/conversation.controller.ts create mode 100644 apps/api/src/conversation/conversation.e2e-spec.ts create mode 100644 apps/api/src/conversation/conversation.module.ts create mode 100644 apps/api/src/conversation/conversation.service.ts create mode 100644 apps/api/src/conversation/dto.ts create mode 100644 prisma/migrations/0010_realtime_conversation/migration.sql diff --git a/apps/api/package.json b/apps/api/package.json index 3df4876..1d578b6 100644 --- a/apps/api/package.json +++ b/apps/api/package.json @@ -7,7 +7,7 @@ "build": "tsc -p tsconfig.build.json", "start": "node dist/main.js", "test": "vitest run --config vitest.config.ts --no-file-parallelism", - "test:e2e": "vitest run --config vitest.config.ts src/health/health.e2e-spec.ts src/auth/auth.e2e-spec.ts src/profile/profile.e2e-spec.ts src/bottle/bottle.e2e-spec.ts src/match/match.e2e-spec.ts --no-file-parallelism", + "test:e2e": "vitest run --config vitest.config.ts src/health/health.e2e-spec.ts src/auth/auth.e2e-spec.ts src/profile/profile.e2e-spec.ts src/bottle/bottle.e2e-spec.ts src/match/match.e2e-spec.ts src/conversation/conversation.e2e-spec.ts --no-file-parallelism", "typecheck": "tsc --noEmit -p tsconfig.json" }, "dependencies": { @@ -15,6 +15,8 @@ "@nestjs/common": "^11.1.6", "@nestjs/core": "^11.1.6", "@nestjs/platform-express": "^11.1.6", + "@nestjs/platform-socket.io": "^11.1.6", + "@nestjs/websockets": "^11.1.6", "@prisma/client": "6.19.0", "class-transformer": "^0.5.1", "class-validator": "^0.14.2", @@ -22,12 +24,14 @@ "helmet": "^8.1.0", "redis": "^5.8.2", "reflect-metadata": "^0.2.2", - "rxjs": "^7.8.2" + "rxjs": "^7.8.2", + "socket.io": "^4.8.1" }, "devDependencies": { "@nestjs/testing": "^11.1.6", "@types/express": "^5.0.3", "@types/supertest": "^6.0.3", + "socket.io-client": "^4.8.1", "supertest": "^7.1.4", "typescript": "^5.6.3", "vitest": "^4.1.11" diff --git a/apps/api/src/app.module.ts b/apps/api/src/app.module.ts index bbb4184..cab43c4 100644 --- a/apps/api/src/app.module.ts +++ b/apps/api/src/app.module.ts @@ -5,9 +5,17 @@ import { AuthModule } from "./auth/auth.module.js"; import { ProfileModule } from "./profile/profile.module.js"; import { BottleModule } from "./bottle/bottle.module.js"; import { MatchModule } from "./match/match.module.js"; +import { ConversationModule } from "./conversation/conversation.module.js"; @Module({ - imports: [HealthModule, AuthModule, ProfileModule, BottleModule, MatchModule], + imports: [ + HealthModule, + AuthModule, + ProfileModule, + BottleModule, + MatchModule, + ConversationModule, + ], }) export class AppModule implements NestModule { configure(consumer: MiddlewareConsumer): void { diff --git a/apps/api/src/auth/auth.guard.ts b/apps/api/src/auth/auth.guard.ts index 6046f35..52bc514 100644 --- a/apps/api/src/auth/auth.guard.ts +++ b/apps/api/src/auth/auth.guard.ts @@ -9,6 +9,7 @@ import { ErrorCode } from "@drift/contracts"; import type { Request } from "express"; import { DomainException } from "../common/domain.exception.js"; import { PrismaService } from "../database/prisma.service.js"; +import type { AccessClaims } from "./token.service.js"; import { TokenService } from "./token.service.js"; import type { AuthenticatedRequest } from "./current-user.decorator.js"; @Injectable() @@ -17,11 +18,8 @@ export class AuthGuard implements CanActivate { @Inject(PrismaService) private readonly prisma: PrismaService, @Inject(TokenService) private readonly tokens: TokenService, ) {} - async canActivate(ctx: ExecutionContext): Promise { - const req = ctx.switchToHttp().getRequest(); - const auth = req.headers.authorization; - if (!auth?.startsWith("Bearer ")) throw this.denied(); - const claims = this.tokens.verifyAccess(auth.slice(7)); + async validateAccess(token: string): Promise { + const claims = this.tokens.verifyAccess(token); const session = await this.prisma.session.findUnique({ where: { id: claims.session_id }, include: { account: true }, @@ -37,7 +35,14 @@ export class AuthGuard implements CanActivate { session.account.tokenVersion !== claims.token_version ) throw this.denied(); - req.user = claims; + return claims; + } + + async canActivate(ctx: ExecutionContext): Promise { + const req = ctx.switchToHttp().getRequest(); + const auth = req.headers.authorization; + if (!auth?.startsWith("Bearer ")) throw this.denied(); + req.user = await this.validateAccess(auth.slice(7)); return true; } private denied() { diff --git a/apps/api/src/conversation/chat-rate-limiter.ts b/apps/api/src/conversation/chat-rate-limiter.ts new file mode 100644 index 0000000..9a83a20 --- /dev/null +++ b/apps/api/src/conversation/chat-rate-limiter.ts @@ -0,0 +1,22 @@ +import { Inject, Injectable } from "@nestjs/common"; +import { RedisService } from "../redis/redis.service.js"; + +@Injectable() +export class ChatRateLimiter { + constructor(@Inject(RedisService) private readonly redis: RedisService) {} + async consume(accountId: string, sessionId: string): Promise { + await this.redis.ensureConnected(); + const prefix = process.env.REDIS_KEY_PREFIX ?? "drift:auth:"; + const max = Number(process.env.CHAT_MESSAGE_RATE_LIMIT ?? 30); + const ttl = Number(process.env.CHAT_MESSAGE_RATE_WINDOW_SECONDS ?? 60); + const keys = [ + `${prefix}chat:account:${accountId}`, + `${prefix}chat:session:${sessionId}`, + ]; + const result = await this.redis.client.eval( + `for _,k in ipairs(KEYS) do if tonumber(redis.call('GET',k) or '0')+1>tonumber(ARGV[1]) then return 0 end end; for _,k in ipairs(KEYS) do local n=redis.call('INCR',k); if n==1 then redis.call('EXPIRE',k,ARGV[2]) end end; return 1`, + { keys, arguments: [String(max), String(ttl)] }, + ); + return Number(result) === 1; + } +} diff --git a/apps/api/src/conversation/chat.gateway.ts b/apps/api/src/conversation/chat.gateway.ts new file mode 100644 index 0000000..a2ec49f --- /dev/null +++ b/apps/api/src/conversation/chat.gateway.ts @@ -0,0 +1,243 @@ +import { HttpStatus, Inject } from "@nestjs/common"; +import { plainToInstance } from "class-transformer"; +import { validate } from "class-validator"; +import { ErrorCode } from "@drift/contracts"; +import { + ConnectedSocket, + MessageBody, + type OnGatewayConnection, + type OnGatewayInit, + SubscribeMessage, + WebSocketGateway, + WebSocketServer, +} from "@nestjs/websockets"; +import type { IncomingMessage } from "node:http"; +import type { Server, Socket } from "socket.io"; +import { webOrigin } from "../auth/auth.config.js"; +import { AuthGuard } from "../auth/auth.guard.js"; +import type { AccessClaims } from "../auth/token.service.js"; +import { DomainException } from "../common/domain.exception.js"; +import { ChatRateLimiter } from "./chat-rate-limiter.js"; +import { + ConversationService, + type MessageResult, +} from "./conversation.service.js"; +import { SendMessageDto, SocketReadDto } from "./dto.js"; + +interface SocketData { + token?: string; + user?: AccessClaims; +} +interface ClientToServerEvents { + "message:send": (dto: SendMessageDto, ack?: Ack) => void; + "conversation:read": (dto: SocketReadDto, ack?: Ack) => void; +} +interface ServerToClientEvents { + "message:ack": (value: SocketResponse) => void; + "message:new": (value: MessageResult["message"]) => void; +} +type ChatServer = Server< + ClientToServerEvents, + ServerToClientEvents, + Record, + SocketData +>; +type ChatSocket = Socket< + ClientToServerEvents, + ServerToClientEvents, + Record, + SocketData +>; +type SocketResponse = + | { ok: true; data: unknown } + | { ok: false; error: { code: ErrorCode; message: string } }; +type Ack = (value: SocketResponse) => void; +type HandshakeAuth = { token?: unknown; authorization?: unknown }; + +function allowOrigin( + request: IncomingMessage, + callback: (error: string | null | undefined, success: boolean) => void, +): void { + const origin = request.headers.origin; + callback(null, origin === undefined || origin === webOrigin()); +} + +@WebSocketGateway({ + namespace: "/chat", + allowRequest: allowOrigin, + cors: { + origin(origin, callback) { + callback(null, origin === undefined || origin === webOrigin()); + }, + credentials: true, + }, +}) +export class ChatGateway implements OnGatewayInit, OnGatewayConnection { + @WebSocketServer() server!: ChatServer; + + constructor( + @Inject(AuthGuard) private readonly auth: AuthGuard, + @Inject(ConversationService) + private readonly conversations: ConversationService, + @Inject(ChatRateLimiter) private readonly limiter: ChatRateLimiter, + ) {} + + afterInit(server: ChatServer): void { + server.use((client, next) => { + void (async () => { + try { + const auth = client.handshake.auth as unknown as HandshakeAuth; + const raw = auth.token ?? auth.authorization; + const token = + typeof raw === "string" && raw.startsWith("Bearer ") + ? raw.slice(7) + : raw; + if (typeof token !== "string") throw new Error("Unauthorized"); + const user = await this.auth.validateAccess(token); + client.data.token = token; + client.data.user = user; + await client.join(`account:${user.sub}`); + next(); + } catch { + next(new Error("Unauthorized")); + } + })(); + }); + } + + handleConnection(client: ChatSocket): void { + // Single-instance MVP: require a Redis adapter before horizontal scaling. + const user = client.data.user; + if (!user || !client.rooms.has(`account:${user.sub}`)) + client.disconnect(true); + } + + @SubscribeMessage("message:send") + async send( + @ConnectedSocket() client: ChatSocket, + @MessageBody() dto: SendMessageDto, + ack?: Ack, + ): Promise { + try { + dto = await this.validatePayload(SendMessageDto, dto); + const user = await this.auth.validateAccess(client.data.token ?? ""); + let allowed: boolean; + try { + allowed = await this.limiter.consume(user.sub, user.session_id); + } catch { + throw new DomainException( + ErrorCode.SERVICE_UNAVAILABLE, + "Service unavailable", + HttpStatus.SERVICE_UNAVAILABLE, + ); + } + if (!allowed) { + throw new DomainException( + ErrorCode.RATE_LIMITED, + "Too many requests", + HttpStatus.TOO_MANY_REQUESTS, + ); + } + const result = await this.conversations.send( + user.sub, + dto.conversationId, + dto.clientMsgId, + dto.text, + ); + const response: SocketResponse = { + ok: true, + data: { + message: result.message, + deduplicated: result.deduplicated, + }, + }; + ack?.(response); + client.emit("message:ack", response); + if (!result.deduplicated) { + for (const member of result.memberIds) { + this.server + .to(`account:${member}`) + .emit("message:new", result.message); + } + } + return response; + } catch (error) { + const response = this.error(error); + ack?.(response); + client.emit("message:ack", response); + return response; + } + } + + @SubscribeMessage("conversation:read") + async read( + @ConnectedSocket() client: ChatSocket, + @MessageBody() body: SocketReadDto, + ack?: Ack, + ): Promise { + try { + body = await this.validatePayload(SocketReadDto, body); + const user = await this.auth.validateAccess(client.data.token ?? ""); + const data = await this.conversations.read( + user.sub, + body.conversationId, + body.seq, + ); + const response: SocketResponse = { ok: true, data }; + ack?.(response); + return response; + } catch (error) { + const response = this.error(error); + ack?.(response); + return response; + } + } + + private error(error: unknown): SocketResponse { + if (error instanceof DomainException) { + return { + ok: false, + error: { code: error.code, message: error.message }, + }; + } + return { + ok: false, + error: { + code: ErrorCode.INTERNAL_ERROR, + message: "Internal server error", + }, + }; + } + + private async validatePayload( + type: new () => T, + value: unknown, + ): Promise { + const dto = plainToInstance(type, value); + const errors = await validate(dto, { + whitelist: true, + forbidNonWhitelisted: true, + }); + if (errors.length) { + throw new DomainException( + ErrorCode.VALIDATION_ERROR, + "Validation failed", + HttpStatus.BAD_REQUEST, + ); + } + return dto; + } +} + +Reflect.defineMetadata( + "design:paramtypes", + [Object, SendMessageDto, Function], + ChatGateway.prototype, + "send", +); +Reflect.defineMetadata( + "design:paramtypes", + [Object, SocketReadDto, Function], + ChatGateway.prototype, + "read", +); diff --git a/apps/api/src/conversation/conversation.controller.ts b/apps/api/src/conversation/conversation.controller.ts new file mode 100644 index 0000000..0ad61ed --- /dev/null +++ b/apps/api/src/conversation/conversation.controller.ts @@ -0,0 +1,119 @@ +import { + Body, + Controller, + Get, + Inject, + Param, + ParseUUIDPipe, + Post, + Query, + UseGuards, +} from "@nestjs/common"; +import { AuthGuard } from "../auth/auth.guard.js"; +import { CurrentUser } from "../auth/current-user.decorator.js"; +import type { AccessClaims } from "../auth/token.service.js"; +import { ConversationService } from "./conversation.service.js"; +import { + ConversationsQueryDto, + MessagesQueryDto, + ReadConversationDto, + ReplyDto, + SendMessageDto, +} from "./dto.js"; + +@Controller() +@UseGuards(AuthGuard) +export class ConversationController { + constructor( + @Inject(ConversationService) + private readonly conversations: ConversationService, + ) {} + @Post("bottles/:id/reply") reply( + @CurrentUser() user: AccessClaims, + @Param("id", new ParseUUIDPipe()) id: string, + @Body() dto: ReplyDto, + ) { + return this.conversations.reply( + user.sub, + id, + dto.leaseId, + dto.leaseToken, + dto.clientMsgId, + dto.text, + ); + } + @Get("conversations") list( + @CurrentUser() user: AccessClaims, + @Query() query: ConversationsQueryDto, + ) { + return this.conversations.list(user.sub, query.cursor, query.limit); + } + @Get("conversations/:id") detail( + @CurrentUser() user: AccessClaims, + @Param("id", new ParseUUIDPipe()) id: string, + ) { + return this.conversations.detail(user.sub, id); + } + @Get("conversations/:id/messages") messages( + @CurrentUser() user: AccessClaims, + @Param("id", new ParseUUIDPipe()) id: string, + @Query() query: MessagesQueryDto, + ) { + return this.conversations.messages( + user.sub, + id, + query.afterSeq, + query.limit, + ); + } + @Post("conversations/:id/messages/prepare") prepare( + @CurrentUser() user: AccessClaims, + @Param("id", new ParseUUIDPipe()) id: string, + @Body() dto: SendMessageDto, + ) { + return this.conversations.send(user.sub, id, dto.clientMsgId, dto.text); + } + @Post("conversations/:id/read") read( + @CurrentUser() user: AccessClaims, + @Param("id", new ParseUUIDPipe()) id: string, + @Body() dto: ReadConversationDto, + ) { + return this.conversations.read(user.sub, id, dto.seq); + } +} +Reflect.defineMetadata( + "design:paramtypes", + [Object, String, ReplyDto], + ConversationController.prototype, + "reply", +); +Reflect.defineMetadata( + "design:paramtypes", + [Object, ConversationsQueryDto], + ConversationController.prototype, + "list", +); +Reflect.defineMetadata( + "design:paramtypes", + [Object, String], + ConversationController.prototype, + "detail", +); +Reflect.defineMetadata( + "design:paramtypes", + [Object, String, MessagesQueryDto], + ConversationController.prototype, + "messages", +); +Reflect.defineMetadata( + "design:paramtypes", + [Object, String, SendMessageDto], + ConversationController.prototype, + "prepare", +); +Reflect.defineMetadata( + "design:paramtypes", + [Object, String, ReadConversationDto], + ConversationController.prototype, + "read", +); diff --git a/apps/api/src/conversation/conversation.e2e-spec.ts b/apps/api/src/conversation/conversation.e2e-spec.ts new file mode 100644 index 0000000..4a673de --- /dev/null +++ b/apps/api/src/conversation/conversation.e2e-spec.ts @@ -0,0 +1,638 @@ +/* eslint-disable @typescript-eslint/no-unsafe-member-access, @typescript-eslint/no-unsafe-assignment, @typescript-eslint/no-unsafe-argument, @typescript-eslint/no-unsafe-call, @typescript-eslint/no-unsafe-return */ +import "reflect-metadata"; +import { type INestApplication } from "@nestjs/common"; +import { Test } from "@nestjs/testing"; +import { PrismaClient } from "@prisma/client"; +import { randomUUID } from "node:crypto"; +import request from "supertest"; +import { io, type Socket } from "socket.io-client"; +import { afterAll, beforeAll, beforeEach, describe, expect, it } from "vitest"; +import { assertSafeTestDatabaseUrl } from "../../../../prisma/database-safety.js"; +import { AppModule } from "../app.module.js"; +import { TokenService } from "../auth/token.service.js"; +import { leaseHmac, encryptLeaseToken } from "../auth/auth.config.js"; +import { configureApp } from "../main.js"; + +process.env.NODE_ENV = "test"; +process.env.WEB_ORIGIN = "http://localhost:3000"; +process.env.PHONE_ENCRYPTION_KEY = + "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA="; +process.env.PHONE_HMAC_KEY = "test-phone-hmac-key-with-at-least-32-bytes"; +process.env.VERIFICATION_CODE_HMAC_KEY = + "test-code-hmac-key-with-at-least-32-bytes"; +process.env.JWT_SECRET = "test-jwt-secret-with-at-least-thirty-two-bytes"; +process.env.REFRESH_TOKEN_HMAC_KEY = + "test-refresh-hmac-key-with-at-least-32-bytes"; +process.env.LEASE_TOKEN_HMAC_KEY = "test-lease-hmac-key-with-at-least-32-bytes"; +process.env.LEASE_TOKEN_ENCRYPTION_KEY = + "BBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBB="; +process.env.MODERATION_REJECT_WORDS = "reject-word"; +process.env.MODERATION_REVIEW_WORDS = "review-word"; + +const prisma = new PrismaClient(); + +describe("conversation API with real PostgreSQL", () => { + let app: INestApplication; + beforeAll(async () => { + assertSafeTestDatabaseUrl(process.env.DATABASE_URL ?? ""); + await prisma.$connect(); + const module = await Test.createTestingModule({ + imports: [AppModule], + }).compile(); + app = module.createNestApplication(); + configureApp(app); + await app.listen(0, "127.0.0.1"); + }); + beforeEach(async () => { + await prisma.$executeRawUnsafe(`TRUNCATE TABLE "accounts" CASCADE`); + }); + afterAll(async () => { + await app?.close(); + await prisma.$disconnect(); + }); + + async function actor(deviceId: string) { + const account = await prisma.account.create({ + data: { + phoneCiphertext: Buffer.from("cipher"), + phoneHmac: randomUUID(), + anonymousProfile: { + create: { + nickname: `anon-${deviceId}`, + avatarColor: "#123456", + reviewStatus: "APPROVED", + }, + }, + }, + }); + const session = await prisma.session.create({ + data: { + accountId: account.id, + refreshTokenHash: randomUUID(), + deviceId: deviceId.padEnd(8, "x"), + expiresAt: new Date(Date.now() + 600_000), + tokenFamily: randomUUID(), + }, + }); + const token = app.get(TokenService).issueAccess({ + sub: account.id, + session_id: session.id, + device_id: deviceId.padEnd(8, "x"), + scopes: ["user"], + token_version: 0, + }); + return { + id: account.id, + sessionId: session.id, + authorization: ["Bear", "er ", token].join(""), + token, + }; + } + async function leased(authorId: string, picker: { id: string }) { + const bottle = await prisma.bottle.create({ + data: { + authorId, + clientRequestId: randomUUID(), + contentText: "source", + reviewStatus: "APPROVED", + poolStatus: "IN_POOL", + approvedAt: new Date(), + }, + }); + const token = randomUUID().replaceAll("-", "") + "abcdefghijk"; + const leaseId = randomUUID(); + const lease = await prisma.bottlePickLease.create({ + data: { + id: leaseId, + bottleId: bottle.id, + pickerId: picker.id, + leaseTokenHash: leaseHmac(token), + leaseTokenCiphertext: encryptLeaseToken(token, leaseId), + expiresAt: new Date(Date.now() + 60_000), + }, + }); + await prisma.bottle.update({ + where: { id: bottle.id }, + data: { poolStatus: "LEASED", activeLeaseId: lease.id }, + }); + return { bottle, lease: { id: lease.id, token } }; + } + const reply = ( + authorization: string, + bottleId: string, + lease: { id: string; token: string }, + clientMsgId: string, + text = "hello", + ) => + request(app.getHttpServer()) + .post(`/api/v1/bottles/${bottleId}/reply`) + .set("Authorization", authorization) + .send({ leaseId: lease.id, leaseToken: lease.token, clientMsgId, text }); + + function socket(token: string, origin = process.env.WEB_ORIGIN): Socket { + const address = app.getHttpServer().address() as { port: number }; + return io(`http://127.0.0.1:${address.port}/chat`, { + auth: { token }, + ...(origin ? { extraHeaders: { Origin: origin } } : {}), + transports: ["websocket"], + forceNew: true, + reconnection: false, + }); + } + + function connected(client: Socket): Promise { + return new Promise((resolve, reject) => { + client.once("connect", resolve); + client.once("connect_error", reject); + }); + } + + function ack(client: Socket, event: string, body: unknown): Promise { + return new Promise((resolve, reject) => { + const timer = setTimeout( + () => reject(new Error(`ACK timeout: ${event}`)), + 2_000, + ); + client.emit(event, body, (value: T) => { + clearTimeout(timer); + resolve(value); + }); + }); + } + + it("atomically consumes a lease and deduplicates concurrent first replies", async () => { + const author = await actor("author"); + const picker = await actor("picker"); + const { bottle, lease } = await leased(author.id, picker); + const clientMsgId = randomUUID(); + const results = await Promise.all( + Array.from({ length: 8 }, () => + reply(picker.authorization, bottle.id, lease, clientMsgId), + ), + ); + if (!results.every((x) => x.status === 201)) + console.error(results.map((x) => [x.status, x.body])); + expect(results.every((x) => x.status === 201)).toBe(true); + expect(new Set(results.map((x) => x.body.data.conversationId)).size).toBe( + 1, + ); + expect( + await prisma.conversation.count({ where: { sourceBottleId: bottle.id } }), + ).toBe(1); + expect(await prisma.message.count()).toBe(1); + expect(await prisma.conversationMember.count()).toBe(2); + expect( + await prisma.bottle.findUniqueOrThrow({ where: { id: bottle.id } }), + ).toMatchObject({ poolStatus: "CONSUMED", activeLeaseId: null }); + expect( + await prisma.bottlePickLease.findUniqueOrThrow({ + where: { id: lease.id }, + }), + ).toMatchObject({ status: "CONSUMED" }); + expect((await prisma.message.findFirstOrThrow()).seq).toBe(1n); + expect((await prisma.conversation.findFirstOrThrow()).nextSeq).toBe(2n); + }); + + it("allows only one first message when distinct client ids race", async () => { + const author = await actor("author"); + const picker = await actor("picker"); + const { bottle, lease } = await leased(author.id, picker); + const results = await Promise.all([ + reply(picker.authorization, bottle.id, lease, randomUUID(), "one"), + reply(picker.authorization, bottle.id, lease, randomUUID(), "two"), + ]); + if (!results.every((x) => x.status === 201)) + console.error(results.map((x) => [x.status, x.body])); + expect(results.every((x) => x.status === 201)).toBe(true); + expect(new Set(results.map((x) => x.body.data.conversationId)).size).toBe( + 1, + ); + expect(await prisma.message.count()).toBe(1); + expect(results.filter((x) => x.body.data.won).length).toBe(1); + expect(results.filter((x) => !x.body.data.won).length).toBe(1); + expect(results.find((x) => !x.body.data.won)!.body.data).toMatchObject({ + created: false, + deduplicated: false, + }); + }); + + it("rejects invalid, expired and foreign leases without mutation", async () => { + const author = await actor("author"); + const picker = await actor("picker"); + const stranger = await actor("stranger"); + const { bottle, lease } = await leased(author.id, picker); + await reply(stranger.authorization, bottle.id, lease, randomUUID()).expect( + 404, + ); + await reply( + picker.authorization, + bottle.id, + { ...lease, token: "x".repeat(43) }, + randomUUID(), + ).expect(404); + await prisma.bottlePickLease.update({ + where: { id: lease.id }, + data: { expiresAt: new Date(Date.now() - 1) }, + }); + await reply(picker.authorization, bottle.id, lease, randomUUID()).expect( + 410, + ); + expect(await prisma.conversation.count()).toBe(0); + }); + + it("fails closed before state changes for blocks, sanctions and moderation", async () => { + const cases = ["block", "sanction", "reject", "review"] as const; + for (const kind of cases) { + await prisma.$executeRawUnsafe(`TRUNCATE TABLE "accounts" CASCADE`); + const author = await actor(`author-${kind}`); + const picker = await actor(`picker-${kind}`); + const { bottle, lease } = await leased(author.id, picker); + if (kind === "block") + await prisma.block.create({ + data: { blockerId: author.id, blockedId: picker.id }, + }); + if (kind === "sanction") + await prisma.sanction.create({ + data: { accountId: picker.id, type: "SUSPENSION", reason: "policy" }, + }); + const text = + kind === "reject" + ? "reject-word" + : kind === "review" + ? "review-word" + : "hello"; + const response = await reply( + picker.authorization, + bottle.id, + lease, + randomUUID(), + text, + ); + expect(response.status).toBe(403); + expect(await prisma.conversation.count()).toBe(0); + expect( + (await prisma.bottle.findUniqueOrThrow({ where: { id: bottle.id } })) + .poolStatus, + ).toBe("LEASED"); + } + }); + + it("authorizes before idempotency lookup and never exposes account ids", async () => { + const author = await actor("author"); + const picker = await actor("picker"); + const stranger = await actor("stranger"); + const { bottle, lease } = await leased(author.id, picker); + const clientMsgId = randomUUID(); + const created = await reply( + picker.authorization, + bottle.id, + lease, + clientMsgId, + ).expect(201); + const conversationId = created.body.data.conversationId as string; + + await request(app.getHttpServer()) + .post(`/api/v1/conversations/${conversationId}/messages/prepare`) + .set("Authorization", stranger.authorization) + .send({ conversationId, clientMsgId, text: "steal" }) + .expect(403); + + const history = await request(app.getHttpServer()) + .get(`/api/v1/conversations/${conversationId}/messages`) + .set("Authorization", picker.authorization) + .expect(200); + expect(history.body.data.items[0].senderId).toBeUndefined(); + expect(history.body.data.items[0].sender).toEqual({ + publicId: expect.any(String), + }); + expect(JSON.stringify(history.body)).not.toContain(picker.id); + }); + + it("writes a text-free message outbox event in the same transaction", async () => { + const author = await actor("author"); + const picker = await actor("picker"); + const { bottle, lease } = await leased(author.id, picker); + const result = await reply( + picker.authorization, + bottle.id, + lease, + randomUUID(), + "private message text", + ).expect(201); + const messageId = result.body.data.message.id as string; + const event = await prisma.outboxEvent.findUniqueOrThrow({ + where: { dedupeKey: `message-created:${messageId}` }, + }); + expect(event.eventType).toBe("MESSAGE_CREATED"); + expect(event.payload).toMatchObject({ messageId }); + expect(JSON.stringify(event.payload)).not.toContain("private message text"); + }); + + it("uses an exact timestamp/id cursor without skipping tied rows", async () => { + const owner = await actor("owner"); + const peer = await actor("peer"); + const tiedAt = new Date("2026-01-02T03:04:05.000Z"); + const ids = [randomUUID(), randomUUID(), randomUUID()].sort().reverse(); + for (const id of ids) { + await prisma.conversation.create({ + data: { + id, + sourceBottle: { + create: { + authorId: peer.id, + clientRequestId: randomUUID(), + contentText: "source", + reviewStatus: "APPROVED", + poolStatus: "CONSUMED", + approvedAt: new Date(), + }, + }, + lastMessageAt: tiedAt, + members: { + create: [ + { accountId: owner.id, peerAliasSnapshot: "peer" }, + { accountId: peer.id, peerAliasSnapshot: "owner" }, + ], + }, + }, + }); + } + const first = await request(app.getHttpServer()) + .get("/api/v1/conversations?limit=2") + .set("Authorization", owner.authorization) + .expect(200); + const second = await request(app.getHttpServer()) + .get(`/api/v1/conversations?limit=2&cursor=${first.body.data.nextCursor}`) + .set("Authorization", owner.authorization) + .expect(200); + expect([ + ...first.body.data.items.map((x: { id: string }) => x.id), + ...second.body.data.items.map((x: { id: string }) => x.id), + ]).toEqual(ids); + await request(app.getHttpServer()) + .get("/api/v1/conversations?cursor=e30") + .set("Authorization", owner.authorization) + .expect(400); + }); + + it("validates sequence bounds and forbids departed members before reads", async () => { + const author = await actor("author"); + const picker = await actor("picker"); + const { bottle, lease } = await leased(author.id, picker); + const created = await reply( + picker.authorization, + bottle.id, + lease, + randomUUID(), + ).expect(201); + const id = created.body.data.conversationId as string; + for (const bad of ["-1", "9223372036854775808", "not-a-number"]) { + await request(app.getHttpServer()) + .get(`/api/v1/conversations/${id}/messages?afterSeq=${bad}`) + .set("Authorization", picker.authorization) + .expect(400); + } + await request(app.getHttpServer()) + .post(`/api/v1/conversations/${id}/read`) + .set("Authorization", picker.authorization) + .send({ seq: "1" }) + .expect(201); + const monotonic = await request(app.getHttpServer()) + .post(`/api/v1/conversations/${id}/read`) + .set("Authorization", picker.authorization) + .send({ seq: "0" }) + .expect(201); + expect(monotonic.body.data.lastReadSeq).toBe("1"); + await prisma.conversationMember.update({ + where: { + conversationId_accountId: { conversationId: id, accountId: picker.id }, + }, + data: { leftAt: new Date() }, + }); + await request(app.getHttpServer()) + .post(`/api/v1/conversations/${id}/read`) + .set("Authorization", picker.authorization) + .send({ seq: "1" }) + .expect(403); + }); + + it("authenticates both socket ends, broadcasts once, revokes live sessions and repairs history", async () => { + const author = await actor("author"); + const picker = await actor("picker"); + const { bottle, lease } = await leased(author.id, picker); + const created = await reply( + picker.authorization, + bottle.id, + lease, + randomUUID(), + ).expect(201); + const conversationId = created.body.data.conversationId as string; + + const unauthenticated = socket("invalid-token"); + const denied = new Promise((resolve) => + unauthenticated.once("connect_error", (error) => resolve(error.message)), + ); + expect(await denied).toBe("Unauthorized"); + unauthenticated.close(); + + const badOrigin = socket(author.token, "https://evil.example"); + const originDenied = new Promise((resolve) => + badOrigin.once("connect_error", () => resolve()), + ); + await originDenied; + badOrigin.close(); + + const authorSocket = socket(`Bearer ${author.token}`); + const pickerSocket = socket(picker.token); + await Promise.all([connected(authorSocket), connected(pickerSocket)]); + const clientMsgId = randomUUID(); + let authorDeliveries = 0; + let pickerDeliveries = 0; + authorSocket.on("message:new", () => authorDeliveries++); + pickerSocket.on("message:new", () => pickerDeliveries++); + const first = await ack<{ + ok: boolean; + data: { message: { seq: string }; deduplicated: boolean }; + }>(pickerSocket, "message:send", { + conversationId, + clientMsgId, + text: "socket hello", + }); + expect(first).toMatchObject({ ok: true, data: { deduplicated: false } }); + await new Promise((resolve) => setTimeout(resolve, 50)); + expect([authorDeliveries, pickerDeliveries]).toEqual([1, 1]); + const duplicate = await ack<{ data: { deduplicated: boolean } }>( + pickerSocket, + "message:send", + { conversationId, clientMsgId, text: "socket hello" }, + ); + expect(duplicate.data.deduplicated).toBe(true); + await new Promise((resolve) => setTimeout(resolve, 50)); + expect([authorDeliveries, pickerDeliveries]).toEqual([1, 1]); + + pickerSocket.close(); + const offlineId = randomUUID(); + await ack(authorSocket, "message:send", { + conversationId, + clientMsgId: offlineId, + text: "while offline", + }); + const history = await request(app.getHttpServer()) + .get( + `/api/v1/conversations/${conversationId}/messages?afterSeq=${first.data.message.seq}`, + ) + .set("Authorization", picker.authorization) + .expect(200); + expect( + history.body.data.items.map( + (item: { clientMsgId: string }) => item.clientMsgId, + ), + ).toEqual([offlineId]); + + await prisma.session.update({ + where: { id: author.sessionId }, + data: { revokedAt: new Date() }, + }); + const revoked = await ack<{ ok: boolean; error: { code: string } }>( + authorSocket, + "message:send", + { + conversationId, + clientMsgId: randomUUID(), + text: "must fail", + }, + ); + expect(revoked).toMatchObject({ + ok: false, + error: { code: "AUTH_UNAUTHORIZED" }, + }); + authorSocket.close(); + }); + + it("always ACKs invalid socket payloads with a sanitized validation error", async () => { + const actorOne = await actor("socket-validation"); + const client = socket(actorOne.token); + await connected(client); + for (const [event, payload] of [ + ["message:send", { conversationId: "not-a-uuid", extra: true }], + [ + "message:send", + { + conversationId: randomUUID(), + clientMsgId: randomUUID(), + text: "valid except extra", + extra: true, + }, + ], + [ + "conversation:read", + { conversationId: randomUUID(), seq: "0", extra: true }, + ], + ["conversation:read", { conversationId: randomUUID(), seq: -1 }], + ] as const) { + const result = await ack<{ + ok: boolean; + error: { code: string; message: string }; + }>(client, event, payload); + expect(result).toEqual({ + ok: false, + error: { code: "VALIDATION_ERROR", message: "Validation failed" }, + }); + } + expect(client.connected).toBe(true); + client.close(); + }); + + it("sanitizes unknown socket failures and blocks nonmembers, blocked peers and rejected content", async () => { + const author = await actor("socket-author"); + const picker = await actor("socket-picker"); + const stranger = await actor("socket-stranger"); + const { bottle, lease } = await leased(author.id, picker); + const created = await reply( + picker.authorization, + bottle.id, + lease, + randomUUID(), + ).expect(201); + const conversationId = created.body.data.conversationId as string; + const strangerSocket = socket(stranger.token); + const pickerSocket = socket(picker.token); + await Promise.all([connected(strangerSocket), connected(pickerSocket)]); + + await expect( + ack(strangerSocket, "message:send", { + conversationId, + clientMsgId: randomUUID(), + text: "not mine", + }), + ).resolves.toMatchObject({ + ok: false, + error: { code: "CONVERSATION_FORBIDDEN" }, + }); + await expect( + ack(pickerSocket, "message:send", { + conversationId, + clientMsgId: randomUUID(), + text: "reject-word", + }), + ).resolves.toMatchObject({ + ok: false, + error: { code: "CONTENT_REJECTED" }, + }); + await prisma.block.create({ + data: { blockerId: author.id, blockedId: picker.id }, + }); + await expect( + ack(pickerSocket, "message:send", { + conversationId, + clientMsgId: randomUUID(), + text: "blocked", + }), + ).resolves.toMatchObject({ + ok: false, + error: { code: "USER_BLOCKED" }, + }); + + const gateway: { conversations: { read: () => Promise } } = app.get( + (await import("./chat.gateway.js")).ChatGateway, + ); + const originalRead = gateway.conversations.read; + gateway.conversations.read = () => + Promise.reject(new Error("database secret")); + const unknown = await ack<{ + ok: boolean; + error: { code: string; message: string }; + }>(pickerSocket, "conversation:read", { conversationId, seq: "1" }); + gateway.conversations.read = originalRead; + expect(unknown).toEqual({ + ok: false, + error: { code: "INTERNAL_ERROR", message: "Internal server error" }, + }); + strangerSocket.close(); + pickerSocket.close(); + }); + + it("keeps the sender public id captured when the message was written", async () => { + const author = await actor("snapshot-author"); + const picker = await actor("snapshot-picker"); + const { bottle, lease } = await leased(author.id, picker); + const created = await reply( + picker.authorization, + bottle.id, + lease, + randomUUID(), + ).expect(201); + const conversationId = created.body.data.conversationId as string; + const originalPublicId = created.body.data.message.sender + .publicId as string; + await prisma.anonymousProfile.update({ + where: { accountId: picker.id }, + data: { publicId: randomUUID() }, + }); + const history = await request(app.getHttpServer()) + .get(`/api/v1/conversations/${conversationId}/messages`) + .set("Authorization", picker.authorization) + .expect(200); + expect(history.body.data.items[0].sender.publicId).toBe(originalPublicId); + }); +}); diff --git a/apps/api/src/conversation/conversation.module.ts b/apps/api/src/conversation/conversation.module.ts new file mode 100644 index 0000000..3991063 --- /dev/null +++ b/apps/api/src/conversation/conversation.module.ts @@ -0,0 +1,16 @@ +import { Module } from "@nestjs/common"; +import { AuthModule } from "../auth/auth.module.js"; +import { DatabaseModule } from "../database/database.module.js"; +import { RedisModule } from "../redis/redis.module.js"; +import { ChatGateway } from "./chat.gateway.js"; +import { ChatRateLimiter } from "./chat-rate-limiter.js"; +import { ConversationController } from "./conversation.controller.js"; +import { ConversationService } from "./conversation.service.js"; + +@Module({ + imports: [DatabaseModule, RedisModule, AuthModule], + controllers: [ConversationController], + providers: [ConversationService, ChatRateLimiter, ChatGateway], + exports: [ConversationService], +}) +export class ConversationModule {} diff --git a/apps/api/src/conversation/conversation.service.ts b/apps/api/src/conversation/conversation.service.ts new file mode 100644 index 0000000..c3d0823 --- /dev/null +++ b/apps/api/src/conversation/conversation.service.ts @@ -0,0 +1,648 @@ +import { HttpStatus, Inject, Injectable } from "@nestjs/common"; +import { ErrorCode } from "@drift/contracts"; +import { Prisma, type PrismaClient } from "@prisma/client"; +import { leaseHmac } from "../auth/auth.config.js"; +import { DomainException } from "../common/domain.exception.js"; +import { PrismaService } from "../database/prisma.service.js"; + +type Db = Prisma.TransactionClient | PrismaClient; +type MessageRow = { + id: string; + conversationId: string; + senderId: string; + senderPublicId: string; + clientMsgId: string; + seq: bigint; + contentText: string; + status: string; + reviewStatus: string; + sentAt: Date; +}; +type ConversationWithFirst = { + id: string; + messages: MessageRow[]; +}; +type Cursor = { t: string; id: string }; + +export interface MessageResult { + message: ReturnType; + deduplicated: boolean; + memberIds: string[]; +} + +const MAX_SEQ = 9_223_372_036_854_775_807n; +const UUID = + /^[0-9a-f]{8}-[0-9a-f]{4}-[1-5][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i; +const firstMessageInclude = { + messages: { + orderBy: { seq: "asc" as const }, + take: 1, + }, +}; + +@Injectable() +export class ConversationService { + constructor(@Inject(PrismaService) private readonly prisma: PrismaService) {} + + async reply( + accountId: string, + bottleId: string, + leaseId: string, + leaseToken: string, + clientMsgId: string, + text: string, + ) { + this.moderate(text); + for (let attempt = 0; attempt < 5; attempt += 1) { + const existing = await this.prisma.conversation.findUnique({ + where: { sourceBottleId: bottleId }, + include: firstMessageInclude, + }); + if (existing) { + await this.assertConversationMember(accountId, existing.id); + return this.replyResult(existing, clientMsgId, false); + } + try { + return await this.prisma.$transaction( + async (tx) => { + await tx.$queryRaw`SELECT "id" FROM "bottles" WHERE "id"=${bottleId}::uuid FOR UPDATE`; + const winner = await tx.conversation.findUnique({ + where: { sourceBottleId: bottleId }, + include: firstMessageInclude, + }); + if (winner) { + await this.assertConversationMember(accountId, winner.id, tx); + return this.replyResult(winner, clientMsgId, false); + } + const lease = await tx.bottlePickLease.findFirst({ + where: { + id: leaseId, + bottleId, + pickerId: accountId, + leaseTokenHash: leaseHmac(leaseToken), + }, + include: { bottle: true }, + }); + if ( + !lease || + lease.status !== "ACTIVE" || + lease.bottle.activeLeaseId !== lease.id || + lease.bottle.poolStatus !== "LEASED" + ) { + throw this.hidden(); + } + if (lease.expiresAt <= new Date()) { + throw new DomainException( + ErrorCode.BOTTLE_LEASE_EXPIRED, + "Bottle lease expired", + HttpStatus.GONE, + ); + } + const authorId = lease.bottle.authorId; + await this.assertAllowed(tx, accountId, authorId); + const profiles = await tx.anonymousProfile.findMany({ + where: { accountId: { in: [accountId, authorId] } }, + select: { accountId: true, nickname: true, publicId: true }, + }); + if (profiles.length !== 2) throw this.forbidden(); + const now = new Date(); + const conversation = await tx.conversation.create({ + data: { + sourceBottleId: bottleId, + nextSeq: 2n, + lastMessageAt: now, + members: { + create: [ + { + accountId, + peerAliasSnapshot: profiles.find( + (profile) => profile.accountId === authorId, + )!.nickname, + }, + { + accountId: authorId, + peerAliasSnapshot: profiles.find( + (profile) => profile.accountId === accountId, + )!.nickname, + }, + ], + }, + messages: { + create: { + senderId: accountId, + senderPublicId: profiles.find( + (profile) => profile.accountId === accountId, + )!.publicId, + clientMsgId, + seq: 1n, + contentText: text, + reviewStatus: "APPROVED", + status: "SENT", + sentAt: now, + }, + }, + }, + include: firstMessageInclude, + }); + const message = conversation.messages[0]!; + await this.createMessageOutbox(tx, message.id, conversation.id, [ + accountId, + authorId, + ]); + const consumed = await tx.bottle.updateMany({ + where: { + id: bottleId, + poolStatus: "LEASED", + activeLeaseId: leaseId, + }, + data: { + poolStatus: "CONSUMED", + activeLeaseId: null, + consumedAt: now, + version: { increment: 1 }, + }, + }); + if (consumed.count !== 1) throw this.hidden(); + await tx.bottlePickLease.update({ + where: { id: leaseId }, + data: { status: "CONSUMED", endedAt: now }, + }); + return this.replyResult(conversation, clientMsgId, true); + }, + { isolationLevel: Prisma.TransactionIsolationLevel.Serializable }, + ); + } catch (error) { + if (error instanceof DomainException) throw error; + if (!this.isReplyRetryable(error)) throw error; + await new Promise((resolve) => setTimeout(resolve, 10 * (attempt + 1))); + const raced = await this.prisma.conversation.findUnique({ + where: { sourceBottleId: bottleId }, + include: firstMessageInclude, + }); + if (raced) { + await this.assertConversationMember(accountId, raced.id); + return this.replyResult(raced, clientMsgId, false); + } + } + } + throw new DomainException( + ErrorCode.SERVICE_UNAVAILABLE, + "Conversation temporarily unavailable", + HttpStatus.SERVICE_UNAVAILABLE, + ); + } + + private replyResult( + conversation: ConversationWithFirst, + clientMsgId: string, + created: boolean, + ) { + const message = conversation.messages[0]!; + const won = message.clientMsgId === clientMsgId; + return { + conversationId: conversation.id, + message: this.serializeMessage(message), + created, + deduplicated: !created && won, + won, + }; + } + + async send( + accountId: string, + conversationId: string, + clientMsgId: string, + text: string, + ): Promise { + this.moderate(text); + try { + return await this.prisma.$transaction(async (tx) => { + const conversation = await this.authorizeConversation( + tx, + accountId, + conversationId, + ); + const previous = await tx.message.findUnique({ + where: { + conversationId_clientMsgId: { conversationId, clientMsgId }, + }, + }); + if (previous) { + return { + message: this.serializeMessage(previous), + deduplicated: true, + memberIds: conversation.members.map((member) => member.accountId), + }; + } + const rows = await tx.$queryRaw>` + UPDATE "conversations" + SET "next_seq"="next_seq"+1, "last_message_at"=now(), "updated_at"=now() + WHERE "id"=${conversationId}::uuid + RETURNING "next_seq"-1 AS seq + `; + const sender = await tx.anonymousProfile.findUnique({ + where: { accountId }, + select: { publicId: true }, + }); + if (!sender) throw this.forbidden(); + const message = await tx.message.create({ + data: { + conversationId, + senderId: accountId, + senderPublicId: sender.publicId, + clientMsgId, + seq: rows[0]!.seq, + contentText: text, + status: "SENT", + reviewStatus: "APPROVED", + }, + }); + const memberIds = conversation.members.map( + (member) => member.accountId, + ); + await this.createMessageOutbox( + tx, + message.id, + conversationId, + memberIds, + ); + return { + message: this.serializeMessage(message), + deduplicated: false, + memberIds, + }; + }); + } catch (error) { + if (this.isMessageIdempotencyRace(error)) { + return this.prisma.$transaction(async (tx) => { + const conversation = await this.authorizeConversation( + tx, + accountId, + conversationId, + ); + const message = await tx.message.findUniqueOrThrow({ + where: { + conversationId_clientMsgId: { conversationId, clientMsgId }, + }, + }); + return { + message: this.serializeMessage(message), + deduplicated: true, + memberIds: conversation.members.map((member) => member.accountId), + }; + }); + } + throw error; + } + } + + async list(accountId: string, cursorText: string | undefined, limit: number) { + const cursor = cursorText ? this.decodeCursor(cursorText) : undefined; + const cursorDate = cursor ? new Date(cursor.t) : undefined; + const rows = await this.prisma.conversation.findMany({ + where: { + members: { some: { accountId, leftAt: null } }, + ...(cursor && cursorDate + ? { + OR: [ + { lastMessageAt: { lt: cursorDate } }, + { lastMessageAt: cursorDate, id: { lt: cursor.id } }, + ], + } + : {}), + }, + orderBy: [{ lastMessageAt: "desc" }, { id: "desc" }], + take: limit + 1, + include: { + members: { where: { accountId } }, + messages: { + orderBy: { seq: "desc" }, + take: 1, + }, + }, + }); + const hasMore = rows.length > limit; + const page = rows.slice(0, limit); + return { + items: page.map((conversation) => { + const last = conversation.messages[0]; + const read = conversation.members[0]!.lastReadSeq; + return { + id: conversation.id, + status: conversation.status, + lastMessageAt: conversation.lastMessageAt, + lastMessage: last ? this.serializeMessage(last) : null, + unread: last && last.seq > read ? (last.seq - read).toString() : "0", + lastReadSeq: read.toString(), + }; + }), + nextCursor: hasMore ? this.encodeCursor(page.at(-1)!) : null, + }; + } + + async detail(accountId: string, id: string) { + const row = await this.prisma.conversation.findFirst({ + where: { id, members: { some: { accountId, leftAt: null } } }, + include: { members: { where: { accountId } } }, + }); + if (!row) throw this.forbidden(); + return { + id: row.id, + status: row.status, + lastMessageAt: row.lastMessageAt, + nextSeq: row.nextSeq.toString(), + lastReadSeq: row.members[0]!.lastReadSeq.toString(), + }; + } + + async messages( + accountId: string, + id: string, + afterSeq: string, + limit: number, + ) { + await this.assertMember(accountId, id); + const after = this.parseSeq(afterSeq); + const items = await this.prisma.message.findMany({ + where: { conversationId: id, seq: { gt: after } }, + orderBy: { seq: "asc" }, + take: limit, + }); + return { items: items.map((message) => this.serializeMessage(message)) }; + } + + async read(accountId: string, id: string, seqText: string) { + const member = await this.assertMember(accountId, id); + const seq = this.parseSeq(seqText); + const conversation = await this.prisma.conversation.findUnique({ + where: { id }, + select: { nextSeq: true }, + }); + if (!conversation) throw this.forbidden(); + if (seq >= conversation.nextSeq) throw this.validation(); + if (seq <= member.lastReadSeq) { + return { conversationId: id, lastReadSeq: member.lastReadSeq.toString() }; + } + const updated = await this.prisma.conversationMember.updateMany({ + where: { + conversationId: id, + accountId, + leftAt: null, + lastReadSeq: { lt: seq }, + }, + data: { lastReadSeq: seq }, + }); + if (updated.count !== 1) throw this.forbidden(); + return { conversationId: id, lastReadSeq: seq.toString() }; + } + + private async authorizeConversation( + tx: Db, + accountId: string, + conversationId: string, + ) { + const conversation = await tx.conversation.findUnique({ + where: { id: conversationId }, + include: { members: true }, + }); + if ( + !conversation || + conversation.status !== "ACTIVE" || + !conversation.members.some( + (member) => member.accountId === accountId && !member.leftAt, + ) + ) { + throw this.forbidden(); + } + const peer = conversation.members.find( + (member) => member.accountId !== accountId && !member.leftAt, + ); + if (!peer) throw this.forbidden(); + await this.assertAllowed(tx, accountId, peer.accountId); + return conversation; + } + + private async createMessageOutbox( + tx: Prisma.TransactionClient, + messageId: string, + conversationId: string, + memberIds: string[], + ) { + await tx.outboxEvent.create({ + data: { + aggregateType: "MESSAGE", + aggregateId: messageId, + eventType: "MESSAGE_CREATED", + dedupeKey: `message-created:${messageId}`, + payload: { messageId, conversationId, memberIds }, + }, + }); + } + + private isReplyRetryable(error: unknown): boolean { + if (!(error instanceof Prisma.PrismaClientKnownRequestError)) return false; + if (error.code === "P2034") return true; + // Prisma wraps PostgreSQL serialization failures from $queryRaw as P2010. + if (error.code === "P2010") { + return error.meta?.code === "40001"; + } + if (error.code !== "P2002") return false; + const fields = this.prismaUniqueTarget(error.meta?.target); + return ( + fields.includes("source_bottle_id") || + fields.includes("client_msg_id") || + fields.includes("conversations_source_bottle_id_key") || + fields.includes("messages_conversation_id_client_msg_id_key") + ); + } + + private isMessageIdempotencyRace(error: unknown): boolean { + if (!(error instanceof Prisma.PrismaClientKnownRequestError)) return false; + if (error.code !== "P2002") return false; + const fields = this.prismaUniqueTarget(error.meta?.target); + return ( + fields.includes("client_msg_id") || + fields.includes("messages_conversation_id_client_msg_id_key") + ); + } + + private prismaUniqueTarget(target: unknown): string[] { + if (typeof target === "string") return [target]; + if ( + Array.isArray(target) && + target.every((item) => typeof item === "string") + ) { + return target; + } + return []; + } + + private encodeCursor(row: { id: string; lastMessageAt: Date }): string { + return Buffer.from( + JSON.stringify({ t: row.lastMessageAt.toISOString(), id: row.id }), + ).toString("base64url"); + } + + private decodeCursor(text: string): Cursor { + if (text.length > 256 || !/^[A-Za-z0-9_-]+$/.test(text)) { + throw this.validation(); + } + try { + const decoded = Buffer.from(text, "base64url").toString("utf8"); + if (Buffer.from(decoded).toString("base64url") !== text) { + throw new Error("non-canonical"); + } + const value: unknown = JSON.parse(decoded); + if ( + !value || + typeof value !== "object" || + Array.isArray(value) || + Object.keys(value).length !== 2 || + typeof (value as Cursor).t !== "string" || + typeof (value as Cursor).id !== "string" || + !UUID.test((value as Cursor).id) + ) { + throw new Error("invalid cursor"); + } + const date = new Date((value as Cursor).t); + if ( + Number.isNaN(date.valueOf()) || + date.toISOString() !== (value as Cursor).t || + JSON.stringify(value) !== decoded + ) { + throw new Error("non-canonical cursor"); + } + return value as Cursor; + } catch { + throw this.validation(); + } + } + + private parseSeq(text: string): bigint { + if (!/^(0|[1-9]\d{0,18})$/.test(text)) throw this.validation(); + const value = BigInt(text); + if (value > MAX_SEQ) throw this.validation(); + return value; + } + + private async assertConversationMember( + accountId: string, + id: string, + tx: Db = this.prisma, + ) { + const found = await tx.conversationMember.findUnique({ + where: { conversationId_accountId: { conversationId: id, accountId } }, + }); + if (!found || found.leftAt) throw this.hidden(); + } + + async assertMember(accountId: string, id: string) { + const found = await this.prisma.conversationMember.findUnique({ + where: { conversationId_accountId: { conversationId: id, accountId } }, + }); + if (!found || found.leftAt) throw this.forbidden(); + return found; + } + + private async assertAllowed(tx: Db, a: string, b: string) { + const now = new Date(); + const [accounts, block, sanction] = await Promise.all([ + tx.account.count({ where: { id: { in: [a, b] }, status: "ACTIVE" } }), + tx.block.findFirst({ + where: { + OR: [ + { blockerId: a, blockedId: b }, + { blockerId: b, blockedId: a }, + ], + }, + select: { id: true }, + }), + tx.sanction.findFirst({ + where: { + accountId: { in: [a, b] }, + type: { in: ["MUTE", "SUSPENSION", "BAN"] }, + revokedAt: null, + startsAt: { lte: now }, + OR: [{ expiresAt: null }, { expiresAt: { gt: now } }], + }, + select: { id: true }, + }), + ]); + if (block) { + throw new DomainException( + ErrorCode.USER_BLOCKED, + "User blocked", + HttpStatus.FORBIDDEN, + ); + } + if (accounts !== 2 || sanction) { + throw new DomainException( + ErrorCode.ACCOUNT_SANCTIONED, + "Account sanctioned", + HttpStatus.FORBIDDEN, + ); + } + } + + private moderate(text: string) { + const reject = (process.env.MODERATION_REJECT_WORDS ?? "reject-word") + .split(",") + .filter(Boolean); + const review = (process.env.MODERATION_REVIEW_WORDS ?? "review-word") + .split(",") + .filter(Boolean); + if ([...reject, ...review].some((word) => text.includes(word))) { + throw new DomainException( + ErrorCode.CONTENT_REJECTED, + "Content rejected", + HttpStatus.FORBIDDEN, + ); + } + } + + serializeMessage(message: MessageRow) { + const publicId = message.senderPublicId; + if (!publicId) { + throw new DomainException( + ErrorCode.INTERNAL_ERROR, + "Message identity unavailable", + HttpStatus.INTERNAL_SERVER_ERROR, + ); + } + return { + id: message.id, + conversationId: message.conversationId, + sender: { publicId }, + clientMsgId: message.clientMsgId, + seq: message.seq.toString(), + text: message.contentText, + status: message.status, + reviewStatus: message.reviewStatus, + sentAt: message.sentAt, + }; + } + + private hidden() { + return new DomainException( + ErrorCode.NOT_FOUND, + "Not Found", + HttpStatus.NOT_FOUND, + ); + } + + private forbidden() { + return new DomainException( + ErrorCode.CONVERSATION_FORBIDDEN, + "Conversation forbidden", + HttpStatus.FORBIDDEN, + ); + } + + private validation() { + return new DomainException( + ErrorCode.VALIDATION_ERROR, + "Validation failed", + HttpStatus.BAD_REQUEST, + ); + } +} diff --git a/apps/api/src/conversation/dto.ts b/apps/api/src/conversation/dto.ts new file mode 100644 index 0000000..7afe61b --- /dev/null +++ b/apps/api/src/conversation/dto.ts @@ -0,0 +1,42 @@ +import { Transform, Type } from "class-transformer"; +import { + IsInt, + IsOptional, + IsString, + IsUUID, + Matches, + Max, + MaxLength, + Min, + MinLength, +} from "class-validator"; + +const trim = ({ value }: { value: unknown }) => + typeof value === "string" ? value.trim() : value; + +export class ReplyDto { + @IsUUID() leaseId!: string; + @IsString() @Matches(/^[A-Za-z0-9_-]{43}$/) leaseToken!: string; + @IsUUID() clientMsgId!: string; + @Transform(trim) @IsString() @MinLength(1) @MaxLength(1000) text!: string; +} +export class SendMessageDto { + @IsUUID() conversationId!: string; + @IsUUID() clientMsgId!: string; + @Transform(trim) @IsString() @MinLength(1) @MaxLength(1000) text!: string; +} +export class MessagesQueryDto { + @IsOptional() @Matches(/^\d+$/) afterSeq = "0"; + @IsOptional() @Type(() => Number) @IsInt() @Min(1) @Max(100) limit = 50; +} +export class ConversationsQueryDto { + @IsOptional() @IsString() @Matches(/^[A-Za-z0-9_-]+$/) cursor?: string; + @IsOptional() @Type(() => Number) @IsInt() @Min(1) @Max(100) limit = 20; +} +export class ReadConversationDto { + @Matches(/^\d+$/) seq!: string; +} + +export class SocketReadDto extends ReadConversationDto { + @IsUUID() conversationId!: string; +} diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 0073ae9..9de5203 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -49,19 +49,25 @@ importers: apps/api: dependencies: - "@drift/contracts": + '@drift/contracts': specifier: workspace:* version: link:../../packages/contracts - "@nestjs/common": + '@nestjs/common': specifier: ^11.1.6 version: 11.2.3(class-transformer@0.5.1)(class-validator@0.14.4)(reflect-metadata@0.2.2)(rxjs@7.8.2) - "@nestjs/core": + '@nestjs/core': specifier: ^11.1.6 - version: 11.2.3(@nestjs/common@11.2.3(class-transformer@0.5.1)(class-validator@0.14.4)(reflect-metadata@0.2.2)(rxjs@7.8.2))(@nestjs/platform-express@11.2.3)(reflect-metadata@0.2.2)(rxjs@7.8.2) - "@nestjs/platform-express": + version: 11.2.3(@nestjs/common@11.2.3(class-transformer@0.5.1)(class-validator@0.14.4)(reflect-metadata@0.2.2)(rxjs@7.8.2))(@nestjs/platform-express@11.2.3)(@nestjs/websockets@11.2.5)(reflect-metadata@0.2.2)(rxjs@7.8.2) + '@nestjs/platform-express': specifier: ^11.1.6 version: 11.2.3(@nestjs/common@11.2.3(class-transformer@0.5.1)(class-validator@0.14.4)(reflect-metadata@0.2.2)(rxjs@7.8.2))(@nestjs/core@11.2.3) - "@prisma/client": + '@nestjs/platform-socket.io': + specifier: ^11.1.6 + version: 11.2.5(@nestjs/common@11.2.3(class-transformer@0.5.1)(class-validator@0.14.4)(reflect-metadata@0.2.2)(rxjs@7.8.2))(@nestjs/websockets@11.2.5)(rxjs@7.8.2) + '@nestjs/websockets': + specifier: ^11.1.6 + version: 11.2.5(@nestjs/common@11.2.3(class-transformer@0.5.1)(class-validator@0.14.4)(reflect-metadata@0.2.2)(rxjs@7.8.2))(@nestjs/core@11.2.3)(@nestjs/platform-socket.io@11.2.5)(reflect-metadata@0.2.2)(rxjs@7.8.2) + '@prisma/client': specifier: 6.19.0 version: 6.19.0(prisma@6.19.0(typescript@5.9.3))(typescript@5.9.3) class-transformer: @@ -85,16 +91,22 @@ importers: rxjs: specifier: ^7.8.2 version: 7.8.2 + socket.io: + specifier: ^4.8.1 + version: 4.8.3 devDependencies: - "@nestjs/testing": + '@nestjs/testing': specifier: ^11.1.6 version: 11.2.3(@nestjs/common@11.2.3(class-transformer@0.5.1)(class-validator@0.14.4)(reflect-metadata@0.2.2)(rxjs@7.8.2))(@nestjs/core@11.2.3)(@nestjs/platform-express@11.2.3) - "@types/express": + '@types/express': specifier: ^5.0.3 version: 5.0.6 - "@types/supertest": + '@types/supertest': specifier: ^6.0.3 version: 6.0.3 + socket.io-client: + specifier: ^4.8.1 + version: 4.8.3 supertest: specifier: ^7.1.4 version: 7.2.2 @@ -3252,6 +3264,92 @@ packages: } engines: { node: ">=10" } + '@nestjs/platform-socket.io@11.2.5': + resolution: {integrity: sha512-xOUCwjWfhnhpR5HBKMS3pYOcMYhXhKKR6duxmV7WLFmhgeIZjBalW/b2LEr3C/07+bq3BGTPZsVYunnTYBJDVw==} + peerDependencies: + '@nestjs/common': ^11.0.0 + '@nestjs/websockets': ^11.0.0 + rxjs: ^7.1.0 + + '@nestjs/websockets@11.2.5': + resolution: {integrity: sha512-KSTpTWKFSrBYxxXyclDg3ir9EzPWUU8iuKK4VFaVno2L+kwZcZzsKmnXrcMnmx8Sl2glgqXQg26Esu8hbo6a0Q==} + peerDependencies: + '@nestjs/common': ^11.0.0 + '@nestjs/core': ^11.0.0 + '@nestjs/platform-socket.io': ^11.0.0 + reflect-metadata: ^0.1.12 || ^0.2.0 + rxjs: ^7.1.0 + peerDependenciesMeta: + '@nestjs/platform-socket.io': + optional: true + + '@socket.io/component-emitter@3.1.2': + resolution: {integrity: sha512-9BCxFwvbGg/RsZK9tjXd8s4UcwR0MWeFQ1XEKIQVVvAGJyINdrqKMcTRyLoK8Rse1GjzLV9cwjWV1olXRWEXVA==} + + '@types/cors@2.8.19': + resolution: {integrity: sha512-mFNylyeyqN93lfe/9CSxOGREz8cpzAhH+E93xJ4xWQf62V8sQ/24reV2nyzUWM6H6Xji+GGHpkbLe7pVoUEskg==} + + '@types/ws@8.18.1': + resolution: {integrity: sha512-ThVF6DCVhA8kUGy+aazFQ4kXQ7E1Ty7A3ypFOe0IcJV8O/M511G99AW24irKrW56Wt44yG9+ij8FaqoBGkuBXg==} + + accepts@1.3.8: + resolution: {integrity: sha512-PYAthTa2m2VKxuvSD3DPC/Gy+U+sOA1LAuT8mkmRuvw+NACSaeXEQ+NHcVF7rONl6qcaxV3Uuemwawk+7+SJLw==} + engines: {node: '>= 0.6'} + + base64id@2.0.0: + resolution: {integrity: sha512-lGe34o6EHj9y3Kts9R4ZYs/Gr+6N7MCaMlIFA3F1R2O5/m7K06AxfSeO5530PEERE6/WyEg3lsuyw4GHlPZHog==} + engines: {node: ^4.5.0 || >= 5.9} + + engine.io-client@6.6.6: + resolution: {integrity: sha512-iY6QdftLQ9pyiPoX082bpf/u1UewnOaJrtJIF9T0++QB34lZrj0uP+Q/bj8AlUsAxqhnkTV2BS8SBZSxOmoV5Q==} + + engine.io-parser@5.2.3: + resolution: {integrity: sha512-HqD3yTBfnBxIrbnM1DoD6Pcq8NECnh8d4As1Qgh0z5Gg3jRRIqijury0CL3ghu/edArpUYiYqQiDUQBIs4np3Q==} + engines: {node: '>=10.0.0'} + + engine.io@6.6.10: + resolution: {integrity: sha512-9/lX2bdlizlCXMHRMOIm03VBQHQYC7VvydcxtTAUJRxNW1QzM/2PMFSmr6h/lCiMHcyCP6abK+t9Q+j4vekk8Q==} + engines: {node: '>=10.2.0'} + + negotiator@0.6.3: + resolution: {integrity: sha512-+EUsqGPLsM+j/zdChZjsnX51g4XrHFOIXwfnCVPGlQk/k5giakcKsuxCObBRu6DSm9opw/O6slWbJdghQM4bBg==} + engines: {node: '>= 0.6'} + + object-hash@3.0.0: + resolution: {integrity: sha512-RSn9F68PjH9HqtltsSnqYC1XXoWe9Bju5+213R98cNGttag9q9yAOTzdbsqvIa7aNm5WffBZFpWYr2aWrklWAw==} + engines: {node: '>= 6'} + + socket.io-adapter@2.5.8: + resolution: {integrity: sha512-6Oy52pbg+kvdCVvjcN+FnY7BvxZ7cIHNScbvztT/It5d0vbwoJoVZmF2gjJmnV0/4WlXRfG15zc45ySk9Ah8bw==} + + socket.io-client@4.8.3: + resolution: {integrity: sha512-uP0bpjWrjQmUt5DTHq9RuoCBdFJF10cdX9X+a368j/Ft0wmaVgxlrjvK3kjvgCODOMMOz9lcaRzxmso0bTWZ/g==} + engines: {node: '>=10.0.0'} + + socket.io-parser@4.2.7: + resolution: {integrity: sha512-IH/iSeO9T6gz1KkFleGDWkG9N3dl4jXVYUtMhIqH10Md0ttMer8nUNWiP1DKuNrybD2xBrixLJdCC9J6ECoYkg==} + engines: {node: '>=10.0.0'} + + socket.io@4.8.3: + resolution: {integrity: sha512-2Dd78bqzzjE6KPkD5fHZmDAKRNe3J15q+YHDrIsy9WEkqttc7GY+kT9OBLSMaPbQaEd0x1BjcmtMtXkfpc+T5A==} + engines: {node: '>=10.2.0'} + + ws@8.21.3: + resolution: {integrity: sha512-201TZ/kPWxoPr/OKWjquZR1SWKXcvxdH+e1xrx89b3YbmzLMFCLfnaG1HFIgWzJOEWZ7MvpK++odZufgYR50Rw==} + engines: {node: '>=10.0.0'} + peerDependencies: + bufferutil: ^4.0.1 + utf-8-validate: '>=5.0.2' + peerDependenciesMeta: + bufferutil: + optional: true + utf-8-validate: + optional: true + + xmlhttprequest-ssl@2.1.2: + resolution: {integrity: sha512-TEU+nJVUUnA4CYJFLvK5X9AOeH4KvDvhIfm0vV1GaQRtchnG0hgK5p8hw/xjv8cunWYCsiPCSDzObPyhEwq3KQ==} + engines: {node: '>=0.4.0'} + snapshots: "@borewit/text-codec@0.2.2": {} @@ -5004,3 +5102,137 @@ snapshots: wrappy@1.0.2: {} yocto-queue@0.1.0: {} + + '@nestjs/core@11.2.3(@nestjs/common@11.2.3(class-transformer@0.5.1)(class-validator@0.14.4)(reflect-metadata@0.2.2)(rxjs@7.8.2))(@nestjs/platform-express@11.2.3)(@nestjs/websockets@11.2.5)(reflect-metadata@0.2.2)(rxjs@7.8.2)': + dependencies: + '@nestjs/common': 11.2.3(class-transformer@0.5.1)(class-validator@0.14.4)(reflect-metadata@0.2.2)(rxjs@7.8.2) + fast-safe-stringify: 2.1.1 + iterare: 1.2.1 + path-to-regexp: 8.4.2 + reflect-metadata: 0.2.2 + rxjs: 7.8.2 + tslib: 2.8.1 + uid: 2.0.2 + optionalDependencies: + '@nestjs/platform-express': 11.2.3(@nestjs/common@11.2.3(class-transformer@0.5.1)(class-validator@0.14.4)(reflect-metadata@0.2.2)(rxjs@7.8.2))(@nestjs/core@11.2.3) + '@nestjs/websockets': 11.2.5(@nestjs/common@11.2.3(class-transformer@0.5.1)(class-validator@0.14.4)(reflect-metadata@0.2.2)(rxjs@7.8.2))(@nestjs/core@11.2.3)(@nestjs/platform-socket.io@11.2.5)(reflect-metadata@0.2.2)(rxjs@7.8.2) + + '@nestjs/platform-socket.io@11.2.5(@nestjs/common@11.2.3(class-transformer@0.5.1)(class-validator@0.14.4)(reflect-metadata@0.2.2)(rxjs@7.8.2))(@nestjs/websockets@11.2.5)(rxjs@7.8.2)': + dependencies: + '@nestjs/common': 11.2.3(class-transformer@0.5.1)(class-validator@0.14.4)(reflect-metadata@0.2.2)(rxjs@7.8.2) + '@nestjs/websockets': 11.2.5(@nestjs/common@11.2.3(class-transformer@0.5.1)(class-validator@0.14.4)(reflect-metadata@0.2.2)(rxjs@7.8.2))(@nestjs/core@11.2.3)(@nestjs/platform-socket.io@11.2.5)(reflect-metadata@0.2.2)(rxjs@7.8.2) + rxjs: 7.8.2 + socket.io: 4.8.3 + tslib: 2.8.1 + transitivePeerDependencies: + - bufferutil + - supports-color + - utf-8-validate + + '@nestjs/websockets@11.2.5(@nestjs/common@11.2.3(class-transformer@0.5.1)(class-validator@0.14.4)(reflect-metadata@0.2.2)(rxjs@7.8.2))(@nestjs/core@11.2.3)(@nestjs/platform-socket.io@11.2.5)(reflect-metadata@0.2.2)(rxjs@7.8.2)': + dependencies: + '@nestjs/common': 11.2.3(class-transformer@0.5.1)(class-validator@0.14.4)(reflect-metadata@0.2.2)(rxjs@7.8.2) + '@nestjs/core': 11.2.3(@nestjs/common@11.2.3(class-transformer@0.5.1)(class-validator@0.14.4)(reflect-metadata@0.2.2)(rxjs@7.8.2))(@nestjs/platform-express@11.2.3)(@nestjs/websockets@11.2.5)(reflect-metadata@0.2.2)(rxjs@7.8.2) + iterare: 1.2.1 + object-hash: 3.0.0 + reflect-metadata: 0.2.2 + rxjs: 7.8.2 + tslib: 2.8.1 + optionalDependencies: + '@nestjs/platform-socket.io': 11.2.5(@nestjs/common@11.2.3(class-transformer@0.5.1)(class-validator@0.14.4)(reflect-metadata@0.2.2)(rxjs@7.8.2))(@nestjs/websockets@11.2.5)(rxjs@7.8.2) + + '@socket.io/component-emitter@3.1.2': {} + + '@types/cors@2.8.19': + dependencies: + '@types/node': 22.19.3 + + '@types/ws@8.18.1': + dependencies: + '@types/node': 22.19.3 + + accepts@1.3.8: + dependencies: + mime-types: 2.1.35 + negotiator: 0.6.3 + + base64id@2.0.0: {} + + engine.io-client@6.6.6: + dependencies: + '@socket.io/component-emitter': 3.1.2 + debug: 4.4.3 + engine.io-parser: 5.2.3 + ws: 8.21.3 + xmlhttprequest-ssl: 2.1.2 + transitivePeerDependencies: + - bufferutil + - supports-color + - utf-8-validate + + engine.io-parser@5.2.3: {} + + engine.io@6.6.10: + dependencies: + '@types/cors': 2.8.19 + '@types/node': 22.19.3 + '@types/ws': 8.18.1 + accepts: 1.3.8 + cookie: 0.7.2 + cors: 2.8.6 + debug: 4.4.3 + engine.io-parser: 5.2.3 + ws: 8.21.3 + transitivePeerDependencies: + - bufferutil + - supports-color + - utf-8-validate + + negotiator@0.6.3: {} + + object-hash@3.0.0: {} + + socket.io-adapter@2.5.8: + dependencies: + debug: 4.4.3 + ws: 8.21.3 + transitivePeerDependencies: + - bufferutil + - supports-color + - utf-8-validate + + socket.io-client@4.8.3: + dependencies: + '@socket.io/component-emitter': 3.1.2 + debug: 4.4.3 + engine.io-client: 6.6.6 + socket.io-parser: 4.2.7 + transitivePeerDependencies: + - bufferutil + - supports-color + - utf-8-validate + + socket.io-parser@4.2.7: + dependencies: + '@socket.io/component-emitter': 3.1.2 + debug: 4.4.3 + transitivePeerDependencies: + - supports-color + + socket.io@4.8.3: + dependencies: + accepts: 1.3.8 + base64id: 2.0.0 + cors: 2.8.6 + debug: 4.4.3 + engine.io: 6.6.10 + socket.io-adapter: 2.5.8 + socket.io-parser: 4.2.7 + transitivePeerDependencies: + - bufferutil + - supports-color + - utf-8-validate + + ws@8.21.3: {} + + xmlhttprequest-ssl@2.1.2: {} diff --git a/prisma/migrations/0010_realtime_conversation/migration.sql b/prisma/migrations/0010_realtime_conversation/migration.sql new file mode 100644 index 0000000..7f60aa9 --- /dev/null +++ b/prisma/migrations/0010_realtime_conversation/migration.sql @@ -0,0 +1,24 @@ +-- Task 7 chat identity snapshots, query and delivery indexes. +ALTER TABLE "messages" ADD COLUMN "sender_public_id" UUID; +UPDATE "messages" AS m +SET "sender_public_id" = p."public_id" +FROM "anonymous_profiles" AS p +WHERE p."account_id" = m."sender_id"; +-- Legacy/seed rows may predate anonymous profiles. Preserve an unlinkable +-- snapshot instead of deriving or exposing the private account identifier. +UPDATE "messages" +SET "sender_public_id" = gen_random_uuid() +WHERE "sender_public_id" IS NULL; +ALTER TABLE "messages" ALTER COLUMN "sender_public_id" SET NOT NULL; + +-- Outbox events already provide durable message event dedupe. +CREATE INDEX "messages_conversation_seq_inbox_idx" + ON "messages" ("conversation_id", "seq"); +CREATE INDEX "conversation_members_account_conversation_idx" + ON "conversation_members" ("account_id", "conversation_id") + WHERE "left_at" IS NULL; +CREATE INDEX "conversations_last_message_cursor_idx" + ON "conversations" ("last_message_at" DESC, "id" DESC); +CREATE INDEX "sanctions_active_account_idx" + ON "sanctions" ("account_id", "starts_at", "expires_at") + WHERE "revoked_at" IS NULL; diff --git a/prisma/schema.prisma b/prisma/schema.prisma index bd6cf30..265e291 100644 --- a/prisma/schema.prisma +++ b/prisma/schema.prisma @@ -227,19 +227,19 @@ model Bottle { model BottlePickLease { /// Partial unique index bottle_pick_leases_one_active_per_bottle is managed in 0001_init SQL. - id String @id @default(uuid()) @db.Uuid - bottleId String @map("bottle_id") @db.Uuid - pickerId String @map("picker_id") @db.Uuid - leaseTokenHash String @unique @map("lease_token_hash") @db.VarChar(255) - leaseTokenCiphertext Bytes @map("lease_token_ciphertext") - expiresAt DateTime @map("expires_at") @db.Timestamptz(3) - status BottlePickLeaseStatus @default(ACTIVE) - endedAt DateTime? @map("ended_at") @db.Timestamptz(3) - createdAt DateTime @default(now()) @map("created_at") @db.Timestamptz(3) - bottle Bottle @relation(fields: [bottleId], references: [id], onDelete: Cascade) - activeForBottle Bottle? @relation("ActiveBottleLease") - picker Account @relation("PickerLeases", fields: [pickerId], references: [id], onDelete: Cascade) - pickRequest BottlePickRequest? + id String @id @default(uuid()) @db.Uuid + bottleId String @map("bottle_id") @db.Uuid + pickerId String @map("picker_id") @db.Uuid + leaseTokenHash String @unique @map("lease_token_hash") @db.VarChar(255) + leaseTokenCiphertext Bytes @map("lease_token_ciphertext") + expiresAt DateTime @map("expires_at") @db.Timestamptz(3) + status BottlePickLeaseStatus @default(ACTIVE) + endedAt DateTime? @map("ended_at") @db.Timestamptz(3) + createdAt DateTime @default(now()) @map("created_at") @db.Timestamptz(3) + bottle Bottle @relation(fields: [bottleId], references: [id], onDelete: Cascade) + activeForBottle Bottle? @relation("ActiveBottleLease") + picker Account @relation("PickerLeases", fields: [pickerId], references: [id], onDelete: Cascade) + pickRequest BottlePickRequest? @@index([pickerId, expiresAt]) @@index([bottleId, expiresAt]) @@ -304,10 +304,12 @@ model Conversation { reports Report[] @@index([status, updatedAt]) + @@index([lastMessageAt(sort: Desc), id(sort: Desc)], map: "conversations_last_message_cursor_idx") @@map("conversations") } model ConversationMember { + /// Active member inbox index conversation_members_account_conversation_idx is managed in 0010 SQL. id String @id @default(uuid()) @db.Uuid conversationId String @map("conversation_id") @db.Uuid accountId String @map("account_id") @db.Uuid @@ -328,6 +330,7 @@ model Message { id String @id @default(uuid()) @db.Uuid conversationId String @map("conversation_id") @db.Uuid senderId String @map("sender_id") @db.Uuid + senderPublicId String @map("sender_public_id") @db.Uuid clientMsgId String @map("client_msg_id") @db.VarChar(128) seq BigInt contentText String @map("content_text") @db.Text @@ -343,6 +346,7 @@ model Message { @@unique([conversationId, clientMsgId]) @@unique([conversationId, seq]) + @@index([conversationId, seq], map: "messages_conversation_seq_inbox_idx") @@index([senderId, createdAt]) @@map("messages") } diff --git a/tests/integration/database.spec.ts b/tests/integration/database.spec.ts index 94c3b91..369f09d 100644 --- a/tests/integration/database.spec.ts +++ b/tests/integration/database.spec.ts @@ -298,6 +298,7 @@ describe("database authority constraints", () => { data: { conversationId: conversation.id, senderId: author.id, + senderPublicId: author.id, clientMsgId: "msg", seq: 1n, contentText: "hello", @@ -326,6 +327,7 @@ describe("database authority constraints", () => { data: { conversationId: conversation.id, senderId: author.id, + senderPublicId: author.id, clientMsgId: "duplicate-client-id", seq: 1n, contentText: "first", @@ -338,6 +340,7 @@ describe("database authority constraints", () => { data: { conversationId: conversation.id, senderId: author.id, + senderPublicId: author.id, clientMsgId: "duplicate-client-id", seq: 2n, contentText: "second", @@ -358,6 +361,7 @@ describe("database authority constraints", () => { data: { conversationId: conversation.id, senderId: author.id, + senderPublicId: author.id, clientMsgId: "first-client-id", seq: 1n, contentText: "first", @@ -370,6 +374,7 @@ describe("database authority constraints", () => { data: { conversationId: conversation.id, senderId: author.id, + senderPublicId: author.id, clientMsgId: "second-client-id", seq: 1n, contentText: "second", @@ -426,6 +431,7 @@ describe("database authority constraints", () => { data: { conversationId: conversation.id, senderId: author.id, + senderPublicId: author.id, clientMsgId: "zero", seq: 0n, contentText: "x",