fix: 加固聊天限流与安全策略串行化

apps/api/src/conversation/chat-rate-limiter.spec.ts

@@ -0,0 +1,56 @@
+import { afterEach, describe, expect, it, vi } from "vitest";
+import { RedisService } from "../redis/redis.service.js";
+import { ChatRateLimiter } from "./chat-rate-limiter.js";
+
+type FakeRedisClient = {
+  isReady: boolean;
+  isOpen: boolean;
+  on: ReturnType<typeof vi.fn>;
+  connect: ReturnType<typeof vi.fn>;
+  eval: ReturnType<typeof vi.fn>;
+  destroy: ReturnType<typeof vi.fn>;
+  quit: ReturnType<typeof vi.fn>;
+};
+
+function fakeClient(evalResult: Promise<unknown>): FakeRedisClient {
+  return {
+    isReady: true,
+    isOpen: true,
+    on: vi.fn(),
+    connect: vi.fn(() => Promise.resolve()),
+    eval: vi.fn(() => evalResult),
+    destroy: vi.fn(),
+    quit: vi.fn(() => Promise.resolve()),
+  };
+}
+
+describe("ChatRateLimiter Redis command deadline", () => {
+  const previousTimeout = process.env.REDIS_OPERATION_TIMEOUT_MS;
+
+  afterEach(() => {
+    if (previousTimeout === undefined)
+      delete process.env.REDIS_OPERATION_TIMEOUT_MS;
+    else process.env.REDIS_OPERATION_TIMEOUT_MS = previousTimeout;
+  });
+
+  it("destroys a client whose eval never settles and recovers with a new client", async () => {
+    process.env.REDIS_OPERATION_TIMEOUT_MS = "20";
+    const stuck = fakeClient(new Promise(() => undefined));
+    const healthy = fakeClient(Promise.resolve(1));
+    const factory = vi
+      .fn()
+      .mockReturnValueOnce(stuck)
+      .mockReturnValueOnce(healthy);
+    const redis = new RedisService(factory as never);
+    const limiter = new ChatRateLimiter(redis);
+
+    await expect(limiter.consume("account", "session")).rejects.toThrow(
+      "Auth Redis operation timed out",
+    );
+    expect(stuck.destroy).toHaveBeenCalledOnce();
+
+    await expect(limiter.consume("account", "session")).resolves.toBe(true);
+    expect(healthy.eval).toHaveBeenCalledOnce();
+    expect(factory).toHaveBeenCalledTimes(2);
+  });
+});

apps/api/src/conversation/chat-rate-limiter.ts

@@ -5,7 +5,6 @@ import { RedisService } from "../redis/redis.service.js";
 export class ChatRateLimiter {
   constructor(@Inject(RedisService) private readonly redis: RedisService) {}
   async consume(accountId: string, sessionId: string): Promise<boolean> {
-    await this.redis.ensureConnected();
     const prefix = process.env.REDIS_KEY_PREFIX ?? "drift:auth:";
     const max = Number(process.env.CHAT_MESSAGE_RATE_LIMIT ?? 30);
     const ttl = Number(process.env.CHAT_MESSAGE_RATE_WINDOW_SECONDS ?? 60);
@@ -13,9 +12,11 @@ export class ChatRateLimiter {
       `${prefix}chat:account:${accountId}`,
       `${prefix}chat:session:${sessionId}`,
     ];
-    const result = await this.redis.client.eval(
-      `for _,k in ipairs(KEYS) do if tonumber(redis.call('GET',k) or '0')+1>tonumber(ARGV[1]) then return 0 end end; for _,k in ipairs(KEYS) do local n=redis.call('INCR',k); if n==1 then redis.call('EXPIRE',k,ARGV[2]) end end; return 1`,
-      { keys, arguments: [String(max), String(ttl)] },
+    const result = await this.redis.executeWithDeadline((client) =>
+      client.eval(
+        `for _,k in ipairs(KEYS) do if tonumber(redis.call('GET',k) or '0')+1>tonumber(ARGV[1]) then return 0 end end; for _,k in ipairs(KEYS) do local n=redis.call('INCR',k); if n==1 then redis.call('EXPIRE',k,ARGV[2]) end end; return 1`,
+        { keys, arguments: [String(max), String(ttl)] },
+      ),
     );
     return Number(result) === 1;
   }

apps/api/src/conversation/chat.gateway.ts

@@ -99,7 +99,6 @@ export class ChatGateway
           const user = await this.auth.validateAccess(token);
           client.data.token = token;
           client.data.user = user;
-          await client.join(`account:${user.sub}`);
           next();
         } catch {
           next(new Error("Unauthorized"));
@@ -108,11 +107,14 @@ export class ChatGateway
     });
   }
 
