feat: 完成举报拉黑和审核处置闭环

apps/api/package.json

@@ -7,7 +7,7 @@
     "build": "tsc -p tsconfig.build.json",
     "start": "node dist/main.js",
     "test": "vitest run --config vitest.config.ts --no-file-parallelism",
-    "test:e2e": "vitest run --config vitest.config.ts src/health/health.e2e-spec.ts src/auth/auth.e2e-spec.ts src/profile/profile.e2e-spec.ts src/bottle/bottle.e2e-spec.ts src/match/match.e2e-spec.ts src/conversation/conversation.e2e-spec.ts --no-file-parallelism",
+    "test:e2e": "vitest run --config vitest.config.ts src/health/health.e2e-spec.ts src/auth/auth.e2e-spec.ts src/profile/profile.e2e-spec.ts src/bottle/bottle.e2e-spec.ts src/match/match.e2e-spec.ts src/conversation/conversation.e2e-spec.ts src/safety/safety-admin.e2e-spec.ts --no-file-parallelism",
     "typecheck": "tsc --noEmit -p tsconfig.json"
   },
   "dependencies": {

apps/api/src/admin/admin.controller.ts

@@ -0,0 +1,88 @@
+import {
+  Body,
+  Controller,
+  Get,
+  Inject,
+  Param,
+  ParseUUIDPipe,
+  Post,
+  Query,
+  UseGuards,
+} from "@nestjs/common";
+import { AuthGuard } from "../auth/auth.guard.js";
+import { CurrentUser } from "../auth/current-user.decorator.js";
+import type { AccessClaims } from "../auth/token.service.js";
+import { StateChangingOriginGuard } from "../conversation/state-changing-origin.guard.js";
+import { AdminGuard } from "./admin.guard.js";
+import { AdminService } from "./admin.service.js";
+import {
+  AdminQueryDto,
+  ModerateDto,
+  ResolveReportDto,
+  SanctionDto,
+} from "./dto.js";
+@Controller("admin")
+@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);
+  }
+  @UseGuards(StateChangingOriginGuard) @Post("reports/:id/resolve") resolve(
+    @CurrentUser() u: AccessClaims,
+    @Param("id", new ParseUUIDPipe()) id: string,
+    @Body() d: ResolveReportDto,
+  ) {
+    return this.admin.resolve(u.sub, id, d);
+  }
+  @Get("moderation") moderation(@Query() q: AdminQueryDto) {
+    return this.admin.moderation(q.limit);
+  }
+  @UseGuards(StateChangingOriginGuard) @Post("moderation/:id/resolve") moderate(
+    @CurrentUser() u: AccessClaims,
+    @Param("id", new ParseUUIDPipe()) id: string,
+    @Body() d: ModerateDto,
+  ) {
+    return this.admin.moderate(u.sub, id, d.decision, d.reason);
+  }
+  @UseGuards(StateChangingOriginGuard)
+  @Post("accounts/:publicId/sanctions")
+  sanction(
+    @CurrentUser() u: AccessClaims,
+    @Param("publicId", new ParseUUIDPipe()) id: string,
+    @Body() d: SanctionDto,
+  ) {
+    return this.admin.sanctions(u.sub, id, d);
+  }
+}
+
+Reflect.defineMetadata(
+  "design:paramtypes",
+  [AdminQueryDto],
+  AdminController.prototype,
+  "reports",
+);
+Reflect.defineMetadata(
+  "design:paramtypes",
+  [Object, String, ResolveReportDto],
+  AdminController.prototype,
+  "resolve",
+);
+Reflect.defineMetadata(
+  "design:paramtypes",
+  [AdminQueryDto],
+  AdminController.prototype,
+  "moderation",
+);
+Reflect.defineMetadata(
+  "design:paramtypes",
+  [Object, String, ModerateDto],
+  AdminController.prototype,
+  "moderate",
+);
+Reflect.defineMetadata(
+  "design:paramtypes",
+  [Object, String, SanctionDto],
+  AdminController.prototype,
+  "sanction",
+);

apps/api/src/admin/admin.guard.ts

@@ -0,0 +1,36 @@
+import {
+  CanActivate,
+  ExecutionContext,
+  HttpStatus,
+  Inject,
+  Injectable,
+} from "@nestjs/common";
+import { ErrorCode } from "@drift/contracts";
+import type { Request } from "express";
+import { PrismaService } from "../database/prisma.service.js";
+import { DomainException } from "../common/domain.exception.js";
+import type { AuthenticatedRequest } from "../auth/current-user.decorator.js";
+@Injectable()
+export class AdminGuard implements CanActivate {
+  constructor(@Inject(PrismaService) private readonly prisma: PrismaService) {}
+  async canActivate(ctx: ExecutionContext) {
+    const req = ctx.switchToHttp().getRequest<Request & AuthenticatedRequest>();
+    if (!req.user)
+      throw new DomainException(
+        ErrorCode.AUTH_UNAUTHORIZED,
+        "Unauthorized",
+        HttpStatus.UNAUTHORIZED,
+      );
+    const account = await this.prisma.account.findUnique({
+      where: { id: req.user.sub },
+      select: { role: true },
+    });
+    if (account?.role !== "ADMIN")
+      throw new DomainException(
+        ErrorCode.CONVERSATION_FORBIDDEN,
+        "Forbidden",
+        HttpStatus.FORBIDDEN,
+      );
+    return true;
+  }
+}

apps/api/src/admin/admin.module.ts

@@ -0,0 +1,14 @@
+import { Module } from "@nestjs/common";
+import { AuthModule } from "../auth/auth.module.js";
+import { DatabaseModule } from "../database/database.module.js";
+import { StateChangingOriginGuard } from "../conversation/state-changing-origin.guard.js";
+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";
+@Module({
+  imports: [DatabaseModule, AuthModule, SafetyModule],
+  controllers: [AdminController],
+  providers: [AdminService, AdminGuard, StateChangingOriginGuard],
+})
+export class AdminModule {}

apps/api/src/admin/admin.service.ts

