diff --git a/apps/worker/src/moderation-worker.spec.ts b/apps/worker/src/moderation-worker.spec.ts index 9367336..983affd 100644 --- a/apps/worker/src/moderation-worker.spec.ts +++ b/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[] = []; diff --git a/apps/worker/src/moderation-worker.ts b/apps/worker/src/moderation-worker.ts index 35a4459..91cbe15 100644 --- a/apps/worker/src/moderation-worker.ts +++ b/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 ); }