From ec573b5ae2c57e6450d12b2cd1673b2a53eb044e Mon Sep 17 00:00:00 2001 From: root Date: Tue, 15 Sep 2026 08:30:39 +0800 Subject: [PATCH] =?UTF-8?q?fix:=20=E5=AE=8C=E5=96=84=E8=B5=84=E6=96=99?= =?UTF-8?q?=E5=AE=A1=E6=A0=B8=E4=B8=8E=20Worker=20=E7=8A=B6=E6=80=81?= =?UTF-8?q?=E4=B8=80=E8=87=B4=E6=80=A7?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- apps/api/src/bottle/bottle.e2e-spec.ts | 25 ++ apps/api/src/bottle/bottle.service.ts | 28 +- apps/api/src/profile/profile.e2e-spec.ts | 61 ++++ apps/api/src/profile/profile.service.ts | 77 +++-- apps/worker/src/moderation-worker.spec.ts | 264 +++++++++++++++++ apps/worker/src/moderation-worker.ts | 273 +++++++++++++----- packages/contracts/src/index.test.ts | 1 + packages/contracts/src/index.ts | 1 + .../migration.sql | 5 + prisma/schema.prisma | 5 + 10 files changed, 639 insertions(+), 101 deletions(-) create mode 100644 prisma/migrations/0006_profile_moderation_version/migration.sql diff --git a/apps/api/src/bottle/bottle.e2e-spec.ts b/apps/api/src/bottle/bottle.e2e-spec.ts index b35b82e..07ca4af 100644 --- a/apps/api/src/bottle/bottle.e2e-spec.ts +++ b/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) => diff --git a/apps/api/src/bottle/bottle.service.ts b/apps/api/src/bottle/bottle.service.ts index c9e7a92..e32c9f2 100644 --- a/apps/api/src/bottle/bottle.service.ts +++ b/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() }, - OR: [{ expiresAt: null }, { expiresAt: { gt: new Date() } }], + startsAt: { lte: now }, + 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>` 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( + 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 }, diff --git a/apps/api/src/profile/profile.e2e-spec.ts b/apps/api/src/profile/profile.e2e-spec.ts index ea05e8c..4b06f10 100644 --- a/apps/api/src/profile/profile.e2e-spec.ts +++ b/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" }, diff --git a/apps/api/src/profile/profile.service.ts b/apps/api/src/profile/profile.service.ts index eaec5a7..d706ac5 100644 --- a/apps/api/src/profile/profile.service.ts +++ b/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({ - where: { accountId }, - create: { - accountId, - nickname: input.nickname, - avatarColor: input.avatarColor, - bio: input.bio ?? null, - reviewStatus: "REVIEWING", - }, - update: { - nickname: input.nickname, - avatarColor: input.avatarColor, - bio: input.bio ?? null, - reviewStatus: "REVIEWING", - }, - select: { - publicId: true, - nickname: true, - avatarColor: true, - bio: true, - reviewStatus: true, - }, + const bio = input.bio ?? null; + const payloadHash = createHash("sha256") + .update(JSON.stringify([input.nickname, input.avatarColor, bio])) + .digest("hex"); + return this.prisma.$transaction(async (tx) => { + const profile = await tx.anonymousProfile.upsert({ + where: { accountId }, + create: { + accountId, + nickname: input.nickname, + avatarColor: input.avatarColor, + bio, + reviewStatus: "REVIEWING", + }, + update: { + nickname: input.nickname, + avatarColor: input.avatarColor, + bio, + reviewStatus: "REVIEWING", + version: { increment: 1 }, + }, + }); + 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; }); } } diff --git a/apps/worker/src/moderation-worker.spec.ts b/apps/worker/src/moderation-worker.spec.ts index 01143df..e6325c1 100644 --- a/apps/worker/src/moderation-worker.spec.ts +++ b/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 }; + } }); diff --git a/apps/worker/src/moderation-worker.ts b/apps/worker/src/moderation-worker.ts index f21769e..21ec020 100644 --- a/apps/worker/src/moderation-worker.ts +++ b/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; -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 { const event = await this.claim(); if (!event) return false; try { - 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"); - 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, - }, - }); - }); + if (event.eventType === "BOTTLE_MODERATION_REQUESTED") + await this.handleBottle(event); + else if (event.eventType === "PROFILE_MODERATION_REQUESTED") + await this.handleProfile(event); + else throw new Error("unsupported moderation event"); } 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 >= maxAttempts; + const exhausted = current.attempts >= this.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 - ? new Date("9999-12-31T23:59:59.999Z") - : new Date(Date.now() + delayMs), + nextRetryAt: exhausted ? NEVER : 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 { 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` - 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' - AND ( - ("status" IN ('PENDING', 'FAILED') AND "next_retry_at" <= now()) - OR ("status" = 'PROCESSING' AND "locked_at" < ${staleBefore}) - ) + WHERE "event_type" IN ('BOTTLE_MODERATION_REQUESTED', 'PROFILE_MODERATION_REQUESTED') + 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`; + FOR UPDATE SKIP LOCKED LIMIT 1`; const row = rows[0]; if (!row) return null; await tx.outboxEvent.update({ diff --git a/packages/contracts/src/index.test.ts b/packages/contracts/src/index.test.ts index fe78a30..56ee358 100644 --- a/packages/contracts/src/index.test.ts +++ b/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", diff --git a/packages/contracts/src/index.ts b/packages/contracts/src/index.ts index 9b5d82f..3222c45 100644 --- a/packages/contracts/src/index.ts +++ b/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", diff --git a/prisma/migrations/0006_profile_moderation_version/migration.sql b/prisma/migrations/0006_profile_moderation_version/migration.sql new file mode 100644 index 0000000..c732e82 --- /dev/null +++ b/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"); \ No newline at end of file diff --git a/prisma/schema.prisma b/prisma/schema.prisma index 5604b8b..2a98fcc 100644 --- a/prisma/schema.prisma +++ b/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") }