@@ -0,0 +1,379 @@
+import { createHash } from "node:crypto";
+import { HttpStatus, Inject, Injectable } from "@nestjs/common";
+import { ErrorCode } from "@drift/contracts";
+import { Prisma } from "@prisma/client";
+import { DomainException } from "../common/domain.exception.js";
+import { PrismaService } from "../database/prisma.service.js";
+import { SafetyLockService } from "../safety/safety-lock.service.js";
+import type { ResolveReportDto, SanctionDto } from "./dto.js";
+@Injectable()
+export class AdminService {
+  constructor(
+    @Inject(PrismaService) private readonly prisma: PrismaService,
+    @Inject(SafetyLockService) private readonly locks: SafetyLockService,
+  ) {}
+  async reports(
+    status: "PENDING" | "REVIEWING" | "RESOLVED" | "DISMISSED" | undefined,
+    limit: number,
+  ) {
+    const rows = await this.prisma.report.findMany({
+      where: status ? { status } : {},
+      orderBy: [{ createdAt: "asc" }, { id: "asc" }],
+      take: limit,
+    });
+    return {
+      items: rows.map((x) => ({
+        id: x.id,
+        status: x.status,
+        reason: x.reason,
+        details: x.details,
+        targetSnapshot: x.targetSnapshot,
+        resolution: x.resolution,
+        createdAt: x.createdAt,
+      })),
+    };
+  }
+  async resolve(actorId: string, id: string, dto: ResolveReportDto) {
+    if (dto.decision === "DISMISSED" && dto.sanction)
+      throw new DomainException(
+        ErrorCode.VALIDATION_ERROR,
+        "Invalid sanction",
+        HttpStatus.BAD_REQUEST,
+      );
+    return this.prisma.$transaction(
+      async (tx) => {
+        const report = await tx.report.findUnique({ where: { id } });
+        if (!report)
+          throw new DomainException(
+            ErrorCode.NOT_FOUND,
+            "Not found",
+            HttpStatus.NOT_FOUND,
+          );
+        const accountId =
+          report.reportedAccountId ??
+          (report.messageId
+            ? (
+                await tx.message.findUnique({
+                  where: { id: report.messageId },
+                  select: { senderId: true },
+                })
+              )?.senderId
+            : report.bottleId
+              ? (
+                  await tx.bottle.findUnique({
+                    where: { id: report.bottleId },
+                    select: { authorId: true },
+                  })
+                )?.authorId
+              : report.conversationId
+                ? (report.targetSnapshot as { reportedAccountId?: string })
+                    .reportedAccountId
+                : undefined);
+        await this.locks.lockAccounts(tx, [
+          actorId,
+          ...(accountId ? [accountId] : []),
+        ]);
+        const changed = await tx.report.updateMany({
+          where: { id, status: { in: ["PENDING", "REVIEWING"] } },
+          data: {
+            status: dto.decision === "UPHELD" ? "RESOLVED" : "DISMISSED",
+            resolution: dto.resolution,
+          },
+        });
+        if (changed.count !== 1)
+          throw new DomainException(
+            ErrorCode.IDEMPOTENCY_CONFLICT,
+            "Report already resolved",
+            HttpStatus.CONFLICT,
+          );
+        const task = await tx.moderationTask.update({
+          where: {
+            targetType_targetId: { targetType: "REPORT", targetId: id },
+          },
+          data: {
+            status: "COMPLETED",
+            decision: dto.decision,
+            result: { decision: dto.decision },
+            assignedToId: actorId,
+            reviewedAt: new Date(),
+          },
+        });
+        if (dto.decision === "UPHELD" && dto.sanction && accountId)
+          await this.issue(tx, actorId, accountId, dto.sanction, task.id);
+        await tx.auditLog.create({
+          data: {
+            actorId,
+            action: "REPORT_RESOLVED",
+            entityType: "REPORT",
+            entityId: id,
+            metadata: {
+              decision: dto.decision,
+              sanctionType: dto.sanction?.type,
+            },
+          },
+        });
+        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(),
+            },
+          });
+        return tx.report.findUniqueOrThrow({
+          where: { id },
+          select: { id: true, status: true, resolution: true },
+        });
+      },
+      { isolationLevel: Prisma.TransactionIsolationLevel.Serializable },
+    );
+  }
+  async sanctions(actorId: string, publicId: string, dto: SanctionDto) {
+    const profile = await this.prisma.anonymousProfile.findUnique({
+      where: { publicId },
+    });
+    if (!profile)
+      throw new DomainException(
+        ErrorCode.NOT_FOUND,
+        "Not found",
+        HttpStatus.NOT_FOUND,
+      );
+    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 tx.auditLog.create({
+        data: {
+          actorId,
+          action: "SANCTION_ISSUED",
+          entityType: "ACCOUNT",
+          entityId: profile.accountId,
+          metadata: { type: dto.type },
+        },
+      });
+      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 moderate(
+    actorId: string,
+    id: string,
+    decision: "APPROVED" | "REJECTED",
+    reason: string,
+  ) {
+    return this.prisma.$transaction(async (tx) => {
+      // Only the worker's terminal MANUAL_REVIEW result may be overridden. The
+      // outbox must already be published, so no leased worker can later write
+      // a stale moderation decision over the administrator's decision.
+      const task = await tx.moderationTask.findUnique({ where: { id } });
+      if (
+        !task ||
+        task.status !== "COMPLETED" ||
+        task.decision !== "MANUAL_REVIEW" ||
+        (task.targetType !== "BOTTLE" && task.targetType !== "PROFILE")
+      )
+        throw new DomainException(
+          ErrorCode.IDEMPOTENCY_CONFLICT,
+          "Task not awaiting manual review",
+          HttpStatus.CONFLICT,
+        );
+      const event = await tx.outboxEvent.findFirst({
+        where: {
+          eventType:
+            task.targetType === "BOTTLE"
+              ? "BOTTLE_MODERATION_REQUESTED"
+              : "PROFILE_MODERATION_REQUESTED",
+          aggregateId: task.targetId,
+          status: "PUBLISHED",
+          payload: { path: ["taskId"], equals: task.id },
+        },
+      });
+      if (!event)
+        throw new DomainException(
+          ErrorCode.IDEMPOTENCY_CONFLICT,
+          "Moderation event not published",
+          HttpStatus.CONFLICT,
+        );
+      if (task.targetType === "BOTTLE") {
+        const bottle = await tx.bottle.findUnique({
+          where: { id: task.targetId },
+        });
+        if (
+          !bottle ||
+          bottle.reviewStatus !== "MANUAL_REVIEW" ||
+          bottle.poolStatus !== "CLOSED" ||
+          task.payloadHash !==
+            createHash("sha256").update(bottle.contentText).digest("hex")
+        )
+          throw new DomainException(
+            ErrorCode.IDEMPOTENCY_CONFLICT,
+            "Bottle state changed",
+            HttpStatus.CONFLICT,
+          );
+        const updated = await tx.bottle.updateMany({
+          where: {
+            id: bottle.id,
+            reviewStatus: "MANUAL_REVIEW",
+            poolStatus: "CLOSED",
+            contentText: bottle.contentText,
+          },
+          data: {
+            reviewStatus: decision,
+            poolStatus: decision === "APPROVED" ? "IN_POOL" : "CLOSED",
+            approvedAt: decision === "APPROVED" ? new Date() : null,
+          },
+        });
+        if (updated.count !== 1)
+          throw new DomainException(
+            ErrorCode.IDEMPOTENCY_CONFLICT,
+            "Bottle state changed",
+            HttpStatus.CONFLICT,
+          );
+      } else {
+        const profile = await tx.anonymousProfile.findUnique({
+          where: { id: task.targetId },
+        });
+        const hash =
+          profile &&
+          createHash("sha256")
+            .update(
+              JSON.stringify([
+                profile.nickname,
+                profile.avatarColor,
+                profile.bio,
+              ]),
+            )
+            .digest("hex");
+        const version = Number(
+          event.dedupeKey.slice(event.dedupeKey.lastIndexOf(":") + 1),
+        );
+        if (
+          !profile ||
+          profile.reviewStatus !== "MANUAL_REVIEW" ||
+          profile.version !== version ||
+          hash !== task.payloadHash
+        )
+          throw new DomainException(
+            ErrorCode.IDEMPOTENCY_CONFLICT,
+            "Profile state changed",
+            HttpStatus.CONFLICT,
+          );
+        const updated = await tx.anonymousProfile.updateMany({
+          where: { id: profile.id, reviewStatus: "MANUAL_REVIEW", version },
+          data: { reviewStatus: decision },
+        });
+        if (updated.count !== 1)
+          throw new DomainException(
+            ErrorCode.IDEMPOTENCY_CONFLICT,
+            "Profile state changed",
+            HttpStatus.CONFLICT,
+          );
+      }
+      const changed = await tx.moderationTask.updateMany({
+        where: {
+          id,
+          status: "COMPLETED",
+          decision: "MANUAL_REVIEW",
+          payloadHash: task.payloadHash,
+        },
+        data: {
+          decision,
+          result: { decision },
+          assignedToId: actorId,
+          reviewedAt: new Date(),
+        },
+      });
+      if (changed.count !== 1)
+        throw new DomainException(
+          ErrorCode.IDEMPOTENCY_CONFLICT,
+          "Task already resolved",
+          HttpStatus.CONFLICT,
+        );
+      await tx.auditLog.create({
+        data: {
+          actorId,
+          action: "MODERATION_RESOLVED",
+          entityType: task.targetType,
+          entityId: task.targetId,
+          metadata: {
+            decision,
+            reason,
+            previousDecision: "MANUAL_REVIEW",
+            eventId: event.id,
+          },
+        },
+      });
+      return { id, status: "COMPLETED", decision };
+    });
+  }
+  private async issue(
+    tx: Prisma.TransactionClient,
+    actorId: string,
+    accountId: string,
+    dto: SanctionDto,
+    moderationTaskId?: string,
+  ) {
+    if (
+      dto.expiresAt &&
+      (new Date(dto.expiresAt).getTime() <= Date.now() ||
+        new Date(dto.expiresAt).getTime() >
+          Date.now() + 365 * 24 * 60 * 60 * 1000)
+    )
+      throw new DomainException(
+        ErrorCode.VALIDATION_ERROR,
+        "Invalid sanction expiry",
+        HttpStatus.BAD_REQUEST,
+      );
+    const sanction = await tx.sanction.create({
+      data: {
+        accountId,
+        issuedById: actorId,
+        moderationTaskId: moderationTaskId ?? null,
+        type: dto.type,
+        reason: dto.reason,
+        expiresAt: dto.expiresAt ? new Date(dto.expiresAt) : null,
+      },
+    });
+    if (dto.type === "SUSPENSION" || dto.type === "BAN")
+      await tx.account.update({
+        where: { id: accountId },
+        data: { status: "SUSPENDED", tokenVersion: { increment: 1 } },
+      });
+    return sanction;
+  }
+}

