diff --git a/apps/api/src/conversation/chat.gateway.ts b/apps/api/src/conversation/chat.gateway.ts index a2ec49f..0353f20 100644 --- a/apps/api/src/conversation/chat.gateway.ts +++ b/apps/api/src/conversation/chat.gateway.ts @@ -23,6 +23,7 @@ import { type MessageResult, } from "./conversation.service.js"; import { SendMessageDto, SocketReadDto } from "./dto.js"; +import type { ChatPublisher } from "./outbox-message-relay.js"; interface SocketData { token?: string; @@ -72,7 +73,9 @@ function allowOrigin( credentials: true, }, }) -export class ChatGateway implements OnGatewayInit, OnGatewayConnection { +export class ChatGateway + implements OnGatewayInit, OnGatewayConnection, ChatPublisher +{ @WebSocketServer() server!: ChatServer; constructor( @@ -153,13 +156,6 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection { }; 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); @@ -193,6 +189,13 @@ export class ChatGateway implements OnGatewayInit, OnGatewayConnection { } } + publishMessage(memberIds: string[], message: MessageResult["message"]): void { + if (!this.server) throw new Error("chat gateway is not ready"); + for (const member of memberIds) { + this.server.to(`account:${member}`).emit("message:new", message); + } + } + private error(error: unknown): SocketResponse { if (error instanceof DomainException) { return { diff --git a/apps/api/src/conversation/conversation.controller.ts b/apps/api/src/conversation/conversation.controller.ts index 0ad61ed..2b0e190 100644 --- a/apps/api/src/conversation/conversation.controller.ts +++ b/apps/api/src/conversation/conversation.controller.ts @@ -2,6 +2,7 @@ import { Body, Controller, Get, + HttpStatus, Inject, Param, ParseUUIDPipe, @@ -9,10 +10,14 @@ import { Query, UseGuards, } from "@nestjs/common"; +import { ErrorCode } from "@drift/contracts"; import { AuthGuard } from "../auth/auth.guard.js"; import { CurrentUser } from "../auth/current-user.decorator.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 } from "./conversation.service.js"; +import { StateChangingOriginGuard } from "./state-changing-origin.guard.js"; import { ConversationsQueryDto, MessagesQueryDto, @@ -27,8 +32,11 @@ export class ConversationController { constructor( @Inject(ConversationService) private readonly conversations: ConversationService, + @Inject(ChatRateLimiter) private readonly limiter: ChatRateLimiter, ) {} - @Post("bottles/:id/reply") reply( + @UseGuards(StateChangingOriginGuard) + @Post("bottles/:id/reply") + reply( @CurrentUser() user: AccessClaims, @Param("id", new ParseUUIDPipe()) id: string, @Body() dto: ReplyDto, @@ -66,20 +74,50 @@ export class ConversationController { query.limit, ); } - @Post("conversations/:id/messages/prepare") prepare( + @UseGuards(StateChangingOriginGuard) + @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); + return this.consumeAndRun(user, () => + this.conversations.send(user.sub, id, dto.clientMsgId, dto.text), + ); } - @Post("conversations/:id/read") read( + @UseGuards(StateChangingOriginGuard) + @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); } + + private async consumeAndRun( + user: AccessClaims, + operation: () => Promise, + ): Promise { + 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, + ); + } + return operation(); + } } Reflect.defineMetadata( "design:paramtypes", diff --git a/apps/api/src/conversation/conversation.e2e-spec.ts b/apps/api/src/conversation/conversation.e2e-spec.ts index 4a673de..b28996b 100644 --- a/apps/api/src/conversation/conversation.e2e-spec.ts +++ b/apps/api/src/conversation/conversation.e2e-spec.ts @@ -12,6 +12,8 @@ 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"; +import { OutboxMessageRelay } from "./outbox-message-relay.js"; +import { ChatRateLimiter } from "./chat-rate-limiter.js"; process.env.NODE_ENV = "test"; process.env.WEB_ORIGIN = "http://localhost:3000"; @@ -44,6 +46,7 @@ describe("conversation API with real PostgreSQL", () => { await app.listen(0, "127.0.0.1"); }); beforeEach(async () => { + await prisma.outboxEvent.deleteMany(); await prisma.$executeRawUnsafe(`TRUNCATE TABLE "accounts" CASCADE`); }); afterAll(async () => { @@ -127,6 +130,7 @@ describe("conversation API with real PostgreSQL", () => { request(app.getHttpServer()) .post(`/api/v1/bottles/${bottleId}/reply`) .set("Authorization", authorization) + .set("Origin", process.env.WEB_ORIGIN!) .send({ leaseId: lease.id, leaseToken: lease.token, clientMsgId, text }); function socket(token: string, origin = process.env.WEB_ORIGIN): Socket { @@ -294,6 +298,7 @@ describe("conversation API with real PostgreSQL", () => { await request(app.getHttpServer()) .post(`/api/v1/conversations/${conversationId}/messages/prepare`) .set("Authorization", stranger.authorization) + .set("Origin", process.env.WEB_ORIGIN!) .send({ conversationId, clientMsgId, text: "steal" }) .expect(403); @@ -328,6 +333,188 @@ describe("conversation API with real PostgreSQL", () => { expect(JSON.stringify(event.payload)).not.toContain("private message text"); }); + it("enforces Origin before REST mutation rate limiting and maps limiter outcomes", async () => { + const author = await actor("rest-origin-author"); + const picker = await actor("rest-origin-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; + const limiter = app.get(ChatRateLimiter); + const originalConsume = limiter.consume.bind(limiter); + let calls = 0; + limiter.consume = () => { + calls += 1; + return Promise.resolve(true); + }; + const body = { + conversationId: id, + clientMsgId: randomUUID(), + text: "rest", + }; + await request(app.getHttpServer()) + .post(`/api/v1/conversations/${id}/messages/prepare`) + .set("Authorization", picker.authorization) + .send(body) + .expect(403); + await request(app.getHttpServer()) + .post(`/api/v1/conversations/${id}/messages/prepare`) + .set("Authorization", picker.authorization) + .set("Origin", "https://evil.example") + .send(body) + .expect(403); + expect(calls).toBe(0); + await request(app.getHttpServer()) + .post(`/api/v1/conversations/${id}/messages/prepare`) + .set("Authorization", picker.authorization) + .set("Origin", process.env.WEB_ORIGIN!) + .send(body) + .expect(201); + expect(calls).toBe(1); + + limiter.consume = () => Promise.resolve(false); + await request(app.getHttpServer()) + .post(`/api/v1/conversations/${id}/messages/prepare`) + .set("Authorization", picker.authorization) + .set("Origin", process.env.WEB_ORIGIN!) + .send({ ...body, clientMsgId: randomUUID() }) + .expect(429); + limiter.consume = () => Promise.reject(new Error("redis secret")); + await request(app.getHttpServer()) + .post(`/api/v1/conversations/${id}/messages/prepare`) + .set("Authorization", picker.authorization) + .set("Origin", process.env.WEB_ORIGIN!) + .send({ ...body, clientMsgId: randomUUID() }) + .expect(503); + limiter.consume = originalConsume; + }); + + it("relays first replies and bidirectional live messages through the durable outbox", async () => { + const author = await actor("relay-author"); + const picker = await actor("relay-picker"); + const { bottle, lease } = await leased(author.id, picker); + const authorSocket = socket(author.token); + const pickerSocket = socket(picker.token); + await Promise.all([connected(authorSocket), connected(pickerSocket)]); + const authorMessages: Array<{ clientMsgId: string }> = []; + const pickerMessages: Array<{ clientMsgId: string }> = []; + authorSocket.on("message:new", (message) => authorMessages.push(message)); + pickerSocket.on("message:new", (message) => pickerMessages.push(message)); + + const firstId = randomUUID(); + const created = await reply( + picker.authorization, + bottle.id, + lease, + firstId, + ).expect(201); + const conversationId = created.body.data.conversationId as string; + expect(await app.get(OutboxMessageRelay).runOnce()).toBe(true); + await new Promise((resolve) => setTimeout(resolve, 25)); + expect(authorMessages.map((message) => message.clientMsgId)).toEqual([ + firstId, + ]); + expect(pickerMessages.map((message) => message.clientMsgId)).toEqual([ + firstId, + ]); + + const pickerToAuthor = randomUUID(); + expect( + ( + await ack<{ ok: boolean }>(pickerSocket, "message:send", { + conversationId, + clientMsgId: pickerToAuthor, + text: "picker to author", + }) + ).ok, + ).toBe(true); + expect(await app.get(OutboxMessageRelay).runOnce()).toBe(true); + const authorToPicker = randomUUID(); + expect( + ( + await ack<{ ok: boolean }>(authorSocket, "message:send", { + conversationId, + clientMsgId: authorToPicker, + text: "author to picker", + }) + ).ok, + ).toBe(true); + expect(await app.get(OutboxMessageRelay).runOnce()).toBe(true); + await new Promise((resolve) => setTimeout(resolve, 25)); + const expected = [firstId, pickerToAuthor, authorToPicker]; + expect(authorMessages.map((message) => message.clientMsgId)).toEqual( + expected, + ); + expect(pickerMessages.map((message) => message.clientMsgId)).toEqual( + expected, + ); + expect( + await prisma.outboxEvent.count({ + where: { eventType: "MESSAGE_CREATED", status: "PUBLISHED" }, + }), + ).toBe(3); + authorSocket.close(); + pickerSocket.close(); + }); + + it("routes relayed messages from canonical membership instead of outbox payload", async () => { + const author = await actor("relay-route-author"); + const picker = await actor("relay-route-picker"); + const stranger = await actor("relay-route-stranger"); + const { bottle, lease } = await leased(author.id, picker); + const authorSocket = socket(author.token); + const pickerSocket = socket(picker.token); + const strangerSocket = socket(stranger.token); + await Promise.all([ + connected(authorSocket), + connected(pickerSocket), + connected(strangerSocket), + ]); + const authorMessages: string[] = []; + const pickerMessages: string[] = []; + const strangerMessages: string[] = []; + authorSocket.on("message:new", (message) => + authorMessages.push(message.id), + ); + pickerSocket.on("message:new", (message) => + pickerMessages.push(message.id), + ); + strangerSocket.on("message:new", (message) => + strangerMessages.push(message.id), + ); + + const created = await reply( + picker.authorization, + bottle.id, + lease, + randomUUID(), + ).expect(201); + const messageId = created.body.data.message.id as string; + await prisma.outboxEvent.update({ + where: { dedupeKey: `message-created:${messageId}` }, + data: { + payload: { + messageId, + conversationId: created.body.data.conversationId, + memberIds: [stranger.id], + }, + }, + }); + + expect(await app.get(OutboxMessageRelay).runOnce()).toBe(true); + await new Promise((resolve) => setTimeout(resolve, 25)); + expect(authorMessages).toEqual([messageId]); + expect(pickerMessages).toEqual([messageId]); + expect(strangerMessages).toEqual([]); + authorSocket.close(); + pickerSocket.close(); + strangerSocket.close(); + }); + it("uses an exact timestamp/id cursor without skipping tied rows", async () => { const owner = await actor("owner"); const peer = await actor("peer"); @@ -395,14 +582,82 @@ describe("conversation API with real PostgreSQL", () => { await request(app.getHttpServer()) .post(`/api/v1/conversations/${id}/read`) .set("Authorization", picker.authorization) + .set("Origin", process.env.WEB_ORIGIN!) .send({ seq: "1" }) .expect(201); const monotonic = await request(app.getHttpServer()) .post(`/api/v1/conversations/${id}/read`) .set("Authorization", picker.authorization) + .set("Origin", process.env.WEB_ORIGIN!) .send({ seq: "0" }) .expect(201); expect(monotonic.body.data.lastReadSeq).toBe("1"); + + await prisma.$transaction([ + prisma.message.create({ + data: { + conversationId: id, + senderId: author.id, + senderPublicId: ( + await prisma.anonymousProfile.findUniqueOrThrow({ + where: { accountId: author.id }, + }) + ).publicId, + clientMsgId: randomUUID(), + seq: 2n, + contentText: "second", + reviewStatus: "APPROVED", + }, + }), + prisma.message.create({ + data: { + conversationId: id, + senderId: author.id, + senderPublicId: ( + await prisma.anonymousProfile.findUniqueOrThrow({ + where: { accountId: author.id }, + }) + ).publicId, + clientMsgId: randomUUID(), + seq: 3n, + contentText: "third", + reviewStatus: "APPROVED", + }, + }), + prisma.conversation.update({ where: { id }, data: { nextSeq: 4n } }), + ]); + const reads = await Promise.all( + ["3", "2"].map((seq) => + request(app.getHttpServer()) + .post(`/api/v1/conversations/${id}/read`) + .set("Authorization", picker.authorization) + .set("Origin", process.env.WEB_ORIGIN!) + .send({ seq }), + ), + ); + expect(reads.map((result) => result.status)).toEqual([201, 201]); + expect( + ( + await prisma.conversationMember.findUniqueOrThrow({ + where: { + conversationId_accountId: { + conversationId: id, + accountId: picker.id, + }, + }, + }) + ).lastReadSeq, + ).toBe(3n); + + const pickerSocket = socket(picker.token); + await connected(pickerSocket); + await expect( + ack(pickerSocket, "conversation:read", { conversationId: id, seq: "2" }), + ).resolves.toMatchObject({ + ok: true, + data: { conversationId: id, lastReadSeq: "3" }, + }); + pickerSocket.close(); await prisma.conversationMember.update({ where: { conversationId_accountId: { conversationId: id, accountId: picker.id }, @@ -412,6 +667,7 @@ describe("conversation API with real PostgreSQL", () => { await request(app.getHttpServer()) .post(`/api/v1/conversations/${id}/read`) .set("Authorization", picker.authorization) + .set("Origin", process.env.WEB_ORIGIN!) .send({ seq: "1" }) .expect(403); }); @@ -427,6 +683,7 @@ describe("conversation API with real PostgreSQL", () => { randomUUID(), ).expect(201); const conversationId = created.body.data.conversationId as string; + expect(await app.get(OutboxMessageRelay).runOnce()).toBe(true); const unauthenticated = socket("invalid-token"); const denied = new Promise((resolve) => @@ -459,6 +716,7 @@ describe("conversation API with real PostgreSQL", () => { text: "socket hello", }); expect(first).toMatchObject({ ok: true, data: { deduplicated: false } }); + expect(await app.get(OutboxMessageRelay).runOnce()).toBe(true); await new Promise((resolve) => setTimeout(resolve, 50)); expect([authorDeliveries, pickerDeliveries]).toEqual([1, 1]); const duplicate = await ack<{ data: { deduplicated: boolean } }>( diff --git a/apps/api/src/conversation/conversation.module.ts b/apps/api/src/conversation/conversation.module.ts index 3991063..6c8d70e 100644 --- a/apps/api/src/conversation/conversation.module.ts +++ b/apps/api/src/conversation/conversation.module.ts @@ -6,11 +6,20 @@ 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"; +import { CHAT_PUBLISHER, OutboxMessageRelay } from "./outbox-message-relay.js"; +import { StateChangingOriginGuard } from "./state-changing-origin.guard.js"; @Module({ imports: [DatabaseModule, RedisModule, AuthModule], controllers: [ConversationController], - providers: [ConversationService, ChatRateLimiter, ChatGateway], - exports: [ConversationService], + providers: [ + ConversationService, + ChatRateLimiter, + ChatGateway, + { provide: CHAT_PUBLISHER, useExisting: ChatGateway }, + OutboxMessageRelay, + StateChangingOriginGuard, + ], + exports: [ConversationService, OutboxMessageRelay], }) export class ConversationModule {} diff --git a/apps/api/src/conversation/conversation.service.ts b/apps/api/src/conversation/conversation.service.ts index c3d0823..02d11e1 100644 --- a/apps/api/src/conversation/conversation.service.ts +++ b/apps/api/src/conversation/conversation.service.ts @@ -145,10 +145,7 @@ export class ConversationService { include: firstMessageInclude, }); const message = conversation.messages[0]!; - await this.createMessageOutbox(tx, message.id, conversation.id, [ - accountId, - authorId, - ]); + await this.createMessageOutbox(tx, message.id, conversation.id); const consumed = await tx.bottle.updateMany({ where: { id: bottleId, @@ -260,12 +257,7 @@ export class ConversationService { const memberIds = conversation.members.map( (member) => member.accountId, ); - await this.createMessageOutbox( - tx, - message.id, - conversationId, - memberIds, - ); + await this.createMessageOutbox(tx, message.id, conversationId); return { message: this.serializeMessage(message), deduplicated: false, @@ -372,28 +364,36 @@ export class ConversationService { } 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: { + const rows = await this.prisma.$queryRaw>` + UPDATE "conversation_members" AS member + SET "last_read_seq"=GREATEST(member."last_read_seq", ${seq}) + FROM "conversations" AS conversation + WHERE member."conversation_id"=${id}::uuid + AND member."account_id"=${accountId}::uuid + AND member."left_at" IS NULL + AND conversation."id"=member."conversation_id" + AND ${seq} < conversation."next_seq" + RETURNING member."last_read_seq" AS "lastReadSeq" + `; + if (rows[0]) { + return { conversationId: id, - accountId, - leftAt: null, - lastReadSeq: { lt: seq }, - }, - data: { lastReadSeq: seq }, - }); - if (updated.count !== 1) throw this.forbidden(); - return { conversationId: id, lastReadSeq: seq.toString() }; + lastReadSeq: rows[0].lastReadSeq.toString(), + }; + } + const [member, conversation] = await Promise.all([ + this.prisma.conversationMember.findUnique({ + where: { conversationId_accountId: { conversationId: id, accountId } }, + }), + this.prisma.conversation.findUnique({ + where: { id }, + select: { nextSeq: true }, + }), + ]); + if (!member || member.leftAt || !conversation) throw this.forbidden(); + if (seq >= conversation.nextSeq) throw this.validation(); + return { conversationId: id, lastReadSeq: member.lastReadSeq.toString() }; } private async authorizeConversation( @@ -426,7 +426,6 @@ export class ConversationService { tx: Prisma.TransactionClient, messageId: string, conversationId: string, - memberIds: string[], ) { await tx.outboxEvent.create({ data: { @@ -434,7 +433,7 @@ export class ConversationService { aggregateId: messageId, eventType: "MESSAGE_CREATED", dedupeKey: `message-created:${messageId}`, - payload: { messageId, conversationId, memberIds }, + payload: { messageId, conversationId }, }, }); } diff --git a/apps/api/src/conversation/outbox-message-relay.ts b/apps/api/src/conversation/outbox-message-relay.ts new file mode 100644 index 0000000..ba93cc5 --- /dev/null +++ b/apps/api/src/conversation/outbox-message-relay.ts @@ -0,0 +1,206 @@ +import { + Inject, + Injectable, + type OnModuleDestroy, + type OnModuleInit, +} from "@nestjs/common"; +import { Prisma } from "@prisma/client"; +import { randomUUID } from "node:crypto"; +import { PrismaService } from "../database/prisma.service.js"; +import { ConversationService } from "./conversation.service.js"; + +export const CHAT_PUBLISHER = Symbol("CHAT_PUBLISHER"); + +export interface ChatPublisher { + publishMessage(memberIds: string[], message: MessagePayload): void; +} + +type MessagePayload = ReturnType; +type ClaimedEvent = { + id: string; + aggregateId: string; + payload: Prisma.JsonValue; + attempts: number; + lockToken: string; +}; + +const NEVER = new Date("9999-12-31T23:59:59.999Z"); + +@Injectable() +export class OutboxMessageRelay implements OnModuleInit, OnModuleDestroy { + private stopped = false; + private loopPromise: Promise | undefined; + private wakeLoop: (() => void) | undefined; + private readonly leaseMs = Number(process.env.OUTBOX_LEASE_MS ?? 30_000); + private readonly maxAttempts = Number(process.env.OUTBOX_MAX_ATTEMPTS ?? 5); + + constructor( + @Inject(PrismaService) private readonly prisma: PrismaService, + @Inject(ConversationService) + private readonly conversations: ConversationService, + @Inject(CHAT_PUBLISHER) private readonly publisher: ChatPublisher, + ) {} + + onModuleInit(): void { + if (process.env.NODE_ENV !== "test") this.loopPromise = this.loop(); + } + + async onModuleDestroy(): Promise { + this.stopped = true; + this.wakeLoop?.(); + await this.loopPromise; + } + + async runOnce(): Promise { + const event = await this.claim(); + if (!event) return false; + try { + const payload = this.payload(event); + const message = await this.prisma.message.findUnique({ + where: { id: payload.messageId }, + include: { + conversation: { + select: { + members: { + where: { leftAt: null }, + select: { accountId: true }, + }, + }, + }, + }, + }); + if ( + !message || + message.id !== event.aggregateId || + message.conversationId !== payload.conversationId + ) { + throw new Error("invalid message event"); + } + this.publisher.publishMessage( + message.conversation.members.map((member) => member.accountId), + this.conversations.serializeMessage(message), + ); + const completed = await this.prisma.outboxEvent.updateMany({ + where: { + id: event.id, + eventType: "MESSAGE_CREATED", + status: "PROCESSING", + lockToken: event.lockToken, + }, + data: { + status: "PUBLISHED", + publishedAt: new Date(), + lockedAt: null, + lockToken: null, + }, + }); + if (completed.count !== 1) throw new Error("outbox lease lost"); + } catch (error) { + const exhausted = event.attempts >= this.maxAttempts; + const delayMs = Math.min(60_000, 1000 * 2 ** (event.attempts - 1)); + await this.prisma.outboxEvent.updateMany({ + where: { + id: event.id, + eventType: "MESSAGE_CREATED", + status: "PROCESSING", + lockToken: event.lockToken, + }, + data: { + status: "FAILED", + lockedAt: null, + lockToken: null, + nextRetryAt: exhausted ? NEVER : new Date(Date.now() + delayMs), + }, + }); + console.error( + JSON.stringify({ + eventId: event.id, + eventType: "MESSAGE_CREATED", + attempt: event.attempts, + errorClass: error instanceof Error ? error.name : "UnknownError", + }), + ); + } + return true; + } + + private async loop(): Promise { + while (!this.stopped) { + const handled = await this.runOnce().catch(() => false); + if (!handled && !this.stopped) await this.waitForWork(); + } + } + + private async waitForWork(): Promise { + await new Promise((resolve) => { + const timer = setTimeout(resolve, 100); + this.wakeLoop = () => { + clearTimeout(timer); + resolve(); + }; + }); + this.wakeLoop = undefined; + } + + private payload(event: ClaimedEvent): { + messageId: string; + conversationId: string; + } { + const value = event.payload; + if (!value || typeof value !== "object" || Array.isArray(value)) + throw new Error("invalid message payload"); + const payload = value as Record; + if ( + typeof payload.messageId !== "string" || + typeof payload.conversationId !== "string" + ) { + throw new Error("invalid message payload"); + } + return { + messageId: payload.messageId, + conversationId: payload.conversationId, + }; + } + + private async claim(): Promise { + const staleBefore = new Date(Date.now() - this.leaseMs); + return this.prisma.$transaction(async (tx) => { + await tx.outboxEvent.updateMany({ + where: { + eventType: "MESSAGE_CREATED", + status: "PROCESSING", + lockedAt: { lt: staleBefore }, + attempts: { gte: this.maxAttempts }, + }, + data: { + status: "FAILED", + lockedAt: null, + lockToken: null, + nextRetryAt: NEVER, + }, + }); + const rows = await tx.$queryRaw>>` + SELECT "id", "aggregate_id" AS "aggregateId", "payload", "attempts" + FROM "outbox_events" + WHERE "event_type" = 'MESSAGE_CREATED' + AND "attempts" < ${this.maxAttempts} + AND (("status" IN ('PENDING', 'FAILED') AND "next_retry_at" <= now()) + OR ("status" = 'PROCESSING' AND "locked_at" < ${staleBefore})) + ORDER BY "created_at" + FOR UPDATE SKIP LOCKED LIMIT 1`; + const row = rows[0]; + if (!row) return null; + const lockToken = randomUUID(); + await tx.outboxEvent.update({ + where: { id: row.id }, + data: { + status: "PROCESSING", + lockedAt: new Date(), + lockToken, + attempts: { increment: 1 }, + }, + }); + return { ...row, attempts: row.attempts + 1, lockToken }; + }); + } +} diff --git a/apps/api/src/conversation/state-changing-origin.guard.ts b/apps/api/src/conversation/state-changing-origin.guard.ts new file mode 100644 index 0000000..8d367e5 --- /dev/null +++ b/apps/api/src/conversation/state-changing-origin.guard.ts @@ -0,0 +1,23 @@ +import { + CanActivate, + ExecutionContext, + HttpStatus, + Injectable, +} from "@nestjs/common"; +import { ErrorCode } from "@drift/contracts"; +import type { Request } from "express"; +import { webOrigin } from "../auth/auth.config.js"; +import { DomainException } from "../common/domain.exception.js"; + +@Injectable() +export class StateChangingOriginGuard implements CanActivate { + canActivate(context: ExecutionContext): boolean { + const request = context.switchToHttp().getRequest(); + if (request.headers.origin === webOrigin()) return true; + throw new DomainException( + ErrorCode.CONVERSATION_FORBIDDEN, + "Forbidden", + HttpStatus.FORBIDDEN, + ); + } +} diff --git a/apps/worker/src/moderation-worker.spec.ts b/apps/worker/src/moderation-worker.spec.ts index 983affd..5f09ea5 100644 --- a/apps/worker/src/moderation-worker.spec.ts +++ b/apps/worker/src/moderation-worker.spec.ts @@ -39,6 +39,42 @@ describe("moderation worker with real PostgreSQL", () => { }, ); + it.each([ + ["PENDING", null, 0], + ["PROCESSING", new Date(Date.now() - 60_000), 5], + ] as const)( + "leaves %s MESSAGE_CREATED for the API relay", + async (status, lockedAt, attempts) => { + await prisma.account.create({ + data: { + phoneCiphertext: Buffer.from("cipher"), + phoneHmac: randomUUID(), + }, + }); + const event = await prisma.outboxEvent.create({ + data: { + aggregateType: "MESSAGE", + aggregateId: randomUUID(), + eventType: "MESSAGE_CREATED", + dedupeKey: randomUUID(), + payload: { + messageId: randomUUID(), + conversationId: randomUUID(), + }, + status, + lockedAt, + lockToken: lockedAt ? randomUUID() : null, + attempts, + }, + }); + + expect(await new ModerationWorker(prisma).runOnce()).toBe(false); + expect( + await prisma.outboxEvent.findUniqueOrThrow({ where: { id: event.id } }), + ).toMatchObject({ status, attempts, lockedAt }); + }, + ); + it("is fail-closed and exponentially reschedules failures", async () => { const { bottle, event } = await fixture("ordinary"); const worker = new ModerationWorker(prisma, () => { diff --git a/apps/worker/src/moderation-worker.ts b/apps/worker/src/moderation-worker.ts index 91cbe15..b2ba0b3 100644 --- a/apps/worker/src/moderation-worker.ts +++ b/apps/worker/src/moderation-worker.ts @@ -382,6 +382,9 @@ export class ModerationWorker { return this.prisma.$transaction(async (tx) => { await tx.outboxEvent.updateMany({ where: { + eventType: { + in: ["BOTTLE_MODERATION_REQUESTED", "PROFILE_MODERATION_REQUESTED"], + }, status: "PROCESSING", lockedAt: { lt: staleBefore }, attempts: { gte: this.maxAttempts },