diff --git a/apps/api/package.json b/apps/api/package.json index dc396a9..9de92b5 100644 --- a/apps/api/package.json +++ b/apps/api/package.json @@ -7,7 +7,7 @@ "build": "tsc -p tsconfig.build.json", "start": "node dist/main.js", "test": "vitest run --config vitest.config.ts --no-file-parallelism", - "test:e2e": "vitest run --config vitest.config.ts src/health/health.e2e-spec.ts src/auth/auth.e2e-spec.ts --no-file-parallelism", + "test:e2e": "vitest run --config vitest.config.ts src/health/health.e2e-spec.ts src/auth/auth.e2e-spec.ts src/profile/profile.e2e-spec.ts src/bottle/bottle.e2e-spec.ts --no-file-parallelism", "typecheck": "tsc --noEmit -p tsconfig.json" }, "dependencies": { diff --git a/apps/api/src/app.module.ts b/apps/api/src/app.module.ts index 43f1937..62b9b6f 100644 --- a/apps/api/src/app.module.ts +++ b/apps/api/src/app.module.ts @@ -2,8 +2,10 @@ import { MiddlewareConsumer, Module, type NestModule } from "@nestjs/common"; import { RequestIdMiddleware } from "./common/request-id.middleware.js"; import { HealthModule } from "./health/health.module.js"; import { AuthModule } from "./auth/auth.module.js"; +import { ProfileModule } from "./profile/profile.module.js"; +import { BottleModule } from "./bottle/bottle.module.js"; -@Module({ imports: [HealthModule, AuthModule] }) +@Module({ imports: [HealthModule, AuthModule, ProfileModule, BottleModule] }) export class AppModule implements NestModule { configure(consumer: MiddlewareConsumer): void { consumer.apply(RequestIdMiddleware).forRoutes("{*path}"); diff --git a/apps/api/src/auth/auth.controller.ts b/apps/api/src/auth/auth.controller.ts index 5bb63ae..aa68cc1 100644 --- a/apps/api/src/auth/auth.controller.ts +++ b/apps/api/src/auth/auth.controller.ts @@ -77,6 +77,9 @@ export class AuthController { accountId: account.id, publicId: account.anonymousProfile?.publicId ?? null, nickname: account.anonymousProfile?.nickname ?? null, + avatarColor: account.anonymousProfile?.avatarColor ?? null, + bio: account.anonymousProfile?.bio ?? null, + profileReviewStatus: account.anonymousProfile?.reviewStatus ?? null, }; } private publicPair(pair: TokenPair) { diff --git a/apps/api/src/bottle/bottle.controller.ts b/apps/api/src/bottle/bottle.controller.ts new file mode 100644 index 0000000..2aefe01 --- /dev/null +++ b/apps/api/src/bottle/bottle.controller.ts @@ -0,0 +1,71 @@ +import { + Body, + Controller, + Get, + Headers, + HttpStatus, + Inject, + Param, + ParseUUIDPipe, + Post, + Query, + UseGuards, +} from "@nestjs/common"; +import { ErrorCode } from "@drift/contracts"; +import { AuthGuard } from "../auth/auth.guard.js"; +import { CurrentUser } from "../auth/current-user.decorator.js"; +import type { AccessClaims } from "../auth/token.service.js"; +import { DomainException } from "../common/domain.exception.js"; +import { BottleService } from "./bottle.service.js"; +import { CreateBottleDto, ListBottlesDto } from "./dto.js"; + +@Controller() +@UseGuards(AuthGuard) +export class BottleController { + constructor(@Inject(BottleService) private readonly bottles: BottleService) {} + + @Post("bottles") + create( + @CurrentUser() user: AccessClaims, + @Headers("idempotency-key") key: string | undefined, + @Body() dto: CreateBottleDto, + ) { + if (!key || !key.trim() || key.length > 128) + throw new DomainException( + ErrorCode.VALIDATION_ERROR, + "Validation failed", + HttpStatus.BAD_REQUEST, + ); + return this.bottles.create(user.sub, key, dto.contentText); + } + @Get("bottles/:id") + get( + @CurrentUser() user: AccessClaims, + @Param("id", new ParseUUIDPipe()) id: string, + ) { + return this.bottles.get(user.sub, id); + } + @Get("me/bottles") + list(@CurrentUser() user: AccessClaims, @Query() query: ListBottlesDto) { + return this.bottles.list(user.sub, query.cursor, query.limit); + } +} + +Reflect.defineMetadata( + "design:paramtypes", + [Object, String, CreateBottleDto], + BottleController.prototype, + "create", +); +Reflect.defineMetadata( + "design:paramtypes", + [Object, String], + BottleController.prototype, + "get", +); +Reflect.defineMetadata( + "design:paramtypes", + [Object, ListBottlesDto], + BottleController.prototype, + "list", +); diff --git a/apps/api/src/bottle/bottle.e2e-spec.ts b/apps/api/src/bottle/bottle.e2e-spec.ts new file mode 100644 index 0000000..b35b82e --- /dev/null +++ b/apps/api/src/bottle/bottle.e2e-spec.ts @@ -0,0 +1,285 @@ +/* eslint-disable @typescript-eslint/no-unsafe-member-access, @typescript-eslint/no-unsafe-assignment, @typescript-eslint/no-unsafe-argument */ +import "reflect-metadata"; +import { type INestApplication } from "@nestjs/common"; +import { Test } from "@nestjs/testing"; +import { PrismaClient } from "@prisma/client"; +import { randomUUID } from "node:crypto"; +import request from "supertest"; +import { afterAll, beforeAll, beforeEach, describe, expect, it } from "vitest"; +import { assertSafeTestDatabaseUrl } from "../../../../prisma/database-safety.js"; +import { AppModule } from "../app.module.js"; +import { TokenService } from "../auth/token.service.js"; +import { configureApp } from "../main.js"; +import { utc8UsageDate } from "./usage-date.js"; + +process.env.NODE_ENV = "test"; +process.env.WEB_ORIGIN = "http://localhost:3000"; +process.env.PHONE_ENCRYPTION_KEY = + "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA="; +process.env.PHONE_HMAC_KEY = "test-phone-hmac-key-with-at-least-32-bytes"; +process.env.VERIFICATION_CODE_HMAC_KEY = + "test-code-hmac-key-with-at-least-32-bytes"; +process.env.JWT_SECRET = "test-jwt-secret-with-at-least-thirty-two-bytes"; +process.env.REFRESH_TOKEN_HMAC_KEY = + "test-refresh-hmac-key-with-at-least-32-bytes"; +const prisma = new PrismaClient(); + +describe("bottles with real PostgreSQL", () => { + let app: INestApplication; + let accountId: string; + let authorization: string; + + beforeAll(async () => { + assertSafeTestDatabaseUrl(process.env.DATABASE_URL ?? ""); + await prisma.$connect(); + const module = await Test.createTestingModule({ + imports: [AppModule], + }).compile(); + app = module.createNestApplication(); + configureApp(app); + await app.init(); + }); + beforeEach(async () => { + await prisma.outboxEvent.deleteMany(); + await prisma.$executeRawUnsafe(`TRUNCATE TABLE "accounts" CASCADE`); + ({ accountId, authorization } = await actor("owner-device")); + }); + afterAll(async () => { + await app?.close(); + await prisma.$disconnect(); + }); + + async function actor(deviceId: string) { + const account = await prisma.account.create({ + data: { phoneCiphertext: Buffer.from("cipher"), phoneHmac: randomUUID() }, + }); + const session = await prisma.session.create({ + data: { + accountId: account.id, + refreshTokenHash: randomUUID(), + deviceId, + expiresAt: new Date(Date.now() + 60_000), + tokenFamily: randomUUID(), + }, + }); + const token = app.get(TokenService).issueAccess({ + sub: account.id, + session_id: session.id, + device_id: deviceId, + scopes: ["user"], + token_version: 0, + }); + return { accountId: account.id, authorization: `Bearer ${token}` }; + } + const create = (key: string, contentText = "hello sea") => + request(app.getHttpServer()) + .post("/api/v1/bottles") + .set("Authorization", authorization) + .set("Idempotency-Key", key) + .send({ contentText }); + + it("creates reviewing/closed bottle, task, and content-free outbox atomically", async () => { + const response = await create(randomUUID(), " hello sea ").expect(201); + expect(response.body.data).toMatchObject({ + contentText: "hello sea", + reviewStatus: "REVIEWING", + poolStatus: "CLOSED", + }); + const bottle = await prisma.bottle.findUniqueOrThrow({ + where: { id: response.body.data.id }, + }); + const [task, event, usage] = await Promise.all([ + prisma.moderationTask.findFirstOrThrow({ + where: { targetId: bottle.id }, + }), + prisma.outboxEvent.findFirstOrThrow({ + where: { aggregateId: bottle.id }, + }), + prisma.dailyUsage.findFirstOrThrow({ where: { accountId } }), + ]); + expect(task.payloadHash).toMatch(/^[a-f0-9]{64}$/); + expect(event.payload).toEqual({ bottleId: bottle.id, taskId: task.id }); + expect(JSON.stringify(event.payload)).not.toContain("hello sea"); + expect(usage.bottlesCreated).toBe(1); + }); + + it("deduplicates retry without incrementing quota or creating events", async () => { + const key = randomUUID(); + const first = await create(key).expect(201); + const retry = await create(key).expect(201); + expect(retry.body.data.id).toBe(first.body.data.id); + expect(await prisma.bottle.count()).toBe(1); + expect(await prisma.moderationTask.count()).toBe(1); + expect(await prisma.outboxEvent.count()).toBe(1); + expect((await prisma.dailyUsage.findFirstOrThrow()).bottlesCreated).toBe(1); + }); + + it("deduplicates concurrent retries without incrementing quota", async () => { + const key = randomUUID(); + const results = await Promise.all( + Array.from({ length: 8 }, () => create(key, "same content")), + ); + expect(results.every((result) => result.status === 201)).toBe(true); + expect( + new Set(results.map((result) => (result.body.data as { id: string }).id)), + ).toHaveLength(1); + expect(await prisma.bottle.count()).toBe(1); + expect(await prisma.moderationTask.count()).toBe(1); + expect(await prisma.outboxEvent.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) => + create(randomUUID(), `b${index}`), + ), + ); + expect(results.filter((x) => x.status === 201)).toHaveLength(10); + const denied = results.filter((x) => x.status === 429); + expect(denied).toHaveLength(2); + expect(denied.every((x) => x.body.code === "BOTTLE_DAILY_LIMIT")).toBe( + true, + ); + expect(await prisma.bottle.count()).toBe(10); + expect(await prisma.outboxEvent.count()).toBe(10); + expect((await prisma.dailyUsage.findFirstOrThrow()).bottlesCreated).toBe( + 10, + ); + }); + + it("validates trimmed content and idempotency key", async () => { + await create(randomUUID(), " ").expect(400); + await create(randomUUID(), "x".repeat(1001)).expect(400); + await request(app.getHttpServer()) + .post("/api/v1/bottles") + .set("Authorization", authorization) + .send({ contentText: "hello" }) + .expect(400); + }); + + it("allows only owner reads and cursor-paginates own bottles", async () => { + const ids: string[] = []; + for (let n = 0; n < 3; n += 1) + ids.push( + (await create(randomUUID(), `message${n}`).expect(201)).body.data.id, + ); + await request(app.getHttpServer()) + .get(`/api/v1/bottles/${ids[0]}`) + .set("Authorization", authorization) + .expect(200); + const stranger = await actor("stranger-device"); + await request(app.getHttpServer()) + .get(`/api/v1/bottles/${ids[0]}`) + .set("Authorization", stranger.authorization) + .expect(404); + const page1 = await request(app.getHttpServer()) + .get("/api/v1/me/bottles?limit=2") + .set("Authorization", authorization) + .expect(200); + expect(page1.body.data.items).toHaveLength(2); + expect(page1.body.data.nextCursor).toEqual(expect.any(String)); + const page2 = await request(app.getHttpServer()) + .get(`/api/v1/me/bottles?limit=2&cursor=${page1.body.data.nextCursor}`) + .set("Authorization", authorization) + .expect(200); + expect(page2.body.data.items).toHaveLength(1); + expect(page2.body.data.nextCursor).toBeNull(); + }); + + it("returns 400 for malformed, missing, too-long, and foreign cursors", async () => { + await request(app.getHttpServer()) + .get("/api/v1/me/bottles?cursor=not-a-uuid") + .set("Authorization", authorization) + .expect(400); + await request(app.getHttpServer()) + .get(`/api/v1/me/bottles?cursor=${randomUUID()}`) + .set("Authorization", authorization) + .expect(400); + const stranger = await actor("cursor-stranger"); + const foreignId = ( + await request(app.getHttpServer()) + .post("/api/v1/bottles") + .set("Authorization", stranger.authorization) + .set("Idempotency-Key", randomUUID()) + .send({ contentText: "foreign" }) + .expect(201) + ).body.data.id; + await request(app.getHttpServer()) + .get(`/api/v1/me/bottles?cursor=${foreignId}`) + .set("Authorization", authorization) + .expect(400); + }); + + it("rejects blank and overlong idempotency keys", async () => { + await create(" ").expect(400); + await create("x".repeat(129)).expect(400); + }); + + it.each(["SUSPENSION", "BAN"] as const)( + "blocks active %s sanction", + async (type) => { + await prisma.sanction.create({ + data: { + accountId, + type, + reason: "policy", + expiresAt: new Date(Date.now() + 60_000), + }, + }); + const denied = await create(randomUUID()).expect(403); + expect(denied.body.code).toBe("ACCOUNT_SANCTIONED"); + expect(await prisma.bottle.count()).toBe(0); + }, + ); + + it("blocks non-active accounts", async () => { + await prisma.account.update({ + where: { id: accountId }, + data: { status: "SUSPENDED" }, + }); + await create(randomUUID()).expect(401); + expect(await prisma.bottle.count()).toBe(0); + }); + + it("blocks sanctions that have started but ignores future sanctions", async () => { + await prisma.sanction.create({ + data: { + accountId, + type: "BAN", + reason: "future", + startsAt: new Date(Date.now() + 60_000), + }, + }); + await create(randomUUID()).expect(201); + }); + + it("ignores expired and revoked sanctions", async () => { + await prisma.sanction.createMany({ + data: [ + { + accountId, + type: "BAN", + reason: "expired", + expiresAt: new Date(Date.now() - 1), + }, + { + accountId, + type: "SUSPENSION", + reason: "revoked", + revokedAt: new Date(), + }, + ], + }); + await create(randomUUID()).expect(201); + }); + + it("computes usage day across UTC+8 midnight", () => { + expect(utc8UsageDate(new Date("2026-09-15T15:59:59.999Z"))).toBe( + "2026-09-15", + ); + expect(utc8UsageDate(new Date("2026-09-15T16:00:00.000Z"))).toBe( + "2026-09-16", + ); + }); +}); diff --git a/apps/api/src/bottle/bottle.module.ts b/apps/api/src/bottle/bottle.module.ts new file mode 100644 index 0000000..4f00fa7 --- /dev/null +++ b/apps/api/src/bottle/bottle.module.ts @@ -0,0 +1,12 @@ +import { Module } from "@nestjs/common"; +import { AuthModule } from "../auth/auth.module.js"; +import { DatabaseModule } from "../database/database.module.js"; +import { BottleController } from "./bottle.controller.js"; +import { BottleService } from "./bottle.service.js"; + +@Module({ + imports: [AuthModule, DatabaseModule], + controllers: [BottleController], + providers: [BottleService], +}) +export class BottleModule {} diff --git a/apps/api/src/bottle/bottle.service.ts b/apps/api/src/bottle/bottle.service.ts new file mode 100644 index 0000000..c9e7a92 --- /dev/null +++ b/apps/api/src/bottle/bottle.service.ts @@ -0,0 +1,138 @@ +import { createHash } from "node:crypto"; +import { HttpStatus, Inject, Injectable } from "@nestjs/common"; +import { ErrorCode } from "@drift/contracts"; +import { Prisma } from "@prisma/client"; +import { DomainException } from "../common/domain.exception.js"; +import { PrismaService } from "../database/prisma.service.js"; +import { utc8UsageDate } from "./usage-date.js"; + +@Injectable() +export class BottleService { + constructor(@Inject(PrismaService) private readonly prisma: PrismaService) {} + + async create(authorId: string, requestId: string, contentText: string) { + const existing = await this.prisma.bottle.findUnique({ + where: { + authorId_clientRequestId: { authorId, clientRequestId: requestId }, + }, + }); + if (existing) return existing; + try { + return await this.prisma.$transaction(async (tx) => { + const duplicate = await tx.bottle.findUnique({ + where: { + authorId_clientRequestId: { authorId, clientRequestId: requestId }, + }, + }); + if (duplicate) return duplicate; + 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() } }], + }, + select: { id: true }, + }); + if (sanctioned) + throw new DomainException( + ErrorCode.ACCOUNT_SANCTIONED, + "Account sanctioned", + HttpStatus.FORBIDDEN, + ); + const usageDate = utc8UsageDate(new Date()); + 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()) + ON CONFLICT ("account_id", "usage_date") DO UPDATE + SET "bottles_created" = "daily_usage"."bottles_created" + 1, "updated_at" = now() + WHERE "daily_usage"."bottles_created" < 10 + RETURNING "bottles_created"`; + if (rows.length === 0) + throw new DomainException( + ErrorCode.BOTTLE_DAILY_LIMIT, + "Daily bottle limit reached", + HttpStatus.TOO_MANY_REQUESTS, + ); + const bottle = await tx.bottle.create({ + data: { + authorId, + clientRequestId: requestId, + contentText, + reviewStatus: "REVIEWING", + poolStatus: "CLOSED", + }, + }); + const task = await tx.moderationTask.create({ + data: { + targetType: "BOTTLE", + targetId: bottle.id, + provider: "SIMULATED", + payloadHash: createHash("sha256").update(contentText).digest("hex"), + riskLabels: [], + }, + }); + await tx.outboxEvent.create({ + data: { + aggregateType: "BOTTLE", + aggregateId: bottle.id, + eventType: "BOTTLE_MODERATION_REQUESTED", + dedupeKey: `bottle-moderation:${bottle.id}`, + payload: { bottleId: bottle.id, taskId: task.id }, + }, + }); + return bottle; + }); + } catch (error) { + if ( + error instanceof Prisma.PrismaClientKnownRequestError && + error.code === "P2002" + ) { + return this.prisma.bottle.findUniqueOrThrow({ + where: { + authorId_clientRequestId: { authorId, clientRequestId: requestId }, + }, + }); + } + throw error; + } + } + + async get(authorId: string, id: string) { + const bottle = await this.prisma.bottle.findFirst({ + where: { id, authorId }, + }); + if (!bottle) + throw new DomainException( + ErrorCode.NOT_FOUND, + "Not Found", + HttpStatus.NOT_FOUND, + ); + return bottle; + } + + async list(authorId: string, cursor: string | undefined, limit: number) { + if (cursor) { + const ownedCursor = await this.prisma.bottle.findFirst({ + where: { id: cursor, authorId }, + select: { id: true }, + }); + if (!ownedCursor) + throw new DomainException( + ErrorCode.VALIDATION_ERROR, + "Invalid cursor", + HttpStatus.BAD_REQUEST, + ); + } + const items = await this.prisma.bottle.findMany({ + where: { authorId }, + orderBy: [{ createdAt: "desc" }, { id: "desc" }], + take: limit + 1, + ...(cursor ? { cursor: { id: cursor }, skip: 1 } : {}), + }); + const hasMore = items.length > limit; + if (hasMore) items.pop(); + return { items, nextCursor: hasMore ? (items.at(-1)?.id ?? null) : null }; + } +} diff --git a/apps/api/src/bottle/dto.ts b/apps/api/src/bottle/dto.ts new file mode 100644 index 0000000..e86c2b2 --- /dev/null +++ b/apps/api/src/bottle/dto.ts @@ -0,0 +1,26 @@ +import { Transform, Type } from "class-transformer"; +import { + IsInt, + IsOptional, + IsString, + IsUUID, + Max, + MaxLength, + Min, + MinLength, +} from "class-validator"; + +export class CreateBottleDto { + @Transform(({ value }: { value: unknown }) => + typeof value === "string" ? value.trim() : value, + ) + @IsString() + @MinLength(1) + @MaxLength(1000) + contentText!: string; +} + +export class ListBottlesDto { + @IsOptional() @IsUUID() cursor?: string; + @IsOptional() @Type(() => Number) @IsInt() @Min(1) @Max(50) limit = 20; +} diff --git a/apps/api/src/bottle/usage-date.ts b/apps/api/src/bottle/usage-date.ts new file mode 100644 index 0000000..59794df --- /dev/null +++ b/apps/api/src/bottle/usage-date.ts @@ -0,0 +1,5 @@ +export function utc8UsageDate(now: Date): string { + return new Date(now.getTime() + 8 * 60 * 60 * 1000) + .toISOString() + .slice(0, 10); +} diff --git a/apps/api/src/profile/dto.ts b/apps/api/src/profile/dto.ts new file mode 100644 index 0000000..19012ca --- /dev/null +++ b/apps/api/src/profile/dto.ts @@ -0,0 +1,31 @@ +import { Transform } from "class-transformer"; +import { + IsHexColor, + IsOptional, + IsString, + MaxLength, + MinLength, +} from "class-validator"; + +const trim = ({ value }: { value: unknown }) => + typeof value === "string" ? value.trim() : value; + +export class UpdateAnonymousProfileDto { + @Transform(trim) + @IsString() + @MinLength(1) + @MaxLength(64) + nickname!: string; + + @Transform(({ value }: { value: unknown }) => + typeof value === "string" ? value.toLowerCase() : value, + ) + @IsHexColor() + avatarColor!: string; + + @Transform(trim) + @IsOptional() + @IsString() + @MaxLength(500) + bio?: string | null; +} diff --git a/apps/api/src/profile/profile.controller.ts b/apps/api/src/profile/profile.controller.ts new file mode 100644 index 0000000..5e54bc7 --- /dev/null +++ b/apps/api/src/profile/profile.controller.ts @@ -0,0 +1,29 @@ +import { Body, Controller, Inject, Patch, UseGuards } from "@nestjs/common"; +import { AuthGuard } from "../auth/auth.guard.js"; +import { CurrentUser } from "../auth/current-user.decorator.js"; +import type { AccessClaims } from "../auth/token.service.js"; +import { UpdateAnonymousProfileDto } from "./dto.js"; +import { ProfileService } from "./profile.service.js"; + +@Controller("me/anonymous-profile") +@UseGuards(AuthGuard) +export class ProfileController { + constructor( + @Inject(ProfileService) private readonly profiles: ProfileService, + ) {} + + @Patch() + update( + @CurrentUser() user: AccessClaims, + @Body() input: UpdateAnonymousProfileDto, + ) { + return this.profiles.update(user.sub, input); + } +} + +Reflect.defineMetadata( + "design:paramtypes", + [Object, UpdateAnonymousProfileDto], + ProfileController.prototype, + "update", +); diff --git a/apps/api/src/profile/profile.e2e-spec.ts b/apps/api/src/profile/profile.e2e-spec.ts new file mode 100644 index 0000000..ea05e8c --- /dev/null +++ b/apps/api/src/profile/profile.e2e-spec.ts @@ -0,0 +1,118 @@ +/* eslint-disable @typescript-eslint/no-unsafe-argument, @typescript-eslint/no-unsafe-member-access, @typescript-eslint/no-unsafe-assignment */ +import "reflect-metadata"; +import { type INestApplication } from "@nestjs/common"; +import { Test } from "@nestjs/testing"; +import { PrismaClient } from "@prisma/client"; +import { randomUUID } from "node:crypto"; +import request from "supertest"; +import { afterAll, beforeAll, beforeEach, describe, expect, it } from "vitest"; +import { assertSafeTestDatabaseUrl } from "../../../../prisma/database-safety.js"; +import { AppModule } from "../app.module.js"; +import { configureApp } from "../main.js"; +import { TokenService } from "../auth/token.service.js"; + +process.env.NODE_ENV = "test"; +process.env.WEB_ORIGIN = "http://localhost:3000"; +process.env.PHONE_ENCRYPTION_KEY = + "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA="; +process.env.PHONE_HMAC_KEY = "test-phone-hmac-key-with-at-least-32-bytes"; +process.env.VERIFICATION_CODE_HMAC_KEY = + "test-code-hmac-key-with-at-least-32-bytes"; +process.env.JWT_SECRET = "test-jwt-secret-with-at-least-thirty-two-bytes"; +process.env.REFRESH_TOKEN_HMAC_KEY = + "test-refresh-hmac-key-with-at-least-32-bytes"; + +const prisma = new PrismaClient(); + +describe("anonymous profile", () => { + let app: INestApplication; + let authorization: string; + let accountId: string; + + beforeAll(async () => { + assertSafeTestDatabaseUrl(process.env.DATABASE_URL ?? ""); + await prisma.$connect(); + const module = await Test.createTestingModule({ + imports: [AppModule], + }).compile(); + app = module.createNestApplication(); + configureApp(app); + await app.init(); + }); + + beforeEach(async () => { + await prisma.$executeRawUnsafe(`TRUNCATE TABLE "accounts" CASCADE`); + const account = await prisma.account.create({ + data: { phoneCiphertext: Buffer.from("cipher"), phoneHmac: randomUUID() }, + }); + accountId = account.id; + const session = await prisma.session.create({ + data: { + accountId, + refreshTokenHash: randomUUID(), + deviceId: "profile-device", + expiresAt: new Date(Date.now() + 60_000), + tokenFamily: randomUUID(), + }, + }); + const token = app.get(TokenService).issueAccess({ + sub: accountId, + session_id: session.id, + device_id: "profile-device", + scopes: ["user"], + token_version: 0, + }); + authorization = `Bearer ${token}`; + }); + + afterAll(async () => { + await app?.close(); + await prisma.$disconnect(); + }); + + it("updates validated anonymous profile, preserves publicId, and exposes pending review to self", async () => { + const first = await request(app.getHttpServer()) + .patch("/api/v1/me/anonymous-profile") + .set("Authorization", authorization) + .send({ nickname: " 海风 ", avatarColor: "#12AbEf", bio: " hello " }) + .expect(200); + expect(first.body.data).toMatchObject({ + nickname: "海风", + avatarColor: "#12abef", + bio: "hello", + reviewStatus: "REVIEWING", + }); + const publicId = first.body.data.publicId as string; + const second = await request(app.getHttpServer()) + .patch("/api/v1/me/anonymous-profile") + .set("Authorization", authorization) + .send({ nickname: "浪花", avatarColor: "#abcdef", bio: null }) + .expect(200); + expect(second.body.data.publicId).toBe(publicId); + expect(second.body.data.bio).toBeNull(); + const me = await request(app.getHttpServer()) + .get("/api/v1/me") + .set("Authorization", authorization) + .expect(200); + expect(me.body.data).toMatchObject({ + publicId, + nickname: "浪花", + avatarColor: "#abcdef", + bio: null, + profileReviewStatus: "REVIEWING", + }); + }); + + it.each([ + { nickname: " ", avatarColor: "#abcdef" }, + { nickname: "x".repeat(65), avatarColor: "#abcdef" }, + { nickname: "ok", avatarColor: "red" }, + { nickname: "ok", avatarColor: "#abcdef", bio: "x".repeat(501) }, + ])("rejects invalid input %#", async (body) => { + await request(app.getHttpServer()) + .patch("/api/v1/me/anonymous-profile") + .set("Authorization", authorization) + .send(body) + .expect(400); + }); +}); diff --git a/apps/api/src/profile/profile.module.ts b/apps/api/src/profile/profile.module.ts new file mode 100644 index 0000000..bf6f483 --- /dev/null +++ b/apps/api/src/profile/profile.module.ts @@ -0,0 +1,12 @@ +import { Module } from "@nestjs/common"; +import { AuthModule } from "../auth/auth.module.js"; +import { DatabaseModule } from "../database/database.module.js"; +import { ProfileController } from "./profile.controller.js"; +import { ProfileService } from "./profile.service.js"; + +@Module({ + imports: [AuthModule, DatabaseModule], + controllers: [ProfileController], + providers: [ProfileService], +}) +export class ProfileModule {} diff --git a/apps/api/src/profile/profile.service.ts b/apps/api/src/profile/profile.service.ts new file mode 100644 index 0000000..eaec5a7 --- /dev/null +++ b/apps/api/src/profile/profile.service.ts @@ -0,0 +1,34 @@ +import { Inject, Injectable } from "@nestjs/common"; +import { PrismaService } from "../database/prisma.service.js"; +import type { UpdateAnonymousProfileDto } from "./dto.js"; + +@Injectable() +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, + }, + }); + } +} diff --git a/apps/worker/package.json b/apps/worker/package.json new file mode 100644 index 0000000..125f589 --- /dev/null +++ b/apps/worker/package.json @@ -0,0 +1,20 @@ +{ + "name": "@drift/worker", + "version": "0.0.0", + "private": true, + "type": "module", + "scripts": { + "build": "tsc -p tsconfig.build.json", + "start": "node dist/main.js", + "run:once": "tsx src/main.ts --once", + "test": "vitest run src/moderation-worker.spec.ts --no-file-parallelism", + "typecheck": "tsc --noEmit -p tsconfig.json" + }, + "dependencies": { + "@prisma/client": "6.19.0" + }, + "devDependencies": { + "typescript": "^5.6.3", + "vitest": "^4.1.11" + } +} diff --git a/apps/worker/src/main.ts b/apps/worker/src/main.ts new file mode 100644 index 0000000..9d6de84 --- /dev/null +++ b/apps/worker/src/main.ts @@ -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()); diff --git a/apps/worker/src/moderation-worker.spec.ts b/apps/worker/src/moderation-worker.spec.ts new file mode 100644 index 0000000..01143df --- /dev/null +++ b/apps/worker/src/moderation-worker.spec.ts @@ -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 }; + } +}); diff --git a/apps/worker/src/moderation-worker.ts b/apps/worker/src/moderation-worker.ts new file mode 100644 index 0000000..f21769e --- /dev/null +++ b/apps/worker/src/moderation-worker.ts @@ -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; +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 { + 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 { + const staleBefore = new Date(Date.now() - this.leaseMs); + return this.prisma.$transaction(async (tx) => { + const rows = await tx.$queryRaw` + 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; + }); + } +} diff --git a/apps/worker/tsconfig.build.json b/apps/worker/tsconfig.build.json new file mode 100644 index 0000000..5fb5003 --- /dev/null +++ b/apps/worker/tsconfig.build.json @@ -0,0 +1,5 @@ +{ + "extends": "./tsconfig.json", + "compilerOptions": { "noEmit": false, "outDir": "dist", "rootDir": "src" }, + "exclude": ["src/**/*.spec.ts"] +} diff --git a/apps/worker/tsconfig.json b/apps/worker/tsconfig.json new file mode 100644 index 0000000..998a209 --- /dev/null +++ b/apps/worker/tsconfig.json @@ -0,0 +1,11 @@ +{ + "extends": "../../tsconfig.base.json", + "compilerOptions": { + "module": "NodeNext", + "moduleResolution": "NodeNext", + "noEmit": true, + "types": ["node", "vitest/globals"] + }, + "include": ["src/**/*.ts"], + "exclude": ["src/**/*spec.ts"] +} diff --git a/packages/contracts/src/index.test.ts b/packages/contracts/src/index.test.ts index 2d8a1c9..fe78a30 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", + "ACCOUNT_SANCTIONED", "BOTTLE_POOL_EMPTY", "BOTTLE_LEASE_EXPIRED", "CONTENT_REJECTED", diff --git a/packages/contracts/src/index.ts b/packages/contracts/src/index.ts index 31ab4da..9b5d82f 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", + ACCOUNT_SANCTIONED = "ACCOUNT_SANCTIONED", BOTTLE_POOL_EMPTY = "BOTTLE_POOL_EMPTY", BOTTLE_LEASE_EXPIRED = "BOTTLE_LEASE_EXPIRED", CONTENT_REJECTED = "CONTENT_REJECTED", diff --git a/pnpm-lock.yaml b/pnpm-lock.yaml index 1413d6c..0073ae9 100644 --- a/pnpm-lock.yaml +++ b/pnpm-lock.yaml @@ -105,6 +105,19 @@ importers: specifier: ^4.1.11 version: 4.1.11(@types/node@22.19.3)(vite@7.3.6(@types/node@22.19.3)(jiti@2.7.0)(tsx@4.20.6)) + apps/worker: + dependencies: + "@prisma/client": + specifier: 6.19.0 + version: 6.19.0(prisma@6.19.0(typescript@5.9.3))(typescript@5.9.3) + devDependencies: + typescript: + specifier: ^5.6.3 + version: 5.9.3 + vitest: + specifier: ^4.1.11 + version: 4.1.11(@types/node@22.19.3)(vite@7.3.6(@types/node@22.19.3)(jiti@2.7.0)(tsx@4.20.6)) + packages/contracts: {} packages: diff --git a/prisma/migrations/0005_bottle_moderation_outbox/migration.sql b/prisma/migrations/0005_bottle_moderation_outbox/migration.sql new file mode 100644 index 0000000..8826a71 --- /dev/null +++ b/prisma/migrations/0005_bottle_moderation_outbox/migration.sql @@ -0,0 +1,16 @@ +ALTER TABLE "bottles" ADD COLUMN "client_request_id" VARCHAR(128); +UPDATE "bottles" SET "client_request_id" = "id"::text WHERE "client_request_id" IS NULL; +ALTER TABLE "bottles" ALTER COLUMN "client_request_id" SET NOT NULL; +CREATE UNIQUE INDEX "bottles_author_id_client_request_id_key" + ON "bottles"("author_id", "client_request_id"); + +CREATE UNIQUE INDEX "moderation_tasks_target_type_target_id_key" + ON "moderation_tasks"("target_type", "target_id"); + +ALTER TABLE "outbox_events" + ADD COLUMN "dedupe_key" VARCHAR(255), + ADD COLUMN "locked_at" TIMESTAMPTZ(3); +UPDATE "outbox_events" SET "dedupe_key" = "id"::text WHERE "dedupe_key" IS NULL; +ALTER TABLE "outbox_events" ALTER COLUMN "dedupe_key" SET NOT NULL; +CREATE UNIQUE INDEX "outbox_events_dedupe_key_key" ON "outbox_events"("dedupe_key"); +CREATE INDEX "outbox_events_processing_lease_idx" ON "outbox_events"("status", "locked_at"); \ No newline at end of file diff --git a/prisma/schema.prisma b/prisma/schema.prisma index 8b4e8ba..5604b8b 100644 --- a/prisma/schema.prisma +++ b/prisma/schema.prisma @@ -193,23 +193,25 @@ model RefreshToken { } model Bottle { - id String @id @default(uuid()) @db.Uuid - authorId String @map("author_id") @db.Uuid - contentText String @map("content_text") @db.Text - reviewStatus ReviewStatus @default(DRAFT) @map("review_status") - poolStatus BottlePoolStatus @default(CLOSED) @map("pool_status") - version Int @default(1) - approvedAt DateTime? @map("approved_at") @db.Timestamptz(3) - consumedAt DateTime? @map("consumed_at") @db.Timestamptz(3) - expiresAt DateTime? @map("expires_at") @db.Timestamptz(3) - createdAt DateTime @default(now()) @map("created_at") @db.Timestamptz(3) - updatedAt DateTime @updatedAt @map("updated_at") @db.Timestamptz(3) - author Account @relation(fields: [authorId], references: [id], onDelete: Restrict) - leases BottlePickLease[] - pickHistory BottlePickHistory[] - conversation Conversation? - reports Report[] + id String @id @default(uuid()) @db.Uuid + authorId String @map("author_id") @db.Uuid + clientRequestId String @default(uuid()) @map("client_request_id") @db.VarChar(128) + contentText String @map("content_text") @db.Text + reviewStatus ReviewStatus @default(DRAFT) @map("review_status") + poolStatus BottlePoolStatus @default(CLOSED) @map("pool_status") + version Int @default(1) + approvedAt DateTime? @map("approved_at") @db.Timestamptz(3) + consumedAt DateTime? @map("consumed_at") @db.Timestamptz(3) + expiresAt DateTime? @map("expires_at") @db.Timestamptz(3) + createdAt DateTime @default(now()) @map("created_at") @db.Timestamptz(3) + updatedAt DateTime @updatedAt @map("updated_at") @db.Timestamptz(3) + author Account @relation(fields: [authorId], references: [id], onDelete: Restrict) + leases BottlePickLease[] + pickHistory BottlePickHistory[] + conversation Conversation? + reports Report[] + @@unique([authorId, clientRequestId]) @@index([poolStatus, createdAt]) @@index([reviewStatus, createdAt]) @@index([authorId, createdAt]) @@ -380,6 +382,7 @@ model ModerationTask { assignedTo Account? @relation("Moderator", fields: [assignedToId], references: [id], onDelete: SetNull) sanctions Sanction[] + @@unique([targetType, targetId]) @@index([status, createdAt]) @@index([targetType, targetId]) @@index([assignedToId, status]) @@ -426,10 +429,12 @@ model OutboxEvent { aggregateType String @map("aggregate_type") @db.VarChar(100) aggregateId String @map("aggregate_id") @db.Uuid eventType String @map("event_type") @db.VarChar(150) + dedupeKey String @unique @default(uuid()) @map("dedupe_key") @db.VarChar(255) payload Json status OutboxStatus @default(PENDING) attempts Int @default(0) nextRetryAt DateTime @default(now()) @map("next_retry_at") @db.Timestamptz(3) + lockedAt DateTime? @map("locked_at") @db.Timestamptz(3) publishedAt DateTime? @map("published_at") @db.Timestamptz(3) createdAt DateTime @default(now()) @map("created_at") @db.Timestamptz(3) updatedAt DateTime @updatedAt @map("updated_at") @db.Timestamptz(3)