apps/api/src/admin/dto.ts

@@ -0,0 +1,43 @@
+import { Transform, Type } from "class-transformer";
+import {
+  IsDateString,
+  IsEnum,
+  IsInt,
+  IsOptional,
+  IsString,
+  Max,
+  MaxLength,
+  Min,
+  MinLength,
+  ValidateNested,
+} from "class-validator";
+const trim = ({ value }: { value: unknown }) =>
+  typeof value === "string" ? value.trim() : value;
+export class SanctionDto {
+  @IsEnum(["WARNING", "MUTE", "SUSPENSION", "BAN"]) type!:
+    "WARNING" | "MUTE" | "SUSPENSION" | "BAN";
+  @Transform(trim) @IsString() @MinLength(1) @MaxLength(500) reason!: string;
+  @IsOptional() @IsDateString() expiresAt?: string;
+}
+export class ResolveReportDto {
+  @IsEnum(["UPHELD", "DISMISSED"]) decision!: "UPHELD" | "DISMISSED";
+  @Transform(trim)
+  @IsString()
+  @MinLength(1)
+  @MaxLength(1000)
+  resolution!: string;
+  @IsOptional()
+  @ValidateNested()
+  @Type(() => SanctionDto)
+  sanction?: SanctionDto;
+}
+export class AdminQueryDto {
+  @IsOptional()
+  @IsEnum(["PENDING", "REVIEWING", "RESOLVED", "DISMISSED"])
+  status?: "PENDING" | "REVIEWING" | "RESOLVED" | "DISMISSED";
+  @IsOptional() @Type(() => Number) @IsInt() @Min(1) @Max(100) limit = 20;
+}
+export class ModerateDto {
+  @IsEnum(["APPROVED", "REJECTED"]) decision!: "APPROVED" | "REJECTED";
+  @Transform(trim) @IsString() @MinLength(1) @MaxLength(500) reason!: string;
+}

apps/api/src/app.module.ts

@@ -6,6 +6,9 @@ import { ProfileModule } from "./profile/profile.module.js";
 import { BottleModule } from "./bottle/bottle.module.js";
 import { MatchModule } from "./match/match.module.js";
 import { ConversationModule } from "./conversation/conversation.module.js";
+import { SafetyModule } from "./safety/safety.module.js";
+import { NotificationModule } from "./notification/notification.module.js";
+import { AdminModule } from "./admin/admin.module.js";
 
 @Module({
   imports: [
@@ -15,6 +18,9 @@ import { ConversationModule } from "./conversation/conversation.module.js";
     BottleModule,
     MatchModule,
     ConversationModule,
+    SafetyModule,
+    NotificationModule,
+    AdminModule,
   ],
 })
 export class AppModule implements NestModule {

apps/api/src/notification/notification.controller.ts

@@ -0,0 +1,74 @@
+import {
+  Body,
+  Controller,
+  Get,
+  Inject,
+  Post,
+  Put,
+  Query,
+  UseGuards,
+} from "@nestjs/common";
+import { AuthGuard } from "../auth/auth.guard.js";
+import { CurrentUser } from "../auth/current-user.decorator.js";
+import type { AccessClaims } from "../auth/token.service.js";
+import { StateChangingOriginGuard } from "../conversation/state-changing-origin.guard.js";
+import {
+  NotificationQueryDto,
+  PushPreferenceDto,
+  ReadNotificationsDto,
+} from "../safety/dto.js";
+import { NotificationService } from "./notification.service.js";
+@Controller()
+@UseGuards(AuthGuard)
+export class NotificationController {
+  constructor(
+    @Inject(NotificationService)
+    private readonly notifications: NotificationService,
+  ) {}
+  @Get("notifications") list(
+    @CurrentUser() u: AccessClaims,
+    @Query() q: NotificationQueryDto,
+  ) {
+    return this.notifications.list(u.sub, q.cursor, q.limit);
+  }
+  @UseGuards(StateChangingOriginGuard) @Post("notifications/read") read(
+    @CurrentUser() u: AccessClaims,
+    @Body() d: ReadNotificationsDto,
+  ) {
+    return this.notifications.read(u.sub, d.ids);
+  }
+  @Get("settings/push") pref(@CurrentUser() u: AccessClaims) {
+    return this.notifications.preference(u.sub);
+  }
+  @UseGuards(StateChangingOriginGuard) @Put("settings/push") update(
+    @CurrentUser() u: AccessClaims,
+    @Body() d: PushPreferenceDto,
+  ) {
+    return this.notifications.updatePreference(u.sub, d.inAppEnabled);
+  }
+}
+
+Reflect.defineMetadata(
+  "design:paramtypes",
+  [Object, NotificationQueryDto],
+  NotificationController.prototype,
+  "list",
+);
+Reflect.defineMetadata(
+  "design:paramtypes",
+  [Object, ReadNotificationsDto],
+  NotificationController.prototype,
+  "read",
+);
+Reflect.defineMetadata(
+  "design:paramtypes",
+  [Object],
+  NotificationController.prototype,
+  "pref",
+);
+Reflect.defineMetadata(
+  "design:paramtypes",
+  [Object, PushPreferenceDto],
+  NotificationController.prototype,
+  "update",
+);

apps/api/src/notification/notification.module.ts

@@ -0,0 +1,12 @@
+import { Module } from "@nestjs/common";
+import { AuthModule } from "../auth/auth.module.js";
+import { DatabaseModule } from "../database/database.module.js";
+import { StateChangingOriginGuard } from "../conversation/state-changing-origin.guard.js";
+import { NotificationController } from "./notification.controller.js";
+import { NotificationService } from "./notification.service.js";
+@Module({
+  imports: [DatabaseModule, AuthModule],
+  controllers: [NotificationController],
+  providers: [NotificationService, StateChangingOriginGuard],
+})
+export class NotificationModule {}

apps/api/src/notification/notification.service.ts

@@ -0,0 +1,107 @@
+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";
+@Injectable()
+export class NotificationService {
+  constructor(@Inject(PrismaService) private readonly prisma: PrismaService) {}
+  async list(accountId: string, cursor: string | undefined, limit: number) {
+    let decoded: { t: string; id: string } | undefined;
+    try {
+      if (cursor) {
+        const value: unknown = JSON.parse(
+          Buffer.from(cursor, "base64url").toString("utf8"),
+        );
+        if (
+          typeof value !== "object" ||
+          value === null ||
+          Array.isArray(value) ||
+          Object.keys(value).sort().join(",") !== "id,t" ||
+          !("t" in value) ||
+          typeof value.t !== "string" ||
+          !("id" in value) ||
+          typeof value.id !== "string" ||
+          !/^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i.test(
+            value.id,
+          ) ||
+          !Number.isFinite(Date.parse(value.t)) ||
+          new Date(value.t).toISOString() !== value.t
+        )
+          throw new Error("Invalid cursor");
+        decoded = { t: value.t, id: value.id };
+      }
+    } catch {
+      throw new DomainException(
+        ErrorCode.VALIDATION_ERROR,
+        "Invalid cursor",
+        HttpStatus.BAD_REQUEST,
+      );
+    }
+    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 page = rows.slice(0, limit);
+    return {
+      items: page.map((x) => ({
+        id: x.id,
+        type: x.type,
+        payload: x.payload,
+        status: x.status,
+        readAt: x.readAt,
+        createdAt: x.createdAt,
+      })),
+      nextCursor:
+        rows.length > limit && page.length
+          ? Buffer.from(
+              JSON.stringify({
+                t: page.at(-1)!.createdAt.toISOString(),
+                id: page.at(-1)!.id,
+              }),
+            ).toString("base64url")
+          : null,
+    };
+  }
+  async read(accountId: string, ids: string[]) {
+    const owned = await this.prisma.notification.count({
+      where: { accountId, id: { in: ids } },
+    });
+    if (owned !== new Set(ids).size)
+      throw new DomainException(
+        ErrorCode.NOT_FOUND,
+        "Not found",
+        HttpStatus.NOT_FOUND,
+      );
+    await this.prisma.notification.updateMany({
+      where: { accountId, id: { in: ids } },
+      data: { status: "READ", readAt: new Date() },
+    });
+    return { read: owned };
+  }
+  async preference(accountId: string) {
+    const row = await this.prisma.pushPreference.findUnique({
+      where: { accountId },
+    });
+    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 },
+    });
+  }
+}

