fix: 完成消息 Outbox 投递与会话状态保护

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 {

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<T>(
+    user: AccessClaims,
+    operation: () => Promise<T>,
+  ): Promise<T> {
+    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",

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<string>((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 } }>(

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 {}

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<Array<{ lastReadSeq: bigint }>>`
+      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 },
       },
     });
   }

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<ConversationService["serializeMessage"]>;
+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<void> | 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<void> {
+    this.stopped = true;
+    this.wakeLoop?.();
+    await this.loopPromise;
+  }
+
+  async runOnce(): Promise<boolean> {
+    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<void> {
+    while (!this.stopped) {
+      const handled = await this.runOnce().catch(() => false);
+      if (!handled && !this.stopped) await this.waitForWork();
+    }
+  }
+
+  private async waitForWork(): Promise<void> {
+    await new Promise<void>((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<string, unknown>;
+    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<ClaimedEvent | null> {
+    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<Array<Omit<ClaimedEvent, "lockToken">>>`
+        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 };
+    });
+  }
+}

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<Request>();
+    if (request.headers.origin === webOrigin()) return true;
+    throw new DomainException(
+      ErrorCode.CONVERSATION_FORBIDDEN,
+      "Forbidden",
+      HttpStatus.FORBIDDEN,
+    );
+  }
+}

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, () => {

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 },