fix: 完善资料审核与 Worker 状态一致性

apps/api/src/bottle/bottle.e2e-spec.ts

@@ -129,6 +129,31 @@ describe("bottles with real PostgreSQL", () => {
     expect((await prisma.dailyUsage.findFirstOrThrow()).bottlesCreated).toBe(1);
   });
 
+  it("returns idempotency conflict for a reused key with different normalized content", async () => {
+    const key = randomUUID();
+    const first = await create(key, "  same content  ").expect(201);
+    const retry = await create(key, "same content").expect(201);
+    expect(retry.body.data.id).toBe(first.body.data.id);
+    const conflict = await create(key, "different content").expect(409);
+    expect(conflict.body.code).toBe("IDEMPOTENCY_CONFLICT");
+    expect(await prisma.bottle.count()).toBe(1);
+    expect((await prisma.dailyUsage.findFirstOrThrow()).bottlesCreated).toBe(1);
+  });
+
+  it("allows one concurrent writer and conflicts the other for different content on one key", async () => {
+    const key = randomUUID();
+    const results = await Promise.all([
+      create(key, "first contender"),
+      create(key, "second contender"),
+    ]);
+    expect(results.map((result) => result.status).sort()).toEqual([201, 409]);
+    expect(results.find((result) => result.status === 409)?.body.code).toBe(
+      "IDEMPOTENCY_CONFLICT",
+    );
+    expect(await prisma.bottle.count()).toBe(1);
+    expect((await prisma.dailyUsage.findFirstOrThrow()).bottlesCreated).toBe(1);
+  });
+
   it("atomically permits exactly ten of twelve concurrent requests", async () => {
     const results = await Promise.all(
       Array.from({ length: 12 }, (_, index) =>

apps/api/src/bottle/bottle.service.ts

@@ -16,22 +16,24 @@ export class BottleService {
         authorId_clientRequestId: { authorId, clientRequestId: requestId },
       },
     });
-    if (existing) return existing;
+    if (existing) return this.resolveIdempotent(existing, contentText);
     try {
       return await this.prisma.$transaction(async (tx) => {
+        await tx.$executeRaw`SELECT pg_advisory_xact_lock(hashtextextended(${`${authorId}:${requestId}`}, 0))`;
         const duplicate = await tx.bottle.findUnique({
           where: {
             authorId_clientRequestId: { authorId, clientRequestId: requestId },
           },
         });
-        if (duplicate) return duplicate;
+        if (duplicate) return this.resolveIdempotent(duplicate, contentText);
+        const now = new Date();
         const sanctioned = await tx.sanction.findFirst({
           where: {
             accountId: authorId,
             type: { in: ["SUSPENSION", "BAN"] },
             revokedAt: null,
-            startsAt: { lte: new Date() },
+            startsAt: { lte: now },
-            OR: [{ expiresAt: null }, { expiresAt: { gt: new Date() } }],
+            OR: [{ expiresAt: null }, { expiresAt: { gt: now } }],
           },
           select: { id: true },
         });
@@ -41,7 +43,7 @@ export class BottleService {
             "Account sanctioned",
             HttpStatus.FORBIDDEN,
           );
-        const usageDate = utc8UsageDate(new Date());
+        const usageDate = utc8UsageDate(now);
         const rows = await tx.$queryRaw<Array<{ bottles_created: number }>>`
           INSERT INTO "daily_usage" ("id", "account_id", "usage_date", "bottles_created", "updated_at")
           VALUES (gen_random_uuid(), ${authorId}::uuid, ${usageDate}::date, 1, now())
@@ -89,16 +91,30 @@ export class BottleService {
         error instanceof Prisma.PrismaClientKnownRequestError &&
         error.code === "P2002"
       ) {
-        return this.prisma.bottle.findUniqueOrThrow({
+        const duplicate = await this.prisma.bottle.findUniqueOrThrow({
           where: {
             authorId_clientRequestId: { authorId, clientRequestId: requestId },
           },
         });
+        return this.resolveIdempotent(duplicate, contentText);
       }
       throw error;
     }
   }
 
+  private resolveIdempotent<T extends { contentText: string }>(
+    bottle: T,
+    contentText: string,
+  ): T {
+    if (bottle.contentText !== contentText)
+      throw new DomainException(
+        ErrorCode.IDEMPOTENCY_CONFLICT,
+        "Idempotency key reused with different content",
+        HttpStatus.CONFLICT,
+      );
+    return bottle;
+  }
+
   async get(authorId: string, id: string) {
     const bottle = await this.prisma.bottle.findFirst({
       where: { id, authorId },

apps/api/src/profile/profile.e2e-spec.ts

@@ -103,6 +103,67 @@ describe("anonymous profile", () => {
     });
   });
 
+  it("atomically queues content-free moderation and versions rapid profile updates", async () => {
+    const first = await request(app.getHttpServer())
+      .patch("/api/v1/me/anonymous-profile")
+      .set("Authorization", authorization)
+      .send({
+        nickname: "  first  ",
+        avatarColor: "#ABCDEF",
+        bio: "  bio one  ",
+      })
+      .expect(200);
+    const profileId = first.body.data.id as string;
+    const task = await prisma.moderationTask.findFirstOrThrow({
+      where: { targetType: "PROFILE", targetId: profileId },
+    });
+    const firstEvent = await prisma.outboxEvent.findFirstOrThrow({
+      where: {
+        eventType: "PROFILE_MODERATION_REQUESTED",
+        aggregateId: profileId,
+      },
+      orderBy: { createdAt: "asc" },
+    });
+    expect(first.body.data).toMatchObject({
+      version: 1,
+      reviewStatus: "REVIEWING",
+    });
+    expect(firstEvent.payload).toEqual({
+      profileId,
+      accountId,
+      taskId: task.id,
+    });
+    expect(JSON.stringify(firstEvent.payload)).not.toContain("first");
+    expect(task.payloadHash).toMatch(/^[a-f0-9]{64}$/);
+
+    const second = await request(app.getHttpServer())
+      .patch("/api/v1/me/anonymous-profile")
+      .set("Authorization", authorization)
+      .send({ nickname: "second", avatarColor: "#abcdef", bio: "bio two" })
+      .expect(200);
+    expect(second.body.data).toMatchObject({ id: profileId, version: 2 });
+    const events = await prisma.outboxEvent.findMany({
+      where: {
+        eventType: "PROFILE_MODERATION_REQUESTED",
+        aggregateId: profileId,
+      },
+      orderBy: { createdAt: "asc" },
+    });
+    expect(events).toHaveLength(2);
+    expect(events[0]?.dedupeKey).toBe(`profile-moderation:${profileId}:1`);
+    expect(events[1]?.dedupeKey).toBe(`profile-moderation:${profileId}:2`);
+    expect(
+      await prisma.moderationTask.count({ where: { targetId: profileId } }),
+    ).toBe(1);
+    expect(
+      await prisma.moderationTask.findUniqueOrThrow({ where: { id: task.id } }),
+    ).toMatchObject({
+      status: "PENDING",
+      decision: null,
+      reviewedAt: null,
+    });
+  });
+
   it.each([
     { nickname: " ", avatarColor: "#abcdef" },
     { nickname: "x".repeat(65), avatarColor: "#abcdef" },

apps/api/src/profile/profile.service.ts

@@ -1,4 +1,6 @@
+import { createHash } from "node:crypto";
 import { Inject, Injectable } from "@nestjs/common";
+import { Prisma } from "@prisma/client";
 import { PrismaService } from "../database/prisma.service.js";
 import type { UpdateAnonymousProfileDto } from "./dto.js";
 
@@ -7,28 +9,59 @@ export class ProfileService {
   constructor(@Inject(PrismaService) private readonly prisma: PrismaService) {}
 
   update(accountId: string, input: UpdateAnonymousProfileDto) {
-    return this.prisma.anonymousProfile.upsert({
+    const bio = input.bio ?? null;
-      where: { accountId },
+    const payloadHash = createHash("sha256")
-      create: {
+      .update(JSON.stringify([input.nickname, input.avatarColor, bio]))
-        accountId,
+      .digest("hex");
-        nickname: input.nickname,
+    return this.prisma.$transaction(async (tx) => {
-        avatarColor: input.avatarColor,
+      const profile = await tx.anonymousProfile.upsert({
-        bio: input.bio ?? null,
+        where: { accountId },
-        reviewStatus: "REVIEWING",
+        create: {
-      },
+          accountId,
-      update: {
+          nickname: input.nickname,
-        nickname: input.nickname,
+          avatarColor: input.avatarColor,
-        avatarColor: input.avatarColor,
+          bio,
-        bio: input.bio ?? null,
+          reviewStatus: "REVIEWING",
-        reviewStatus: "REVIEWING",
+        },
-      },
+        update: {
-      select: {
+          nickname: input.nickname,
-        publicId: true,
+          avatarColor: input.avatarColor,
-        nickname: true,
+          bio,
-        avatarColor: true,
+          reviewStatus: "REVIEWING",
-        bio: true,
+          version: { increment: 1 },
-        reviewStatus: true,
+        },
-      },
+      });
+      const task = await tx.moderationTask.upsert({
+        where: {
+          targetType_targetId: { targetType: "PROFILE", targetId: profile.id },
+        },
+        create: {
+          targetType: "PROFILE",
+          targetId: profile.id,
+          provider: "SIMULATED",
+          payloadHash,
+          riskLabels: [],
+        },
+        update: {
+          provider: "SIMULATED",
+          payloadHash,
+          riskLabels: [],
+          result: Prisma.JsonNull,
+          status: "PENDING",
+          decision: null,
+          reviewedAt: null,
+        },
+      });
+      await tx.outboxEvent.create({
+        data: {
+          aggregateType: "PROFILE",
+          aggregateId: profile.id,
+          eventType: "PROFILE_MODERATION_REQUESTED",
+          dedupeKey: `profile-moderation:${profile.id}:${profile.version}`,
+          payload: { profileId: profile.id, accountId, taskId: task.id },
+        },
+      });
+      return profile;
     });
   }
 }

apps/worker/src/moderation-worker.spec.ts

@@ -133,6 +133,230 @@ describe("moderation worker with real PostgreSQL", () => {
     ).toBe("PUBLISHED");
   });
 
+  it("publishes a replay when the matching bottle task is already completed", async () => {
+    const { bottle, event, task } = await fixture("ordinary");
+    await prisma.moderationTask.update({
+      where: { id: task.id },
+      data: {
+        status: "COMPLETED",
+        decision: "APPROVED",
+        result: { decision: "APPROVED" },
+        reviewedAt: new Date(),
+      },
+    });
+    const decide = vi.fn(decideModeration);
+
+    expect(await new ModerationWorker(prisma, decide).runOnce()).toBe(true);
+    expect(decide).not.toHaveBeenCalled();
+    expect(
+      await prisma.outboxEvent.findUniqueOrThrow({ where: { id: event.id } }),
+    ).toMatchObject({ status: "PUBLISHED" });
+    expect(
+      await prisma.bottle.findUniqueOrThrow({ where: { id: bottle.id } }),
+    ).toMatchObject({ reviewStatus: "REVIEWING", poolStatus: "CLOSED" });
+  });
+
+  it("fails the event and leaves the task open when bottle state drift prevents the update", async () => {
+    const { bottle, event, task } = await fixture("ordinary");
+    await prisma.bottle.update({
+      where: { id: bottle.id },
+      data: { reviewStatus: "APPROVED", poolStatus: "IN_POOL" },
+    });
+    expect(await new ModerationWorker(prisma).runOnce()).toBe(true);
+    expect(
+      await prisma.outboxEvent.findUniqueOrThrow({ where: { id: event.id } }),
+    ).toMatchObject({ status: "FAILED", attempts: 1 });
+    expect(
+      await prisma.moderationTask.findUniqueOrThrow({ where: { id: task.id } }),
+    ).toMatchObject({ status: "PENDING", decision: null });
+  });
+
+  it("reaps a stale PROCESSING event already at max attempts without handling it", async () => {
+    const { event } = await fixture("ordinary");
+    await prisma.outboxEvent.update({
+      where: { id: event.id },
+      data: {
+        status: "PROCESSING",
+        attempts: 5,
+        lockedAt: new Date(Date.now() - 60_000),
+      },
+    });
+    const decide = vi.fn(decideModeration);
+    expect(await new ModerationWorker(prisma, decide, 1000).runOnce()).toBe(
+      false,
+    );
+    expect(decide).not.toHaveBeenCalled();
+    expect(
+      await prisma.outboxEvent.findUniqueOrThrow({ where: { id: event.id } }),
+    ).toMatchObject({ status: "FAILED", attempts: 5, lockedAt: null });
+  });
+
+  it.each([
+    ["ordinary", "APPROVED"],
+    ["reject-word", "REJECTED"],
+    ["review-word", "MANUAL_REVIEW"],
+  ] as const)(
+    "moderates normalized profile content with rule %#",
+    async (nickname, status) => {
+      const { profile, event, task } = await profileFixture(
+        nickname,
+        "profile bio",
+      );
+      expect(await new ModerationWorker(prisma).runOnce()).toBe(true);
+      expect(
+        (
+          await prisma.anonymousProfile.findUniqueOrThrow({
+            where: { id: profile.id },
+          })
+        ).reviewStatus,
+      ).toBe(status);
+      expect(
+        await prisma.moderationTask.findUniqueOrThrow({
+          where: { id: task.id },
+        }),
+      ).toMatchObject({ status: "COMPLETED", decision: status });
+      expect(
+        (
+          await prisma.outboxEvent.findUniqueOrThrow({
+            where: { id: event.id },
+          })
+        ).status,
+      ).toBe("PUBLISHED");
+    },
+  );
+
+  it("publishes a profile event that becomes stale while moderation is running", async () => {
+    const { createHash } = await import("node:crypto");
+    const { profile, event, task } = await profileFixture("ordinary", "first");
+    const currentText = JSON.stringify(["reject-word", "#abcdef", "second"]);
+    const decide = vi.fn(async (text: string) => {
+      await prisma.$transaction([
+        prisma.anonymousProfile.update({
+          where: { id: profile.id },
+          data: {
+            nickname: "reject-word",
+            bio: "second",
+            version: 2,
+            reviewStatus: "REVIEWING",
+          },
+        }),
+        prisma.moderationTask.update({
+          where: { id: task.id },
+          data: {
+            payloadHash: createHash("sha256").update(currentText).digest("hex"),
+            status: "PENDING",
+          },
+        }),
+      ]);
+      return decideModeration(text);
+    });
+
+    expect(await new ModerationWorker(prisma, decide).runOnce()).toBe(true);
+    expect(decide).toHaveBeenCalledTimes(1);
+    expect(
+      await prisma.outboxEvent.findUniqueOrThrow({ where: { id: event.id } }),
+    ).toMatchObject({ status: "PUBLISHED" });
+    expect(
+      await prisma.anonymousProfile.findUniqueOrThrow({
+        where: { id: profile.id },
+      }),
+    ).toMatchObject({ version: 2, reviewStatus: "REVIEWING" });
+    expect(
+      await prisma.moderationTask.findUniqueOrThrow({ where: { id: task.id } }),
+    ).toMatchObject({ status: "PENDING", decision: null });
+  });
+
+  it("publishes a replay when the matching profile task is already completed", async () => {
+    const { profile, event, task } = await profileFixture("ordinary", "first");
+    await prisma.moderationTask.update({
+      where: { id: task.id },
+      data: {
+        status: "COMPLETED",
+        decision: "APPROVED",
+        result: { decision: "APPROVED" },
+        reviewedAt: new Date(),
+      },
+    });
+    const decide = vi.fn(decideModeration);
+
+    expect(await new ModerationWorker(prisma, decide).runOnce()).toBe(true);
+    expect(
+      await prisma.outboxEvent.findUniqueOrThrow({ where: { id: event.id } }),
+    ).toMatchObject({ status: "PUBLISHED" });
+    expect(
+      await prisma.anonymousProfile.findUniqueOrThrow({
+        where: { id: profile.id },
+      }),
+    ).toMatchObject({ reviewStatus: "REVIEWING" });
+    expect(
+      await prisma.moderationTask.findUniqueOrThrow({ where: { id: task.id } }),
+    ).toMatchObject({ status: "COMPLETED", decision: "APPROVED" });
+  });
+
+  it("publishes an old profile event without letting it overwrite a newer version", async () => {
+    const { createHash } = await import("node:crypto");
+    const { profile, event, task } = await profileFixture("ordinary", "first");
+    const currentText = JSON.stringify(["reject-word", "#abcdef", "second"]);
+    await prisma.$transaction([
+      prisma.anonymousProfile.update({
+        where: { id: profile.id },
+        data: {
+          nickname: "reject-word",
+          bio: "second",
+          version: 2,
+          reviewStatus: "REVIEWING",
+        },
+      }),
+      prisma.moderationTask.update({
+        where: { id: task.id },
+        data: {
+          payloadHash: createHash("sha256").update(currentText).digest("hex"),
+        },
+      }),
+      prisma.outboxEvent.create({
+        data: {
+          aggregateType: "PROFILE",
+          aggregateId: profile.id,
+          eventType: "PROFILE_MODERATION_REQUESTED",
+          dedupeKey: `profile-moderation:${profile.id}:2`,
+          payload: {
+            profileId: profile.id,
+            accountId: profile.accountId,
+            taskId: task.id,
+          },
+        },
+      }),
+    ]);
+    const worker = new ModerationWorker(prisma);
+    expect(await worker.runOnce()).toBe(true);
+    expect(
+      (await prisma.outboxEvent.findUniqueOrThrow({ where: { id: event.id } }))
+        .status,
+    ).toBe("PUBLISHED");
+    expect(
+      (
+        await prisma.anonymousProfile.findUniqueOrThrow({
+          where: { id: profile.id },
+        })
+      ).reviewStatus,
+    ).toBe("REVIEWING");
+    expect(
+      (
+        await prisma.moderationTask.findUniqueOrThrow({
+          where: { id: task.id },
+        })
+      ).status,
+    ).toBe("PENDING");
+    expect(await worker.runOnce()).toBe(true);
+    expect(
+      (
+        await prisma.anonymousProfile.findUniqueOrThrow({
+          where: { id: profile.id },
+        })
+      ).reviewStatus,
+    ).toBe("REJECTED");
+  });
+
   async function fixture(contentText: string) {
     const account = await prisma.account.create({
       data: { phoneCiphertext: Buffer.from("cipher"), phoneHmac: randomUUID() },
@@ -166,4 +390,44 @@ describe("moderation worker with real PostgreSQL", () => {
     });
     return { bottle, task, event };
   }
+
+  async function profileFixture(nickname: string, bio: string) {
+    const { createHash } = await import("node:crypto");
+    const account = await prisma.account.create({
+      data: { phoneCiphertext: Buffer.from("cipher"), phoneHmac: randomUUID() },
+    });
+    const profile = await prisma.anonymousProfile.create({
+      data: {
+        accountId: account.id,
+        nickname,
+        avatarColor: "#abcdef",
+        bio,
+        version: 1,
+      },
+    });
+    const text = JSON.stringify([nickname, "#abcdef", bio]);
+    const task = await prisma.moderationTask.create({
+      data: {
+        targetType: "PROFILE",
+        targetId: profile.id,
+        provider: "SIMULATED",
+        payloadHash: createHash("sha256").update(text).digest("hex"),
+        riskLabels: [],
+      },
+    });
+    const event = await prisma.outboxEvent.create({
+      data: {
+        aggregateType: "PROFILE",
+        aggregateId: profile.id,
+        eventType: "PROFILE_MODERATION_REQUESTED",
+        dedupeKey: `profile-moderation:${profile.id}:1`,
+        payload: {
+          profileId: profile.id,
+          accountId: account.id,
+          taskId: task.id,
+        },
+      },
+    });
+    return { profile, task, event };
+  }
 });

apps/worker/src/moderation-worker.ts

@@ -1,8 +1,9 @@
+import { createHash } from "node:crypto";
 import {
   Prisma,
   PrismaClient,
-  type ReviewStatus,
   type BottlePoolStatus,
+  type ReviewStatus,
 } from "@prisma/client";
 
 type Decision = {
@@ -11,7 +12,25 @@ type Decision = {
   labels: string[];
 };
 type Decide = (text: string) => Decision | Promise<Decision>;
-type Claimed = { id: string; aggregateId: string; payload: Prisma.JsonValue };
+type Claimed = {
+  id: string;
+  aggregateId: string;
+  eventType: string;
+  dedupeKey: string;
+  payload: Prisma.JsonValue;
+};
+const NEVER = new Date("9999-12-31T23:59:59.999Z");
+
+function positiveInt(
+  value: string | undefined,
+  fallback: number,
+  name: string,
+) {
+  const parsed = value === undefined ? fallback : Number(value);
+  if (!Number.isInteger(parsed) || parsed <= 0)
+    throw new Error(`${name} must be a positive integer`);
+  return parsed;
+}
 
 export function decideModeration(text: string): Decision {
   const rejectWords = (process.env.MODERATION_REJECT_WORDS ?? "reject-word")
@@ -36,78 +55,38 @@ export function decideModeration(text: string): Decision {
 }
 
 export class ModerationWorker {
+  private readonly maxAttempts: number;
+
   constructor(
     private readonly prisma: PrismaClient,
     private readonly decide: Decide = decideModeration,
-    private readonly leaseMs = Number(process.env.OUTBOX_LEASE_MS ?? 30_000),
+    private readonly leaseMs = positiveInt(
-  ) {}
+      process.env.OUTBOX_LEASE_MS,
+      30_000,
+      "OUTBOX_LEASE_MS",
+    ),
+  ) {
+    this.maxAttempts = positiveInt(
+      process.env.OUTBOX_MAX_ATTEMPTS,
+      5,
+      "OUTBOX_MAX_ATTEMPTS",
+    );
+  }
 
   async runOnce(): Promise<boolean> {
     const event = await this.claim();
     if (!event) return false;
     try {
-      const payload = event.payload as { bottleId?: unknown; taskId?: unknown };
+      if (event.eventType === "BOTTLE_MODERATION_REQUESTED")
-      if (
+        await this.handleBottle(event);
-        typeof payload.bottleId !== "string" ||
+      else if (event.eventType === "PROFILE_MODERATION_REQUESTED")
-        typeof payload.taskId !== "string" ||
+        await this.handleProfile(event);
-        payload.bottleId !== event.aggregateId
+      else throw new Error("unsupported moderation event");
-      )
-        throw new Error("invalid event payload");
-      const [bottle, task] = await Promise.all([
-        this.prisma.bottle.findUniqueOrThrow({
-          where: { id: payload.bottleId },
-        }),
-        this.prisma.moderationTask.findUniqueOrThrow({
-          where: { id: payload.taskId },
-        }),
-      ]);
-      if (task.targetType !== "BOTTLE" || task.targetId !== bottle.id)
-        throw new Error("mismatched moderation task");
-      const decision = await this.decide(bottle.contentText);
-      await this.prisma.$transaction(async (tx) => {
-        const currentTask = await tx.moderationTask.findUniqueOrThrow({
-          where: { id: task.id },
-        });
-        if (currentTask.status !== "COMPLETED") {
-          await tx.bottle.updateMany({
-            where: {
-              id: bottle.id,
-              reviewStatus: "REVIEWING",
-              poolStatus: "CLOSED",
-            },
-            data: {
-              reviewStatus: decision.reviewStatus,
-              poolStatus: decision.poolStatus,
-              approvedAt:
-                decision.reviewStatus === "APPROVED" ? new Date() : null,
-            },
-          });
-          await tx.moderationTask.update({
-            where: { id: task.id },
-            data: {
-              status: "COMPLETED",
-              decision: decision.reviewStatus,
-              riskLabels: decision.labels,
-              result: { decision: decision.reviewStatus },
-              reviewedAt: new Date(),
-            },
-          });
-        }
-        await tx.outboxEvent.update({
-          where: { id: event.id },
-          data: {
-            status: "PUBLISHED",
-            publishedAt: new Date(),
-            lockedAt: null,
-          },
-        });
-      });
     } catch {
       const current = await this.prisma.outboxEvent.findUniqueOrThrow({
         where: { id: event.id },
       });
-      const maxAttempts = Number(process.env.OUTBOX_MAX_ATTEMPTS ?? 5);
+      const exhausted = current.attempts >= this.maxAttempts;
-      const exhausted = current.attempts >= maxAttempts;
       const delayMs = Math.min(
         60_000,
         1000 * 2 ** Math.max(0, current.attempts - 1),
@@ -117,29 +96,177 @@ export class ModerationWorker {
         data: {
           status: "FAILED",
           lockedAt: null,
-          nextRetryAt: exhausted
+          nextRetryAt: exhausted ? NEVER : new Date(Date.now() + delayMs),
-            ? new Date("9999-12-31T23:59:59.999Z")
-            : new Date(Date.now() + delayMs),
         },
       });
     }
     return true;
   }
 
+  private async handleBottle(event: Claimed) {
+    const payload = event.payload as { bottleId?: unknown; taskId?: unknown };
+    if (
+      typeof payload.bottleId !== "string" ||
+      typeof payload.taskId !== "string" ||
+      payload.bottleId !== event.aggregateId
+    )
+      throw new Error("invalid event payload");
+    const [bottle, task] = await Promise.all([
+      this.prisma.bottle.findUniqueOrThrow({ where: { id: payload.bottleId } }),
+      this.prisma.moderationTask.findUniqueOrThrow({
+        where: { id: payload.taskId },
+      }),
+    ]);
+    if (task.targetType !== "BOTTLE" || task.targetId !== bottle.id)
+      throw new Error("mismatched moderation task");
+    if (task.status === "COMPLETED") {
+      await this.prisma.$transaction((tx) => this.publish(tx, event.id));
+      return;
+    }
+    const decision = await this.decide(bottle.contentText);
+    await this.prisma.$transaction(async (tx) => {
+      const currentTask = await tx.moderationTask.findUniqueOrThrow({
+        where: { id: task.id },
+      });
+      if (currentTask.status !== "COMPLETED") {
+        const updated = await tx.bottle.updateMany({
+          where: {
+            id: bottle.id,
+            reviewStatus: "REVIEWING",
+            poolStatus: "CLOSED",
+          },
+          data: {
+            reviewStatus: decision.reviewStatus,
+            poolStatus: decision.poolStatus,
+            approvedAt:
+              decision.reviewStatus === "APPROVED" ? new Date() : null,
+          },
+        });
+        if (updated.count !== 1)
+          throw new Error("bottle moderation state drift");
+        await this.completeTask(tx, task.id, decision);
+      }
+      await this.publish(tx, event.id);
+    });
+  }
+
+  private async handleProfile(event: Claimed) {
+    const payload = event.payload as {
+      profileId?: unknown;
+      accountId?: unknown;
+      taskId?: unknown;
+    };
+    if (
+      typeof payload.profileId !== "string" ||
+      typeof payload.accountId !== "string" ||
+      typeof payload.taskId !== "string" ||
+      payload.profileId !== event.aggregateId
+    )
+      throw new Error("invalid event payload");
+    const [profile, task] = await Promise.all([
+      this.prisma.anonymousProfile.findUniqueOrThrow({
+        where: { id: payload.profileId },
+      }),
+      this.prisma.moderationTask.findUniqueOrThrow({
+        where: { id: payload.taskId },
+      }),
+    ]);
+    if (
+      profile.accountId !== payload.accountId ||
+      task.targetType !== "PROFILE" ||
+      task.targetId !== profile.id
+    )
+      throw new Error("mismatched moderation task");
+    const version = Number(
+      event.dedupeKey.slice(event.dedupeKey.lastIndexOf(":") + 1),
+    );
+    const text = JSON.stringify([
+      profile.nickname,
+      profile.avatarColor,
+      profile.bio,
+    ]);
+    const hash = createHash("sha256").update(text).digest("hex");
+    if (profile.version !== version || task.payloadHash !== hash) {
+      await this.prisma.outboxEvent.update({
+        where: { id: event.id },
+        data: { status: "PUBLISHED", publishedAt: new Date(), lockedAt: null },
+      });
+      return;
+    }
+    if (task.status === "COMPLETED") {
+      await this.prisma.$transaction((tx) => this.publish(tx, event.id));
+      return;
+    }
+    const decision = await this.decide(text);
+    await this.prisma.$transaction(async (tx) => {
+      const [currentProfile, currentTask] = await Promise.all([
+        tx.anonymousProfile.findUniqueOrThrow({ where: { id: profile.id } }),
+        tx.moderationTask.findUniqueOrThrow({ where: { id: task.id } }),
+      ]);
+      if (
+        currentProfile.version !== version ||
+        currentTask.payloadHash !== hash ||
+        currentTask.status === "COMPLETED"
+      ) {
+        await this.publish(tx, event.id);
+        return;
+      }
+      const updated = await tx.anonymousProfile.updateMany({
+        where: { id: profile.id, version, reviewStatus: "REVIEWING" },
+        data: { reviewStatus: decision.reviewStatus },
+      });
+      if (updated.count !== 1)
+        throw new Error("profile moderation state drift");
+      await this.completeTask(tx, task.id, decision);
+      await this.publish(tx, event.id);
+    });
+  }
+
+  private completeTask(
+    tx: Prisma.TransactionClient,
+    id: string,
+    decision: Decision,
+  ) {
+    return tx.moderationTask.update({
+      where: { id },
+      data: {
+        status: "COMPLETED",
+        decision: decision.reviewStatus,
+        riskLabels: decision.labels,
+        result: { decision: decision.reviewStatus },
+        reviewedAt: new Date(),
+      },
+    });
+  }
+
+  private publish(tx: Prisma.TransactionClient, id: string) {
+    return tx.outboxEvent.update({
+      where: { id },
+      data: { status: "PUBLISHED", publishedAt: new Date(), lockedAt: null },
+    });
+  }
+
   private async claim(): Promise<Claimed | null> {
     const staleBefore = new Date(Date.now() - this.leaseMs);
     return this.prisma.$transaction(async (tx) => {
+      await tx.outboxEvent.updateMany({
+        where: {
+          status: "PROCESSING",
+          lockedAt: { lt: staleBefore },
+          attempts: { gte: this.maxAttempts },
+        },
+        data: { status: "FAILED", lockedAt: null, nextRetryAt: NEVER },
+      });
       const rows = await tx.$queryRaw<Claimed[]>`
-        SELECT "id", "aggregate_id" AS "aggregateId", "payload"
+        SELECT "id", "aggregate_id" AS "aggregateId", "event_type" AS "eventType",
+               "dedupe_key" AS "dedupeKey", "payload"
         FROM "outbox_events"
-        WHERE "event_type" = 'BOTTLE_MODERATION_REQUESTED'
+        WHERE "event_type" IN ('BOTTLE_MODERATION_REQUESTED', 'PROFILE_MODERATION_REQUESTED')
-          AND (
+          AND "attempts" < ${this.maxAttempts}
-            ("status" IN ('PENDING', 'FAILED') AND "next_retry_at" <= now())
+          AND (("status" IN ('PENDING', 'FAILED') AND "next_retry_at" <= now())
-            OR ("status" = 'PROCESSING' AND "locked_at" < ${staleBefore})
+            OR ("status" = 'PROCESSING' AND "locked_at" < ${staleBefore}))
-          )
         ORDER BY "created_at"
-        FOR UPDATE SKIP LOCKED
+        FOR UPDATE SKIP LOCKED LIMIT 1`;
-        LIMIT 1`;
       const row = rows[0];
       if (!row) return null;
       await tx.outboxEvent.update({

packages/contracts/src/index.test.ts

@@ -33,6 +33,7 @@ describe("contracts", () => {
       "RATE_LIMITED",
       "AUTH_ORIGIN_FORBIDDEN",
       "BOTTLE_DAILY_LIMIT",
+      "IDEMPOTENCY_CONFLICT",
       "ACCOUNT_SANCTIONED",
       "BOTTLE_POOL_EMPTY",
       "BOTTLE_LEASE_EXPIRED",

packages/contracts/src/index.ts

@@ -11,6 +11,7 @@ export enum ErrorCode {
   RATE_LIMITED = "RATE_LIMITED",
   AUTH_ORIGIN_FORBIDDEN = "AUTH_ORIGIN_FORBIDDEN",
   BOTTLE_DAILY_LIMIT = "BOTTLE_DAILY_LIMIT",
+  IDEMPOTENCY_CONFLICT = "IDEMPOTENCY_CONFLICT",
   ACCOUNT_SANCTIONED = "ACCOUNT_SANCTIONED",
   BOTTLE_POOL_EMPTY = "BOTTLE_POOL_EMPTY",
   BOTTLE_LEASE_EXPIRED = "BOTTLE_LEASE_EXPIRED",

prisma/migrations/0006_profile_moderation_version/migration.sql

@@ -0,0 +1,5 @@
+ALTER TABLE "anonymous_profiles"
+  ADD COLUMN "version" INTEGER NOT NULL DEFAULT 1;
+
+CREATE INDEX "anonymous_profiles_review_status_updated_at_idx"
+  ON "anonymous_profiles"("review_status", "updated_at");

prisma/schema.prisma

@@ -123,6 +123,7 @@ model Account {
 }
 
 model AnonymousProfile {
+  /// Review queue index anonymous_profiles_review_status_updated_at_idx is managed in 0006 SQL.
   id           String       @id @default(uuid()) @db.Uuid
   accountId    String       @unique @map("account_id") @db.Uuid
   publicId     String       @unique @default(uuid()) @map("public_id") @db.Uuid
@@ -131,10 +132,12 @@ model AnonymousProfile {
   avatarColor  String       @map("avatar_color") @db.VarChar(16)
   bio          String?      @db.VarChar(500)
   reviewStatus ReviewStatus @default(REVIEWING) @map("review_status")
+  version      Int          @default(1)
   createdAt    DateTime     @default(now()) @map("created_at") @db.Timestamptz(3)
   updatedAt    DateTime     @updatedAt @map("updated_at") @db.Timestamptz(3)
   account      Account      @relation(fields: [accountId], references: [id], onDelete: Cascade)
 
+  @@index([reviewStatus, updatedAt], map: "anonymous_profiles_review_status_updated_at_idx")
   @@map("anonymous_profiles")
 }
 
@@ -425,6 +428,7 @@ model Notification {
 }
 
 model OutboxEvent {
+  /// Processing lease index outbox_events_processing_lease_idx is managed in 0005 SQL.
   id            String       @id @default(uuid()) @db.Uuid
   aggregateType String       @map("aggregate_type") @db.VarChar(100)
   aggregateId   String       @map("aggregate_id") @db.Uuid
@@ -440,6 +444,7 @@ model OutboxEvent {
   updatedAt     DateTime     @updatedAt @map("updated_at") @db.Timestamptz(3)
 
   @@index([status, nextRetryAt])
+  @@index([status, lockedAt], map: "outbox_events_processing_lease_idx")
   @@index([aggregateType, aggregateId])
   @@map("outbox_events")
 }