apps/api/src/safety/dto.ts

@@ -0,0 +1,53 @@
+import { Transform, Type } from "class-transformer";
+import {
+  ArrayMaxSize,
+  IsArray,
+  IsEnum,
+  IsInt,
+  IsOptional,
+  IsString,
+  IsUUID,
+  Matches,
+  Max,
+  MaxLength,
+  Min,
+  MinLength,
+} from "class-validator";
+
+const trim = ({ value }: { value: unknown }) =>
+  typeof value === "string" ? value.trim() : value;
+export enum ReportTargetType {
+  ACCOUNT = "ACCOUNT",
+  BOTTLE = "BOTTLE",
+  CONVERSATION = "CONVERSATION",
+  MESSAGE = "MESSAGE",
+}
+export enum ReportReason {
+  HARASSMENT = "HARASSMENT",
+  SPAM = "SPAM",
+  SEXUAL = "SEXUAL",
+  VIOLENCE = "VIOLENCE",
+  FRAUD = "FRAUD",
+  OTHER = "OTHER",
+}
+export class CreateReportDto {
+  @IsEnum(ReportTargetType) targetType!: ReportTargetType;
+  @IsUUID() targetId!: string;
+  @IsEnum(ReportReason) reason!: ReportReason;
+  @IsOptional()
+  @Transform(trim)
+  @IsString()
+  @MinLength(1)
+  @MaxLength(1000)
+  details?: string;
+}
+export class ReadNotificationsDto {
+  @IsArray() @ArrayMaxSize(100) @IsUUID("4", { each: true }) ids!: string[];
+}
+export class NotificationQueryDto {
+  @IsOptional() @Matches(/^[A-Za-z0-9_-]{1,512}$/) cursor?: string;
+  @IsOptional() @Type(() => Number) @IsInt() @Min(1) @Max(100) limit = 20;
+}
+export class PushPreferenceDto {
+  @IsEnum([true, false]) inAppEnabled!: boolean;
+}

apps/api/src/safety/safety-admin.e2e-spec.ts

