fix: 严格校验资料审核终态

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

@@ -282,7 +282,12 @@ describe("moderation worker with real PostgreSQL", () => {
     expect(decide).not.toHaveBeenCalled();
     expect(
       await prisma.outboxEvent.findUniqueOrThrow({ where: { id: event.id } }),
-    ).toMatchObject({ status: "FAILED", attempts: 5, lockedAt: null });
+    ).toMatchObject({
+      status: "FAILED",
+      attempts: 5,
+      lockToken: null,
+      lockedAt: null,
+    });
   });
 
   it.each([
@@ -349,7 +354,11 @@ describe("moderation worker with real PostgreSQL", () => {
     expect(decide).toHaveBeenCalledTimes(1);
     expect(
       await prisma.outboxEvent.findUniqueOrThrow({ where: { id: event.id } }),
-    ).toMatchObject({ status: "PUBLISHED" });
+    ).toMatchObject({
+      status: "PUBLISHED",
+      lockToken: null,
+      lockedAt: null,
+    });
     expect(
       await prisma.anonymousProfile.findUniqueOrThrow({
         where: { id: profile.id },
@@ -360,6 +369,32 @@ describe("moderation worker with real PostgreSQL", () => {
     ).toMatchObject({ status: "PENDING", decision: null });
   });
 
+  it("fails a same-version profile event whose task hash drifted", async () => {
+    const { profile, event, task } = await profileFixture("ordinary", "first");
+    await prisma.moderationTask.update({
+      where: { id: task.id },
+      data: { payloadHash: "0".repeat(64) },
+    });
+
+    expect(await new ModerationWorker(prisma).runOnce()).toBe(true);
+    expect(
+      await prisma.outboxEvent.findUniqueOrThrow({ where: { id: event.id } }),
+    ).toMatchObject({
+      status: "FAILED",
+      attempts: 1,
+      lockToken: null,
+      lockedAt: null,
+    });
+    expect(
+      await prisma.anonymousProfile.findUniqueOrThrow({
+        where: { id: profile.id },
+      }),
+    ).toMatchObject({ version: 1, reviewStatus: "REVIEWING" });
+    expect(
+      await prisma.moderationTask.findUniqueOrThrow({ where: { id: task.id } }),
+    ).toMatchObject({ status: "PENDING", decision: null });
+  });
+
   it("fails a replay when the completed profile task drifted from current state", async () => {
     const { profile, event, task } = await profileFixture("ordinary", "first");
     await prisma.moderationTask.update({
@@ -387,6 +422,37 @@ describe("moderation worker with real PostgreSQL", () => {
     ).toMatchObject({ status: "COMPLETED", decision: "APPROVED" });
   });
 
+  it("fails a completed profile task with a non-terminal reviewing decision", async () => {
+    const { profile, event, task } = await profileFixture("ordinary", "first");
+    await prisma.moderationTask.update({
+      where: { id: task.id },
+      data: {
+        status: "COMPLETED",
+        decision: "REVIEWING",
+        result: { decision: "REVIEWING" },
+        reviewedAt: new Date(),
+      },
+    });
+
+    expect(await new ModerationWorker(prisma).runOnce()).toBe(true);
+    expect(
+      await prisma.outboxEvent.findUniqueOrThrow({ where: { id: event.id } }),
+    ).toMatchObject({
+      status: "FAILED",
+      attempts: 1,
+      lockToken: null,
+      lockedAt: null,
+    });
+    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: "REVIEWING" });
+  });
+
   it("logs a safe error category without payload or exception message", async () => {
     const { event } = await fixture("secret正文");
     const entries: unknown[] = [];

apps/worker/src/moderation-worker.ts

@@ -269,10 +269,12 @@ export class ModerationWorker {
       profile.bio,
     ]);
     const hash = createHash("sha256").update(text).digest("hex");
-    if (profile.version !== version || task.payloadHash !== hash) {
+    if (profile.version !== version) {
       await this.prisma.$transaction((tx) => this.publish(tx, event));
       return;
     }
+    if (task.payloadHash !== hash)
+      throw new Error("profile moderation payload drift");
     if (task.status === "COMPLETED") {
       if (!this.completedProfileMatches(profile, task, hash, version))
         throw new Error("completed profile moderation state drift");
@@ -294,12 +296,8 @@ export class ModerationWorker {
       const currentHash = createHash("sha256")
         .update(currentText)
         .digest("hex");
-      if (
-        currentProfile.version !== version ||
-        currentTask.payloadHash !== hash
-      )
-        return;
-      if (currentHash !== hash)
+      if (currentProfile.version !== version) return;
+      if (currentHash !== hash || currentTask.payloadHash !== currentHash)
         throw new Error("profile moderation payload drift");
       if (currentTask.status === "COMPLETED") {
         if (
@@ -333,12 +331,15 @@ export class ModerationWorker {
     hash: string,
     version: number,
   ) {
+    const decision = task.decision;
     return (
       profile.version === version &&
       task.payloadHash === hash &&
-      task.decision !== null &&
-      resultDecision(task.result) === task.decision &&
-      profile.reviewStatus === task.decision
+      (decision === "APPROVED" ||
+        decision === "REJECTED" ||
+        decision === "MANUAL_REVIEW") &&
+      resultDecision(task.result) === decision &&
+      profile.reviewStatus === decision
     );
   }