jkjoy/plp
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(治理): 完成任务 8 安全与通知闭环修复

Browse Source 
- 串行化拉黑、处罚、投瓶和匹配策略检查\n- 完成异步站内通知、未读统计和偏好并发语义\n- 补齐后台查询审计、处罚恢复和隐私测试\n- 稳定 Redis 恢复、匹配锁序及超时测试
This commit is contained in:
root
2026-09-17 13:08:55 +08:00
parent ce8e1db279
commit fa0fa78312
29 changed files with 1231 additions and 153 deletions
Download Patch File Download Diff File Expand all files Collapse all files
apps/api/src/admin/admin.controller.ts
+12 -6
View File
@@ -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",
);
apps/api/src/admin/admin.module.ts
+2 -1
View File
@@ -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],
})
apps/api/src/admin/admin.service.ts
+81 -48
View File
@@ -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;
}
}
apps/api/src/auth/auth.e2e-spec.ts
+30
View File
@@ -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<Socket>();
const server = createServer((socket) => {
apps/api/src/auth/auth.guard.ts
+12
View File
@@ -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;
}
apps/api/src/auth/auth.service.ts
+12
View File
@@ -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<Account> {
apps/api/src/bottle/bottle.e2e-spec.ts
+2 -2
View File
@@ -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);
},
);
apps/api/src/bottle/bottle.module.ts
+2 -1
View File
@@ -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],
})
apps/api/src/bottle/bottle.service.ts
+11 -8
View File
@@ -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",
apps/api/src/conversation/conversation.e2e-spec.ts
+37 -1
View File
@@ -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 } }))
apps/api/src/conversation/conversation.module.ts
+8 -1
View File
@@ -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,
apps/api/src/conversation/conversation.service.ts
+10
View File
@@ -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,
apps/api/src/health/health.e2e-spec.ts
+3 -1
View File
@@ -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);
apps/api/src/match/match-retry-policy.spec.ts
+3 -3
View File
@@ -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,
apps/api/src/match/match-retry-policy.ts
+3 -3
View File
@@ -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,
),
};
apps/api/src/match/match.e2e-spec.ts
+17 -2
View File
@@ -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);
},
);
apps/api/src/match/match.module.ts
+2 -1
View File
@@ -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],
})
apps/api/src/match/match.service.spec.ts
+84 -13
View File
@@ -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<void>((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<unknown>,
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<unknown>,
) => 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]]);
});
});
apps/api/src/match/match.service.ts
+120 -33
View File
@@ -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,
apps/api/src/notification/notification.module.ts
+1
View File
@@ -8,5 +8,6 @@ import { NotificationService } from "./notification.service.js";
imports: [DatabaseModule, AuthModule],
controllers: [NotificationController],
providers: [NotificationService, StateChangingOriginGuard],
exports: [NotificationService],
})
export class NotificationModule {}
apps/api/src/notification/notification.service.ts
+91 -22
View File
@@ -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 },
});
});
}
}
apps/api/src/safety/safety-admin.e2e-spec.ts
+483
View File
@@ -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<typeof request>[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<typeof request>[0])
.get("/api/v1/admin/reports?status=PENDING&limit=1")
.set("Authorization", bob.authorization)
.expect(403);
const reports = await request(
app.getHttpServer() as Parameters<typeof request>[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<typeof request>[0])
.get("/api/v1/admin/moderation?limit=2")
.set("Authorization", bob.authorization)
.expect(403);
await request(app.getHttpServer() as Parameters<typeof request>[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<typeof request>[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<typeof request>[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<typeof request>[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<typeof request>[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<typeof request>[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<void>((resolve) => {
locked = resolve;
});
const gate = new Promise<void>((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<typeof request>[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<Array<{ ready: boolean }>>`
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<typeof request>[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<typeof request>[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<typeof request>[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<typeof request>[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<typeof request>[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<typeof request>[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<typeof request>[0])
.post("/api/v1/notifications/read")
.set("Authorization", alice.authorization)
.set(origin)
.send({ ids: [pending.id] })
.expect(404);
await request(app.getHttpServer() as Parameters<typeof request>[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<typeof request>[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<typeof request>[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<typeof request>[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<typeof request>[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<typeof request>[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<typeof request>[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<typeof request>[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<typeof request>[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<typeof request>[0],
)
.get("/api/v1/admin/reports")
.set("Authorization", admin.authorization)
.expect(200);
expect(JSON.stringify(adminList.body.data)).not.toContain(bob.id);
});
});
apps/api/src/safety/safety.module.ts
+2 -1
View File
@@ -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],
apps/api/src/safety/safety.service.ts
+17 -3
View File
@@ -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,
},
};
apps/worker/src/main.spec.ts
+8
View File
@@ -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(
apps/worker/src/main.ts
+6 -2
View File
@@ -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<boolean> };
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(),
apps/worker/src/notification.processor.spec.ts
+132
View File
@@ -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<void>((resolve) => {
locked = resolve;
});
const gate = new Promise<void>((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 } });
}
});
});
apps/worker/src/notification.processor.ts
+38
View File
@@ -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<boolean> {
// 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<Array<{ enabled: boolean }>>`
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;
});
}
}
package.json
+2 -1
View File
@@ -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",