@@ -0,0 +1,507 @@
+/* eslint-disable @typescript-eslint/no-unsafe-member-access */
+import "reflect-metadata";
+import { type INestApplication } from "@nestjs/common";
+import { Test } from "@nestjs/testing";
+import { PrismaClient } from "@prisma/client";
+import { randomUUID } from "node:crypto";
+import request from "supertest";
+import { afterAll, beforeAll, beforeEach, describe, expect, it } from "vitest";
+import { AppModule } from "../app.module.js";
+import { TokenService } from "../auth/token.service.js";
+import { configureApp } from "../main.js";
+import { assertSafeTestDatabaseUrl } from "../../../../prisma/database-safety.js";
+
+process.env.NODE_ENV = "test";
+process.env.WEB_ORIGIN = "http://localhost:3000";
+process.env.PHONE_ENCRYPTION_KEY =
+  "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=";
+process.env.PHONE_HMAC_KEY = "test-phone-hmac-key-with-at-least-32-bytes";
+process.env.VERIFICATION_CODE_HMAC_KEY =
+  "test-code-hmac-key-with-at-least-32-bytes";
+process.env.LEASE_TOKEN_HMAC_KEY = "test-lease-hmac-key-with-at-least-32-bytes";
+process.env.LEASE_TOKEN_ENCRYPTION_KEY =
+  "BBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBB=";
+process.env.JWT_SECRET = "test-jwt-secret-with-at-least-thirty-two-bytes";
+process.env.REFRESH_TOKEN_HMAC_KEY =
+  "test-refresh-hmac-key-with-at-least-32-bytes";
+const prisma = new PrismaClient();
+const origin = { Origin: process.env.WEB_ORIGIN };
+
+type Actor = { id: string; publicId: string; authorization: string };
+
+describe("safety, notification and admin governance", () => {
+  let app: INestApplication;
+  let alice: Actor;
+  let bob: Actor;
+  let admin: Actor;
+  let conversationId: string;
+  let messageId: string;
+
+  async function actor(
+    name: string,
+    role: "USER" | "ADMIN" = "USER",
+  ): Promise<Actor> {
+    const account = await prisma.account.create({
+      data: {
+        phoneCiphertext: Buffer.from(`cipher-${name}`),
+        phoneHmac: randomUUID(),
+        role,
+        anonymousProfile: {
+          create: {
+            nickname: name,
+            avatarColor: "#66CCFF",
+            reviewStatus: "APPROVED",
+          },
+        },
+      },
+      include: { anonymousProfile: true },
+    });
+    const session = await prisma.session.create({
+      data: {
+        accountId: account.id,
+        refreshTokenHash: randomUUID(),
+        deviceId: name.padEnd(8, "x"),
+        tokenVersion: 0,
+        expiresAt: new Date(Date.now() + 60_000),
+        tokenFamily: randomUUID(),
+      },
+    });
+    const token = app.get(TokenService).issueAccess({
+      sub: account.id,
+      session_id: session.id,
+      device_id: name.padEnd(8, "x"),
+      scopes: ["user"],
+      token_version: 0,
+    });
+    return {
+      id: account.id,
+      publicId: account.anonymousProfile!.publicId,
+      authorization: `Bearer ${token}`,
+    };
+  }
+
+  beforeAll(async () => {
+    assertSafeTestDatabaseUrl(process.env.DATABASE_URL ?? "");
+    await prisma.$connect();
+    const module = await Test.createTestingModule({
+      imports: [AppModule],
+    }).compile();
+    app = module.createNestApplication();
+    configureApp(app);
+    await app.init();
+  });
+  beforeEach(async () => {
+    await prisma.outboxEvent.deleteMany();
+    await prisma.$executeRawUnsafe('TRUNCATE TABLE "accounts" CASCADE');
+    alice = await actor("alice");
+    bob = await actor("bob");
+    admin = await actor("admin", "ADMIN");
+    const bottle = await prisma.bottle.create({
+      data: {
+        authorId: alice.id,
+        contentText: "private bottle",
+        reviewStatus: "APPROVED",
+        poolStatus: "CONSUMED",
+      },
+    });
+    const conversation = await prisma.conversation.create({
+      data: {
+        sourceBottleId: bottle.id,
+        nextSeq: 2n,
+        members: {
+          create: [
+            { accountId: alice.id, peerAliasSnapshot: "bob" },
+            { accountId: bob.id, peerAliasSnapshot: "alice" },
+          ],
+        },
+        messages: {
+          create: {
+            senderId: bob.id,
+            senderPublicId: bob.publicId,
+            clientMsgId: randomUUID(),
+            seq: 1n,
+            contentText: "reported secret",
+            reviewStatus: "APPROVED",
+          },
+        },
+      },
+      include: { messages: true },
+    });
+    conversationId = conversation.id;
+    messageId = conversation.messages[0]!.id;
+  });
+  afterAll(async () => {
+    await app?.close();
+    await prisma.$disconnect();
+  });
+
+  it("creates an authorized immutable message report exactly once without leaking body to outbox/audit", async () => {
+    const key = randomUUID();
+    const body = {
+      targetType: "MESSAGE",
+      targetId: messageId,
+      reason: "HARASSMENT",
+      details: "please review",
+    };
+    const first = await request(
+      app.getHttpServer() as Parameters<typeof request>[0],
+    )
+      .post("/api/v1/reports")
+      .set("Authorization", alice.authorization)
+      .set(origin)
+      .set("Idempotency-Key", key)
+      .send(body)
+      .expect(201);
+    const retry = await request(
+      app.getHttpServer() as Parameters<typeof request>[0],
+    )
+      .post("/api/v1/reports")
+      .set("Authorization", alice.authorization)
+      .set(origin)
+      .set("Idempotency-Key", key)
+      .send(body)
+      .expect(201);
+    expect(retry.body.data.id).toBe(first.body.data.id);
+    const report = await prisma.report.findUniqueOrThrow({
+      where: { id: first.body.data.id as string },
+    });
+    expect(report.targetSnapshot).toMatchObject({
+      targetType: "MESSAGE",
+      contentText: "reported secret",
+      senderPublicId: bob.publicId,
+    });
+    expect(JSON.stringify(first.body)).not.toContain("reported secret");
+    expect(
+      await prisma.moderationTask.count({ where: { reportId: report.id } }),
+    ).toBe(1);
+    expect(
+      await prisma.auditLog.count({
+        where: { entityId: report.id, action: "REPORT_CREATED" },
+      }),
+    ).toBe(1);
+    expect(
+      JSON.stringify(
+        await prisma.auditLog.findMany({ where: { entityId: report.id } }),
+      ),
+    ).not.toContain("reported secret");
+    const events = await prisma.outboxEvent.findMany({
+      where: { aggregateId: report.id },
+    });
+    expect(events).toHaveLength(0);
+    expect(JSON.stringify(events)).not.toContain("reported secret");
+    await request(app.getHttpServer() as Parameters<typeof request>[0])
+      .post("/api/v1/reports")
+      .set("Authorization", admin.authorization)
+      .set(origin)
+      .set("Idempotency-Key", randomUUID())
+      .send(body)
+      .expect(403);
+  });
+
+  it("blocks only a conversation peer immediately, lists/unblocks by public id, and is idempotent", async () => {
+    const endpoint = `/api/v1/conversations/${conversationId}/block`;
+    await request(app.getHttpServer() as Parameters<typeof request>[0])
+      .post(endpoint)
+      .set("Authorization", alice.authorization)
+      .set(origin)
+      .send({})
+      .expect(201);
+    await request(app.getHttpServer() as Parameters<typeof request>[0])
+      .post(endpoint)
+      .set("Authorization", alice.authorization)
+      .set(origin)
+      .send({})
+      .expect(201);
+    expect(await prisma.block.count()).toBe(1);
+    await request(app.getHttpServer() as Parameters<typeof request>[0])
+      .post(`/api/v1/conversations/${conversationId}/messages/prepare`)
+      .set("Authorization", bob.authorization)
+      .set(origin)
+      .send({ conversationId, clientMsgId: randomUUID(), text: "blocked" })
+      .expect(403);
+    const listed = await request(
+      app.getHttpServer() as Parameters<typeof request>[0],
+    )
+      .get("/api/v1/me/blocks")
+      .set("Authorization", alice.authorization)
+      .expect(200);
+    expect(listed.body.data.items).toEqual([
+      expect.objectContaining({ publicId: bob.publicId }),
+    ]);
+    await request(app.getHttpServer() as Parameters<typeof request>[0])
+      .delete(`/api/v1/me/blocks/${bob.publicId}`)
+      .set("Authorization", alice.authorization)
+      .set(origin)
+      .expect(200);
+    expect(await prisma.block.count()).toBe(0);
+    expect(
+      await prisma.conversationMember.count({
+        where: { conversationId, blockedAt: { not: null } },
+      }),
+    ).toBe(0);
+  });
+
+  it("requires admin, resolves once with sanction, audit and reporter/target notifications", 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: "HARASSMENT",
+      })
+      .expect(201);
+    await request(app.getHttpServer() as Parameters<typeof request>[0])
+      .get("/api/v1/admin/reports")
+      .set("Authorization", bob.authorization)
+      .expect(403);
+    const resolved = 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: "confirmed",
+        sanction: {
+          type: "MUTE",
+          reason: "harassment",
+          expiresAt: new Date(Date.now() + 60_000).toISOString(),
+        },
+      })
+      .expect(201);
+    expect(resolved.body.data.status).toBe("RESOLVED");
+    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: "second" })
+      .expect(409);
+    expect(
+      await prisma.sanction.count({
+        where: { accountId: bob.id, type: "MUTE" },
+      }),
+    ).toBe(1);
+    expect(
+      await prisma.auditLog.count({
+        where: {
+          action: "REPORT_RESOLVED",
+          entityId: report.body.data.id as string,
+        },
+      }),
+    ).toBe(1);
+    expect(
+      await prisma.notification.count({
+        where: { accountId: { in: [alice.id, bob.id] } },
+      }),
+    ).toBe(2);
+  });
+
+  it("rejects conflicting report replay and unauthorized targets without storing new reports", async () => {
+    const key = randomUUID();
+    const create = (targetId: string) =>
+      request(app.getHttpServer() as Parameters<typeof request>[0])
+        .post("/api/v1/reports")
+        .set("Authorization", alice.authorization)
+        .set(origin)
+        .set("Idempotency-Key", key)
+        .send({ targetType: "MESSAGE", targetId, reason: "SPAM" });
+    await create(messageId).expect(201);
+    await create(randomUUID()).expect(409);
+    expect(await prisma.report.count()).toBe(1);
+    await request(app.getHttpServer() as Parameters<typeof request>[0])
+      .post("/api/v1/reports")
+      .set("Authorization", admin.authorization)
+      .set(origin)
+      .set("Idempotency-Key", randomUUID())
+      .send({ targetType: "ACCOUNT", targetId: bob.publicId, reason: "SPAM" })
+      .expect(403);
+  });
+
+  it("rejects expired sanctions, dismissed sanctions, and demoted admins", async () => {
+    const sanction = (expiresAt: string) =>
+      request(app.getHttpServer() as Parameters<typeof request>[0])
+        .post(`/api/v1/admin/accounts/${bob.publicId}/sanctions`)
+        .set("Authorization", admin.authorization)
+        .set(origin)
+        .send({ type: "MUTE", reason: "reason", expiresAt });
+    await sanction(new Date(Date.now() - 60_000).toISOString()).expect(400);
+    await request(app.getHttpServer() as Parameters<typeof request>[0])
+      .post(`/api/v1/admin/reports/${randomUUID()}/resolve`)
+      .set("Authorization", admin.authorization)
+      .set(origin)
+      .send({
+        decision: "DISMISSED",
+        resolution: "no",
+        sanction: { type: "BAN", reason: "no" },
+      })
+      .expect(400);
+    await prisma.account.update({
+      where: { id: admin.id },
+      data: { role: "USER" },
+    });
+    await request(app.getHttpServer() as Parameters<typeof request>[0])
+      .get("/api/v1/admin/reports")
+      .set("Authorization", admin.authorization)
+      .expect(403);
+  });
+
+  it("rejects pending moderation and overrides only a published, intact manual-review bottle", async () => {
+    const bottle = await prisma.bottle.create({
+      data: {
+        authorId: alice.id,
+        contentText: "review-word admin",
+        reviewStatus: "REVIEWING",
+        poolStatus: "CLOSED",
+      },
+    });
+    const { createHash } = await import("node:crypto");
+    const task = await prisma.moderationTask.create({
+      data: {
+        targetType: "BOTTLE",
+        targetId: bottle.id,
+        provider: "SIMULATED",
+        riskLabels: [],
+        payloadHash: createHash("sha256")
+          .update(bottle.contentText)
+          .digest("hex"),
+      },
+    });
+    const event = await prisma.outboxEvent.create({
+      data: {
+        aggregateType: "BOTTLE",
+        aggregateId: bottle.id,
+        eventType: "BOTTLE_MODERATION_REQUESTED",
+        dedupeKey: `bottle:${bottle.id}:moderation`,
+        payload: { bottleId: bottle.id, taskId: task.id },
+      },
+    });
+    const endpoint = `/api/v1/admin/moderation/${task.id}/resolve`;
+    const moderate = () =>
+      request(app.getHttpServer() as Parameters<typeof request>[0])
+        .post(endpoint)
+        .set("Authorization", admin.authorization)
+        .set(origin)
+        .send({ decision: "APPROVED", reason: "human verified" });
+    await moderate().expect(409);
+    expect(
+      (await prisma.outboxEvent.findUniqueOrThrow({ where: { id: event.id } }))
+        .status,
+    ).toBe("PENDING");
+    await prisma.bottle.update({
+      where: { id: bottle.id },
+      data: { reviewStatus: "MANUAL_REVIEW" },
+    });
+    await prisma.moderationTask.update({
+      where: { id: task.id },
+      data: {
+        status: "COMPLETED",
+        decision: "MANUAL_REVIEW",
+        result: { decision: "MANUAL_REVIEW" },
+      },
+    });
+    await moderate().expect(409);
+    await prisma.outboxEvent.update({
+      where: { id: event.id },
+      data: { status: "PUBLISHED", publishedAt: new Date() },
+    });
+    const queue = await request(
+      app.getHttpServer() as Parameters<typeof request>[0],
+    )
+      .get("/api/v1/admin/moderation")
+      .set("Authorization", admin.authorization)
+      .expect(200);
+    expect(queue.body.data.items).toEqual(
+      expect.arrayContaining([expect.objectContaining({ id: task.id })]),
+    );
+    await moderate().expect(201);
+    await moderate().expect(409);
+    expect(
+      (await prisma.bottle.findUniqueOrThrow({ where: { id: bottle.id } }))
+        .reviewStatus,
+    ).toBe("APPROVED");
+    expect(
+      (
+        await prisma.moderationTask.findUniqueOrThrow({
+          where: { id: task.id },
+        })
+      ).result,
+    ).toEqual({ decision: "APPROVED" });
+    expect(
+      (await prisma.outboxEvent.findUniqueOrThrow({ where: { id: event.id } }))
+        .status,
+    ).toBe("PUBLISHED");
+    expect(
+      await prisma.auditLog.count({
+        where: { action: "MODERATION_RESOLVED", entityId: bottle.id },
+      }),
+    ).toBe(1);
+  });
+
+  it("keeps notifications private, cursor-paged/readable, and stores in-app preferences", async () => {
+    const own = await prisma.notification.create({
+      data: {
+        accountId: alice.id,
+        dedupeKey: randomUUID(),
+        type: "TEST",
+        payload: { safe: true },
+        status: "SENT",
+      },
+    });
+    await prisma.notification.create({
+      data: {
+        accountId: bob.id,
+        dedupeKey: randomUUID(),
+        type: "SECRET",
+        payload: { safe: false },
+        status: "SENT",
+      },
+    });
+    const list = await request(
+      app.getHttpServer() as Parameters<typeof request>[0],
+    )
+      .get("/api/v1/notifications?limit=1")
+      .set("Authorization", alice.authorization)
+      .expect(200);
+    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])
+      .post("/api/v1/notifications/read")
+      .set("Authorization", alice.authorization)
+      .set(origin)
+      .send({ ids: [own.id] })
+      .expect(201);
+    await request(app.getHttpServer() as Parameters<typeof request>[0])
+      .post("/api/v1/notifications/read")
+      .set("Authorization", alice.authorization)
+      .set(origin)
+      .send({
+        ids: [
+          (
+            await prisma.notification.findFirstOrThrow({
+              where: { accountId: bob.id },
+            })
+          ).id,
+        ],
+      })
+      .expect(404);
+    await request(app.getHttpServer() as Parameters<typeof request>[0])
+      .put("/api/v1/settings/push")
+      .set("Authorization", alice.authorization)
+      .set(origin)
+      .send({ inAppEnabled: false })
+      .expect(200);
+    const pref = await request(
+      app.getHttpServer() as Parameters<typeof request>[0],
+    )
+      .get("/api/v1/settings/push")
+      .set("Authorization", alice.authorization)
+      .expect(200);
+    expect(pref.body.data).toEqual({ inAppEnabled: false });
+  });
+});