-  handleConnection(client: ChatSocket): void {
+  async handleConnection(client: ChatSocket): Promise<void> {
     // Single-instance MVP: require a Redis adapter before horizontal scaling.
     const user = client.data.user;
-    if (!user || !client.rooms.has(`account:${user.sub}`))
+    if (!user) {
       client.disconnect(true);
+      return;
+    }
+    await client.join(`account:${user.sub}`);
   }
 
   @SubscribeMessage("message:send")

apps/api/src/conversation/conversation.e2e-spec.ts

@@ -14,6 +14,8 @@ import { leaseHmac, encryptLeaseToken } from "../auth/auth.config.js";
 import { configureApp } from "../main.js";
 import { OutboxMessageRelay } from "./outbox-message-relay.js";
 import { ChatRateLimiter } from "./chat-rate-limiter.js";
+import { ConversationService } from "./conversation.service.js";
+import { SafetyLockService } from "../safety/safety-lock.service.js";
 
 process.env.NODE_ENV = "test";
 process.env.WEB_ORIGIN = "http://localhost:3000";
@@ -281,6 +283,101 @@ describe("conversation API with real PostgreSQL", () => {
     }
   });
 
+  it("linearizes a send before a later block using canonical subject locks", async () => {
+    const author = await actor("lock-send-author");
+    const picker = await actor("lock-send-picker");
+    const { bottle, lease } = await leased(author.id, picker);
+    const created = await reply(
+      picker.authorization,
+      bottle.id,
+      lease,
+      randomUUID(),
+    ).expect(201);
+    const conversationId = created.body.data.conversationId as string;
+    const locks = app.get(SafetyLockService);
+    const originalLock = locks.lockAccounts.bind(locks);
+    let announceLocked!: () => void;
+    let releaseSend!: () => void;
+    const sendLocked = new Promise<void>(
+      (resolve) => (announceLocked = resolve),
+    );
+    const sendRelease = new Promise<void>((resolve) => (releaseSend = resolve));
+    let pause = true;
+    locks.lockAccounts = async (tx, accountIds) => {
+      await originalLock(tx, accountIds);
+      if (pause) {
+        pause = false;
+        announceLocked();
+        await sendRelease;
+      }
+    };
+
+    const send = app
+      .get(ConversationService)
+      .send(picker.id, conversationId, randomUUID(), "before block");
+    await sendLocked;
+    let blockCommitted = false;
+    const block = prisma
+      .$transaction(async (tx) => {
+        await locks.lockAccounts(tx, [author.id, picker.id]);
+        await tx.block.create({
+          data: { blockerId: author.id, blockedId: picker.id },
+        });
+      })
+      .then(() => {
+        blockCommitted = true;
+      });
+    await new Promise((resolve) => setTimeout(resolve, 30));
+    expect(blockCommitted).toBe(false);
+    releaseSend();
+    await expect(send).resolves.toMatchObject({ deduplicated: false });
+    await block;
+    locks.lockAccounts = originalLock;
+  });
+
+  it("waits for an earlier block and then rejects the message", async () => {
+    const author = await actor("lock-block-author");
+    const picker = await actor("lock-block-picker");
+    const { bottle, lease } = await leased(author.id, picker);
+    const created = await reply(
+      picker.authorization,
+      bottle.id,
+      lease,
+      randomUUID(),
+    ).expect(201);
+    const conversationId = created.body.data.conversationId as string;
+    const locks = app.get(SafetyLockService);
+    let announceLocked!: () => void;
+    let releaseBlock!: () => void;
+    const blockLocked = new Promise<void>(
+      (resolve) => (announceLocked = resolve),
+    );
+    const blockRelease = new Promise<void>(
+      (resolve) => (releaseBlock = resolve),
+    );
+    const block = prisma.$transaction(async (tx) => {
+      await locks.lockAccounts(tx, [picker.id, author.id]);
+      await tx.block.create({
+        data: { blockerId: author.id, blockedId: picker.id },
+      });
+      announceLocked();
+      await blockRelease;
+    });
+    await blockLocked;
+    let sendSettled = false;
+    const send = app
+      .get(ConversationService)
+      .send(picker.id, conversationId, randomUUID(), "after block")
+      .finally(() => {
+        sendSettled = true;
+      });
+    await new Promise((resolve) => setTimeout(resolve, 30));
+    expect(sendSettled).toBe(false);
+    releaseBlock();
+    await block;
+    await expect(send).rejects.toMatchObject({ code: "USER_BLOCKED" });
+  });
+
   it("authorizes before idempotency lookup and never exposes account ids", async () => {
     const author = await actor("author");
     const picker = await actor("picker");

apps/api/src/conversation/conversation.module.ts

@@ -2,6 +2,7 @@ import { Module } from "@nestjs/common";
 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 { ChatGateway } from "./chat.gateway.js";
 import { ChatRateLimiter } from "./chat-rate-limiter.js";
 import { ConversationController } from "./conversation.controller.js";
@@ -10,7 +11,7 @@ import { CHAT_PUBLISHER, OutboxMessageRelay } from "./outbox-message-relay.js";
 import { StateChangingOriginGuard } from "./state-changing-origin.guard.js";
 
 @Module({
-  imports: [DatabaseModule, RedisModule, AuthModule],
+  imports: [DatabaseModule, RedisModule, AuthModule, SafetyModule],
   controllers: [ConversationController],
   providers: [
     ConversationService,

apps/api/src/conversation/conversation.service.ts

@@ -4,6 +4,7 @@ import { Prisma, type PrismaClient } from "@prisma/client";
 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";
 
 type Db = Prisma.TransactionClient | PrismaClient;
 type MessageRow = {
@@ -42,7 +43,10 @@ const firstMessageInclude = {
 
 @Injectable()
 export class ConversationService {
-  constructor(@Inject(PrismaService) private readonly prisma: PrismaService) {}
+  constructor(
+    @Inject(PrismaService) private readonly prisma: PrismaService,
+    @Inject(SafetyLockService) private readonly safetyLocks: SafetyLockService,
+  ) {}
 
   async reply(
     accountId: string,
@@ -59,8 +63,7 @@ export class ConversationService {
         include: firstMessageInclude,
       });
       if (existing) {
-        await this.assertConversationMember(accountId, existing.id);
-        return this.replyResult(existing, clientMsgId, false);
+        return this.resolveExistingReply(accountId, existing, clientMsgId);
       }
       try {
         return await this.prisma.$transaction(
@@ -71,8 +74,12 @@ export class ConversationService {
               include: firstMessageInclude,
             });
             if (winner) {
-              await this.assertConversationMember(accountId, winner.id, tx);
-              return this.replyResult(winner, clientMsgId, false);
+              return this.resolveExistingReply(
+                accountId,
+                winner,
+                clientMsgId,
+                tx,
+              );
             }
             const lease = await tx.bottlePickLease.findFirst({
               where: {
@@ -99,6 +106,7 @@ export class ConversationService {
               );
             }
             const authorId = lease.bottle.authorId;
+            await this.safetyLocks.lockAccounts(tx, [accountId, authorId]);
             await this.assertAllowed(tx, accountId, authorId);
             const profiles = await tx.anonymousProfile.findMany({
               where: { accountId: { in: [accountId, authorId] } },
@@ -177,8 +185,7 @@ export class ConversationService {
           include: firstMessageInclude,
         });
         if (raced) {
-          await this.assertConversationMember(accountId, raced.id);
-          return this.replyResult(raced, clientMsgId, false);
+          return this.resolveExistingReply(accountId, raced, clientMsgId);
         }
       }
     }
@@ -189,6 +196,26 @@ export class ConversationService {
     );
   }
 
+  private async resolveExistingReply(
+    accountId: string,
+    conversation: ConversationWithFirst,
+    clientMsgId: string,
+    existingTx?: Prisma.TransactionClient,
+  ) {
+    const resolve = async (tx: Prisma.TransactionClient) => {
+      const authorized = await this.authorizeConversationMembership(
+        tx,
+        accountId,
+        conversation.id,
+      );
+      const peer = this.activePeer(authorized.members, accountId);
+      await this.safetyLocks.lockAccounts(tx, [accountId, peer.accountId]);
+      await this.assertAllowed(tx, accountId, peer.accountId);
+      return this.replyResult(conversation, clientMsgId, false);
+    };
+    return existingTx ? resolve(existingTx) : this.prisma.$transaction(resolve);
+  }
+
   private replyResult(
     conversation: ConversationWithFirst,
     clientMsgId: string,
@@ -214,11 +241,14 @@ export class ConversationService {
     this.moderate(text);
     try {
       return await this.prisma.$transaction(async (tx) => {
-        const conversation = await this.authorizeConversation(
+        const conversation = await this.authorizeConversationMembership(
           tx,
           accountId,
           conversationId,
         );
+        const peer = this.activePeer(conversation.members, accountId);
+        await this.safetyLocks.lockAccounts(tx, [accountId, peer.accountId]);
+        await this.assertAllowed(tx, accountId, peer.accountId);
         const previous = await tx.message.findUnique({
           where: {
             conversationId_clientMsgId: { conversationId, clientMsgId },
@@ -267,11 +297,14 @@ export class ConversationService {
     } catch (error) {
       if (this.isMessageIdempotencyRace(error)) {
         return this.prisma.$transaction(async (tx) => {
-          const conversation = await this.authorizeConversation(
+          const conversation = await this.authorizeConversationMembership(
             tx,
             accountId,
             conversationId,
           );
+          const peer = this.activePeer(conversation.members, accountId);
+          await this.safetyLocks.lockAccounts(tx, [accountId, peer.accountId]);
+          await this.assertAllowed(tx, accountId, peer.accountId);
           const message = await tx.message.findUniqueOrThrow({
             where: {
               conversationId_clientMsgId: { conversationId, clientMsgId },
@@ -396,7 +429,7 @@ export class ConversationService {
     return { conversationId: id, lastReadSeq: member.lastReadSeq.toString() };
   }
 
-  private async authorizeConversation(
+  private async authorizeConversationMembership(
     tx: Db,
     accountId: string,
     conversationId: string,
@@ -414,12 +447,18 @@ export class ConversationService {
     ) {
       throw this.forbidden();
     }
-    const peer = conversation.members.find(
+    return conversation;
+  }
+
+  private activePeer<T extends { accountId: string; leftAt: Date | null }>(
+    members: T[],
+    accountId: string,
+  ): T {
+    const peer = members.find(
       (member) => member.accountId !== accountId && !member.leftAt,
     );
     if (!peer) throw this.forbidden();
-    await this.assertAllowed(tx, accountId, peer.accountId);
-    return conversation;
+    return peer;
   }
 
   private async createMessageOutbox(
@@ -524,17 +563,6 @@ export class ConversationService {
     return value;
   }
 
-  private async assertConversationMember(
-    accountId: string,
-    id: string,
-    tx: Db = this.prisma,
-  ) {
-    const found = await tx.conversationMember.findUnique({
-      where: { conversationId_accountId: { conversationId: id, accountId } },
-    });
-    if (!found || found.leftAt) throw this.hidden();
-  }
-
   async assertMember(accountId: string, id: string) {
     const found = await this.prisma.conversationMember.findUnique({
       where: { conversationId_accountId: { conversationId: id, accountId } },

apps/api/src/conversation/outbox-message-relay.ts

@@ -1,6 +1,7 @@
 import {
   Inject,
   Injectable,
+  Logger,
   type OnModuleDestroy,
   type OnModuleInit,
 } from "@nestjs/common";
@@ -26,13 +27,22 @@ type ClaimedEvent = {
 
 const NEVER = new Date("9999-12-31T23:59:59.999Z");
 
+function positiveInt(value: string | undefined, fallback: number): number {
+  const parsed = Number(value);
+  return Number.isSafeInteger(parsed) && parsed > 0 ? parsed : fallback;
+}
+
 @Injectable()
 export class OutboxMessageRelay implements OnModuleInit, OnModuleDestroy {
+  private readonly logger = new Logger(OutboxMessageRelay.name);
   private stopped = false;
   private loopPromise: Promise<void> | undefined;
   private wakeLoop: (() => void) | undefined;
-  private readonly leaseMs = Number(process.env.OUTBOX_LEASE_MS ?? 30_000);
-  private readonly maxAttempts = Number(process.env.OUTBOX_MAX_ATTEMPTS ?? 5);
+  private readonly leaseMs = positiveInt(process.env.OUTBOX_LEASE_MS, 30_000);
+  private readonly maxAttempts = positiveInt(
+    process.env.OUTBOX_MAX_ATTEMPTS,
+    5,
+  );
 
   constructor(
     @Inject(PrismaService) private readonly prisma: PrismaService,
@@ -112,21 +122,25 @@ export class OutboxMessageRelay implements OnModuleInit, OnModuleDestroy {
           nextRetryAt: exhausted ? NEVER : new Date(Date.now() + delayMs),
         },
       });
-      console.error(
-        JSON.stringify({
-          eventId: event.id,
-          eventType: "MESSAGE_CREATED",
-          attempt: event.attempts,
-          errorClass: error instanceof Error ? error.name : "UnknownError",
-        }),
-      );
+      this.logger.error({
+        eventId: event.id,
+        eventType: "MESSAGE_CREATED",
+        attempt: event.attempts,
+        errorClass: error instanceof Error ? error.name : "UnknownError",
+      });
     }
     return true;
   }
 
   private async loop(): Promise<void> {
     while (!this.stopped) {
-      const handled = await this.runOnce().catch(() => false);
+      const handled = await this.runOnce().catch((error: unknown) => {
+        this.logger.error({
+          event: "Message outbox relay iteration failed",
+          errorClass: error instanceof Error ? error.name : "UnknownError",
+        });
+        return false;
+      });
       if (!handled && !this.stopped) await this.waitForWork();
     }
   }

apps/api/src/redis/redis.service.ts

@@ -74,6 +74,27 @@ export class RedisService implements OnModuleInit, OnModuleDestroy {
     }
   }
 
+  /**
+   * Runs an already-connected Redis command under the same hard deadline as
+   * connect/quit. On timeout the exact client generation is destroyed, which
+   * rejects queued commands and lets the next call create a fresh client.
+   * Callers must create the command from the provided client: retaining
+   * `redis.client` across calls would bypass generation-safe replacement.
+   */
+  async executeWithDeadline<T>(
+    operation: (client: RedisClient) => Promise<T>,
+  ): Promise<T> {
+    await this.ensureConnected();
+    const client = this.client;
+    const generation = this.generation;
+    try {
+      return await this.withDeadline(client, generation, operation(client));
+    } catch (error) {
+      this.destroyClient(client, generation);
+      throw error;
+    }
+  }
+
   async onModuleInit(): Promise<void> {
     /* Auth Redis connects lazily so unrelated modules remain isolated. */
   }

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

@@ -0,0 +1,28 @@
+import { Injectable } from "@nestjs/common";
+import { type Prisma } from "@prisma/client";
+
+/**
+ * Transaction-scoped serialization protocol for policy subjects.
+ *
+ * Every transaction that reads or writes block, sanction, or account-status
+ * policy MUST call this method first for every affected account. IDs are
+ * de-duplicated and sorted so overlapping multi-account operations cannot
+ * deadlock. The lock is released by PostgreSQL only when the transaction ends;
+ * callers must therefore pass the transaction client, never PrismaService.
+ */
+@Injectable()
+export class SafetyLockService {
+  async lockAccounts(
+    tx: Prisma.TransactionClient,
+    accountIds: readonly string[],
+  ): Promise<void> {
+    const canonicalIds = Array.from(new Set(accountIds)).sort();
+    for (const accountId of canonicalIds) {
+      await tx.$executeRaw`
+        SELECT pg_advisory_xact_lock(
+          hashtextextended(${`safety:${accountId}`}, 0)
+        )
+      `;
+    }
+  }
+}

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

@@ -0,0 +1,8 @@
+import { Module } from "@nestjs/common";
+import { SafetyLockService } from "./safety-lock.service.js";
+
+@Module({
+  providers: [SafetyLockService],
+  exports: [SafetyLockService],
+})
+export class SafetyModule {}