feat: 实现投瓶和异步内容审核

2026-09-15 01:28:11 +08:00
parent b480d2c91f
commit f455552e92
25 changed files with 1204 additions and 18 deletions
@@ -0,0 +1,23 @@
+import { PrismaClient } from "@prisma/client";
+import { ModerationWorker } from "./moderation-worker.js";
+
+const prisma = new PrismaClient();
+
+async function main() {
+  await prisma.$connect();
+  const worker = new ModerationWorker(prisma);
+  if (process.argv.includes("--once")) {
+    await worker.runOnce();
+    return;
+  }
+  for (;;) {
+    const handled = await worker.runOnce();
+    if (!handled) await new Promise((resolve) => setTimeout(resolve, 1000));
+  }
+}
+
+main()
+  .catch(() => {
+    process.exitCode = 1;
+  })
+  .finally(() => prisma.$disconnect());
@@ -0,0 +1,169 @@
+import { PrismaClient } from "@prisma/client";
+import { randomUUID } from "node:crypto";
+import { afterAll, beforeEach, describe, expect, it, vi } from "vitest";
+import { assertSafeTestDatabaseUrl } from "../../../prisma/database-safety.js";
+import { ModerationWorker, decideModeration } from "./moderation-worker.js";
+
+const prisma = new PrismaClient();
+
+describe("moderation worker with real PostgreSQL", () => {
+  beforeEach(async () => {
+    assertSafeTestDatabaseUrl(process.env.DATABASE_URL ?? "");
+    await prisma.$connect();
+    await prisma.outboxEvent.deleteMany();
+    await prisma.$executeRawUnsafe(`TRUNCATE TABLE "accounts" CASCADE`);
+  });
+  afterAll(() => prisma.$disconnect());
+
+  it.each([
+    ["ordinary sea note", "APPROVED", "IN_POOL"],
+    ["contains reject-word", "REJECTED", "CLOSED"],
+    ["contains review-word", "MANUAL_REVIEW", "CLOSED"],
+  ] as const)(
+    "maps simulated rule %#",
+    async (content, reviewStatus, poolStatus) => {
+      const { bottle } = await fixture(content);
+      expect(await new ModerationWorker(prisma).runOnce()).toBe(true);
+      const updated = await prisma.bottle.findUniqueOrThrow({
+        where: { id: bottle.id },
+      });
+      expect(updated).toMatchObject({ reviewStatus, poolStatus });
+      expect(updated.approvedAt !== null).toBe(reviewStatus === "APPROVED");
+      expect((await prisma.moderationTask.findFirstOrThrow()).status).toBe(
+        "COMPLETED",
+      );
+      expect((await prisma.outboxEvent.findFirstOrThrow()).status).toBe(
+        "PUBLISHED",
+      );
+      expect(await new ModerationWorker(prisma).runOnce()).toBe(false);
+    },
+  );
+
+  it("is fail-closed and exponentially reschedules failures", async () => {
+    const { bottle, event } = await fixture("ordinary");
+    const worker = new ModerationWorker(prisma, () => {
+      throw new Error("provider unavailable");
+    });
+    await expect(worker.runOnce()).resolves.toBe(true);
+    expect(
+      await prisma.bottle.findUniqueOrThrow({ where: { id: bottle.id } }),
+    ).toMatchObject({
+      reviewStatus: "REVIEWING",
+      poolStatus: "CLOSED",
+    });
+    const failed = await prisma.outboxEvent.findUniqueOrThrow({
+      where: { id: event.id },
+    });
+    expect(failed.status).toBe("FAILED");
+    expect(failed.attempts).toBe(1);
+    expect(failed.nextRetryAt.getTime()).toBeGreaterThan(Date.now());
+  });
+
+  it("stops retrying at max attempts while remaining fail-closed", async () => {
+    const { bottle, event } = await fixture("ordinary");
+    await prisma.outboxEvent.update({
+      where: { id: event.id },
+      data: { attempts: 4 },
+    });
+    const worker = new ModerationWorker(prisma, () => {
+      throw new Error("provider unavailable");
+    });
+    await worker.runOnce();
+    const failed = await prisma.outboxEvent.findUniqueOrThrow({
+      where: { id: event.id },
+    });
+    expect(failed).toMatchObject({ status: "FAILED", attempts: 5 });
+    expect(failed.nextRetryAt.getUTCFullYear()).toBe(9999);
+    expect(await worker.runOnce()).toBe(false);
+    expect(
+      await prisma.bottle.findUniqueOrThrow({ where: { id: bottle.id } }),
+    ).toMatchObject({ reviewStatus: "REVIEWING", poolStatus: "CLOSED" });
+  });
+
+  it("does not publish when the task payload points at another task", async () => {
+    const first = await fixture("ordinary");
+    const second = await fixture("another");
+    await prisma.outboxEvent.delete({ where: { id: second.event.id } });
+    await prisma.outboxEvent.update({
+      where: { id: first.event.id },
+      data: { payload: { bottleId: first.bottle.id, taskId: second.task.id } },
+    });
+    await new ModerationWorker(prisma).runOnce();
+    expect(
+      (
+        await prisma.outboxEvent.findUniqueOrThrow({
+          where: { id: first.event.id },
+        })
+      ).status,
+    ).toBe("FAILED");
+    expect(
+      await prisma.bottle.findUniqueOrThrow({ where: { id: first.bottle.id } }),
+    ).toMatchObject({ reviewStatus: "REVIEWING", poolStatus: "CLOSED" });
+  });
+
+  it("lets two workers claim one event only once", async () => {
+    await fixture("ordinary");
+    const entered = vi.fn();
+    const decide = async (text: string) => {
+      entered();
+      await new Promise((resolve) => setTimeout(resolve, 100));
+      return decideModeration(text);
+    };
+    const outcomes = await Promise.all([
+      new ModerationWorker(prisma, decide).runOnce(),
+      new ModerationWorker(prisma, decide).runOnce(),
+    ]);
+    expect(outcomes.sort()).toEqual([false, true]);
+    expect(entered).toHaveBeenCalledTimes(1);
+    expect((await prisma.outboxEvent.findFirstOrThrow()).attempts).toBe(1);
+  });
+
+  it("recovers an expired PROCESSING lease", async () => {
+    const { event } = await fixture("ordinary");
+    await prisma.outboxEvent.update({
+      where: { id: event.id },
+      data: { status: "PROCESSING", lockedAt: new Date(Date.now() - 60_000) },
+    });
+    expect(await new ModerationWorker(prisma, undefined, 1000).runOnce()).toBe(
+      true,
+    );
+    expect(
+      (await prisma.outboxEvent.findUniqueOrThrow({ where: { id: event.id } }))
+        .status,
+    ).toBe("PUBLISHED");
+  });
+
+  async function fixture(contentText: string) {
+    const account = await prisma.account.create({
+      data: { phoneCiphertext: Buffer.from("cipher"), phoneHmac: randomUUID() },
+    });
+    const bottle = await prisma.bottle.create({
+      data: {
+        authorId: account.id,
+        clientRequestId: randomUUID(),
+        contentText,
+        reviewStatus: "REVIEWING",
+        poolStatus: "CLOSED",
+      },
+    });
+    const task = await prisma.moderationTask.create({
+      data: {
+        targetType: "BOTTLE",
+        targetId: bottle.id,
+        provider: "SIMULATED",
+        payloadHash: "a".repeat(64),
+        riskLabels: [],
+      },
+    });
+    const event = await prisma.outboxEvent.create({
+      data: {
+        aggregateType: "BOTTLE",
+        aggregateId: bottle.id,
+        eventType: "BOTTLE_MODERATION_REQUESTED",
+        dedupeKey: randomUUID(),
+        payload: { bottleId: bottle.id, taskId: task.id },
+      },
+    });
+    return { bottle, task, event };
+  }
+});
@@ -0,0 +1,156 @@
+import {
+  Prisma,
+  PrismaClient,
+  type ReviewStatus,
+  type BottlePoolStatus,
+} from "@prisma/client";
+
+type Decision = {
+  reviewStatus: ReviewStatus;
+  poolStatus: BottlePoolStatus;
+  labels: string[];
+};
+type Decide = (text: string) => Decision | Promise<Decision>;
+type Claimed = { id: string; aggregateId: string; payload: Prisma.JsonValue };
+
+export function decideModeration(text: string): Decision {
+  const rejectWords = (process.env.MODERATION_REJECT_WORDS ?? "reject-word")
+    .split(",")
+    .filter(Boolean);
+  const reviewWords = (process.env.MODERATION_REVIEW_WORDS ?? "review-word")
+    .split(",")
+    .filter(Boolean);
+  if (rejectWords.some((word) => text.includes(word)))
+    return {
+      reviewStatus: "REJECTED",
+      poolStatus: "CLOSED",
+      labels: ["SIMULATED_REJECT"],
+    };
+  if (reviewWords.some((word) => text.includes(word)))
+    return {
+      reviewStatus: "MANUAL_REVIEW",
+      poolStatus: "CLOSED",
+      labels: ["SIMULATED_REVIEW"],
+    };
+  return { reviewStatus: "APPROVED", poolStatus: "IN_POOL", labels: [] };
+}
+
+export class ModerationWorker {
+  constructor(
+    private readonly prisma: PrismaClient,
+    private readonly decide: Decide = decideModeration,
+    private readonly leaseMs = Number(process.env.OUTBOX_LEASE_MS ?? 30_000),
+  ) {}
+
+  async runOnce(): Promise<boolean> {
+    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,
+          },
+        });
+      });
+    } 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 delayMs = Math.min(
+        60_000,
+        1000 * 2 ** Math.max(0, current.attempts - 1),
+      );
+      await this.prisma.outboxEvent.update({
+        where: { id: event.id },
+        data: {
+          status: "FAILED",
+          lockedAt: null,
+          nextRetryAt: exhausted
+            ? new Date("9999-12-31T23:59:59.999Z")
+            : new Date(Date.now() + delayMs),
+        },
+      });
+    }
+    return true;
+  }
+
+  private async claim(): Promise<Claimed | null> {
+    const staleBefore = new Date(Date.now() - this.leaseMs);
+    return this.prisma.$transaction(async (tx) => {
+      const rows = await tx.$queryRaw<Claimed[]>`
+        SELECT "id", "aggregate_id" AS "aggregateId", "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})
+          )
+        ORDER BY "created_at"
+        FOR UPDATE SKIP LOCKED
+        LIMIT 1`;
+      const row = rows[0];
+      if (!row) return null;
+      await tx.outboxEvent.update({
+        where: { id: row.id },
+        data: {
+          status: "PROCESSING",
+          lockedAt: new Date(),
+          attempts: { increment: 1 },
+        },
+      });
+      return row;
+    });
+  }
+}