apps/api/src/safety/safety.controller.ts

@@ -0,0 +1,79 @@
+import {
+  Body,
+  Controller,
+  Delete,
+  Get,
+  Headers,
+  HttpStatus,
+  Inject,
+  Param,
+  ParseUUIDPipe,
+  Post,
+  UseGuards,
+} from "@nestjs/common";
+import { ErrorCode } from "@drift/contracts";
+import { AuthGuard } from "../auth/auth.guard.js";
+import { CurrentUser } from "../auth/current-user.decorator.js";
+import type { AccessClaims } from "../auth/token.service.js";
+import { DomainException } from "../common/domain.exception.js";
+import { StateChangingOriginGuard } from "../conversation/state-changing-origin.guard.js";
+import { CreateReportDto } from "./dto.js";
+import { SafetyService } from "./safety.service.js";
+@Controller()
+@UseGuards(AuthGuard)
+export class SafetyController {
+  constructor(@Inject(SafetyService) private readonly safety: SafetyService) {}
+  @UseGuards(StateChangingOriginGuard) @Post("reports") report(
+    @CurrentUser() user: AccessClaims,
+    @Headers("idempotency-key") key: string,
+    @Body() dto: CreateReportDto,
+  ) {
+    if (!key || key.length > 128)
+      throw new DomainException(
+        ErrorCode.VALIDATION_ERROR,
+        "Invalid idempotency key",
+        HttpStatus.BAD_REQUEST,
+      );
+    return this.safety.report(user.sub, key, dto);
+  }
+  @UseGuards(StateChangingOriginGuard) @Post("conversations/:id/block") block(
+    @CurrentUser() user: AccessClaims,
+    @Param("id", new ParseUUIDPipe()) id: string,
+  ) {
+    return this.safety.block(user.sub, id);
+  }
+  @Get("me/blocks") blocks(@CurrentUser() user: AccessClaims) {
+    return this.safety.blocks(user.sub);
+  }
+  @UseGuards(StateChangingOriginGuard) @Delete("me/blocks/:publicId") unblock(
+    @CurrentUser() user: AccessClaims,
+    @Param("publicId", new ParseUUIDPipe()) id: string,
+  ) {
+    return this.safety.unblock(user.sub, id);
+  }
+}
+
+Reflect.defineMetadata(
+  "design:paramtypes",
+  [Object, String, CreateReportDto],
+  SafetyController.prototype,
+  "report",
+);
+Reflect.defineMetadata(
+  "design:paramtypes",
+  [Object, String],
+  SafetyController.prototype,
+  "block",
+);
+Reflect.defineMetadata(
+  "design:paramtypes",
+  [Object],
+  SafetyController.prototype,
+  "blocks",
+);
+Reflect.defineMetadata(
+  "design:paramtypes",
+  [Object, String],
+  SafetyController.prototype,
+  "unblock",
+);

apps/api/src/safety/safety.module.ts

@@ -1,8 +1,14 @@
 import { Module } from "@nestjs/common";
+import { AuthModule } from "../auth/auth.module.js";
+import { DatabaseModule } from "../database/database.module.js";
+import { StateChangingOriginGuard } from "../conversation/state-changing-origin.guard.js";
+import { SafetyController } from "./safety.controller.js";
 import { SafetyLockService } from "./safety-lock.service.js";
-
+import { SafetyService } from "./safety.service.js";
 @Module({
-  providers: [SafetyLockService],
+  imports: [DatabaseModule, AuthModule],
+  controllers: [SafetyController],
+  providers: [SafetyLockService, SafetyService, StateChangingOriginGuard],
   exports: [SafetyLockService],
 })
 export class SafetyModule {}

apps/api/src/safety/safety.service.ts

