From fa0fa78312830cb516d8bae3b22408b6973ff55d Mon Sep 17 00:00:00 2001 From: root Date: Thu, 17 Sep 2026 13:08:55 +0800 Subject: [PATCH] =?UTF-8?q?fix(=E6=B2=BB=E7=90=86):=20=E5=AE=8C=E6=88=90?= =?UTF-8?q?=E4=BB=BB=E5=8A=A1=208=20=E5=AE=89=E5=85=A8=E4=B8=8E=E9=80=9A?= =?UTF-8?q?=E7=9F=A5=E9=97=AD=E7=8E=AF=E4=BF=AE=E5=A4=8D?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 串行化拉黑、处罚、投瓶和匹配策略检查\n- 完成异步站内通知、未读统计和偏好并发语义\n- 补齐后台查询审计、处罚恢复和隐私测试\n- 稳定 Redis 恢复、匹配锁序及超时测试 --- apps/api/src/admin/admin.controller.ts | 18 +- apps/api/src/admin/admin.module.ts | 3 +- apps/api/src/admin/admin.service.ts | 129 +++-- apps/api/src/auth/auth.e2e-spec.ts | 30 ++ apps/api/src/auth/auth.guard.ts | 12 + apps/api/src/auth/auth.service.ts | 12 + apps/api/src/bottle/bottle.e2e-spec.ts | 4 +- apps/api/src/bottle/bottle.module.ts | 3 +- apps/api/src/bottle/bottle.service.ts | 19 +- .../src/conversation/conversation.e2e-spec.ts | 38 +- .../src/conversation/conversation.module.ts | 9 +- .../src/conversation/conversation.service.ts | 10 + apps/api/src/health/health.e2e-spec.ts | 4 +- apps/api/src/match/match-retry-policy.spec.ts | 6 +- apps/api/src/match/match-retry-policy.ts | 6 +- apps/api/src/match/match.e2e-spec.ts | 19 +- apps/api/src/match/match.module.ts | 3 +- apps/api/src/match/match.service.spec.ts | 97 +++- apps/api/src/match/match.service.ts | 153 ++++-- .../src/notification/notification.module.ts | 1 + .../src/notification/notification.service.ts | 113 +++- apps/api/src/safety/safety-admin.e2e-spec.ts | 483 ++++++++++++++++++ apps/api/src/safety/safety.module.ts | 3 +- apps/api/src/safety/safety.service.ts | 20 +- apps/worker/src/main.spec.ts | 8 + apps/worker/src/main.ts | 8 +- .../worker/src/notification.processor.spec.ts | 132 +++++ apps/worker/src/notification.processor.ts | 38 ++ package.json | 3 +- 29 files changed, 1231 insertions(+), 153 deletions(-) create mode 100644 apps/worker/src/notification.processor.spec.ts create mode 100644 apps/worker/src/notification.processor.ts diff --git a/apps/api/src/admin/admin.controller.ts b/apps/api/src/admin/admin.controller.ts index 11b4b4d..f922e78 100644 --- a/apps/api/src/admin/admin.controller.ts +++ b/apps/api/src/admin/admin.controller.ts @@ -25,8 +25,11 @@ import { @UseGuards(AuthGuard, AdminGuard) export class AdminController { constructor(@Inject(AdminService) private readonly admin: AdminService) {} - @Get("reports") reports(@Query() q: AdminQueryDto) { - return this.admin.reports(q.status, q.limit); + @Get("reports") reports( + @CurrentUser() u: AccessClaims, + @Query() q: AdminQueryDto, + ) { + return this.admin.reports(u.sub, q.status, q.limit); } @UseGuards(StateChangingOriginGuard) @Post("reports/:id/resolve") resolve( @CurrentUser() u: AccessClaims, @@ -35,8 +38,11 @@ export class AdminController { ) { return this.admin.resolve(u.sub, id, d); } - @Get("moderation") moderation(@Query() q: AdminQueryDto) { - return this.admin.moderation(q.limit); + @Get("moderation") moderation( + @CurrentUser() u: AccessClaims, + @Query() q: AdminQueryDto, + ) { + return this.admin.moderation(u.sub, q.limit); } @UseGuards(StateChangingOriginGuard) @Post("moderation/:id/resolve") moderate( @CurrentUser() u: AccessClaims, @@ -58,7 +64,7 @@ export class AdminController { Reflect.defineMetadata( "design:paramtypes", - [AdminQueryDto], + [Object, AdminQueryDto], AdminController.prototype, "reports", ); @@ -70,7 +76,7 @@ Reflect.defineMetadata( ); Reflect.defineMetadata( "design:paramtypes", - [AdminQueryDto], + [Object, AdminQueryDto], AdminController.prototype, "moderation", ); diff --git a/apps/api/src/admin/admin.module.ts b/apps/api/src/admin/admin.module.ts index 31afa9c..7fed667 100644 --- a/apps/api/src/admin/admin.module.ts +++ b/apps/api/src/admin/admin.module.ts @@ -6,8 +6,9 @@ import { SafetyModule } from "../safety/safety.module.js"; import { AdminController } from "./admin.controller.js"; import { AdminGuard } from "./admin.guard.js"; import { AdminService } from "./admin.service.js"; +import { NotificationModule } from "../notification/notification.module.js"; @Module({ - imports: [DatabaseModule, AuthModule, SafetyModule], + imports: [DatabaseModule, AuthModule, SafetyModule, NotificationModule], controllers: [AdminController], providers: [AdminService, AdminGuard, StateChangingOriginGuard], }) diff --git a/apps/api/src/admin/admin.service.ts b/apps/api/src/admin/admin.service.ts index ae1be9a..8b4be56 100644 --- a/apps/api/src/admin/admin.service.ts +++ b/apps/api/src/admin/admin.service.ts @@ -6,13 +6,17 @@ import { DomainException } from "../common/domain.exception.js"; import { PrismaService } from "../database/prisma.service.js"; import { SafetyLockService } from "../safety/safety-lock.service.js"; import type { ResolveReportDto, SanctionDto } from "./dto.js"; +import { NotificationService } from "../notification/notification.service.js"; @Injectable() export class AdminService { constructor( @Inject(PrismaService) private readonly prisma: PrismaService, @Inject(SafetyLockService) private readonly locks: SafetyLockService, + @Inject(NotificationService) + private readonly notifications: NotificationService, ) {} async reports( + actorId: string, status: "PENDING" | "REVIEWING" | "RESOLVED" | "DISMISSED" | undefined, limit: number, ) { @@ -21,6 +25,14 @@ export class AdminService { orderBy: [{ createdAt: "asc" }, { id: "asc" }], take: limit, }); + await this.prisma.auditLog.create({ + data: { + actorId, + action: "REPORTS_QUERIED", + entityType: "REPORT", + metadata: { status: status ?? null, limit, resultCount: rows.length }, + }, + }); return { items: rows.map((x) => ({ id: x.id, @@ -66,8 +78,20 @@ export class AdminService { }) )?.authorId : report.conversationId - ? (report.targetSnapshot as { reportedAccountId?: string }) - .reportedAccountId + ? ( + await tx.anonymousProfile.findUnique({ + where: { + publicId: + ( + report.targetSnapshot as { + subjectPublicId?: string; + } + ).subjectPublicId ?? + "00000000-0000-0000-0000-000000000000", + }, + select: { accountId: true }, + }) + )?.accountId : undefined); await this.locks.lockAccounts(tx, [ actorId, @@ -112,31 +136,21 @@ export class AdminService { }, }, }); - await tx.notification.create({ - data: { - accountId: report.reporterId, - dedupeKey: `report:${id}:reporter`, - type: "REPORT_RESOLVED", - payload: { reportId: id, decision: dto.decision }, - status: "SENT", - sentAt: new Date(), - }, - }); - if (accountId) - await tx.notification.create({ - data: { - accountId, - dedupeKey: `report:${id}:target`, - type: "MODERATION_DECISION", - payload: { - reportId: id, - decision: dto.decision, - sanctionType: dto.sanction?.type ?? null, - }, - status: "SENT", - sentAt: new Date(), - }, - }); + await this.notifications.createInApp( + tx, + report.reporterId, + `report:${id}:reporter`, + "REPORT_RESOLVED", + { reportId: id, decision: dto.decision }, + ); + if (dto.decision === "UPHELD" && dto.sanction && accountId) + await this.notifications.createInApp( + tx, + accountId, + `report:${id}:target`, + "ACCOUNT_ACTION", + { type: "ACCOUNT_ACTION" }, + ); return tx.report.findUniqueOrThrow({ where: { id }, select: { id: true, status: true, resolution: true }, @@ -158,6 +172,13 @@ export class AdminService { return this.prisma.$transaction(async (tx) => { await this.locks.lockAccounts(tx, [actorId, profile.accountId]); const sanction = await this.issue(tx, actorId, profile.accountId, dto); + await this.notifications.createInApp( + tx, + profile.accountId, + `sanction:${sanction.id}:account-action`, + "ACCOUNT_ACTION", + { type: "ACCOUNT_ACTION" }, + ); await tx.auditLog.create({ data: { actorId, @@ -170,25 +191,32 @@ export class AdminService { return sanction; }); } - async moderation(limit: number) { - return { - items: await this.prisma.moderationTask.findMany({ - where: { - status: "COMPLETED", - decision: "MANUAL_REVIEW", - targetType: { in: ["PROFILE", "BOTTLE"] }, - }, - select: { - id: true, - targetType: true, - targetId: true, - riskLabels: true, - createdAt: true, - }, - orderBy: { createdAt: "asc" }, - take: limit, - }), - }; + async moderation(actorId: string, limit: number) { + const items = await this.prisma.moderationTask.findMany({ + where: { + status: "COMPLETED", + decision: "MANUAL_REVIEW", + targetType: { in: ["PROFILE", "BOTTLE"] }, + }, + select: { + id: true, + targetType: true, + targetId: true, + riskLabels: true, + createdAt: true, + }, + orderBy: { createdAt: "asc" }, + take: limit, + }); + await this.prisma.auditLog.create({ + data: { + actorId, + action: "MODERATION_QUERIED", + entityType: "MODERATION_TASK", + metadata: { limit, resultCount: items.length }, + }, + }); + return { items }; } async moderate( actorId: string, @@ -369,11 +397,16 @@ export class AdminService { expiresAt: dto.expiresAt ? new Date(dto.expiresAt) : null, }, }); - if (dto.type === "SUSPENSION" || dto.type === "BAN") + if (dto.type === "SUSPENSION" || dto.type === "BAN") { await tx.account.update({ where: { id: accountId }, - data: { status: "SUSPENDED", tokenVersion: { increment: 1 } }, + data: { tokenVersion: { increment: 1 } }, }); + await tx.session.updateMany({ + where: { accountId, revokedAt: null }, + data: { revokedAt: new Date() }, + }); + } return sanction; } } diff --git a/apps/api/src/auth/auth.e2e-spec.ts b/apps/api/src/auth/auth.e2e-spec.ts index d299e29..8d7d75a 100644 --- a/apps/api/src/auth/auth.e2e-spec.ts +++ b/apps/api/src/auth/auth.e2e-spec.ts @@ -780,6 +780,36 @@ describe("auth real PostgreSQL/Redis", () => { .expect(401); }); + it("rejects login while a suspension is active and permits a fresh login after expiry", async () => { + const initial = await login(); + expect(initial.status).toBe(201); + const account = await prisma.account.findFirstOrThrow(); + const expiresAt = new Date(Date.now() + 60_000); + const sanction = await prisma.sanction.create({ + data: { + accountId: account.id, + type: "SUSPENSION", + reason: "test", + startsAt: new Date(Date.now() - 1_000), + expiresAt, + }, + }); + const sent = await send(); + const blocked = await request(app.getHttpServer()) + .post("/api/v1/auth/sms/login") + .send({ phone, deviceId, code: sent.body.data.debugCode }); + expect(blocked.status).toBe(401); + await prisma.sanction.update({ + where: { id: sanction.id }, + data: { expiresAt: new Date(Date.now() - 1_000) }, + }); + const recovered = await login(); + expect(recovered.status).toBe(201); + expect( + await prisma.account.findUniqueOrThrow({ where: { id: account.id } }), + ).toMatchObject({ status: "ACTIVE" }); + }); + it("recovers Auth Redis after a connection deadline and closes its sockets", async () => { const sockets = new Set(); const server = createServer((socket) => { diff --git a/apps/api/src/auth/auth.guard.ts b/apps/api/src/auth/auth.guard.ts index 52bc514..b70d557 100644 --- a/apps/api/src/auth/auth.guard.ts +++ b/apps/api/src/auth/auth.guard.ts @@ -35,6 +35,18 @@ export class AuthGuard implements CanActivate { session.account.tokenVersion !== claims.token_version ) throw this.denied(); + const now = new Date(); + const sanction = await this.prisma.sanction.findFirst({ + where: { + accountId: claims.sub, + type: { in: ["SUSPENSION", "BAN"] }, + revokedAt: null, + startsAt: { lte: now }, + OR: [{ expiresAt: null }, { expiresAt: { gt: now } }], + }, + select: { id: true }, + }); + if (sanction) throw this.denied(); return claims; } diff --git a/apps/api/src/auth/auth.service.ts b/apps/api/src/auth/auth.service.ts index 51f4603..2bd9be4 100644 --- a/apps/api/src/auth/auth.service.ts +++ b/apps/api/src/auth/auth.service.ts @@ -148,6 +148,18 @@ export class AuthService { }); if (!account) account = await this.createAccount(phone, digest); if (account.status !== "ACTIVE") throw this.invalid(); + const now = new Date(); + const sanction = await this.prisma.sanction.findFirst({ + where: { + accountId: account.id, + type: { in: ["SUSPENSION", "BAN"] }, + revokedAt: null, + startsAt: { lte: now }, + OR: [{ expiresAt: null }, { expiresAt: { gt: now } }], + }, + select: { id: true }, + }); + if (sanction) throw this.invalid(); return this.createSession(account, deviceId); } private async createAccount(phone: string, digest: string): Promise { diff --git a/apps/api/src/bottle/bottle.e2e-spec.ts b/apps/api/src/bottle/bottle.e2e-spec.ts index dafa779..1255329 100644 --- a/apps/api/src/bottle/bottle.e2e-spec.ts +++ b/apps/api/src/bottle/bottle.e2e-spec.ts @@ -255,8 +255,8 @@ describe("bottles with real PostgreSQL", () => { expiresAt: new Date(Date.now() + 60_000), }, }); - const denied = await create(randomUUID()).expect(403); - expect(denied.body.code).toBe("ACCOUNT_SANCTIONED"); + const denied = await create(randomUUID()).expect(401); + expect(denied.body.code).toBe("AUTH_UNAUTHORIZED"); expect(await prisma.bottle.count()).toBe(0); }, ); diff --git a/apps/api/src/bottle/bottle.module.ts b/apps/api/src/bottle/bottle.module.ts index 4f00fa7..ea392dd 100644 --- a/apps/api/src/bottle/bottle.module.ts +++ b/apps/api/src/bottle/bottle.module.ts @@ -3,9 +3,10 @@ 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"; +import { SafetyModule } from "../safety/safety.module.js"; @Module({ - imports: [AuthModule, DatabaseModule], + imports: [AuthModule, DatabaseModule, SafetyModule], controllers: [BottleController], providers: [BottleService], }) diff --git a/apps/api/src/bottle/bottle.service.ts b/apps/api/src/bottle/bottle.service.ts index e32c9f2..f06c017 100644 --- a/apps/api/src/bottle/bottle.service.ts +++ b/apps/api/src/bottle/bottle.service.ts @@ -5,20 +5,19 @@ 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"; +import { SafetyLockService } from "../safety/safety-lock.service.js"; @Injectable() export class BottleService { - constructor(@Inject(PrismaService) private readonly prisma: PrismaService) {} + constructor( + @Inject(PrismaService) private readonly prisma: PrismaService, + @Inject(SafetyLockService) private readonly locks: SafetyLockService, + ) {} async create(authorId: string, requestId: string, contentText: string) { - const existing = await this.prisma.bottle.findUnique({ - where: { - authorId_clientRequestId: { authorId, clientRequestId: requestId }, - }, - }); - if (existing) return this.resolveIdempotent(existing, contentText); try { return await this.prisma.$transaction(async (tx) => { + await this.locks.lockAccounts(tx, [authorId]); await tx.$executeRaw`SELECT pg_advisory_xact_lock(hashtextextended(${`${authorId}:${requestId}`}, 0))`; const duplicate = await tx.bottle.findUnique({ where: { @@ -27,6 +26,10 @@ export class BottleService { }); if (duplicate) return this.resolveIdempotent(duplicate, contentText); const now = new Date(); + const account = await tx.account.findUnique({ + where: { id: authorId }, + select: { status: true }, + }); const sanctioned = await tx.sanction.findFirst({ where: { accountId: authorId, @@ -37,7 +40,7 @@ export class BottleService { }, select: { id: true }, }); - if (sanctioned) + if (account?.status !== "ACTIVE" || sanctioned) throw new DomainException( ErrorCode.ACCOUNT_SANCTIONED, "Account sanctioned", diff --git a/apps/api/src/conversation/conversation.e2e-spec.ts b/apps/api/src/conversation/conversation.e2e-spec.ts index 9e877f7..4a6d09b 100644 --- a/apps/api/src/conversation/conversation.e2e-spec.ts +++ b/apps/api/src/conversation/conversation.e2e-spec.ts @@ -166,6 +166,42 @@ describe("conversation API with real PostgreSQL", () => { }); } + it("queues a first-reply notification without message text or internal account ids", async () => { + const author = await actor("author"); + const picker = await actor("picker"); + const { bottle, lease } = await leased(author.id, picker); + const key = randomUUID(); + const first = await reply( + picker.authorization, + bottle.id, + lease, + key, + "very private body", + ).expect(201); + await reply( + picker.authorization, + bottle.id, + lease, + key, + "very private body", + ).expect(201); + const rows = await prisma.notification.findMany({ + where: { accountId: author.id }, + }); + expect(rows).toHaveLength(1); + expect(rows[0]).toMatchObject({ + type: "FIRST_REPLY", + status: "PENDING", + sentAt: null, + }); + expect(rows[0]!.payload).toEqual({ + conversationId: first.body.data.conversationId, + }); + expect(JSON.stringify(rows[0]!.payload)).not.toContain(picker.id); + expect(JSON.stringify(rows[0]!.payload)).not.toContain(author.id); + expect(JSON.stringify(rows[0]!.payload)).not.toContain("very private body"); + }); + it("atomically consumes a lease and deduplicates concurrent first replies", async () => { const author = await actor("author"); const picker = await actor("picker"); @@ -274,7 +310,7 @@ describe("conversation API with real PostgreSQL", () => { randomUUID(), text, ); - expect(response.status).toBe(403); + expect(response.status).toBe(kind === "sanction" ? 401 : 403); expect(await prisma.conversation.count()).toBe(0); expect( (await prisma.bottle.findUniqueOrThrow({ where: { id: bottle.id } })) diff --git a/apps/api/src/conversation/conversation.module.ts b/apps/api/src/conversation/conversation.module.ts index f945e15..67013cc 100644 --- a/apps/api/src/conversation/conversation.module.ts +++ b/apps/api/src/conversation/conversation.module.ts @@ -3,6 +3,7 @@ import { AuthModule } from "../auth/auth.module.js"; import { DatabaseModule } from "../database/database.module.js"; import { RedisModule } from "../redis/redis.module.js"; import { SafetyModule } from "../safety/safety.module.js"; +import { NotificationModule } from "../notification/notification.module.js"; import { ChatGateway } from "./chat.gateway.js"; import { ChatRateLimiter } from "./chat-rate-limiter.js"; import { ConversationController } from "./conversation.controller.js"; @@ -11,7 +12,13 @@ import { CHAT_PUBLISHER, OutboxMessageRelay } from "./outbox-message-relay.js"; import { StateChangingOriginGuard } from "./state-changing-origin.guard.js"; @Module({ - imports: [DatabaseModule, RedisModule, AuthModule, SafetyModule], + imports: [ + DatabaseModule, + RedisModule, + AuthModule, + SafetyModule, + NotificationModule, + ], controllers: [ConversationController], providers: [ ConversationService, diff --git a/apps/api/src/conversation/conversation.service.ts b/apps/api/src/conversation/conversation.service.ts index 7eea09c..028eccc 100644 --- a/apps/api/src/conversation/conversation.service.ts +++ b/apps/api/src/conversation/conversation.service.ts @@ -5,6 +5,7 @@ import { leaseHmac } from "../auth/auth.config.js"; import { DomainException } from "../common/domain.exception.js"; import { PrismaService } from "../database/prisma.service.js"; import { SafetyLockService } from "../safety/safety-lock.service.js"; +import { NotificationService } from "../notification/notification.service.js"; type Db = Prisma.TransactionClient | PrismaClient; type MessageRow = { @@ -46,6 +47,8 @@ export class ConversationService { constructor( @Inject(PrismaService) private readonly prisma: PrismaService, @Inject(SafetyLockService) private readonly safetyLocks: SafetyLockService, + @Inject(NotificationService) + private readonly notifications: NotificationService, ) {} async reply( @@ -154,6 +157,13 @@ export class ConversationService { }); const message = conversation.messages[0]!; await this.createMessageOutbox(tx, message.id, conversation.id); + await this.notifications.createInApp( + tx, + authorId, + `first-reply:${conversation.id}`, + "FIRST_REPLY", + { conversationId: conversation.id }, + ); const consumed = await tx.bottle.updateMany({ where: { id: bottleId, diff --git a/apps/api/src/health/health.e2e-spec.ts b/apps/api/src/health/health.e2e-spec.ts index 6350663..2841894 100644 --- a/apps/api/src/health/health.e2e-spec.ts +++ b/apps/api/src/health/health.e2e-spec.ts @@ -390,7 +390,9 @@ describe("API infrastructure", () => { const previousUrl = process.env.REDIS_URL; const previousTimeout = process.env.REDIS_PROBE_TIMEOUT_MS; process.env.REDIS_URL = "redis://127.0.0.1:1"; - process.env.REDIS_PROBE_TIMEOUT_MS = "100"; + // A real reconnect plus PING can exceed 100 ms under full-suite load. + // Keep the strict deadline in the dedicated never-answering-connection test. + process.env.REDIS_PROBE_TIMEOUT_MS = "1000"; const warning = vi .spyOn(Logger.prototype, "warn") .mockImplementation(() => undefined); diff --git a/apps/api/src/match/match-retry-policy.spec.ts b/apps/api/src/match/match-retry-policy.spec.ts index 7ebdb70..57a09b6 100644 --- a/apps/api/src/match/match-retry-policy.spec.ts +++ b/apps/api/src/match/match-retry-policy.spec.ts @@ -18,10 +18,10 @@ describe("match pick retry policy", () => { it("uses bounded defaults and caps each transaction by the overall deadline", () => { const config = readMatchRetryConfig({}); expect(config).toEqual({ - budgetMs: 5_000, + budgetMs: 30_000, maxAttempts: 64, - transactionMaxWaitMs: 1_000, - transactionTimeoutMs: 2_000, + transactionMaxWaitMs: 10_000, + transactionTimeoutMs: 10_000, }); expect(transactionLimits(config, 14_500, 14_750)).toEqual({ maxWait: 125, diff --git a/apps/api/src/match/match-retry-policy.ts b/apps/api/src/match/match-retry-policy.ts index e036a3f..8bcee92 100644 --- a/apps/api/src/match/match-retry-policy.ts +++ b/apps/api/src/match/match-retry-policy.ts @@ -36,20 +36,20 @@ export function readMatchRetryConfig( budgetMs: boundedInt( environment, "MATCH_PICK_RETRY_BUDGET_MS", - 5_000, + 30_000, 30_000, ), maxAttempts: boundedInt(environment, "MATCH_PICK_MAX_ATTEMPTS", 64, 100), transactionMaxWaitMs: boundedInt( environment, "MATCH_PICK_TRANSACTION_MAX_WAIT_MS", - 1_000, + 10_000, 10_000, ), transactionTimeoutMs: boundedInt( environment, "MATCH_PICK_TRANSACTION_TIMEOUT_MS", - 2_000, + 10_000, 10_000, ), }; diff --git a/apps/api/src/match/match.e2e-spec.ts b/apps/api/src/match/match.e2e-spec.ts index 2103381..5168747 100644 --- a/apps/api/src/match/match.e2e-spec.ts +++ b/apps/api/src/match/match.e2e-spec.ts @@ -325,6 +325,21 @@ describe("match API with real PostgreSQL", () => { expect((await prisma.dailyUsage.findFirstOrThrow()).bottlesPicked).toBe(1); }); + it("does not reveal an idempotent bottle or lease after either user blocks the other", async () => { + const author = await actor("author-device"); + const picker = await actor("picker-device"); + await bottle(author.id, "sensitive replay body"); + const key = randomUUID(); + await pick(picker.authorization, key).expect(201); + await prisma.block.create({ + data: { blockerId: author.id, blockedId: picker.id }, + }); + + const retry = await pick(picker.authorization, key).expect(403); + expect(retry.body.code).toBe("USER_BLOCKED"); + expect(JSON.stringify(retry.body)).not.toContain("sensitive replay body"); + }); + it.each(["RETURNED", "EXPIRED"] as const)( "returns the original pick response when the idempotent lease is %s", async (status) => { @@ -512,8 +527,8 @@ describe("match API with real PostgreSQL", () => { await prisma.sanction.create({ data: { accountId: picker.id, type, reason: "policy" }, }); - const response = await pick(picker.authorization).expect(403); - expect(response.body.code).toBe("ACCOUNT_SANCTIONED"); + const response = await pick(picker.authorization).expect(401); + expect(response.body.code).toBe("AUTH_UNAUTHORIZED"); expect(await prisma.bottlePickLease.count()).toBe(0); }, ); diff --git a/apps/api/src/match/match.module.ts b/apps/api/src/match/match.module.ts index a36d979..b6e9a63 100644 --- a/apps/api/src/match/match.module.ts +++ b/apps/api/src/match/match.module.ts @@ -4,9 +4,10 @@ import { DatabaseModule } from "../database/database.module.js"; import { CandidateRepository } from "./candidate.repository.js"; import { MatchController } from "./match.controller.js"; import { MatchService } from "./match.service.js"; +import { SafetyModule } from "../safety/safety.module.js"; @Module({ - imports: [DatabaseModule, AuthModule], + imports: [DatabaseModule, AuthModule, SafetyModule], controllers: [MatchController], providers: [CandidateRepository, MatchService], }) diff --git a/apps/api/src/match/match.service.spec.ts b/apps/api/src/match/match.service.spec.ts index c7caebb..e5c86f3 100644 --- a/apps/api/src/match/match.service.spec.ts +++ b/apps/api/src/match/match.service.spec.ts @@ -5,6 +5,8 @@ import { DomainException } from "../common/domain.exception.js"; import type { PrismaService } from "../database/prisma.service.js"; import type { CandidateRepository } from "./candidate.repository.js"; import { MatchService } from "./match.service.js"; +import { SafetyLockService } from "../safety/safety-lock.service.js"; +import type { Candidate } from "./candidate.repository.js"; const pickRequestRace = () => new Prisma.PrismaClientKnownRequestError("test", { @@ -46,9 +48,15 @@ describe("MatchService winner reads", () => { process.env.MATCH_PICK_TRANSACTION_MAX_WAIT_MS = "20"; process.env.MATCH_PICK_TRANSACTION_TIMEOUT_MS = "20"; + const clock = vi.spyOn(performance, "now").mockReturnValue(0); try { const transactionOptions: Array<{ maxWait: number; timeout: number }> = []; + let rejectWinnerRead!: (reason: unknown) => void; + let markWinnerReadStarted!: () => void; + const winnerReadStarted = new Promise((resolve) => { + markWinnerReadStarted = resolve; + }); const rootRead = vi.fn(() => { throw new Error("winner read escaped its bounded transaction"); }); @@ -56,18 +64,18 @@ describe("MatchService winner reads", () => { .fn() .mockRejectedValueOnce(initialError) .mockImplementationOnce( - async ( + ( callback: (tx: unknown) => Promise, options: { maxWait: number; timeout: number }, ) => { transactionOptions.push(options); + markWinnerReadStarted(); void callback({ bottlePickRequest: { findUnique: () => new Promise(() => {}) }, }); - await new Promise((resolve) => - setTimeout(resolve, options.timeout), - ); - throw transactionTimeout(); + return new Promise((_, reject) => { + rejectWinnerRead = reject; + }); }, ); const prisma = { @@ -77,25 +85,28 @@ describe("MatchService winner reads", () => { const service = new MatchService( prisma, {} as unknown as CandidateRepository, + {} as unknown as SafetyLockService, ); - const startedAt = performance.now(); - await expect(service.pick("picker", "request")).rejects.toMatchObject({ - code: expectedCode, - status: expectedStatus, - }); - const elapsed = performance.now() - startedAt; + const result = service.pick("picker", "request"); + await winnerReadStarted; - expect(rootRead).not.toHaveBeenCalled(); expect(transaction).toHaveBeenCalledTimes(2); + expect(rootRead).not.toHaveBeenCalled(); expect(transactionOptions).toHaveLength(1); expect(transactionOptions[0]!.maxWait).toBeGreaterThan(0); expect(transactionOptions[0]!.timeout).toBeGreaterThan(0); expect( transactionOptions[0]!.maxWait + transactionOptions[0]!.timeout, ).toBeLessThanOrEqual(40); - expect(elapsed).toBeLessThan(250); + + rejectWinnerRead(transactionTimeout()); + await expect(result).rejects.toMatchObject({ + code: expectedCode, + status: expectedStatus, + }); } finally { + clock.mockRestore(); const restore = (name: string, value: string | undefined) => { if (value === undefined) delete process.env[name]; else process.env[name] = value; @@ -114,3 +125,63 @@ describe("MatchService winner reads", () => { }, ); }); + +describe("MatchService candidate policy locking", () => { + it("locks the complete batch in canonical order before trying reverse-ordered candidates", async () => { + const picker = "00000000-0000-4000-8000-000000000002"; + const authorA = "00000000-0000-4000-8000-000000000001"; + const authorB = "00000000-0000-4000-8000-000000000003"; + const candidate = (authorId: string): Candidate => ({ + id: authorId, + authorId, + version: 1, + contentText: "test", + publicId: authorId, + nickname: "test", + avatarColor: "#ffffff", + bio: null, + }); + const locked: string[][] = []; + const tx = { + $executeRaw: vi.fn().mockResolvedValue(1), + bottlePickRequest: { findUnique: vi.fn().mockResolvedValue(null) }, + account: { + findUnique: vi.fn().mockResolvedValue({ status: "ACTIVE" }), + count: vi.fn().mockResolvedValue(2), + }, + sanction: { findFirst: vi.fn().mockResolvedValue(null) }, + block: { findFirst: vi.fn().mockResolvedValue(null) }, + bottle: { findFirst: vi.fn().mockResolvedValue(null) }, + }; + const transactionClient = tx; + const prisma = { + $transaction: async ( + fn: (client: typeof transactionClient) => Promise, + ) => fn(transactionClient), + } as unknown as PrismaService; + const candidates = { + findBatch: vi + .fn() + .mockResolvedValue([candidate(authorB), candidate(authorA)]), + } as unknown as CandidateRepository; + const locks = new SafetyLockService(); + const realLockAccounts = locks.lockAccounts.bind(locks); + const lockAccounts = vi + .spyOn(locks, "lockAccounts") + .mockImplementation(async (client, ids) => { + const before = tx.$executeRaw.mock.calls.length; + await realLockAccounts(client, ids); + const queries = tx.$executeRaw.mock.calls.slice(before) as Array< + [TemplateStringsArray, string] + >; + locked.push(queries.map(([, value]) => value.slice("safety:".length))); + }); + const service = new MatchService(prisma, candidates, locks); + await expect(service.pick(picker, "request")).rejects.toMatchObject({ + code: ErrorCode.BOTTLE_POOL_EMPTY, + }); + expect(tx.bottle.findFirst).toHaveBeenCalledTimes(2); + expect(lockAccounts).toHaveBeenCalledTimes(2); + expect(locked).toEqual([[authorA, picker, authorB], [picker]]); + }); +}); diff --git a/apps/api/src/match/match.service.ts b/apps/api/src/match/match.service.ts index 4329708..a0ecd2a 100644 --- a/apps/api/src/match/match.service.ts +++ b/apps/api/src/match/match.service.ts @@ -17,6 +17,7 @@ import { readMatchRetryConfig, transactionLimits, } from "./match-retry-policy.js"; +import { SafetyLockService } from "../safety/safety-lock.service.js"; type LeaseWithBottle = { id: string; @@ -24,6 +25,7 @@ type LeaseWithBottle = { leaseTokenCiphertext: Uint8Array; bottle: { id: string; + authorId: string; contentText: string; author: { anonymousProfile: { @@ -60,6 +62,7 @@ export class MatchService { @Inject(PrismaService) private readonly prisma: PrismaService, @Inject(CandidateRepository) private readonly candidates: CandidateRepository, + @Inject(SafetyLockService) private readonly locks: SafetyLockService, ) {} async pick(pickerId: string, requestId: string) { @@ -72,30 +75,41 @@ export class MatchService { async (tx) => { const previous = await this.findPrevious(tx, pickerId, requestId); if (previous?.lease) { - let token: string; - try { - token = decryptLeaseToken( - previous.lease.leaseTokenCiphertext, - previous.lease.id, - ); - } catch { - throw new DomainException( - ErrorCode.SERVICE_UNAVAILABLE, - "Bottle pick temporarily unavailable", - HttpStatus.SERVICE_UNAVAILABLE, - ); - } - return this.response(previous.lease, token); + await this.locks.lockAccounts(tx, [ + pickerId, + previous.lease.bottle.authorId, + ]); + await this.assertPairAllowed( + tx, + pickerId, + previous.lease.bottle.authorId, + ); + return this.replayResponse(previous.lease); } await this.assertEligible(tx, pickerId); const usageDate = utc8UsageDate(new Date()); const batch = await this.candidates.findBatch(tx, pickerId); - if (!batch.length) + if (!batch.length) { + // An in-flight same-key winner may have claimed the sole bottle + // without committing its request row yet. Let it commit before + // deciding this request actually saw an empty pool. + await this.locks.lockAccounts(tx, [pickerId]); + if (await this.findPrevious(tx, pickerId, requestId)) + throw new CandidateBatchContended(); throw new DomainException( ErrorCode.BOTTLE_POOL_EMPTY, "Bottle pool empty", HttpStatus.NOT_FOUND, ); + } + // Lock the entire subject set once, in global order. Locking pairs + // across candidates could otherwise form A→B / B→A cycles. This + // trades wider short-lived locks for correctness; the concurrent + // sole-bottle and same-key retry E2E cases exercise contention. + await this.locks.lockAccounts(tx, [ + pickerId, + ...batch.map((candidate) => candidate.authorId), + ]); while (batch.length) { const index = Math.floor(Math.random() * batch.length); const candidate = batch.splice(index, 1)[0]!; @@ -193,26 +207,24 @@ export class MatchService { ) { const limits = transactionLimits(this.retry, performance.now(), deadline); if (!limits) return null; - const previous = await this.prisma.$transaction( - (tx) => this.findPrevious(tx, pickerId, requestId), + const result = await this.prisma.$transaction( + async (tx) => { + const previous = await this.findPrevious(tx, pickerId, requestId); + if (!previous?.lease) return null; + await this.locks.lockAccounts(tx, [ + pickerId, + previous.lease.bottle.authorId, + ]); + await this.assertPairAllowed( + tx, + pickerId, + previous.lease.bottle.authorId, + ); + return this.replayResponse(previous.lease); + }, { isolationLevel: "ReadCommitted", ...limits }, ); - if (previous?.lease) { - try { - const token = decryptLeaseToken( - previous.lease.leaseTokenCiphertext, - previous.lease.id, - ); - return this.response(previous.lease, token); - } catch { - throw new DomainException( - ErrorCode.SERVICE_UNAVAILABLE, - "Bottle pick temporarily unavailable", - HttpStatus.SERVICE_UNAVAILABLE, - ); - } - } - return null; + return result; } private async readWinnerUntil( @@ -235,6 +247,21 @@ export class MatchService { usageDate: string, candidate: Candidate, ) { + await this.assertPairAllowed(tx, pickerId, candidate.authorId); + const eligible = await tx.bottle.findFirst({ + where: { + id: candidate.id, + authorId: candidate.authorId, + version: candidate.version, + reviewStatus: "APPROVED", + poolStatus: "IN_POOL", + activeLeaseId: null, + author: { anonymousProfile: { reviewStatus: "APPROVED" } }, + pickHistory: { none: { pickerId } }, + }, + select: { id: true }, + }); + if (!eligible) return null; const token = randomBytes(32).toString("base64url"); const expiresAt = new Date(Date.now() + this.ttlMs); const leaseId = randomUUID(); @@ -312,6 +339,66 @@ export class MatchService { ); } + private async assertPairAllowed( + tx: Prisma.TransactionClient, + pickerId: string, + authorId: string, + ) { + const now = new Date(); + const [accounts, block, sanction] = await Promise.all([ + tx.account.count({ + where: { id: { in: [pickerId, authorId] }, status: "ACTIVE" }, + }), + tx.block.findFirst({ + where: { + OR: [ + { blockerId: pickerId, blockedId: authorId }, + { blockerId: authorId, blockedId: pickerId }, + ], + }, + select: { id: true }, + }), + tx.sanction.findFirst({ + where: { + accountId: { in: [pickerId, authorId] }, + type: { in: ["SUSPENSION", "BAN"] }, + revokedAt: null, + startsAt: { lte: now }, + OR: [{ expiresAt: null }, { expiresAt: { gt: now } }], + }, + select: { id: true }, + }), + ]); + if (block) + throw new DomainException( + ErrorCode.USER_BLOCKED, + "User blocked", + HttpStatus.FORBIDDEN, + ); + if (accounts !== 2 || sanction) + throw new DomainException( + ErrorCode.ACCOUNT_SANCTIONED, + "Account sanctioned", + HttpStatus.FORBIDDEN, + ); + } + + private replayResponse(lease: LeaseWithBottle) { + try { + return this.response( + lease, + decryptLeaseToken(lease.leaseTokenCiphertext, lease.id), + ); + } catch (error) { + if (error instanceof DomainException) throw error; + throw new DomainException( + ErrorCode.SERVICE_UNAVAILABLE, + "Bottle pick temporarily unavailable", + HttpStatus.SERVICE_UNAVAILABLE, + ); + } + } + async returnBottle( pickerId: string, bottleId: string, diff --git a/apps/api/src/notification/notification.module.ts b/apps/api/src/notification/notification.module.ts index 1021839..ef204dc 100644 --- a/apps/api/src/notification/notification.module.ts +++ b/apps/api/src/notification/notification.module.ts @@ -8,5 +8,6 @@ import { NotificationService } from "./notification.service.js"; imports: [DatabaseModule, AuthModule], controllers: [NotificationController], providers: [NotificationService, StateChangingOriginGuard], + exports: [NotificationService], }) export class NotificationModule {} diff --git a/apps/api/src/notification/notification.service.ts b/apps/api/src/notification/notification.service.ts index 2029df0..78bc6b9 100644 --- a/apps/api/src/notification/notification.service.ts +++ b/apps/api/src/notification/notification.service.ts @@ -2,10 +2,66 @@ import { HttpStatus, Inject, Injectable } from "@nestjs/common"; import { ErrorCode } from "@drift/contracts"; import { PrismaService } from "../database/prisma.service.js"; import { DomainException } from "../common/domain.exception.js"; +import type { Prisma, PrismaClient } from "@prisma/client"; +type Db = Prisma.TransactionClient | PrismaClient; @Injectable() export class NotificationService { constructor(@Inject(PrismaService) private readonly prisma: PrismaService) {} + async createInApp( + tx: Db, + accountId: string, + dedupeKey: string, + type: string, + payload: Prisma.InputJsonObject, + ) { + const preference = await tx.pushPreference.findUnique({ + where: { accountId }, + select: { inAppEnabled: true }, + }); + if (preference?.inAppEnabled === false) return null; + return tx.notification.upsert({ + where: { accountId_dedupeKey: { accountId, dedupeKey } }, + create: { + accountId, + dedupeKey, + type, + payload, + status: "PENDING", + }, + update: {}, + }); + } + async notifyAdmins(tx: Db, reportId: string) { + const admins = await tx.account.findMany({ + where: { + role: "ADMIN", + status: "ACTIVE", + OR: [ + { pushPreference: null }, + { pushPreference: { is: { inAppEnabled: true } } }, + ], + }, + select: { id: true }, + }); + await Promise.all( + admins.map(({ id }) => + this.createInApp( + tx, + id, + `report:${reportId}:admin-pending`, + "ADMIN_REPORT_PENDING", + { reportId }, + ), + ), + ); + } async list(accountId: string, cursor: string | undefined, limit: number) { + const preference = await this.prisma.pushPreference.findUnique({ + where: { accountId }, + select: { inAppEnabled: true }, + }); + if (preference?.inAppEnabled === false) + return { items: [], nextCursor: null, unreadCount: 0 }; let decoded: { t: string; id: string } | undefined; try { if (cursor) { @@ -38,23 +94,30 @@ export class NotificationService { ); } const date = decoded ? new Date(decoded.t) : undefined; - const rows = await this.prisma.notification.findMany({ - where: { - accountId, - ...(decoded && date - ? { - OR: [ - { createdAt: { lt: date } }, - { createdAt: date, id: { lt: decoded.id } }, - ], - } - : {}), - }, - orderBy: [{ createdAt: "desc" }, { id: "desc" }], - take: limit + 1, - }); + const [rows, unreadCount] = await Promise.all([ + this.prisma.notification.findMany({ + where: { + accountId, + status: { in: ["SENT", "READ"] }, + ...(decoded && date + ? { + OR: [ + { createdAt: { lt: date } }, + { createdAt: date, id: { lt: decoded.id } }, + ], + } + : {}), + }, + orderBy: [{ createdAt: "desc" }, { id: "desc" }], + take: limit + 1, + }), + this.prisma.notification.count({ + where: { accountId, status: "SENT", readAt: null }, + }), + ]); const page = rows.slice(0, limit); return { + unreadCount, items: page.map((x) => ({ id: x.id, type: x.type, @@ -76,7 +139,7 @@ export class NotificationService { } async read(accountId: string, ids: string[]) { const owned = await this.prisma.notification.count({ - where: { accountId, id: { in: ids } }, + where: { accountId, id: { in: ids }, status: { in: ["SENT", "READ"] } }, }); if (owned !== new Set(ids).size) throw new DomainException( @@ -85,7 +148,7 @@ export class NotificationService { HttpStatus.NOT_FOUND, ); await this.prisma.notification.updateMany({ - where: { accountId, id: { in: ids } }, + where: { accountId, id: { in: ids }, status: { in: ["SENT", "READ"] } }, data: { status: "READ", readAt: new Date() }, }); return { read: owned }; @@ -97,11 +160,17 @@ export class NotificationService { return { inAppEnabled: row?.inAppEnabled ?? true }; } async updatePreference(accountId: string, inAppEnabled: boolean) { - return this.prisma.pushPreference.upsert({ - where: { accountId }, - create: { accountId, inAppEnabled }, - update: { inAppEnabled }, - select: { inAppEnabled: true }, + return this.prisma.$transaction(async (tx) => { + // The worker locks notification -> account -> preference. A settings + // change has no notification row; it locks account -> preference too. + // Account locking covers the initially absent preference row. + await tx.$queryRaw`SELECT "id" FROM "accounts" WHERE "id"=${accountId}::uuid FOR UPDATE`; + return tx.pushPreference.upsert({ + where: { accountId }, + create: { accountId, inAppEnabled }, + update: { inAppEnabled }, + select: { inAppEnabled: true }, + }); }); } } diff --git a/apps/api/src/safety/safety-admin.e2e-spec.ts b/apps/api/src/safety/safety-admin.e2e-spec.ts index fab592d..9f228e0 100644 --- a/apps/api/src/safety/safety-admin.e2e-spec.ts +++ b/apps/api/src/safety/safety-admin.e2e-spec.ts @@ -7,6 +7,7 @@ import { randomUUID } from "node:crypto"; import request from "supertest"; import { afterAll, beforeAll, beforeEach, describe, expect, it } from "vitest"; import { AppModule } from "../app.module.js"; +import { NotificationProcessor } from "../../../worker/src/notification.processor.js"; import { TokenService } from "../auth/token.service.js"; import { configureApp } from "../main.js"; import { assertSafeTestDatabaseUrl } from "../../../../prisma/database-safety.js"; @@ -241,6 +242,74 @@ describe("safety, notification and admin governance", () => { ).toBe(0); }); + it("audits authorized report and moderation queries without sensitive metadata", async () => { + const report = await request( + app.getHttpServer() as Parameters[0], + ) + .post("/api/v1/reports") + .set("Authorization", alice.authorization) + .set(origin) + .set("Idempotency-Key", randomUUID()) + .send({ targetType: "MESSAGE", targetId: messageId, reason: "SPAM" }) + .expect(201); + const reportId = report.body.data.id as string; + await request(app.getHttpServer() as Parameters[0]) + .get("/api/v1/admin/reports?status=PENDING&limit=1") + .set("Authorization", bob.authorization) + .expect(403); + const reports = await request( + app.getHttpServer() as Parameters[0], + ) + .get("/api/v1/admin/reports?status=PENDING&limit=1") + .set("Authorization", admin.authorization) + .expect(200); + expect(reports.body.data.items).toHaveLength(1); + await request(app.getHttpServer() as Parameters[0]) + .get("/api/v1/admin/moderation?limit=2") + .set("Authorization", bob.authorization) + .expect(403); + await request(app.getHttpServer() as Parameters[0]) + .get("/api/v1/admin/moderation?limit=2") + .set("Authorization", admin.authorization) + .expect(200); + const logs = await prisma.auditLog.findMany({ + where: { action: { in: ["REPORTS_QUERIED", "MODERATION_QUERIED"] } }, + orderBy: { createdAt: "asc" }, + }); + expect(logs).toHaveLength(2); + expect( + logs.map(({ actorId, action, entityType, metadata }) => ({ + actorId, + action, + entityType, + metadata, + })), + ).toEqual([ + { + actorId: admin.id, + action: "REPORTS_QUERIED", + entityType: "REPORT", + metadata: { status: "PENDING", limit: 1, resultCount: 1 }, + }, + { + actorId: admin.id, + action: "MODERATION_QUERIED", + entityType: "MODERATION_TASK", + metadata: { limit: 2, resultCount: 0 }, + }, + ]); + const serialized = JSON.stringify(logs); + for (const secret of [ + "reported secret", + "private bottle", + bob.id, + alice.id, + reportId, + messageId, + ]) + expect(serialized).not.toContain(secret); + }); + it("requires admin, resolves once with sanction, audit and reporter/target notifications", async () => { const report = await request( app.getHttpServer() as Parameters[0], @@ -302,6 +371,37 @@ describe("safety, notification and admin governance", () => { ).toBe(2); }); + it("upholds a report without sanction without notifying the target of account action", async () => { + const report = await request( + app.getHttpServer() as Parameters[0], + ) + .post("/api/v1/reports") + .set("Authorization", alice.authorization) + .set(origin) + .set("Idempotency-Key", randomUUID()) + .send({ targetType: "MESSAGE", targetId: messageId, reason: "SPAM" }) + .expect(201); + await request(app.getHttpServer() as Parameters[0]) + .post(`/api/v1/admin/reports/${report.body.data.id}/resolve`) + .set("Authorization", admin.authorization) + .set(origin) + .send({ decision: "UPHELD", resolution: "upheld without penalty" }) + .expect(201); + expect(await prisma.sanction.count({ where: { accountId: bob.id } })).toBe( + 0, + ); + expect( + await prisma.notification.count({ + where: { accountId: bob.id, type: "ACCOUNT_ACTION" }, + }), + ).toBe(0); + expect( + await prisma.notification.count({ + where: { accountId: alice.id, type: "REPORT_RESOLVED" }, + }), + ).toBe(1); + }); + it("rejects conflicting report replay and unauthorized targets without storing new reports", async () => { const key = randomUUID(); const create = (targetId: string) => @@ -323,6 +423,32 @@ describe("safety, notification and admin governance", () => { .expect(403); }); + it("queues a minimal account action for an independent sanction, respecting opt-out", async () => { + const sanction = () => + request(app.getHttpServer() as Parameters[0]) + .post(`/api/v1/admin/accounts/${bob.publicId}/sanctions`) + .set("Authorization", admin.authorization) + .set(origin) + .send({ type: "WARNING", reason: "private investigation" }); + await sanction().expect(201); + const first = await prisma.notification.findFirstOrThrow({ + where: { accountId: bob.id, type: "ACCOUNT_ACTION" }, + }); + expect(first.status).toBe("PENDING"); + expect(first.sentAt).toBeNull(); + expect(first.payload).toEqual({ type: "ACCOUNT_ACTION" }); + expect(JSON.stringify(first)).not.toContain("private investigation"); + await prisma.pushPreference.create({ + data: { accountId: bob.id, inAppEnabled: false }, + }); + await sanction().expect(201); + expect( + await prisma.notification.count({ + where: { accountId: bob.id, type: "ACCOUNT_ACTION" }, + }), + ).toBe(1); + }); + it("rejects expired sanctions, dismissed sanctions, and demoted admins", async () => { const sanction = (expiresAt: string) => request(app.getHttpServer() as Parameters[0]) @@ -443,6 +569,193 @@ describe("safety, notification and admin governance", () => { ).toBe(1); }); + it("serializes settings changes behind worker delivery without leaking after opt-out", async () => { + const pending = await prisma.notification.create({ + data: { + accountId: alice.id, + dedupeKey: randomUUID(), + type: "TEST", + payload: {}, + }, + }); + let release!: () => void; + let locked!: () => void; + const held = new Promise((resolve) => { + locked = resolve; + }); + const gate = new Promise((resolve) => { + release = resolve; + }); + const blocker = prisma.$transaction(async (tx) => { + await tx.$queryRaw`SELECT "id" FROM "accounts" WHERE "id"=${alice.id}::uuid FOR UPDATE`; + locked(); + await gate; + }); + try { + await held; + // Hold the shared account lock while both API and worker try to acquire + // it. PostgreSQL's lock queue gives the earlier API request priority. + const preference = request( + app.getHttpServer() as Parameters[0], + ) + .put("/api/v1/settings/push") + .set("Authorization", alice.authorization) + .set(origin) + .send({ inAppEnabled: false }); + const updating = preference.then((response) => response); + // Poll database lock state, not wall-clock time, so the API is known to + // be waiting for the account lock before the worker starts. + let apiWaiting = false; + for (let attempt = 0; attempt < 100 && !apiWaiting; attempt += 1) { + const waiting = await prisma.$queryRaw>` + SELECT EXISTS ( + SELECT 1 FROM pg_stat_activity + WHERE datname = current_database() + AND pid <> pg_backend_pid() + AND wait_event_type = 'Lock' + AND query LIKE '%FROM "accounts"%FOR UPDATE%' + ) AS ready`; + apiWaiting = waiting[0]?.ready ?? false; + if (!apiWaiting) await new Promise((resolve) => setImmediate(resolve)); + } + expect(apiWaiting).toBe(true); + // The API is queued on the shared account lock before worker delivery + // begins; neither operation has completed when the gate opens. + const worker = new NotificationProcessor(prisma).runOnce(); + let workerClaimed = false; + for (let attempt = 0; attempt < 100 && !workerClaimed; attempt += 1) { + try { + await prisma.$transaction(async (probe) => { + await probe.$queryRaw`SELECT "id" FROM "notifications" WHERE "id"=${pending.id}::uuid FOR UPDATE NOWAIT`; + }); + } catch { + workerClaimed = true; + } + if (!workerClaimed) + await new Promise((resolve) => setImmediate(resolve)); + } + expect(workerClaimed).toBe(true); + release(); + await blocker; + const concurrent = await Promise.all([updating, worker]); + expect(concurrent[0].status).toBe(200); + expect(concurrent[1]).toBe(true); + expect( + ( + await prisma.notification.findUniqueOrThrow({ + where: { id: pending.id }, + }) + ).status, + ).toBe("FAILED"); + const visible = await request( + app.getHttpServer() as Parameters[0], + ) + .get("/api/v1/notifications") + .set("Authorization", alice.authorization) + .expect(200); + expect(visible.body.data).toMatchObject({ items: [], unreadCount: 0 }); + } finally { + release(); + await blocker; + } + }); + + it("delivers new API notifications only in the worker and hides pending rows", async () => { + const report = await request( + app.getHttpServer() as Parameters[0], + ) + .post("/api/v1/reports") + .set("Authorization", alice.authorization) + .set(origin) + .set("Idempotency-Key", randomUUID()) + .send({ targetType: "MESSAGE", targetId: messageId, reason: "SPAM" }) + .expect(201); + const before = await request( + app.getHttpServer() as Parameters[0], + ) + .get("/api/v1/notifications") + .set("Authorization", admin.authorization) + .expect(200); + expect(before.body.data).toMatchObject({ items: [], unreadCount: 0 }); + const pending = await prisma.notification.findFirstOrThrow({ + where: { accountId: admin.id, type: "ADMIN_REPORT_PENDING" }, + }); + expect(pending).toMatchObject({ status: "PENDING", sentAt: null }); + expect(pending.payload).toEqual({ + reportId: report.body.data.id as string, + }); + expect(await new NotificationProcessor(prisma).runOnce()).toBe(true); + const after = await request( + app.getHttpServer() as Parameters[0], + ) + .get("/api/v1/notifications") + .set("Authorization", admin.authorization) + .expect(200); + expect(after.body.data).toMatchObject({ + unreadCount: 1, + items: [expect.objectContaining({ id: pending.id, status: "SENT" })], + }); + }); + + it("counts only visible delivered unread notifications across pagination and read boundaries", async () => { + const make = ( + status: "PENDING" | "SENT" | "FAILED" | "READ", + readAt?: Date, + ) => + prisma.notification.create({ + data: { + accountId: alice.id, + dedupeKey: randomUUID(), + type: "TEST", + payload: {}, + status, + ...(readAt ? { readAt } : {}), + }, + }); + const pending = await make("PENDING"); + const failed = await make("FAILED"); + const unread = await make("SENT"); + const alreadyRead = await make("READ", new Date()); + const sentButRead = await make("SENT", new Date()); + const list = () => + request(app.getHttpServer() as Parameters[0]) + .get("/api/v1/notifications?limit=1") + .set("Authorization", alice.authorization) + .expect(200); + const initial = await list(); + expect(initial.body.data.unreadCount).toBe(1); + expect(initial.body.data.items[0].id).toBe(sentButRead.id); + expect(JSON.stringify(initial.body.data)).not.toContain(pending.id); + expect(JSON.stringify(initial.body.data)).not.toContain(failed.id); + const ids: string[] = [initial.body.data.items[0].id as string]; + let cursor = initial.body.data.nextCursor as string | null; + while (cursor) { + const page = await request( + app.getHttpServer() as Parameters[0], + ) + .get(`/api/v1/notifications?limit=1&cursor=${cursor}`) + .set("Authorization", alice.authorization) + .expect(200); + expect(page.body.data.unreadCount).toBe(1); + ids.push(page.body.data.items[0].id as string); + cursor = page.body.data.nextCursor as string | null; + } + expect(ids).toEqual([sentButRead.id, alreadyRead.id, unread.id]); + await request(app.getHttpServer() as Parameters[0]) + .post("/api/v1/notifications/read") + .set("Authorization", alice.authorization) + .set(origin) + .send({ ids: [pending.id] }) + .expect(404); + await request(app.getHttpServer() as Parameters[0]) + .post("/api/v1/notifications/read") + .set("Authorization", alice.authorization) + .set(origin) + .send({ ids: [unread.id] }) + .expect(201); + expect((await list()).body.data.unreadCount).toBe(0); + }); + it("keeps notifications private, cursor-paged/readable, and stores in-app preferences", async () => { const own = await prisma.notification.create({ data: { @@ -468,6 +781,7 @@ describe("safety, notification and admin governance", () => { .get("/api/v1/notifications?limit=1") .set("Authorization", alice.authorization) .expect(200); + expect(list.body.data.unreadCount).toBe(1); expect(list.body.data.items).toHaveLength(1); expect(list.body.data.items[0].id).toBe(own.id); await request(app.getHttpServer() as Parameters[0]) @@ -476,6 +790,13 @@ describe("safety, notification and admin governance", () => { .set(origin) .send({ ids: [own.id] }) .expect(201); + const afterRead = await request( + app.getHttpServer() as Parameters[0], + ) + .get("/api/v1/notifications?limit=1") + .set("Authorization", alice.authorization) + .expect(200); + expect(afterRead.body.data.unreadCount).toBe(0); await request(app.getHttpServer() as Parameters[0]) .post("/api/v1/notifications/read") .set("Authorization", alice.authorization) @@ -503,5 +824,167 @@ describe("safety, notification and admin governance", () => { .set("Authorization", alice.authorization) .expect(200); expect(pref.body.data).toEqual({ inAppEnabled: false }); + const hidden = await request( + app.getHttpServer() as Parameters[0], + ) + .get("/api/v1/notifications") + .set("Authorization", alice.authorization) + .expect(200); + expect(hidden.body.data).toEqual({ + items: [], + nextCursor: null, + unreadCount: 0, + }); + }); + + it("honors notification preferences and never links a dismissed report to its target", async () => { + await prisma.pushPreference.create({ + data: { accountId: admin.id, inAppEnabled: false }, + }); + const report = await request( + app.getHttpServer() as Parameters[0], + ) + .post("/api/v1/reports") + .set("Authorization", alice.authorization) + .set(origin) + .set("Idempotency-Key", randomUUID()) + .send({ targetType: "MESSAGE", targetId: messageId, reason: "SPAM" }) + .expect(201); + expect( + await prisma.notification.count({ + where: { accountId: admin.id, type: "ADMIN_REPORT_PENDING" }, + }), + ).toBe(0); + await request(app.getHttpServer() as Parameters[0]) + .post(`/api/v1/admin/reports/${report.body.data.id}/resolve`) + .set("Authorization", admin.authorization) + .set(origin) + .send({ decision: "DISMISSED", resolution: "not upheld" }) + .expect(201); + expect( + await prisma.notification.count({ where: { accountId: bob.id } }), + ).toBe(0); + }); + + it("enforces visibility and idempotency independently for each report target", async () => { + const bottle = await prisma.bottle.findFirstOrThrow({ + where: { authorId: alice.id }, + }); + const targets = [ + ["ACCOUNT", bob.publicId], + ["BOTTLE", bottle.id], + ["CONVERSATION", conversationId], + ["MESSAGE", messageId], + ] as const; + const submit = ( + authorization: string, + targetType: (typeof targets)[number][0], + targetId: string, + key: string, + details = "original detail", + ) => + request(app.getHttpServer() as Parameters[0]) + .post("/api/v1/reports") + .set("Authorization", authorization) + .set(origin) + .set("Idempotency-Key", key) + .send({ targetType, targetId, reason: "SPAM", details }); + for (const [targetType, targetId] of targets) { + const key = randomUUID(); + const first = await submit( + alice.authorization, + targetType, + targetId, + key, + ).expect(201); + await submit(alice.authorization, targetType, targetId, key).expect(201); + await submit( + alice.authorization, + targetType, + targetId, + key, + "changed", + ).expect(409); + await submit( + admin.authorization, + targetType, + targetId, + randomUUID(), + ).expect(403); + const snapshot = ( + await prisma.report.findUniqueOrThrow({ + where: { id: first.body.data.id as string }, + }) + ).targetSnapshot; + expect(snapshot).toMatchObject({ targetType }); + expect( + await prisma.report.count({ where: { reporterId: alice.id } }), + ).toBe(targets.findIndex(([kind]) => kind === targetType) + 1); + } + const snapshotBefore = ( + await prisma.report.findFirstOrThrow({ + where: { messageId }, + }) + ).targetSnapshot; + await prisma.message.update({ + where: { id: messageId }, + data: { contentText: "edited after report" }, + }); + expect( + (await prisma.report.findFirstOrThrow({ where: { messageId } })) + .targetSnapshot, + ).toEqual(snapshotBefore); + await prisma.conversationMember.updateMany({ + where: { conversationId, accountId: alice.id }, + data: { leftAt: new Date() }, + }); + for (const targetType of ["CONVERSATION", "MESSAGE"] as const) + await submit( + alice.authorization, + targetType, + targetType === "MESSAGE" ? messageId : conversationId, + randomUUID(), + ).expect(403); + }); + + it("stores deidentified immutable snapshots for all four report target kinds", async () => { + const bottle = await prisma.bottle.findFirstOrThrow({ + where: { authorId: alice.id }, + }); + const targets = [ + ["MESSAGE", messageId], + ["CONVERSATION", conversationId], + ["BOTTLE", bottle.id], + ["ACCOUNT", bob.publicId], + ] as const; + for (const [targetType, targetId] of targets) { + await request(app.getHttpServer() as Parameters[0]) + .post("/api/v1/reports") + .set("Authorization", alice.authorization) + .set(origin) + .set("Idempotency-Key", randomUUID()) + .send({ targetType, targetId, reason: "SPAM" }) + .expect(201); + } + const reports = await prisma.report.findMany({ + orderBy: { createdAt: "asc" }, + }); + expect(reports).toHaveLength(4); + const serialized = JSON.stringify(reports.map((row) => row.targetSnapshot)); + expect(serialized).not.toContain(alice.id); + expect(serialized).not.toContain(bob.id); + expect(serialized).not.toContain("accountId"); + expect(serialized).not.toContain("reportedAccountId"); + expect(serialized).toContain("senderPublicId"); + expect(serialized).toContain("ownerPublicId"); + expect(serialized).toContain("members"); + expect(serialized).toContain(bob.publicId); + const adminList = await request( + app.getHttpServer() as Parameters[0], + ) + .get("/api/v1/admin/reports") + .set("Authorization", admin.authorization) + .expect(200); + expect(JSON.stringify(adminList.body.data)).not.toContain(bob.id); }); }); diff --git a/apps/api/src/safety/safety.module.ts b/apps/api/src/safety/safety.module.ts index 51f1aeb..23843f4 100644 --- a/apps/api/src/safety/safety.module.ts +++ b/apps/api/src/safety/safety.module.ts @@ -5,8 +5,9 @@ import { StateChangingOriginGuard } from "../conversation/state-changing-origin. import { SafetyController } from "./safety.controller.js"; import { SafetyLockService } from "./safety-lock.service.js"; import { SafetyService } from "./safety.service.js"; +import { NotificationModule } from "../notification/notification.module.js"; @Module({ - imports: [DatabaseModule, AuthModule], + imports: [DatabaseModule, AuthModule, NotificationModule], controllers: [SafetyController], providers: [SafetyLockService, SafetyService, StateChangingOriginGuard], exports: [SafetyLockService], diff --git a/apps/api/src/safety/safety.service.ts b/apps/api/src/safety/safety.service.ts index 54fb62f..b388554 100644 --- a/apps/api/src/safety/safety.service.ts +++ b/apps/api/src/safety/safety.service.ts @@ -6,6 +6,7 @@ import { DomainException } from "../common/domain.exception.js"; import { PrismaService } from "../database/prisma.service.js"; import { SafetyLockService } from "./safety-lock.service.js"; import type { CreateReportDto, ReportTargetType } from "./dto.js"; +import { NotificationService } from "../notification/notification.service.js"; type Db = Prisma.TransactionClient | PrismaClient; const targetType = (dto: CreateReportDto): string => dto.targetType; @@ -14,6 +15,8 @@ export class SafetyService { constructor( @Inject(PrismaService) private readonly prisma: PrismaService, @Inject(SafetyLockService) private readonly locks: SafetyLockService, + @Inject(NotificationService) + private readonly notifications: NotificationService, ) {} async report(reporterId: string, key: string, dto: CreateReportDto) { const prior = await this.prisma.report.findUnique({ @@ -65,6 +68,7 @@ export class SafetyService { metadata: { targetType: dto.targetType, reason: dto.reason }, }, }); + await this.notifications.notifyAdmins(tx, report.id); return this.publicReport(report); }, { isolationLevel: Prisma.TransactionIsolationLevel.Serializable }, @@ -187,7 +191,11 @@ export class SafetyService { id, members: { some: { accountId: reporterId, leftAt: null } }, }, - include: { members: true }, + include: { + members: { + include: { account: { select: { anonymousProfile: true } } }, + }, + }, }); if (!row) throw this.forbidden(); const accountId = row.members.find( @@ -198,7 +206,13 @@ export class SafetyService { snapshot: { targetType: type, conversationId: id, - reportedAccountId: accountId, + members: row.members.map((member) => ({ + publicId: member.account.anonymousProfile?.publicId, + peerAlias: member.peerAliasSnapshot, + })), + subjectPublicId: row.members.find( + (member) => member.accountId !== reporterId, + )?.account.anonymousProfile?.publicId, }, }; } @@ -219,7 +233,7 @@ export class SafetyService { snapshot: { targetType: type, bottleId: id, - authorPublicId: row.author.anonymousProfile?.publicId, + ownerPublicId: row.author.anonymousProfile?.publicId, contentText: row.contentText, }, }; diff --git a/apps/worker/src/main.spec.ts b/apps/worker/src/main.spec.ts index be4bc94..5ae4ff5 100644 --- a/apps/worker/src/main.spec.ts +++ b/apps/worker/src/main.spec.ts @@ -17,6 +17,14 @@ describe("worker lifecycle", () => { expect(reaper.runOnce).toHaveBeenCalledTimes(1); expect(moderation.runOnce).toHaveBeenCalledTimes(1); }); + it("invokes reaping, moderation and notification each iteration", async () => { + const reaper = { runOnce: vi.fn().mockResolvedValue(true) }; + const moderation = { runOnce: vi.fn().mockResolvedValue(false) }; + const notification = { runOnce: vi.fn().mockResolvedValue(true) }; + const worker = createCombinedWorker(reaper, moderation, notification); + await expect(worker.runOnce()).resolves.toBe(true); + expect(notification.runOnce).toHaveBeenCalledTimes(1); + }); it("waits for the current run before disconnecting on SIGTERM", async () => { let finish!: () => void; const runOnce = vi.fn( diff --git a/apps/worker/src/main.ts b/apps/worker/src/main.ts index df895b1..475e664 100644 --- a/apps/worker/src/main.ts +++ b/apps/worker/src/main.ts @@ -2,6 +2,7 @@ import { pathToFileURL } from "node:url"; import { PrismaClient } from "@prisma/client"; import { ModerationWorker } from "./moderation-worker.js"; import { LeaseReaper } from "./lease-reaper.processor.js"; +import { NotificationProcessor } from "./notification.processor.js"; type SignalSource = { once(signal: "SIGTERM" | "SIGINT", listener: () => void): unknown; @@ -22,12 +23,14 @@ type Worker = { runOnce(): Promise }; export function createCombinedWorker( reaper: Worker, moderation: Worker, + notification?: Worker, ): Worker { return { async runOnce() { const reaped = await reaper.runOnce(); const moderated = await moderation.runOnce(); - return reaped || moderated; + const notified = notification ? await notification.runOnce() : false; + return reaped || moderated || notified; }, }; } @@ -65,7 +68,8 @@ export async function main() { const prisma = new PrismaClient(); const moderation = new ModerationWorker(prisma); const reaper = new LeaseReaper(prisma); - const worker = createCombinedWorker(reaper, moderation); + const notification = new NotificationProcessor(prisma); + const worker = createCombinedWorker(reaper, moderation, notification); await runWorker({ worker, connect: () => prisma.$connect(), diff --git a/apps/worker/src/notification.processor.spec.ts b/apps/worker/src/notification.processor.spec.ts new file mode 100644 index 0000000..6cc06cd --- /dev/null +++ b/apps/worker/src/notification.processor.spec.ts @@ -0,0 +1,132 @@ +import { randomUUID } from "node:crypto"; +import { PrismaClient } from "@prisma/client"; +import { afterAll, beforeAll, describe, expect, it } from "vitest"; +import { assertSafeTestDatabaseUrl } from "../../../prisma/database-safety.js"; +import { NotificationProcessor } from "./notification.processor.js"; + +const prisma = new PrismaClient(); + +describe("NotificationProcessor", () => { + beforeAll(async () => { + assertSafeTestDatabaseUrl(process.env.DATABASE_URL ?? ""); + await prisma.$connect(); + }); + afterAll(async () => prisma.$disconnect()); + + it("lets a preference disable committed before worker delivery win the race", async () => { + const account = await prisma.account.create({ + data: { phoneCiphertext: Buffer.from("race"), phoneHmac: randomUUID() }, + }); + const preference = await prisma.pushPreference.create({ + data: { accountId: account.id, inAppEnabled: true }, + }); + const notification = await prisma.notification.create({ + data: { + accountId: account.id, + dedupeKey: randomUUID(), + type: "TEST", + payload: {}, + }, + }); + await prisma.notification.updateMany({ + where: { status: "PENDING", id: { not: notification.id } }, + data: { status: "FAILED" }, + }); + let release!: () => void; + let locked!: () => void; + const lockHeld = new Promise((resolve) => { + locked = resolve; + }); + const gate = new Promise((resolve) => { + release = resolve; + }); + const disabling = prisma.$transaction(async (tx) => { + await tx.$queryRaw`SELECT "account_id" FROM "push_preferences" WHERE "account_id"=${preference.accountId}::uuid FOR UPDATE`; + locked(); + await gate; + await tx.pushPreference.update({ + where: { accountId: account.id }, + data: { inAppEnabled: false }, + }); + }); + try { + await lockHeld; + const delivery = new NotificationProcessor(prisma).runOnce(); + // Wait until the worker holds the notification lock and has reached the + // preference boundary; an unlocked preference would expose stale true. + let claimed = false; + for (let attempt = 0; attempt < 100 && !claimed; attempt += 1) { + try { + await prisma.$transaction(async (probe) => { + await probe.$queryRaw`SELECT "id" FROM "notifications" WHERE "id"=${notification.id}::uuid FOR UPDATE NOWAIT`; + }); + } catch { + claimed = true; + } + if (!claimed) await new Promise((resolve) => setTimeout(resolve, 20)); + } + expect(claimed).toBe(true); + release(); + await disabling; + expect(await delivery).toBe(true); + expect( + ( + await prisma.notification.findUniqueOrThrow({ + where: { id: notification.id }, + }) + ).status, + ).toBe("FAILED"); + } finally { + release(); + await disabling; + await prisma.account.delete({ where: { id: account.id } }); + } + }); + + it("delivers a pending in-app notification once and skips disabled preferences", async () => { + const account = await prisma.account.create({ + data: { phoneCiphertext: Buffer.from("test"), phoneHmac: randomUUID() }, + }); + try { + const enabled = await prisma.notification.create({ + data: { + accountId: account.id, + type: "TEST", + dedupeKey: randomUUID(), + payload: { safe: true }, + }, + }); + await prisma.notification.updateMany({ + where: { status: "PENDING", id: { not: enabled.id } }, + data: { status: "FAILED" }, + }); + const worker = new NotificationProcessor(prisma); + expect(await worker.runOnce()).toBe(true); + const delivered = await prisma.notification.findUniqueOrThrow({ + where: { id: enabled.id }, + }); + expect(delivered).toMatchObject({ status: "SENT", readAt: null }); + expect(delivered.sentAt).toBeInstanceOf(Date); + expect(await worker.runOnce()).toBe(false); + await prisma.pushPreference.create({ + data: { accountId: account.id, inAppEnabled: false }, + }); + const disabled = await prisma.notification.create({ + data: { + accountId: account.id, + type: "TEST", + dedupeKey: randomUUID(), + payload: { safe: true }, + }, + }); + expect(await worker.runOnce()).toBe(true); + expect( + await prisma.notification.findUniqueOrThrow({ + where: { id: disabled.id }, + }), + ).toMatchObject({ status: "FAILED", sentAt: null }); + } finally { + await prisma.account.delete({ where: { id: account.id } }); + } + }); +}); diff --git a/apps/worker/src/notification.processor.ts b/apps/worker/src/notification.processor.ts new file mode 100644 index 0000000..f75b400 --- /dev/null +++ b/apps/worker/src/notification.processor.ts @@ -0,0 +1,38 @@ +import { PrismaClient } from "@prisma/client"; + +/** Deliver committed PENDING in-app rows after serializing against opt-out. */ +export class NotificationProcessor { + constructor(private readonly prisma: PrismaClient) {} + + async runOnce(): Promise { + // Claim and update share a transaction; the row lock lives until commit. + return this.prisma.$transaction(async (tx) => { + const claimed = await tx.$queryRaw< + Array<{ id: string; accountId: string }> + >` + SELECT n."id", n."account_id" AS "accountId" + FROM "notifications" n + WHERE n."status" = 'PENDING' + ORDER BY n."created_at", n."id" + FOR UPDATE OF n SKIP LOCKED LIMIT 1`; + const notification = claimed[0]; + if (!notification) return false; + // n -> account -> preference is the fixed lock order. Locking the + // account also serializes preference-row insertion when none exists. + await tx.$queryRaw` + SELECT "id" FROM "accounts" + WHERE "id"=${notification.accountId}::uuid FOR UPDATE`; + const preference = await tx.$queryRaw>` + SELECT "in_app_enabled" AS "enabled" FROM "push_preferences" + WHERE "account_id"=${notification.accountId}::uuid FOR UPDATE`; + await tx.notification.update({ + where: { id: notification.id }, + data: + preference[0]?.enabled !== false + ? { status: "SENT", sentAt: new Date() } + : { status: "FAILED" }, + }); + return true; + }); + } +} diff --git a/package.json b/package.json index 46e07fc..1142a68 100644 --- a/package.json +++ b/package.json @@ -6,7 +6,8 @@ "scripts": { "test": "vitest run --no-file-parallelism", "test:database": "vitest run tests/integration/database.spec.ts", - "test:integration": "vitest run tests/integration --no-file-parallelism", + "test:integration": "vitest run tests/integration --no-file-parallelism && corepack pnpm --filter @drift/api exec vitest run --config vitest.config.ts src/safety/safety-admin.e2e-spec.ts --no-file-parallelism", + "test:security": "corepack pnpm --filter @drift/api exec vitest run --config vitest.config.ts src/safety/safety-admin.e2e-spec.ts src/auth/auth.e2e-spec.ts src/conversation/conversation.e2e-spec.ts --no-file-parallelism", "typecheck": "tsc --noEmit -p tsconfig.base.json", "lint": "eslint . --ext .ts --max-warnings 0 && prettier --check .", "build": "corepack pnpm prisma generate && corepack pnpm --recursive run build",