@@ -0,0 +1,325 @@
+import { createHash } from "node:crypto";
+import { HttpStatus, Inject, Injectable } from "@nestjs/common";
+import { ErrorCode } from "@drift/contracts";
+import { Prisma, type PrismaClient } from "@prisma/client";
+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";
+
+type Db = Prisma.TransactionClient | PrismaClient;
+const targetType = (dto: CreateReportDto): string => dto.targetType;
+@Injectable()
+export class SafetyService {
+  constructor(
+    @Inject(PrismaService) private readonly prisma: PrismaService,
+    @Inject(SafetyLockService) private readonly locks: SafetyLockService,
+  ) {}
+  async report(reporterId: string, key: string, dto: CreateReportDto) {
+    const prior = await this.prisma.report.findUnique({
+      where: { reporterId_idempotencyKey: { reporterId, idempotencyKey: key } },
+    });
+    if (prior) return this.replay(prior, dto);
+    return this.prisma.$transaction(
+      async (tx) => {
+        const existing = await tx.report.findUnique({
+          where: {
+            reporterId_idempotencyKey: { reporterId, idempotencyKey: key },
+          },
+        });
+        if (existing) return this.replay(existing, dto);
+        const target = await this.snapshot(
+          tx,
+          reporterId,
+          dto.targetType,
+          dto.targetId,
+        );
+        const report = await tx.report.create({
+          data: {
+            reporterId,
+            idempotencyKey: key,
+            reason: dto.reason,
+            details: dto.details ?? null,
+            targetSnapshot: target.snapshot,
+            reportedAccountId:
+              targetType(dto) === "ACCOUNT" ? (target.accountId ?? null) : null,
+            ...this.targetFk(dto.targetType, dto.targetId),
+          },
+        });
+        await tx.moderationTask.create({
+          data: {
+            targetType: "REPORT",
+            targetId: report.id,
+            provider: "ADMIN",
+            riskLabels: [],
+            payloadHash: createHash("sha256").update(report.id).digest("hex"),
+            reportId: report.id,
+          },
+        });
+        await tx.auditLog.create({
+          data: {
+            actorId: reporterId,
+            action: "REPORT_CREATED",
+            entityType: "REPORT",
+            entityId: report.id,
+            metadata: { targetType: dto.targetType, reason: dto.reason },
+          },
+        });
+        return this.publicReport(report);
+      },
+      { isolationLevel: Prisma.TransactionIsolationLevel.Serializable },
+    );
+  }
+  async block(accountId: string, conversationId: string) {
+    return this.prisma.$transaction(async (tx) => {
+      const members = await tx.conversationMember.findMany({
+        where: { conversationId, leftAt: null },
+      });
+      if (
+        !members.some((x) => x.accountId === accountId) ||
+        members.length !== 2
+      )
+        throw this.forbidden();
+      const peer = members.find((x) => x.accountId !== accountId)!;
+      await this.locks.lockAccounts(tx, [accountId, peer.accountId]);
+      await tx.block.upsert({
+        where: {
+          blockerId_blockedId: {
+            blockerId: accountId,
+            blockedId: peer.accountId,
+          },
+        },
+        create: { blockerId: accountId, blockedId: peer.accountId },
+        update: {},
+      });
+      await tx.conversationMember.updateMany({
+        where: { conversationId },
+        data: { blockedAt: new Date() },
+      });
+      return {
+        conversationId,
+        blockedPublicId: (
+          await tx.anonymousProfile.findUniqueOrThrow({
+            where: { accountId: peer.accountId },
+          })
+        ).publicId,
+      };
+    });
+  }
+  async blocks(accountId: string) {
+    const rows = await this.prisma.block.findMany({
+      where: { blockerId: accountId },
+      orderBy: [{ createdAt: "desc" }, { id: "desc" }],
+      include: { blocked: { select: { anonymousProfile: true } } },
+    });
+    return {
+      items: rows.map((x) => ({
+        publicId: x.blocked.anonymousProfile!.publicId,
+        createdAt: x.createdAt,
+      })),
+    };
+  }
+  async unblock(accountId: string, publicId: string) {
+    return this.prisma.$transaction(async (tx) => {
+      const peer = await tx.anonymousProfile.findUnique({
+        where: { publicId },
+      });
+      if (!peer) throw this.notFound();
+      await this.locks.lockAccounts(tx, [accountId, peer.accountId]);
+      const removed = await tx.block.deleteMany({
+        where: { blockerId: accountId, blockedId: peer.accountId },
+      });
+      if (!removed.count) throw this.notFound();
+      const conversations = await tx.conversation.findMany({
+        where: {
+          AND: [
+            { members: { some: { accountId } } },
+            { members: { some: { accountId: peer.accountId } } },
+          ],
+        },
+        select: { id: true },
+      });
+      const ids = conversations.map((x) => x.id);
+      if (ids.length) {
+        const reverse = await tx.block.count({
+          where: { blockerId: peer.accountId, blockedId: accountId },
+        });
+        if (!reverse)
+          await tx.conversationMember.updateMany({
+            where: { conversationId: { in: ids } },
+            data: { blockedAt: null },
+          });
+      }
+      return { publicId, unblocked: true };
+    });
+  }
+  private async snapshot(
+    tx: Db,
+    reporterId: string,
+    type: ReportTargetType,
+    id: string,
+  ): Promise<{ snapshot: Prisma.InputJsonObject; accountId?: string }> {
+    if (String(type) === "MESSAGE") {
+      const row = await tx.message.findFirst({
+        where: {
+          id,
+          conversation: {
+            members: { some: { accountId: reporterId, leftAt: null } },
+          },
+        },
+      });
+      if (!row) throw this.forbidden();
+      return {
+        accountId: row.senderId,
+        snapshot: {
+          targetType: type,
+          messageId: row.id,
+          conversationId: row.conversationId,
+          senderPublicId: row.senderPublicId,
+          contentText: row.contentText,
+          sentAt: row.sentAt.toISOString(),
+        },
+      };
+    }
+    if (String(type) === "CONVERSATION") {
+      const row = await tx.conversation.findFirst({
+        where: {
+          id,
+          members: { some: { accountId: reporterId, leftAt: null } },
+        },
+        include: { members: true },
+      });
+      if (!row) throw this.forbidden();
+      const accountId = row.members.find(
+        (x) => x.accountId !== reporterId,
+      )?.accountId;
+      return {
+        ...(accountId ? { accountId } : {}),
+        snapshot: {
+          targetType: type,
+          conversationId: id,
+          reportedAccountId: accountId,
+        },
+      };
+    }
+    if (String(type) === "BOTTLE") {
+      const row = await tx.bottle.findFirst({
+        where: {
+          id,
+          OR: [
+            { authorId: reporterId },
+            { leases: { some: { pickerId: reporterId } } },
+          ],
+        },
+        include: { author: { select: { anonymousProfile: true } } },
+      });
+      if (!row) throw this.forbidden();
+      return {
+        accountId: row.authorId,
+        snapshot: {
+          targetType: type,
+          bottleId: id,
+          authorPublicId: row.author.anonymousProfile?.publicId,
+          contentText: row.contentText,
+        },
+      };
+    }
+    const profile = await tx.anonymousProfile.findUnique({
+      where: { publicId: id },
+    });
+    if (!profile || profile.accountId === reporterId) throw this.forbidden();
+    const related = await tx.conversation.count({
+      where: {
+        AND: [
+          { members: { some: { accountId: reporterId } } },
+          { members: { some: { accountId: profile.accountId } } },
+        ],
+      },
+    });
+    if (!related) throw this.forbidden();
+    return {
+      accountId: profile.accountId,
+      snapshot: {
+        targetType: type,
+        publicId: profile.publicId,
+        nickname: profile.nickname,
+      },
+    };
+  }
+  private targetFk(type: ReportTargetType, id: string) {
+    return String(type) === "MESSAGE"
+      ? { messageId: id }
+      : String(type) === "CONVERSATION"
+        ? { conversationId: id }
+        : String(type) === "BOTTLE"
+          ? { bottleId: id }
+          : {};
+  }
+  private replay(
+    row: {
+      id: string;
+      status: string;
+      reason: string;
+      details: string | null;
+      targetSnapshot: Prisma.JsonValue;
+      reportedAccountId: string | null;
+      bottleId: string | null;
+      conversationId: string | null;
+      messageId: string | null;
+      createdAt: Date;
+    },
+    dto: CreateReportDto,
+  ) {
+    const snap = row.targetSnapshot as {
+      targetType?: string;
+      publicId?: string;
+    };
+    const id =
+      targetType(dto) === "ACCOUNT"
+        ? snap.publicId
+        : targetType(dto) === "BOTTLE"
+          ? row.bottleId
+          : targetType(dto) === "CONVERSATION"
+            ? row.conversationId
+            : row.messageId;
+    if (
+      snap.targetType !== dto.targetType ||
+      id !== dto.targetId ||
+      row.reason !== String(dto.reason) ||
+      row.details !== (dto.details ?? null)
+    )
+      throw new DomainException(
+        ErrorCode.IDEMPOTENCY_CONFLICT,
+        "Conflicting request",
+        HttpStatus.CONFLICT,
+      );
+    return this.publicReport(row);
+  }
+  private publicReport(row: {
+    id: string;
+    status: string;
+    reason: string;
+    createdAt: Date;
+  }) {
+    return {
+      id: row.id,
+      status: row.status,
+      reason: row.reason,
+      createdAt: row.createdAt,
+    };
+  }
+  private forbidden() {
+    return new DomainException(
+      ErrorCode.CONVERSATION_FORBIDDEN,
+      "Forbidden",
+      HttpStatus.FORBIDDEN,
+    );
+  }
+  private notFound() {
+    return new DomainException(
+      ErrorCode.NOT_FOUND,
+      "Not found",
+      HttpStatus.NOT_FOUND,
+    );
+  }
+}

prisma/migrations/0011_safety_admin_notification/migration.sql

@@ -0,0 +1,17 @@
+CREATE TYPE "AccountRole" AS ENUM ('USER', 'ADMIN');
+ALTER TABLE "accounts" ADD COLUMN "role" "AccountRole" NOT NULL DEFAULT 'USER';
+ALTER TABLE "reports" ADD COLUMN "idempotency_key" VARCHAR(128) DEFAULT gen_random_uuid()::text;
+UPDATE "reports" SET "idempotency_key" = "id"::text WHERE "idempotency_key" IS NULL;
+ALTER TABLE "reports" ALTER COLUMN "idempotency_key" SET NOT NULL;
+CREATE UNIQUE INDEX "reports_reporter_id_idempotency_key_key" ON "reports"("reporter_id", "idempotency_key");
+ALTER TABLE "notifications" ADD COLUMN "dedupe_key" VARCHAR(255);
+UPDATE "notifications" SET "dedupe_key" = "id"::text WHERE "dedupe_key" IS NULL;
+ALTER TABLE "notifications" ALTER COLUMN "dedupe_key" SET NOT NULL;
+CREATE UNIQUE INDEX "notifications_account_id_dedupe_key_key" ON "notifications"("account_id", "dedupe_key");
+CREATE TABLE "push_preferences" (
+  "account_id" UUID NOT NULL,
+  "in_app_enabled" BOOLEAN NOT NULL DEFAULT true,
+  "updated_at" TIMESTAMPTZ(3) NOT NULL DEFAULT CURRENT_TIMESTAMP,
+  CONSTRAINT "push_preferences_pkey" PRIMARY KEY ("account_id"),
+  CONSTRAINT "push_preferences_account_id_fkey" FOREIGN KEY ("account_id") REFERENCES "accounts"("id") ON DELETE CASCADE ON UPDATE CASCADE
+);

prisma/schema.prisma

@@ -13,6 +13,11 @@ enum AccountStatus {
   DELETED
 }
 
+enum AccountRole {
+  USER
+  ADMIN
+}
+
 enum AuthProvider {
   PHONE
   APPLE
@@ -97,6 +102,7 @@ model Account {
   phoneCiphertext     Bytes                @map("phone_ciphertext")
   phoneHmac           String               @unique @map("phone_hmac") @db.VarChar(128)
   status              AccountStatus        @default(ACTIVE)
+  role                AccountRole          @default(USER)
   tokenVersion        Int                  @default(0) @map("token_version")
   createdAt           DateTime             @default(now()) @map("created_at") @db.Timestamptz(3)
   updatedAt           DateTime             @updatedAt @map("updated_at") @db.Timestamptz(3)
@@ -119,6 +125,7 @@ model Account {
   auditLogs           AuditLog[]           @relation("AuditActor")
   dailyUsage          DailyUsage[]
   pickRequests        BottlePickRequest[]
+  pushPreference      PushPreference?
 
   @@map("accounts")
 }
@@ -368,6 +375,7 @@ model Report {
   /// Exactly one nullable target foreign key must be set; enforced in 0001_init SQL.
   id                String           @id @default(uuid()) @db.Uuid
   reporterId        String           @map("reporter_id") @db.Uuid
+  idempotencyKey    String           @default(uuid()) @map("idempotency_key") @db.VarChar(128)
   reportedAccountId String?          @map("reported_account_id") @db.Uuid
   bottleId          String?          @map("bottle_id") @db.Uuid
   conversationId    String?          @map("conversation_id") @db.Uuid
@@ -388,6 +396,7 @@ model Report {
 
   @@index([status, createdAt])
   @@index([reportedAccountId, createdAt])
+  @@unique([reporterId, idempotencyKey])
   @@map("reports")
 }
 
@@ -439,6 +448,7 @@ model Sanction {
 model Notification {
   id        String             @id @default(uuid()) @db.Uuid
   accountId String             @map("account_id") @db.Uuid
+  dedupeKey String             @map("dedupe_key") @db.VarChar(255)
   type      String             @db.VarChar(100)
   payload   Json
   status    NotificationStatus @default(PENDING)
@@ -449,9 +459,19 @@ model Notification {
   account   Account            @relation(fields: [accountId], references: [id], onDelete: Cascade)
 
   @@index([accountId, status, createdAt])
+  @@unique([accountId, dedupeKey])
   @@map("notifications")
 }
 
+model PushPreference {
+  accountId    String   @id @map("account_id") @db.Uuid
+  inAppEnabled Boolean  @default(true) @map("in_app_enabled")
+  updatedAt    DateTime @updatedAt @map("updated_at") @db.Timestamptz(3)
+  account      Account  @relation(fields: [accountId], references: [id], onDelete: Cascade)
+
+  @@map("push_preferences")
+}
+
 model OutboxEvent {
   /// Processing lease index outbox_events_processing_lease_idx is managed in 0007 SQL.
   id            String       @id @default(uuid()) @db.Uuid

prisma/seed.ts

@@ -23,10 +23,16 @@ async function seed(): Promise<void> {
         nickname: "星河",
         publicId: "00000000-0000-4000-8000-000000000002",
       },
+      {
+        key: "demo-admin",
+        nickname: "管理员",
+        publicId: "00000000-0000-4000-8000-000000000003",
+      },
     ].map(({ key, nickname, publicId }) =>
       prisma.account.upsert({
         where: { phoneHmac: id(`drift-bottle:${key}:phone-hmac`) },
         update: {
+          role: key === "demo-admin" ? "ADMIN" : "USER",
           anonymousProfile: {
             upsert: {
               create: {
@@ -44,6 +50,7 @@ async function seed(): Promise<void> {
           },
         },
         create: {
+          role: key === "demo-admin" ? "ADMIN" : "USER",
           phoneCiphertext: Buffer.from(
             id(`drift-bottle:${key}:ciphertext`),
             "hex",
@@ -78,7 +85,7 @@ async function seed(): Promise<void> {
   });
 
   console.info(
-    "Seed complete: 2 demo accounts, 2 anonymous profiles, 1 in-pool bottle.",
+    "Seed complete: 3 demo accounts (including admin), 3 anonymous profiles, 1 in-pool bottle.",
   );
 }
 

tests/integration/seed.spec.ts

@@ -28,6 +28,6 @@ describe("seed", () => {
         prisma.anonymousProfile.count(),
         prisma.bottle.count({ where: { poolStatus: "IN_POOL" } }),
       ]),
-    ).resolves.toEqual([2, 2, 1]);
+    ).resolves.toEqual([3, 3, 1]);
   }, 30_000);
 });