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 --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",
     "typecheck": "tsc --noEmit -p tsconfig.json"
   },
   "dependencies": {
@@ -15,6 +15,8 @@
     "@nestjs/common": "^11.1.6",
     "@nestjs/core": "^11.1.6",
     "@nestjs/platform-express": "^11.1.6",
+    "@nestjs/platform-socket.io": "^11.1.6",
+    "@nestjs/websockets": "^11.1.6",
     "@prisma/client": "6.19.0",
     "class-transformer": "^0.5.1",
     "class-validator": "^0.14.2",
@@ -22,12 +24,14 @@
     "helmet": "^8.1.0",
     "redis": "^5.8.2",
     "reflect-metadata": "^0.2.2",
-    "rxjs": "^7.8.2"
+    "rxjs": "^7.8.2",
+    "socket.io": "^4.8.1"
   },
   "devDependencies": {
     "@nestjs/testing": "^11.1.6",
     "@types/express": "^5.0.3",
     "@types/supertest": "^6.0.3",
+    "socket.io-client": "^4.8.1",
     "supertest": "^7.1.4",
     "typescript": "^5.6.3",
     "vitest": "^4.1.11"

apps/api/src/app.module.ts

@@ -5,9 +5,17 @@ import { AuthModule } from "./auth/auth.module.js";
 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";
 
 @Module({
-  imports: [HealthModule, AuthModule, ProfileModule, BottleModule, MatchModule],
+  imports: [
+    HealthModule,
+    AuthModule,
+    ProfileModule,
+    BottleModule,
+    MatchModule,
+    ConversationModule,
+  ],
 })
 export class AppModule implements NestModule {
   configure(consumer: MiddlewareConsumer): void {

apps/api/src/auth/auth.guard.ts

@@ -9,6 +9,7 @@ import { ErrorCode } from "@drift/contracts";
 import type { Request } from "express";
 import { DomainException } from "../common/domain.exception.js";
 import { PrismaService } from "../database/prisma.service.js";
+import type { AccessClaims } from "./token.service.js";
 import { TokenService } from "./token.service.js";
 import type { AuthenticatedRequest } from "./current-user.decorator.js";
 @Injectable()
@@ -17,11 +18,8 @@ export class AuthGuard implements CanActivate {
     @Inject(PrismaService) private readonly prisma: PrismaService,
     @Inject(TokenService) private readonly tokens: TokenService,
   ) {}
-  async canActivate(ctx: ExecutionContext): Promise<boolean> {
+  async validateAccess(token: string): Promise<AccessClaims> {
-    const req = ctx.switchToHttp().getRequest<Request & AuthenticatedRequest>();
+    const claims = this.tokens.verifyAccess(token);
-    const auth = req.headers.authorization;
-    if (!auth?.startsWith("Bearer ")) throw this.denied();
-    const claims = this.tokens.verifyAccess(auth.slice(7));
     const session = await this.prisma.session.findUnique({
       where: { id: claims.session_id },
       include: { account: true },
@@ -37,7 +35,14 @@ export class AuthGuard implements CanActivate {
       session.account.tokenVersion !== claims.token_version
     )
       throw this.denied();
-    req.user = claims;
+    return claims;
+  }
+
+  async canActivate(ctx: ExecutionContext): Promise<boolean> {
+    const req = ctx.switchToHttp().getRequest<Request & AuthenticatedRequest>();
+    const auth = req.headers.authorization;
+    if (!auth?.startsWith("Bearer ")) throw this.denied();
+    req.user = await this.validateAccess(auth.slice(7));
     return true;
   }
   private denied() {

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

@@ -0,0 +1,22 @@
+import { Inject, Injectable } from "@nestjs/common";
+import { RedisService } from "../redis/redis.service.js";
+
+@Injectable()
+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);
+    const keys = [
+      `${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)] },
+    );
+    return Number(result) === 1;
+  }
+}

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

@@ -0,0 +1,243 @@
+import { HttpStatus, Inject } from "@nestjs/common";
+import { plainToInstance } from "class-transformer";
+import { validate } from "class-validator";
+import { ErrorCode } from "@drift/contracts";
+import {
+  ConnectedSocket,
+  MessageBody,
+  type OnGatewayConnection,
+  type OnGatewayInit,
+  SubscribeMessage,
+  WebSocketGateway,
+  WebSocketServer,
+} from "@nestjs/websockets";
+import type { IncomingMessage } from "node:http";
+import type { Server, Socket } from "socket.io";
+import { webOrigin } from "../auth/auth.config.js";
+import { AuthGuard } from "../auth/auth.guard.js";
+import type { AccessClaims } from "../auth/token.service.js";
+import { DomainException } from "../common/domain.exception.js";
+import { ChatRateLimiter } from "./chat-rate-limiter.js";
+import {
+  ConversationService,
+  type MessageResult,
+} from "./conversation.service.js";
+import { SendMessageDto, SocketReadDto } from "./dto.js";
+
+interface SocketData {
+  token?: string;
+  user?: AccessClaims;
+}
+interface ClientToServerEvents {
+  "message:send": (dto: SendMessageDto, ack?: Ack) => void;
+  "conversation:read": (dto: SocketReadDto, ack?: Ack) => void;
+}
+interface ServerToClientEvents {
+  "message:ack": (value: SocketResponse) => void;
+  "message:new": (value: MessageResult["message"]) => void;
+}
+type ChatServer = Server<
+  ClientToServerEvents,
+  ServerToClientEvents,
+  Record<string, never>,
+  SocketData
+>;
+type ChatSocket = Socket<
+  ClientToServerEvents,
+  ServerToClientEvents,
+  Record<string, never>,
+  SocketData
+>;
+type SocketResponse =
+  | { ok: true; data: unknown }
+  | { ok: false; error: { code: ErrorCode; message: string } };
+type Ack = (value: SocketResponse) => void;
+type HandshakeAuth = { token?: unknown; authorization?: unknown };
+
+function allowOrigin(
+  request: IncomingMessage,
+  callback: (error: string | null | undefined, success: boolean) => void,
+): void {
+  const origin = request.headers.origin;
+  callback(null, origin === undefined || origin === webOrigin());
+}
+
+@WebSocketGateway({
+  namespace: "/chat",
+  allowRequest: allowOrigin,
+  cors: {
+    origin(origin, callback) {
+      callback(null, origin === undefined || origin === webOrigin());
+    },
+    credentials: true,
+  },
+})
+export class ChatGateway implements OnGatewayInit, OnGatewayConnection {
+  @WebSocketServer() server!: ChatServer;
+
+  constructor(
+    @Inject(AuthGuard) private readonly auth: AuthGuard,
+    @Inject(ConversationService)
+    private readonly conversations: ConversationService,
+    @Inject(ChatRateLimiter) private readonly limiter: ChatRateLimiter,
+  ) {}
+
+  afterInit(server: ChatServer): void {
+    server.use((client, next) => {
+      void (async () => {
+        try {
+          const auth = client.handshake.auth as unknown as HandshakeAuth;
+          const raw = auth.token ?? auth.authorization;
+          const token =
+            typeof raw === "string" && raw.startsWith("Bearer ")
+              ? raw.slice(7)
+              : raw;
+          if (typeof token !== "string") throw new Error("Unauthorized");
+          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"));
+        }
+      })();
+    });
+  }
+
+  handleConnection(client: ChatSocket): void {
+    // Single-instance MVP: require a Redis adapter before horizontal scaling.
+    const user = client.data.user;
+    if (!user || !client.rooms.has(`account:${user.sub}`))
+      client.disconnect(true);
+  }
+
+  @SubscribeMessage("message:send")
+  async send(
+    @ConnectedSocket() client: ChatSocket,
+    @MessageBody() dto: SendMessageDto,
+    ack?: Ack,
+  ): Promise<SocketResponse> {
+    try {
+      dto = await this.validatePayload(SendMessageDto, dto);
+      const user = await this.auth.validateAccess(client.data.token ?? "");
+      let allowed: boolean;
+      try {
+        allowed = await this.limiter.consume(user.sub, user.session_id);
+      } catch {
+        throw new DomainException(
+          ErrorCode.SERVICE_UNAVAILABLE,
+          "Service unavailable",
+          HttpStatus.SERVICE_UNAVAILABLE,
+        );
+      }
+      if (!allowed) {
+        throw new DomainException(
+          ErrorCode.RATE_LIMITED,
+          "Too many requests",
+          HttpStatus.TOO_MANY_REQUESTS,
+        );
+      }
+      const result = await this.conversations.send(
+        user.sub,
+        dto.conversationId,
+        dto.clientMsgId,
+        dto.text,
+      );
+      const response: SocketResponse = {
+        ok: true,
+        data: {
+          message: result.message,
+          deduplicated: result.deduplicated,
+        },
+      };
+      ack?.(response);
+      client.emit("message:ack", response);
+      if (!result.deduplicated) {
+        for (const member of result.memberIds) {
+          this.server
+            .to(`account:${member}`)
+            .emit("message:new", result.message);
+        }
+      }
+      return response;
+    } catch (error) {
+      const response = this.error(error);
+      ack?.(response);
+      client.emit("message:ack", response);
+      return response;
+    }
+  }
+
+  @SubscribeMessage("conversation:read")
+  async read(
+    @ConnectedSocket() client: ChatSocket,
+    @MessageBody() body: SocketReadDto,
+    ack?: Ack,
+  ): Promise<SocketResponse> {
+    try {
+      body = await this.validatePayload(SocketReadDto, body);
+      const user = await this.auth.validateAccess(client.data.token ?? "");
+      const data = await this.conversations.read(
+        user.sub,
+        body.conversationId,
+        body.seq,
+      );
+      const response: SocketResponse = { ok: true, data };
+      ack?.(response);
+      return response;
+    } catch (error) {
+      const response = this.error(error);
+      ack?.(response);
+      return response;
+    }
+  }
+
+  private error(error: unknown): SocketResponse {
+    if (error instanceof DomainException) {
+      return {
+        ok: false,
+        error: { code: error.code, message: error.message },
+      };
+    }
+    return {
+      ok: false,
+      error: {
+        code: ErrorCode.INTERNAL_ERROR,
+        message: "Internal server error",
+      },
+    };
+  }
+
+  private async validatePayload<T extends object>(
+    type: new () => T,
+    value: unknown,
+  ): Promise<T> {
+    const dto = plainToInstance(type, value);
+    const errors = await validate(dto, {
+      whitelist: true,
+      forbidNonWhitelisted: true,
+    });
+    if (errors.length) {
+      throw new DomainException(
+        ErrorCode.VALIDATION_ERROR,
+        "Validation failed",
+        HttpStatus.BAD_REQUEST,
+      );
+    }
+    return dto;
+  }
+}
+
+Reflect.defineMetadata(
+  "design:paramtypes",
+  [Object, SendMessageDto, Function],
+  ChatGateway.prototype,
+  "send",
+);
+Reflect.defineMetadata(
+  "design:paramtypes",
+  [Object, SocketReadDto, Function],
+  ChatGateway.prototype,
+  "read",
+);

apps/api/src/conversation/conversation.controller.ts

@@ -0,0 +1,119 @@
+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 { ConversationService } from "./conversation.service.js";
+import {
+  ConversationsQueryDto,
+  MessagesQueryDto,
+  ReadConversationDto,
+  ReplyDto,
+  SendMessageDto,
+} from "./dto.js";
+
+@Controller()
+@UseGuards(AuthGuard)
+export class ConversationController {
+  constructor(
+    @Inject(ConversationService)
+    private readonly conversations: ConversationService,
+  ) {}
+  @Post("bottles/:id/reply") reply(
+    @CurrentUser() user: AccessClaims,
+    @Param("id", new ParseUUIDPipe()) id: string,
+    @Body() dto: ReplyDto,
+  ) {
+    return this.conversations.reply(
+      user.sub,
+      id,
+      dto.leaseId,
+      dto.leaseToken,
+      dto.clientMsgId,
+      dto.text,
+    );
+  }
+  @Get("conversations") list(
+    @CurrentUser() user: AccessClaims,
+    @Query() query: ConversationsQueryDto,
+  ) {
+    return this.conversations.list(user.sub, query.cursor, query.limit);
+  }
+  @Get("conversations/:id") detail(
+    @CurrentUser() user: AccessClaims,
+    @Param("id", new ParseUUIDPipe()) id: string,
+  ) {
+    return this.conversations.detail(user.sub, id);
+  }
+  @Get("conversations/:id/messages") messages(
+    @CurrentUser() user: AccessClaims,
+    @Param("id", new ParseUUIDPipe()) id: string,
+    @Query() query: MessagesQueryDto,
+  ) {
+    return this.conversations.messages(
+      user.sub,
+      id,
+      query.afterSeq,
+      query.limit,
+    );
+  }
+  @Post("conversations/:id/messages/prepare") prepare(
+    @CurrentUser() user: AccessClaims,
+    @Param("id", new ParseUUIDPipe()) id: string,
+    @Body() dto: SendMessageDto,
+  ) {
+    return this.conversations.send(user.sub, id, dto.clientMsgId, dto.text);
+  }
+  @Post("conversations/:id/read") read(
+    @CurrentUser() user: AccessClaims,
+    @Param("id", new ParseUUIDPipe()) id: string,
+    @Body() dto: ReadConversationDto,
+  ) {
+    return this.conversations.read(user.sub, id, dto.seq);
+  }
+}
+Reflect.defineMetadata(
+  "design:paramtypes",
+  [Object, String, ReplyDto],
+  ConversationController.prototype,
+  "reply",
+);
+Reflect.defineMetadata(
+  "design:paramtypes",
+  [Object, ConversationsQueryDto],
+  ConversationController.prototype,
+  "list",
+);
+Reflect.defineMetadata(
+  "design:paramtypes",
+  [Object, String],
+  ConversationController.prototype,
+  "detail",
+);
+Reflect.defineMetadata(
+  "design:paramtypes",
+  [Object, String, MessagesQueryDto],
+  ConversationController.prototype,
+  "messages",
+);
+Reflect.defineMetadata(
+  "design:paramtypes",
+  [Object, String, SendMessageDto],
+  ConversationController.prototype,
+  "prepare",
+);
+Reflect.defineMetadata(
+  "design:paramtypes",
+  [Object, String, ReadConversationDto],
+  ConversationController.prototype,
+  "read",
+);

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

@@ -0,0 +1,638 @@
+/* eslint-disable @typescript-eslint/no-unsafe-member-access, @typescript-eslint/no-unsafe-assignment, @typescript-eslint/no-unsafe-argument, @typescript-eslint/no-unsafe-call, @typescript-eslint/no-unsafe-return */
+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 { io, type Socket } from "socket.io-client";
+import { afterAll, beforeAll, beforeEach, describe, expect, it } from "vitest";
+import { assertSafeTestDatabaseUrl } from "../../../../prisma/database-safety.js";
+import { AppModule } from "../app.module.js";
+import { TokenService } from "../auth/token.service.js";
+import { leaseHmac, encryptLeaseToken } from "../auth/auth.config.js";
+import { configureApp } from "../main.js";
+
+process.env.NODE_ENV = "test";
+process.env.WEB_ORIGIN = "http://localhost:3000";
+process.env.PHONE_ENCRYPTION_KEY =
+  "AAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAAA=";
+process.env.PHONE_HMAC_KEY = "test-phone-hmac-key-with-at-least-32-bytes";
+process.env.VERIFICATION_CODE_HMAC_KEY =
+  "test-code-hmac-key-with-at-least-32-bytes";
+process.env.JWT_SECRET = "test-jwt-secret-with-at-least-thirty-two-bytes";
+process.env.REFRESH_TOKEN_HMAC_KEY =
+  "test-refresh-hmac-key-with-at-least-32-bytes";
+process.env.LEASE_TOKEN_HMAC_KEY = "test-lease-hmac-key-with-at-least-32-bytes";
+process.env.LEASE_TOKEN_ENCRYPTION_KEY =
+  "BBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBBB=";
+process.env.MODERATION_REJECT_WORDS = "reject-word";
+process.env.MODERATION_REVIEW_WORDS = "review-word";
+
+const prisma = new PrismaClient();
+
+describe("conversation API with real PostgreSQL", () => {
+  let app: INestApplication;
+  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.listen(0, "127.0.0.1");
+  });
+  beforeEach(async () => {
+    await prisma.$executeRawUnsafe(`TRUNCATE TABLE "accounts" CASCADE`);
+  });
+  afterAll(async () => {
+    await app?.close();
+    await prisma.$disconnect();
+  });
+
+  async function actor(deviceId: string) {
+    const account = await prisma.account.create({
+      data: {
+        phoneCiphertext: Buffer.from("cipher"),
+        phoneHmac: randomUUID(),
+        anonymousProfile: {
+          create: {
+            nickname: `anon-${deviceId}`,
+            avatarColor: "#123456",
+            reviewStatus: "APPROVED",
+          },
+        },
+      },
+    });
+    const session = await prisma.session.create({
+      data: {
+        accountId: account.id,
+        refreshTokenHash: randomUUID(),
+        deviceId: deviceId.padEnd(8, "x"),
+        expiresAt: new Date(Date.now() + 600_000),
+        tokenFamily: randomUUID(),
+      },
+    });
+    const token = app.get(TokenService).issueAccess({
+      sub: account.id,
+      session_id: session.id,
+      device_id: deviceId.padEnd(8, "x"),
+      scopes: ["user"],
+      token_version: 0,
+    });
+    return {
+      id: account.id,
+      sessionId: session.id,
+      authorization: ["Bear", "er ", token].join(""),
+      token,
+    };
+  }
+  async function leased(authorId: string, picker: { id: string }) {
+    const bottle = await prisma.bottle.create({
+      data: {
+        authorId,
+        clientRequestId: randomUUID(),
+        contentText: "source",
+        reviewStatus: "APPROVED",
+        poolStatus: "IN_POOL",
+        approvedAt: new Date(),
+      },
+    });
+    const token = randomUUID().replaceAll("-", "") + "abcdefghijk";
+    const leaseId = randomUUID();
+    const lease = await prisma.bottlePickLease.create({
+      data: {
+        id: leaseId,
+        bottleId: bottle.id,
+        pickerId: picker.id,
+        leaseTokenHash: leaseHmac(token),
+        leaseTokenCiphertext: encryptLeaseToken(token, leaseId),
+        expiresAt: new Date(Date.now() + 60_000),
+      },
+    });
+    await prisma.bottle.update({
+      where: { id: bottle.id },
+      data: { poolStatus: "LEASED", activeLeaseId: lease.id },
+    });
+    return { bottle, lease: { id: lease.id, token } };
+  }
+  const reply = (
+    authorization: string,
+    bottleId: string,
+    lease: { id: string; token: string },
+    clientMsgId: string,
+    text = "hello",
+  ) =>
+    request(app.getHttpServer())
+      .post(`/api/v1/bottles/${bottleId}/reply`)
+      .set("Authorization", authorization)
+      .send({ leaseId: lease.id, leaseToken: lease.token, clientMsgId, text });
+
+  function socket(token: string, origin = process.env.WEB_ORIGIN): Socket {
+    const address = app.getHttpServer().address() as { port: number };
+    return io(`http://127.0.0.1:${address.port}/chat`, {
+      auth: { token },
+      ...(origin ? { extraHeaders: { Origin: origin } } : {}),
+      transports: ["websocket"],
+      forceNew: true,
+      reconnection: false,
+    });
+  }
+
+  function connected(client: Socket): Promise<void> {
+    return new Promise((resolve, reject) => {
+      client.once("connect", resolve);
+      client.once("connect_error", reject);
+    });
+  }
+
+  function ack<T>(client: Socket, event: string, body: unknown): Promise<T> {
+    return new Promise((resolve, reject) => {
+      const timer = setTimeout(
+        () => reject(new Error(`ACK timeout: ${event}`)),
+        2_000,
+      );
+      client.emit(event, body, (value: T) => {
+        clearTimeout(timer);
+        resolve(value);
+      });
+    });
+  }
+
+  it("atomically consumes a lease and deduplicates concurrent first replies", async () => {
+    const author = await actor("author");
+    const picker = await actor("picker");
+    const { bottle, lease } = await leased(author.id, picker);
+    const clientMsgId = randomUUID();
+    const results = await Promise.all(
+      Array.from({ length: 8 }, () =>
+        reply(picker.authorization, bottle.id, lease, clientMsgId),
+      ),
+    );
+    if (!results.every((x) => x.status === 201))
+      console.error(results.map((x) => [x.status, x.body]));
+    expect(results.every((x) => x.status === 201)).toBe(true);
+    expect(new Set(results.map((x) => x.body.data.conversationId)).size).toBe(
+      1,
+    );
+    expect(
+      await prisma.conversation.count({ where: { sourceBottleId: bottle.id } }),
+    ).toBe(1);
+    expect(await prisma.message.count()).toBe(1);
+    expect(await prisma.conversationMember.count()).toBe(2);
+    expect(
+      await prisma.bottle.findUniqueOrThrow({ where: { id: bottle.id } }),
+    ).toMatchObject({ poolStatus: "CONSUMED", activeLeaseId: null });
+    expect(
+      await prisma.bottlePickLease.findUniqueOrThrow({
+        where: { id: lease.id },
+      }),
+    ).toMatchObject({ status: "CONSUMED" });
+    expect((await prisma.message.findFirstOrThrow()).seq).toBe(1n);
+    expect((await prisma.conversation.findFirstOrThrow()).nextSeq).toBe(2n);
+  });
+
+  it("allows only one first message when distinct client ids race", async () => {
+    const author = await actor("author");
+    const picker = await actor("picker");
+    const { bottle, lease } = await leased(author.id, picker);
+    const results = await Promise.all([
+      reply(picker.authorization, bottle.id, lease, randomUUID(), "one"),
+      reply(picker.authorization, bottle.id, lease, randomUUID(), "two"),
+    ]);
+    if (!results.every((x) => x.status === 201))
+      console.error(results.map((x) => [x.status, x.body]));
+    expect(results.every((x) => x.status === 201)).toBe(true);
+    expect(new Set(results.map((x) => x.body.data.conversationId)).size).toBe(
+      1,
+    );
+    expect(await prisma.message.count()).toBe(1);
+    expect(results.filter((x) => x.body.data.won).length).toBe(1);
+    expect(results.filter((x) => !x.body.data.won).length).toBe(1);
+    expect(results.find((x) => !x.body.data.won)!.body.data).toMatchObject({
+      created: false,
+      deduplicated: false,
+    });
+  });
+
+  it("rejects invalid, expired and foreign leases without mutation", async () => {
+    const author = await actor("author");
+    const picker = await actor("picker");
+    const stranger = await actor("stranger");
+    const { bottle, lease } = await leased(author.id, picker);
+    await reply(stranger.authorization, bottle.id, lease, randomUUID()).expect(
+      404,
+    );
+    await reply(
+      picker.authorization,
+      bottle.id,
+      { ...lease, token: "x".repeat(43) },
+      randomUUID(),
+    ).expect(404);
+    await prisma.bottlePickLease.update({
+      where: { id: lease.id },
+      data: { expiresAt: new Date(Date.now() - 1) },
+    });
+    await reply(picker.authorization, bottle.id, lease, randomUUID()).expect(
+      410,
+    );
+    expect(await prisma.conversation.count()).toBe(0);
+  });
+
+  it("fails closed before state changes for blocks, sanctions and moderation", async () => {
+    const cases = ["block", "sanction", "reject", "review"] as const;
+    for (const kind of cases) {
+      await prisma.$executeRawUnsafe(`TRUNCATE TABLE "accounts" CASCADE`);
+      const author = await actor(`author-${kind}`);
+      const picker = await actor(`picker-${kind}`);
+      const { bottle, lease } = await leased(author.id, picker);
+      if (kind === "block")
+        await prisma.block.create({
+          data: { blockerId: author.id, blockedId: picker.id },
+        });
+      if (kind === "sanction")
+        await prisma.sanction.create({
+          data: { accountId: picker.id, type: "SUSPENSION", reason: "policy" },
+        });
+      const text =
+        kind === "reject"
+          ? "reject-word"
+          : kind === "review"
+            ? "review-word"
+            : "hello";
+      const response = await reply(
+        picker.authorization,
+        bottle.id,
+        lease,
+        randomUUID(),
+        text,
+      );
+      expect(response.status).toBe(403);
+      expect(await prisma.conversation.count()).toBe(0);
+      expect(
+        (await prisma.bottle.findUniqueOrThrow({ where: { id: bottle.id } }))
+          .poolStatus,
+      ).toBe("LEASED");
+    }
+  });
+
+  it("authorizes before idempotency lookup and never exposes account ids", async () => {
+    const author = await actor("author");
+    const picker = await actor("picker");
+    const stranger = await actor("stranger");
+    const { bottle, lease } = await leased(author.id, picker);
+    const clientMsgId = randomUUID();
+    const created = await reply(
+      picker.authorization,
+      bottle.id,
+      lease,
+      clientMsgId,
+    ).expect(201);
+    const conversationId = created.body.data.conversationId as string;
+
+    await request(app.getHttpServer())
+      .post(`/api/v1/conversations/${conversationId}/messages/prepare`)
+      .set("Authorization", stranger.authorization)
+      .send({ conversationId, clientMsgId, text: "steal" })
+      .expect(403);
+
+    const history = await request(app.getHttpServer())
+      .get(`/api/v1/conversations/${conversationId}/messages`)
+      .set("Authorization", picker.authorization)
+      .expect(200);
+    expect(history.body.data.items[0].senderId).toBeUndefined();
+    expect(history.body.data.items[0].sender).toEqual({
+      publicId: expect.any(String),
+    });
+    expect(JSON.stringify(history.body)).not.toContain(picker.id);
+  });
+
+  it("writes a text-free message outbox event in the same transaction", async () => {
+    const author = await actor("author");
+    const picker = await actor("picker");
+    const { bottle, lease } = await leased(author.id, picker);
+    const result = await reply(
+      picker.authorization,
+      bottle.id,
+      lease,
+      randomUUID(),
+      "private message text",
+    ).expect(201);
+    const messageId = result.body.data.message.id as string;
+    const event = await prisma.outboxEvent.findUniqueOrThrow({
+      where: { dedupeKey: `message-created:${messageId}` },
+    });
+    expect(event.eventType).toBe("MESSAGE_CREATED");
+    expect(event.payload).toMatchObject({ messageId });
+    expect(JSON.stringify(event.payload)).not.toContain("private message text");
+  });
+
+  it("uses an exact timestamp/id cursor without skipping tied rows", async () => {
+    const owner = await actor("owner");
+    const peer = await actor("peer");
+    const tiedAt = new Date("2026-01-02T03:04:05.000Z");
+    const ids = [randomUUID(), randomUUID(), randomUUID()].sort().reverse();
+    for (const id of ids) {
+      await prisma.conversation.create({
+        data: {
+          id,
+          sourceBottle: {
+            create: {
+              authorId: peer.id,
+              clientRequestId: randomUUID(),
+              contentText: "source",
+              reviewStatus: "APPROVED",
+              poolStatus: "CONSUMED",
+              approvedAt: new Date(),
+            },
+          },
+          lastMessageAt: tiedAt,
+          members: {
+            create: [
+              { accountId: owner.id, peerAliasSnapshot: "peer" },
+              { accountId: peer.id, peerAliasSnapshot: "owner" },
+            ],
+          },
+        },
+      });
+    }
+    const first = await request(app.getHttpServer())
+      .get("/api/v1/conversations?limit=2")
+      .set("Authorization", owner.authorization)
+      .expect(200);
+    const second = await request(app.getHttpServer())
+      .get(`/api/v1/conversations?limit=2&cursor=${first.body.data.nextCursor}`)
+      .set("Authorization", owner.authorization)
+      .expect(200);
+    expect([
+      ...first.body.data.items.map((x: { id: string }) => x.id),
+      ...second.body.data.items.map((x: { id: string }) => x.id),
+    ]).toEqual(ids);
+    await request(app.getHttpServer())
+      .get("/api/v1/conversations?cursor=e30")
+      .set("Authorization", owner.authorization)
+      .expect(400);
+  });
+
+  it("validates sequence bounds and forbids departed members before reads", async () => {
+    const author = await actor("author");
+    const picker = await actor("picker");
+    const { bottle, lease } = await leased(author.id, picker);
+    const created = await reply(
+      picker.authorization,
+      bottle.id,
+      lease,
+      randomUUID(),
+    ).expect(201);
+    const id = created.body.data.conversationId as string;
+    for (const bad of ["-1", "9223372036854775808", "not-a-number"]) {
+      await request(app.getHttpServer())
+        .get(`/api/v1/conversations/${id}/messages?afterSeq=${bad}`)
+        .set("Authorization", picker.authorization)
+        .expect(400);
+    }
+    await request(app.getHttpServer())
+      .post(`/api/v1/conversations/${id}/read`)
+      .set("Authorization", picker.authorization)
+      .send({ seq: "1" })
+      .expect(201);
+    const monotonic = await request(app.getHttpServer())
+      .post(`/api/v1/conversations/${id}/read`)
+      .set("Authorization", picker.authorization)
+      .send({ seq: "0" })
+      .expect(201);
+    expect(monotonic.body.data.lastReadSeq).toBe("1");
+    await prisma.conversationMember.update({
+      where: {
+        conversationId_accountId: { conversationId: id, accountId: picker.id },
+      },
+      data: { leftAt: new Date() },
+    });
+    await request(app.getHttpServer())
+      .post(`/api/v1/conversations/${id}/read`)
+      .set("Authorization", picker.authorization)
+      .send({ seq: "1" })
+      .expect(403);
+  });
+
+  it("authenticates both socket ends, broadcasts once, revokes live sessions and repairs history", async () => {
+    const author = await actor("author");
+    const picker = await actor("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 unauthenticated = socket("invalid-token");
+    const denied = new Promise<string>((resolve) =>
+      unauthenticated.once("connect_error", (error) => resolve(error.message)),
+    );
+    expect(await denied).toBe("Unauthorized");
+    unauthenticated.close();
+
+    const badOrigin = socket(author.token, "https://evil.example");
+    const originDenied = new Promise<void>((resolve) =>
+      badOrigin.once("connect_error", () => resolve()),
+    );
+    await originDenied;
+    badOrigin.close();
+
+    const authorSocket = socket(`Bearer ${author.token}`);
+    const pickerSocket = socket(picker.token);
+    await Promise.all([connected(authorSocket), connected(pickerSocket)]);
+    const clientMsgId = randomUUID();
+    let authorDeliveries = 0;
+    let pickerDeliveries = 0;
+    authorSocket.on("message:new", () => authorDeliveries++);
+    pickerSocket.on("message:new", () => pickerDeliveries++);
+    const first = await ack<{
+      ok: boolean;
+      data: { message: { seq: string }; deduplicated: boolean };
+    }>(pickerSocket, "message:send", {
+      conversationId,
+      clientMsgId,
+      text: "socket hello",
+    });
+    expect(first).toMatchObject({ ok: true, data: { deduplicated: false } });
+    await new Promise((resolve) => setTimeout(resolve, 50));
+    expect([authorDeliveries, pickerDeliveries]).toEqual([1, 1]);
+    const duplicate = await ack<{ data: { deduplicated: boolean } }>(
+      pickerSocket,
+      "message:send",
+      { conversationId, clientMsgId, text: "socket hello" },
+    );
+    expect(duplicate.data.deduplicated).toBe(true);
+    await new Promise((resolve) => setTimeout(resolve, 50));
+    expect([authorDeliveries, pickerDeliveries]).toEqual([1, 1]);
+
+    pickerSocket.close();
+    const offlineId = randomUUID();
+    await ack(authorSocket, "message:send", {
+      conversationId,
+      clientMsgId: offlineId,
+      text: "while offline",
+    });
+    const history = await request(app.getHttpServer())
+      .get(
+        `/api/v1/conversations/${conversationId}/messages?afterSeq=${first.data.message.seq}`,
+      )
+      .set("Authorization", picker.authorization)
+      .expect(200);
+    expect(
+      history.body.data.items.map(
+        (item: { clientMsgId: string }) => item.clientMsgId,
+      ),
+    ).toEqual([offlineId]);
+
+    await prisma.session.update({
+      where: { id: author.sessionId },
+      data: { revokedAt: new Date() },
+    });
+    const revoked = await ack<{ ok: boolean; error: { code: string } }>(
+      authorSocket,
+      "message:send",
+      {
+        conversationId,
+        clientMsgId: randomUUID(),
+        text: "must fail",
+      },
+    );
+    expect(revoked).toMatchObject({
+      ok: false,
+      error: { code: "AUTH_UNAUTHORIZED" },
+    });
+    authorSocket.close();
+  });
+
+  it("always ACKs invalid socket payloads with a sanitized validation error", async () => {
+    const actorOne = await actor("socket-validation");
+    const client = socket(actorOne.token);
+    await connected(client);
+    for (const [event, payload] of [
+      ["message:send", { conversationId: "not-a-uuid", extra: true }],
+      [
+        "message:send",
+        {
+          conversationId: randomUUID(),
+          clientMsgId: randomUUID(),
+          text: "valid except extra",
+          extra: true,
+        },
+      ],
+      [
+        "conversation:read",
+        { conversationId: randomUUID(), seq: "0", extra: true },
+      ],
+      ["conversation:read", { conversationId: randomUUID(), seq: -1 }],
+    ] as const) {
+      const result = await ack<{
+        ok: boolean;
+        error: { code: string; message: string };
+      }>(client, event, payload);
+      expect(result).toEqual({
+        ok: false,
+        error: { code: "VALIDATION_ERROR", message: "Validation failed" },
+      });
+    }
+    expect(client.connected).toBe(true);
+    client.close();
+  });
+
+  it("sanitizes unknown socket failures and blocks nonmembers, blocked peers and rejected content", async () => {
+    const author = await actor("socket-author");
+    const picker = await actor("socket-picker");
+    const stranger = await actor("socket-stranger");
+    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 strangerSocket = socket(stranger.token);
+    const pickerSocket = socket(picker.token);
+    await Promise.all([connected(strangerSocket), connected(pickerSocket)]);
+
+    await expect(
+      ack(strangerSocket, "message:send", {
+        conversationId,
+        clientMsgId: randomUUID(),
+        text: "not mine",
+      }),
+    ).resolves.toMatchObject({
+      ok: false,
+      error: { code: "CONVERSATION_FORBIDDEN" },
+    });
+    await expect(
+      ack(pickerSocket, "message:send", {
+        conversationId,
+        clientMsgId: randomUUID(),
+        text: "reject-word",
+      }),
+    ).resolves.toMatchObject({
+      ok: false,
+      error: { code: "CONTENT_REJECTED" },
+    });
+    await prisma.block.create({
+      data: { blockerId: author.id, blockedId: picker.id },
+    });
+    await expect(
+      ack(pickerSocket, "message:send", {
+        conversationId,
+        clientMsgId: randomUUID(),
+        text: "blocked",
+      }),
+    ).resolves.toMatchObject({
+      ok: false,
+      error: { code: "USER_BLOCKED" },
+    });
+
+    const gateway: { conversations: { read: () => Promise<never> } } = app.get(
+      (await import("./chat.gateway.js")).ChatGateway,
+    );
+    const originalRead = gateway.conversations.read;
+    gateway.conversations.read = () =>
+      Promise.reject(new Error("database secret"));
+    const unknown = await ack<{
+      ok: boolean;
+      error: { code: string; message: string };
+    }>(pickerSocket, "conversation:read", { conversationId, seq: "1" });
+    gateway.conversations.read = originalRead;
+    expect(unknown).toEqual({
+      ok: false,
+      error: { code: "INTERNAL_ERROR", message: "Internal server error" },
+    });
+    strangerSocket.close();
+    pickerSocket.close();
+  });
+
+  it("keeps the sender public id captured when the message was written", async () => {
+    const author = await actor("snapshot-author");
+    const picker = await actor("snapshot-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 originalPublicId = created.body.data.message.sender
+      .publicId as string;
+    await prisma.anonymousProfile.update({
+      where: { accountId: picker.id },
+      data: { publicId: randomUUID() },
+    });
+    const history = await request(app.getHttpServer())
+      .get(`/api/v1/conversations/${conversationId}/messages`)
+      .set("Authorization", picker.authorization)
+      .expect(200);
+    expect(history.body.data.items[0].sender.publicId).toBe(originalPublicId);
+  });
+});

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

@@ -0,0 +1,16 @@
+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 { ChatGateway } from "./chat.gateway.js";
+import { ChatRateLimiter } from "./chat-rate-limiter.js";
+import { ConversationController } from "./conversation.controller.js";
+import { ConversationService } from "./conversation.service.js";
+
+@Module({
+  imports: [DatabaseModule, RedisModule, AuthModule],
+  controllers: [ConversationController],
+  providers: [ConversationService, ChatRateLimiter, ChatGateway],
+  exports: [ConversationService],
+})
+export class ConversationModule {}

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

@@ -0,0 +1,648 @@
+import { HttpStatus, Inject, Injectable } from "@nestjs/common";
+import { ErrorCode } from "@drift/contracts";
+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";
+
+type Db = Prisma.TransactionClient | PrismaClient;
+type MessageRow = {
+  id: string;
+  conversationId: string;
+  senderId: string;
+  senderPublicId: string;
+  clientMsgId: string;
+  seq: bigint;
+  contentText: string;
+  status: string;
+  reviewStatus: string;
+  sentAt: Date;
+};
+type ConversationWithFirst = {
+  id: string;
+  messages: MessageRow[];
+};
+type Cursor = { t: string; id: string };
+
+export interface MessageResult {
+  message: ReturnType<ConversationService["serializeMessage"]>;
+  deduplicated: boolean;
+  memberIds: string[];
+}
+
+const MAX_SEQ = 9_223_372_036_854_775_807n;
+const UUID =
+  /^[0-9a-f]{8}-[0-9a-f]{4}-[1-5][0-9a-f]{3}-[89ab][0-9a-f]{3}-[0-9a-f]{12}$/i;
+const firstMessageInclude = {
+  messages: {
+    orderBy: { seq: "asc" as const },
+    take: 1,
+  },
+};
+
+@Injectable()
+export class ConversationService {
+  constructor(@Inject(PrismaService) private readonly prisma: PrismaService) {}
+
+  async reply(
+    accountId: string,
+    bottleId: string,
+    leaseId: string,
+    leaseToken: string,
+    clientMsgId: string,
+    text: string,
+  ) {
+    this.moderate(text);
+    for (let attempt = 0; attempt < 5; attempt += 1) {
+      const existing = await this.prisma.conversation.findUnique({
+        where: { sourceBottleId: bottleId },
+        include: firstMessageInclude,
+      });
+      if (existing) {
+        await this.assertConversationMember(accountId, existing.id);
+        return this.replyResult(existing, clientMsgId, false);
+      }
+      try {
+        return await this.prisma.$transaction(
+          async (tx) => {
+            await tx.$queryRaw`SELECT "id" FROM "bottles" WHERE "id"=${bottleId}::uuid FOR UPDATE`;
+            const winner = await tx.conversation.findUnique({
+              where: { sourceBottleId: bottleId },
+              include: firstMessageInclude,
+            });
+            if (winner) {
+              await this.assertConversationMember(accountId, winner.id, tx);
+              return this.replyResult(winner, clientMsgId, false);
+            }
+            const lease = await tx.bottlePickLease.findFirst({
+              where: {
+                id: leaseId,
+                bottleId,
+                pickerId: accountId,
+                leaseTokenHash: leaseHmac(leaseToken),
+              },
+              include: { bottle: true },
+            });
+            if (
+              !lease ||
+              lease.status !== "ACTIVE" ||
+              lease.bottle.activeLeaseId !== lease.id ||
+              lease.bottle.poolStatus !== "LEASED"
+            ) {
+              throw this.hidden();
+            }
+            if (lease.expiresAt <= new Date()) {
+              throw new DomainException(
+                ErrorCode.BOTTLE_LEASE_EXPIRED,
+                "Bottle lease expired",
+                HttpStatus.GONE,
+              );
+            }
+            const authorId = lease.bottle.authorId;
+            await this.assertAllowed(tx, accountId, authorId);
+            const profiles = await tx.anonymousProfile.findMany({
+              where: { accountId: { in: [accountId, authorId] } },
+              select: { accountId: true, nickname: true, publicId: true },
+            });
+            if (profiles.length !== 2) throw this.forbidden();
+            const now = new Date();
+            const conversation = await tx.conversation.create({
+              data: {
+                sourceBottleId: bottleId,
+                nextSeq: 2n,
+                lastMessageAt: now,
+                members: {
+                  create: [
+                    {
+                      accountId,
+                      peerAliasSnapshot: profiles.find(
+                        (profile) => profile.accountId === authorId,
+                      )!.nickname,
+                    },
+                    {
+                      accountId: authorId,
+                      peerAliasSnapshot: profiles.find(
+                        (profile) => profile.accountId === accountId,
+                      )!.nickname,
+                    },
+                  ],
+                },
+                messages: {
+                  create: {
+                    senderId: accountId,
+                    senderPublicId: profiles.find(
+                      (profile) => profile.accountId === accountId,
+                    )!.publicId,
+                    clientMsgId,
+                    seq: 1n,
+                    contentText: text,
+                    reviewStatus: "APPROVED",
+                    status: "SENT",
+                    sentAt: now,
+                  },
+                },
+              },
+              include: firstMessageInclude,
+            });
+            const message = conversation.messages[0]!;
+            await this.createMessageOutbox(tx, message.id, conversation.id, [
+              accountId,
+              authorId,
+            ]);
+            const consumed = await tx.bottle.updateMany({
+              where: {
+                id: bottleId,
+                poolStatus: "LEASED",
+                activeLeaseId: leaseId,
+              },
+              data: {
+                poolStatus: "CONSUMED",
+                activeLeaseId: null,
+                consumedAt: now,
+                version: { increment: 1 },
+              },
+            });
+            if (consumed.count !== 1) throw this.hidden();
+            await tx.bottlePickLease.update({
+              where: { id: leaseId },
+              data: { status: "CONSUMED", endedAt: now },
+            });
+            return this.replyResult(conversation, clientMsgId, true);
+          },
+          { isolationLevel: Prisma.TransactionIsolationLevel.Serializable },
+        );
+      } catch (error) {
+        if (error instanceof DomainException) throw error;
+        if (!this.isReplyRetryable(error)) throw error;
+        await new Promise((resolve) => setTimeout(resolve, 10 * (attempt + 1)));
+        const raced = await this.prisma.conversation.findUnique({
+          where: { sourceBottleId: bottleId },
+          include: firstMessageInclude,
+        });
+        if (raced) {
+          await this.assertConversationMember(accountId, raced.id);
+          return this.replyResult(raced, clientMsgId, false);
+        }
+      }
+    }
+    throw new DomainException(
+      ErrorCode.SERVICE_UNAVAILABLE,
+      "Conversation temporarily unavailable",
+      HttpStatus.SERVICE_UNAVAILABLE,
+    );
+  }
+
+  private replyResult(
+    conversation: ConversationWithFirst,
+    clientMsgId: string,
+    created: boolean,
+  ) {
+    const message = conversation.messages[0]!;
+    const won = message.clientMsgId === clientMsgId;
+    return {
+      conversationId: conversation.id,
+      message: this.serializeMessage(message),
+      created,
+      deduplicated: !created && won,
+      won,
+    };
+  }
+
+  async send(
+    accountId: string,
+    conversationId: string,
+    clientMsgId: string,
+    text: string,
+  ): Promise<MessageResult> {
+    this.moderate(text);
+    try {
+      return await this.prisma.$transaction(async (tx) => {
+        const conversation = await this.authorizeConversation(
+          tx,
+          accountId,
+          conversationId,
+        );
+        const previous = await tx.message.findUnique({
+          where: {
+            conversationId_clientMsgId: { conversationId, clientMsgId },
+          },
+        });
+        if (previous) {
+          return {
+            message: this.serializeMessage(previous),
+            deduplicated: true,
+            memberIds: conversation.members.map((member) => member.accountId),
+          };
+        }
+        const rows = await tx.$queryRaw<Array<{ seq: bigint }>>`
+          UPDATE "conversations"
+          SET "next_seq"="next_seq"+1, "last_message_at"=now(), "updated_at"=now()
+          WHERE "id"=${conversationId}::uuid
+          RETURNING "next_seq"-1 AS seq
+        `;
+        const sender = await tx.anonymousProfile.findUnique({
+          where: { accountId },
+          select: { publicId: true },
+        });
+        if (!sender) throw this.forbidden();
+        const message = await tx.message.create({
+          data: {
+            conversationId,
+            senderId: accountId,
+            senderPublicId: sender.publicId,
+            clientMsgId,
+            seq: rows[0]!.seq,
+            contentText: text,
+            status: "SENT",
+            reviewStatus: "APPROVED",
+          },
+        });
+        const memberIds = conversation.members.map(
+          (member) => member.accountId,
+        );
+        await this.createMessageOutbox(
+          tx,
+          message.id,
+          conversationId,
+          memberIds,
+        );
+        return {
+          message: this.serializeMessage(message),
+          deduplicated: false,
+          memberIds,
+        };
+      });
+    } catch (error) {
+      if (this.isMessageIdempotencyRace(error)) {
+        return this.prisma.$transaction(async (tx) => {
+          const conversation = await this.authorizeConversation(
+            tx,
+            accountId,
+            conversationId,
+          );
+          const message = await tx.message.findUniqueOrThrow({
+            where: {
+              conversationId_clientMsgId: { conversationId, clientMsgId },
+            },
+          });
+          return {
+            message: this.serializeMessage(message),
+            deduplicated: true,
+            memberIds: conversation.members.map((member) => member.accountId),
+          };
+        });
+      }
+      throw error;
+    }
+  }
+
+  async list(accountId: string, cursorText: string | undefined, limit: number) {
+    const cursor = cursorText ? this.decodeCursor(cursorText) : undefined;
+    const cursorDate = cursor ? new Date(cursor.t) : undefined;
+    const rows = await this.prisma.conversation.findMany({
+      where: {
+        members: { some: { accountId, leftAt: null } },
+        ...(cursor && cursorDate
+          ? {
+              OR: [
+                { lastMessageAt: { lt: cursorDate } },
+                { lastMessageAt: cursorDate, id: { lt: cursor.id } },
+              ],
+            }
+          : {}),
+      },
+      orderBy: [{ lastMessageAt: "desc" }, { id: "desc" }],
+      take: limit + 1,
+      include: {
+        members: { where: { accountId } },
+        messages: {
+          orderBy: { seq: "desc" },
+          take: 1,
+        },
+      },
+    });
+    const hasMore = rows.length > limit;
+    const page = rows.slice(0, limit);
+    return {
+      items: page.map((conversation) => {
+        const last = conversation.messages[0];
+        const read = conversation.members[0]!.lastReadSeq;
+        return {
+          id: conversation.id,
+          status: conversation.status,
+          lastMessageAt: conversation.lastMessageAt,
+          lastMessage: last ? this.serializeMessage(last) : null,
+          unread: last && last.seq > read ? (last.seq - read).toString() : "0",
+          lastReadSeq: read.toString(),
+        };
+      }),
+      nextCursor: hasMore ? this.encodeCursor(page.at(-1)!) : null,
+    };
+  }
+
+  async detail(accountId: string, id: string) {
+    const row = await this.prisma.conversation.findFirst({
+      where: { id, members: { some: { accountId, leftAt: null } } },
+      include: { members: { where: { accountId } } },
+    });
+    if (!row) throw this.forbidden();
+    return {
+      id: row.id,
+      status: row.status,
+      lastMessageAt: row.lastMessageAt,
+      nextSeq: row.nextSeq.toString(),
+      lastReadSeq: row.members[0]!.lastReadSeq.toString(),
+    };
+  }
+
+  async messages(
+    accountId: string,
+    id: string,
+    afterSeq: string,
+    limit: number,
+  ) {
+    await this.assertMember(accountId, id);
+    const after = this.parseSeq(afterSeq);
+    const items = await this.prisma.message.findMany({
+      where: { conversationId: id, seq: { gt: after } },
+      orderBy: { seq: "asc" },
+      take: limit,
+    });
+    return { items: items.map((message) => this.serializeMessage(message)) };
+  }
+
+  async read(accountId: string, id: string, seqText: string) {
+    const member = await this.assertMember(accountId, id);
+    const seq = this.parseSeq(seqText);
+    const conversation = await this.prisma.conversation.findUnique({
+      where: { id },
+      select: { nextSeq: true },
+    });
+    if (!conversation) throw this.forbidden();
+    if (seq >= conversation.nextSeq) throw this.validation();
+    if (seq <= member.lastReadSeq) {
+      return { conversationId: id, lastReadSeq: member.lastReadSeq.toString() };
+    }
+    const updated = await this.prisma.conversationMember.updateMany({
+      where: {
+        conversationId: id,
+        accountId,
+        leftAt: null,
+        lastReadSeq: { lt: seq },
+      },
+      data: { lastReadSeq: seq },
+    });
+    if (updated.count !== 1) throw this.forbidden();
+    return { conversationId: id, lastReadSeq: seq.toString() };
+  }
+
+  private async authorizeConversation(
+    tx: Db,
+    accountId: string,
+    conversationId: string,
+  ) {
+    const conversation = await tx.conversation.findUnique({
+      where: { id: conversationId },
+      include: { members: true },
+    });
+    if (
+      !conversation ||
+      conversation.status !== "ACTIVE" ||
+      !conversation.members.some(
+        (member) => member.accountId === accountId && !member.leftAt,
+      )
+    ) {
+      throw this.forbidden();
+    }
+    const peer = conversation.members.find(
+      (member) => member.accountId !== accountId && !member.leftAt,
+    );
+    if (!peer) throw this.forbidden();
+    await this.assertAllowed(tx, accountId, peer.accountId);
+    return conversation;
+  }
+
+  private async createMessageOutbox(
+    tx: Prisma.TransactionClient,
+    messageId: string,
+    conversationId: string,
+    memberIds: string[],
+  ) {
+    await tx.outboxEvent.create({
+      data: {
+        aggregateType: "MESSAGE",
+        aggregateId: messageId,
+        eventType: "MESSAGE_CREATED",
+        dedupeKey: `message-created:${messageId}`,
+        payload: { messageId, conversationId, memberIds },
+      },
+    });
+  }
+
+  private isReplyRetryable(error: unknown): boolean {
+    if (!(error instanceof Prisma.PrismaClientKnownRequestError)) return false;
+    if (error.code === "P2034") return true;
+    // Prisma wraps PostgreSQL serialization failures from $queryRaw as P2010.
+    if (error.code === "P2010") {
+      return error.meta?.code === "40001";
+    }
+    if (error.code !== "P2002") return false;
+    const fields = this.prismaUniqueTarget(error.meta?.target);
+    return (
+      fields.includes("source_bottle_id") ||
+      fields.includes("client_msg_id") ||
+      fields.includes("conversations_source_bottle_id_key") ||
+      fields.includes("messages_conversation_id_client_msg_id_key")
+    );
+  }
+
+  private isMessageIdempotencyRace(error: unknown): boolean {
+    if (!(error instanceof Prisma.PrismaClientKnownRequestError)) return false;
+    if (error.code !== "P2002") return false;
+    const fields = this.prismaUniqueTarget(error.meta?.target);
+    return (
+      fields.includes("client_msg_id") ||
+      fields.includes("messages_conversation_id_client_msg_id_key")
+    );
+  }
+
+  private prismaUniqueTarget(target: unknown): string[] {
+    if (typeof target === "string") return [target];
+    if (
+      Array.isArray(target) &&
+      target.every((item) => typeof item === "string")
+    ) {
+      return target;
+    }
+    return [];
+  }
+
+  private encodeCursor(row: { id: string; lastMessageAt: Date }): string {
+    return Buffer.from(
+      JSON.stringify({ t: row.lastMessageAt.toISOString(), id: row.id }),
+    ).toString("base64url");
+  }
+
+  private decodeCursor(text: string): Cursor {
+    if (text.length > 256 || !/^[A-Za-z0-9_-]+$/.test(text)) {
+      throw this.validation();
+    }
+    try {
+      const decoded = Buffer.from(text, "base64url").toString("utf8");
+      if (Buffer.from(decoded).toString("base64url") !== text) {
+        throw new Error("non-canonical");
+      }
+      const value: unknown = JSON.parse(decoded);
+      if (
+        !value ||
+        typeof value !== "object" ||
+        Array.isArray(value) ||
+        Object.keys(value).length !== 2 ||
+        typeof (value as Cursor).t !== "string" ||
+        typeof (value as Cursor).id !== "string" ||
+        !UUID.test((value as Cursor).id)
+      ) {
+        throw new Error("invalid cursor");
+      }
+      const date = new Date((value as Cursor).t);
+      if (
+        Number.isNaN(date.valueOf()) ||
+        date.toISOString() !== (value as Cursor).t ||
+        JSON.stringify(value) !== decoded
+      ) {
+        throw new Error("non-canonical cursor");
+      }
+      return value as Cursor;
+    } catch {
+      throw this.validation();
+    }
+  }
+
+  private parseSeq(text: string): bigint {
+    if (!/^(0|[1-9]\d{0,18})$/.test(text)) throw this.validation();
+    const value = BigInt(text);
+    if (value > MAX_SEQ) throw this.validation();
+    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 } },
+    });
+    if (!found || found.leftAt) throw this.forbidden();
+    return found;
+  }
+
+  private async assertAllowed(tx: Db, a: string, b: string) {
+    const now = new Date();
+    const [accounts, block, sanction] = await Promise.all([
+      tx.account.count({ where: { id: { in: [a, b] }, status: "ACTIVE" } }),
+      tx.block.findFirst({
+        where: {
+          OR: [
+            { blockerId: a, blockedId: b },
+            { blockerId: b, blockedId: a },
+          ],
+        },
+        select: { id: true },
+      }),
+      tx.sanction.findFirst({
+        where: {
+          accountId: { in: [a, b] },
+          type: { in: ["MUTE", "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 moderate(text: string) {
+    const reject = (process.env.MODERATION_REJECT_WORDS ?? "reject-word")
+      .split(",")
+      .filter(Boolean);
+    const review = (process.env.MODERATION_REVIEW_WORDS ?? "review-word")
+      .split(",")
+      .filter(Boolean);
+    if ([...reject, ...review].some((word) => text.includes(word))) {
+      throw new DomainException(
+        ErrorCode.CONTENT_REJECTED,
+        "Content rejected",
+        HttpStatus.FORBIDDEN,
+      );
+    }
+  }
+
+  serializeMessage(message: MessageRow) {
+    const publicId = message.senderPublicId;
+    if (!publicId) {
+      throw new DomainException(
+        ErrorCode.INTERNAL_ERROR,
+        "Message identity unavailable",
+        HttpStatus.INTERNAL_SERVER_ERROR,
+      );
+    }
+    return {
+      id: message.id,
+      conversationId: message.conversationId,
+      sender: { publicId },
+      clientMsgId: message.clientMsgId,
+      seq: message.seq.toString(),
+      text: message.contentText,
+      status: message.status,
+      reviewStatus: message.reviewStatus,
+      sentAt: message.sentAt,
+    };
+  }
+
+  private hidden() {
+    return new DomainException(
+      ErrorCode.NOT_FOUND,
+      "Not Found",
+      HttpStatus.NOT_FOUND,
+    );
+  }
+
+  private forbidden() {
+    return new DomainException(
+      ErrorCode.CONVERSATION_FORBIDDEN,
+      "Conversation forbidden",
+      HttpStatus.FORBIDDEN,
+    );
+  }
+
+  private validation() {
+    return new DomainException(
+      ErrorCode.VALIDATION_ERROR,
+      "Validation failed",
+      HttpStatus.BAD_REQUEST,
+    );
+  }
+}

apps/api/src/conversation/dto.ts

@@ -0,0 +1,42 @@
+import { Transform, Type } from "class-transformer";
+import {
+  IsInt,
+  IsOptional,
+  IsString,
+  IsUUID,
+  Matches,
+  Max,
+  MaxLength,
+  Min,
+  MinLength,
+} from "class-validator";
+
+const trim = ({ value }: { value: unknown }) =>
+  typeof value === "string" ? value.trim() : value;
+
+export class ReplyDto {
+  @IsUUID() leaseId!: string;
+  @IsString() @Matches(/^[A-Za-z0-9_-]{43}$/) leaseToken!: string;
+  @IsUUID() clientMsgId!: string;
+  @Transform(trim) @IsString() @MinLength(1) @MaxLength(1000) text!: string;
+}
+export class SendMessageDto {
+  @IsUUID() conversationId!: string;
+  @IsUUID() clientMsgId!: string;
+  @Transform(trim) @IsString() @MinLength(1) @MaxLength(1000) text!: string;
+}
+export class MessagesQueryDto {
+  @IsOptional() @Matches(/^\d+$/) afterSeq = "0";
+  @IsOptional() @Type(() => Number) @IsInt() @Min(1) @Max(100) limit = 50;
+}
+export class ConversationsQueryDto {
+  @IsOptional() @IsString() @Matches(/^[A-Za-z0-9_-]+$/) cursor?: string;
+  @IsOptional() @Type(() => Number) @IsInt() @Min(1) @Max(100) limit = 20;
+}
+export class ReadConversationDto {
+  @Matches(/^\d+$/) seq!: string;
+}
+
+export class SocketReadDto extends ReadConversationDto {
+  @IsUUID() conversationId!: string;
+}

pnpm-lock.yaml

@@ -49,19 +49,25 @@ importers:
 
   apps/api:
     dependencies:
-      "@drift/contracts":
+      '@drift/contracts':
         specifier: workspace:*
         version: link:../../packages/contracts
-      "@nestjs/common":
+      '@nestjs/common':
         specifier: ^11.1.6
         version: 11.2.3(class-transformer@0.5.1)(class-validator@0.14.4)(reflect-metadata@0.2.2)(rxjs@7.8.2)
-      "@nestjs/core":
+      '@nestjs/core':
         specifier: ^11.1.6
-        version: 11.2.3(@nestjs/common@11.2.3(class-transformer@0.5.1)(class-validator@0.14.4)(reflect-metadata@0.2.2)(rxjs@7.8.2))(@nestjs/platform-express@11.2.3)(reflect-metadata@0.2.2)(rxjs@7.8.2)
+        version: 11.2.3(@nestjs/common@11.2.3(class-transformer@0.5.1)(class-validator@0.14.4)(reflect-metadata@0.2.2)(rxjs@7.8.2))(@nestjs/platform-express@11.2.3)(@nestjs/websockets@11.2.5)(reflect-metadata@0.2.2)(rxjs@7.8.2)
-      "@nestjs/platform-express":
+      '@nestjs/platform-express':
         specifier: ^11.1.6
         version: 11.2.3(@nestjs/common@11.2.3(class-transformer@0.5.1)(class-validator@0.14.4)(reflect-metadata@0.2.2)(rxjs@7.8.2))(@nestjs/core@11.2.3)
-      "@prisma/client":
+      '@nestjs/platform-socket.io':
+        specifier: ^11.1.6
+        version: 11.2.5(@nestjs/common@11.2.3(class-transformer@0.5.1)(class-validator@0.14.4)(reflect-metadata@0.2.2)(rxjs@7.8.2))(@nestjs/websockets@11.2.5)(rxjs@7.8.2)
+      '@nestjs/websockets':
+        specifier: ^11.1.6
+        version: 11.2.5(@nestjs/common@11.2.3(class-transformer@0.5.1)(class-validator@0.14.4)(reflect-metadata@0.2.2)(rxjs@7.8.2))(@nestjs/core@11.2.3)(@nestjs/platform-socket.io@11.2.5)(reflect-metadata@0.2.2)(rxjs@7.8.2)
+      '@prisma/client':
         specifier: 6.19.0
         version: 6.19.0(prisma@6.19.0(typescript@5.9.3))(typescript@5.9.3)
       class-transformer:
@@ -85,16 +91,22 @@ importers:
       rxjs:
         specifier: ^7.8.2
         version: 7.8.2
+      socket.io:
+        specifier: ^4.8.1
+        version: 4.8.3
     devDependencies:
-      "@nestjs/testing":
+      '@nestjs/testing':
         specifier: ^11.1.6
         version: 11.2.3(@nestjs/common@11.2.3(class-transformer@0.5.1)(class-validator@0.14.4)(reflect-metadata@0.2.2)(rxjs@7.8.2))(@nestjs/core@11.2.3)(@nestjs/platform-express@11.2.3)
-      "@types/express":
+      '@types/express':
         specifier: ^5.0.3
         version: 5.0.6
-      "@types/supertest":
+      '@types/supertest':
         specifier: ^6.0.3
         version: 6.0.3
+      socket.io-client:
+        specifier: ^4.8.1
+        version: 4.8.3
       supertest:
         specifier: ^7.1.4
         version: 7.2.2
@@ -3252,6 +3264,92 @@ packages:
       }
     engines: { node: ">=10" }
 
+  '@nestjs/platform-socket.io@11.2.5':
+    resolution: {integrity: sha512-xOUCwjWfhnhpR5HBKMS3pYOcMYhXhKKR6duxmV7WLFmhgeIZjBalW/b2LEr3C/07+bq3BGTPZsVYunnTYBJDVw==}
+    peerDependencies:
+      '@nestjs/common': ^11.0.0
+      '@nestjs/websockets': ^11.0.0
+      rxjs: ^7.1.0
+
+  '@nestjs/websockets@11.2.5':
+    resolution: {integrity: sha512-KSTpTWKFSrBYxxXyclDg3ir9EzPWUU8iuKK4VFaVno2L+kwZcZzsKmnXrcMnmx8Sl2glgqXQg26Esu8hbo6a0Q==}
+    peerDependencies:
+      '@nestjs/common': ^11.0.0
+      '@nestjs/core': ^11.0.0
+      '@nestjs/platform-socket.io': ^11.0.0
+      reflect-metadata: ^0.1.12 || ^0.2.0
+      rxjs: ^7.1.0
+    peerDependenciesMeta:
+      '@nestjs/platform-socket.io':
+        optional: true
+
+  '@socket.io/component-emitter@3.1.2':
+    resolution: {integrity: sha512-9BCxFwvbGg/RsZK9tjXd8s4UcwR0MWeFQ1XEKIQVVvAGJyINdrqKMcTRyLoK8Rse1GjzLV9cwjWV1olXRWEXVA==}
+
+  '@types/cors@2.8.19':
+    resolution: {integrity: sha512-mFNylyeyqN93lfe/9CSxOGREz8cpzAhH+E93xJ4xWQf62V8sQ/24reV2nyzUWM6H6Xji+GGHpkbLe7pVoUEskg==}
+
+  '@types/ws@8.18.1':
+    resolution: {integrity: sha512-ThVF6DCVhA8kUGy+aazFQ4kXQ7E1Ty7A3ypFOe0IcJV8O/M511G99AW24irKrW56Wt44yG9+ij8FaqoBGkuBXg==}
+
+  accepts@1.3.8:
+    resolution: {integrity: sha512-PYAthTa2m2VKxuvSD3DPC/Gy+U+sOA1LAuT8mkmRuvw+NACSaeXEQ+NHcVF7rONl6qcaxV3Uuemwawk+7+SJLw==}
+    engines: {node: '>= 0.6'}
+
+  base64id@2.0.0:
+    resolution: {integrity: sha512-lGe34o6EHj9y3Kts9R4ZYs/Gr+6N7MCaMlIFA3F1R2O5/m7K06AxfSeO5530PEERE6/WyEg3lsuyw4GHlPZHog==}
+    engines: {node: ^4.5.0 || >= 5.9}
+
+  engine.io-client@6.6.6:
+    resolution: {integrity: sha512-iY6QdftLQ9pyiPoX082bpf/u1UewnOaJrtJIF9T0++QB34lZrj0uP+Q/bj8AlUsAxqhnkTV2BS8SBZSxOmoV5Q==}
+
+  engine.io-parser@5.2.3:
+    resolution: {integrity: sha512-HqD3yTBfnBxIrbnM1DoD6Pcq8NECnh8d4As1Qgh0z5Gg3jRRIqijury0CL3ghu/edArpUYiYqQiDUQBIs4np3Q==}
+    engines: {node: '>=10.0.0'}
+
+  engine.io@6.6.10:
+    resolution: {integrity: sha512-9/lX2bdlizlCXMHRMOIm03VBQHQYC7VvydcxtTAUJRxNW1QzM/2PMFSmr6h/lCiMHcyCP6abK+t9Q+j4vekk8Q==}
+    engines: {node: '>=10.2.0'}
+
+  negotiator@0.6.3:
+    resolution: {integrity: sha512-+EUsqGPLsM+j/zdChZjsnX51g4XrHFOIXwfnCVPGlQk/k5giakcKsuxCObBRu6DSm9opw/O6slWbJdghQM4bBg==}
+    engines: {node: '>= 0.6'}
+
+  object-hash@3.0.0:
+    resolution: {integrity: sha512-RSn9F68PjH9HqtltsSnqYC1XXoWe9Bju5+213R98cNGttag9q9yAOTzdbsqvIa7aNm5WffBZFpWYr2aWrklWAw==}
+    engines: {node: '>= 6'}
+
+  socket.io-adapter@2.5.8:
+    resolution: {integrity: sha512-6Oy52pbg+kvdCVvjcN+FnY7BvxZ7cIHNScbvztT/It5d0vbwoJoVZmF2gjJmnV0/4WlXRfG15zc45ySk9Ah8bw==}
+
+  socket.io-client@4.8.3:
+    resolution: {integrity: sha512-uP0bpjWrjQmUt5DTHq9RuoCBdFJF10cdX9X+a368j/Ft0wmaVgxlrjvK3kjvgCODOMMOz9lcaRzxmso0bTWZ/g==}
+    engines: {node: '>=10.0.0'}
+
+  socket.io-parser@4.2.7:
+    resolution: {integrity: sha512-IH/iSeO9T6gz1KkFleGDWkG9N3dl4jXVYUtMhIqH10Md0ttMer8nUNWiP1DKuNrybD2xBrixLJdCC9J6ECoYkg==}
+    engines: {node: '>=10.0.0'}
+
+  socket.io@4.8.3:
+    resolution: {integrity: sha512-2Dd78bqzzjE6KPkD5fHZmDAKRNe3J15q+YHDrIsy9WEkqttc7GY+kT9OBLSMaPbQaEd0x1BjcmtMtXkfpc+T5A==}
+    engines: {node: '>=10.2.0'}
+
+  ws@8.21.3:
+    resolution: {integrity: sha512-201TZ/kPWxoPr/OKWjquZR1SWKXcvxdH+e1xrx89b3YbmzLMFCLfnaG1HFIgWzJOEWZ7MvpK++odZufgYR50Rw==}
+    engines: {node: '>=10.0.0'}
+    peerDependencies:
+      bufferutil: ^4.0.1
+      utf-8-validate: '>=5.0.2'
+    peerDependenciesMeta:
+      bufferutil:
+        optional: true
+      utf-8-validate:
+        optional: true
+
+  xmlhttprequest-ssl@2.1.2:
+    resolution: {integrity: sha512-TEU+nJVUUnA4CYJFLvK5X9AOeH4KvDvhIfm0vV1GaQRtchnG0hgK5p8hw/xjv8cunWYCsiPCSDzObPyhEwq3KQ==}
+    engines: {node: '>=0.4.0'}
+
 snapshots:
   "@borewit/text-codec@0.2.2": {}
 
@@ -5004,3 +5102,137 @@ snapshots:
   wrappy@1.0.2: {}
 
   yocto-queue@0.1.0: {}
+
+  '@nestjs/core@11.2.3(@nestjs/common@11.2.3(class-transformer@0.5.1)(class-validator@0.14.4)(reflect-metadata@0.2.2)(rxjs@7.8.2))(@nestjs/platform-express@11.2.3)(@nestjs/websockets@11.2.5)(reflect-metadata@0.2.2)(rxjs@7.8.2)':
+    dependencies:
+      '@nestjs/common': 11.2.3(class-transformer@0.5.1)(class-validator@0.14.4)(reflect-metadata@0.2.2)(rxjs@7.8.2)
+      fast-safe-stringify: 2.1.1
+      iterare: 1.2.1
+      path-to-regexp: 8.4.2
+      reflect-metadata: 0.2.2
+      rxjs: 7.8.2
+      tslib: 2.8.1
+      uid: 2.0.2
+    optionalDependencies:
+      '@nestjs/platform-express': 11.2.3(@nestjs/common@11.2.3(class-transformer@0.5.1)(class-validator@0.14.4)(reflect-metadata@0.2.2)(rxjs@7.8.2))(@nestjs/core@11.2.3)
+      '@nestjs/websockets': 11.2.5(@nestjs/common@11.2.3(class-transformer@0.5.1)(class-validator@0.14.4)(reflect-metadata@0.2.2)(rxjs@7.8.2))(@nestjs/core@11.2.3)(@nestjs/platform-socket.io@11.2.5)(reflect-metadata@0.2.2)(rxjs@7.8.2)
+
+  '@nestjs/platform-socket.io@11.2.5(@nestjs/common@11.2.3(class-transformer@0.5.1)(class-validator@0.14.4)(reflect-metadata@0.2.2)(rxjs@7.8.2))(@nestjs/websockets@11.2.5)(rxjs@7.8.2)':
+    dependencies:
+      '@nestjs/common': 11.2.3(class-transformer@0.5.1)(class-validator@0.14.4)(reflect-metadata@0.2.2)(rxjs@7.8.2)
+      '@nestjs/websockets': 11.2.5(@nestjs/common@11.2.3(class-transformer@0.5.1)(class-validator@0.14.4)(reflect-metadata@0.2.2)(rxjs@7.8.2))(@nestjs/core@11.2.3)(@nestjs/platform-socket.io@11.2.5)(reflect-metadata@0.2.2)(rxjs@7.8.2)
+      rxjs: 7.8.2
+      socket.io: 4.8.3
+      tslib: 2.8.1
+    transitivePeerDependencies:
+      - bufferutil
+      - supports-color
+      - utf-8-validate
+
+  '@nestjs/websockets@11.2.5(@nestjs/common@11.2.3(class-transformer@0.5.1)(class-validator@0.14.4)(reflect-metadata@0.2.2)(rxjs@7.8.2))(@nestjs/core@11.2.3)(@nestjs/platform-socket.io@11.2.5)(reflect-metadata@0.2.2)(rxjs@7.8.2)':
+    dependencies:
+      '@nestjs/common': 11.2.3(class-transformer@0.5.1)(class-validator@0.14.4)(reflect-metadata@0.2.2)(rxjs@7.8.2)
+      '@nestjs/core': 11.2.3(@nestjs/common@11.2.3(class-transformer@0.5.1)(class-validator@0.14.4)(reflect-metadata@0.2.2)(rxjs@7.8.2))(@nestjs/platform-express@11.2.3)(@nestjs/websockets@11.2.5)(reflect-metadata@0.2.2)(rxjs@7.8.2)
+      iterare: 1.2.1
+      object-hash: 3.0.0
+      reflect-metadata: 0.2.2
+      rxjs: 7.8.2
+      tslib: 2.8.1
+    optionalDependencies:
+      '@nestjs/platform-socket.io': 11.2.5(@nestjs/common@11.2.3(class-transformer@0.5.1)(class-validator@0.14.4)(reflect-metadata@0.2.2)(rxjs@7.8.2))(@nestjs/websockets@11.2.5)(rxjs@7.8.2)
+
+  '@socket.io/component-emitter@3.1.2': {}
+
+  '@types/cors@2.8.19':
+    dependencies:
+      '@types/node': 22.19.3
+
+  '@types/ws@8.18.1':
+    dependencies:
+      '@types/node': 22.19.3
+
+  accepts@1.3.8:
+    dependencies:
+      mime-types: 2.1.35
+      negotiator: 0.6.3
+
+  base64id@2.0.0: {}
+
+  engine.io-client@6.6.6:
+    dependencies:
+      '@socket.io/component-emitter': 3.1.2
+      debug: 4.4.3
+      engine.io-parser: 5.2.3
+      ws: 8.21.3
+      xmlhttprequest-ssl: 2.1.2
+    transitivePeerDependencies:
+      - bufferutil
+      - supports-color
+      - utf-8-validate
+
+  engine.io-parser@5.2.3: {}
+
+  engine.io@6.6.10:
+    dependencies:
+      '@types/cors': 2.8.19
+      '@types/node': 22.19.3
+      '@types/ws': 8.18.1
+      accepts: 1.3.8
+      cookie: 0.7.2
+      cors: 2.8.6
+      debug: 4.4.3
+      engine.io-parser: 5.2.3
+      ws: 8.21.3
+    transitivePeerDependencies:
+      - bufferutil
+      - supports-color
+      - utf-8-validate
+
+  negotiator@0.6.3: {}
+
+  object-hash@3.0.0: {}
+
+  socket.io-adapter@2.5.8:
+    dependencies:
+      debug: 4.4.3
+      ws: 8.21.3
+    transitivePeerDependencies:
+      - bufferutil
+      - supports-color
+      - utf-8-validate
+
+  socket.io-client@4.8.3:
+    dependencies:
+      '@socket.io/component-emitter': 3.1.2
+      debug: 4.4.3
+      engine.io-client: 6.6.6
+      socket.io-parser: 4.2.7
+    transitivePeerDependencies:
+      - bufferutil
+      - supports-color
+      - utf-8-validate
+
+  socket.io-parser@4.2.7:
+    dependencies:
+      '@socket.io/component-emitter': 3.1.2
+      debug: 4.4.3
+    transitivePeerDependencies:
+      - supports-color
+
+  socket.io@4.8.3:
+    dependencies:
+      accepts: 1.3.8
+      base64id: 2.0.0
+      cors: 2.8.6
+      debug: 4.4.3
+      engine.io: 6.6.10
+      socket.io-adapter: 2.5.8
+      socket.io-parser: 4.2.7
+    transitivePeerDependencies:
+      - bufferutil
+      - supports-color
+      - utf-8-validate
+
+  ws@8.21.3: {}
+
+  xmlhttprequest-ssl@2.1.2: {}

prisma/migrations/0010_realtime_conversation/migration.sql

@@ -0,0 +1,24 @@
+-- Task 7 chat identity snapshots, query and delivery indexes.
+ALTER TABLE "messages" ADD COLUMN "sender_public_id" UUID;
+UPDATE "messages" AS m
+SET "sender_public_id" = p."public_id"
+FROM "anonymous_profiles" AS p
+WHERE p."account_id" = m."sender_id";
+-- Legacy/seed rows may predate anonymous profiles. Preserve an unlinkable
+-- snapshot instead of deriving or exposing the private account identifier.
+UPDATE "messages"
+SET "sender_public_id" = gen_random_uuid()
+WHERE "sender_public_id" IS NULL;
+ALTER TABLE "messages" ALTER COLUMN "sender_public_id" SET NOT NULL;
+
+-- Outbox events already provide durable message event dedupe.
+CREATE INDEX "messages_conversation_seq_inbox_idx"
+  ON "messages" ("conversation_id", "seq");
+CREATE INDEX "conversation_members_account_conversation_idx"
+  ON "conversation_members" ("account_id", "conversation_id")
+  WHERE "left_at" IS NULL;
+CREATE INDEX "conversations_last_message_cursor_idx"
+  ON "conversations" ("last_message_at" DESC, "id" DESC);
+CREATE INDEX "sanctions_active_account_idx"
+  ON "sanctions" ("account_id", "starts_at", "expires_at")
+  WHERE "revoked_at" IS NULL;

prisma/schema.prisma

@@ -227,19 +227,19 @@ model Bottle {
 
 model BottlePickLease {
   /// Partial unique index bottle_pick_leases_one_active_per_bottle is managed in 0001_init SQL.
-  id              String                @id @default(uuid()) @db.Uuid
+  id                   String                @id @default(uuid()) @db.Uuid
-  bottleId        String                @map("bottle_id") @db.Uuid
+  bottleId             String                @map("bottle_id") @db.Uuid
-  pickerId        String                @map("picker_id") @db.Uuid
+  pickerId             String                @map("picker_id") @db.Uuid
-  leaseTokenHash  String                @unique @map("lease_token_hash") @db.VarChar(255)
+  leaseTokenHash       String                @unique @map("lease_token_hash") @db.VarChar(255)
-  leaseTokenCiphertext Bytes             @map("lease_token_ciphertext")
+  leaseTokenCiphertext Bytes                 @map("lease_token_ciphertext")
-  expiresAt       DateTime              @map("expires_at") @db.Timestamptz(3)
+  expiresAt            DateTime              @map("expires_at") @db.Timestamptz(3)
-  status          BottlePickLeaseStatus @default(ACTIVE)
+  status               BottlePickLeaseStatus @default(ACTIVE)
-  endedAt         DateTime?             @map("ended_at") @db.Timestamptz(3)
+  endedAt              DateTime?             @map("ended_at") @db.Timestamptz(3)
-  createdAt       DateTime              @default(now()) @map("created_at") @db.Timestamptz(3)
+  createdAt            DateTime              @default(now()) @map("created_at") @db.Timestamptz(3)
-  bottle          Bottle                @relation(fields: [bottleId], references: [id], onDelete: Cascade)
+  bottle               Bottle                @relation(fields: [bottleId], references: [id], onDelete: Cascade)
-  activeForBottle Bottle?               @relation("ActiveBottleLease")
+  activeForBottle      Bottle?               @relation("ActiveBottleLease")
-  picker          Account               @relation("PickerLeases", fields: [pickerId], references: [id], onDelete: Cascade)
+  picker               Account               @relation("PickerLeases", fields: [pickerId], references: [id], onDelete: Cascade)
-  pickRequest     BottlePickRequest?
+  pickRequest          BottlePickRequest?
 
   @@index([pickerId, expiresAt])
   @@index([bottleId, expiresAt])
@@ -304,10 +304,12 @@ model Conversation {
   reports        Report[]
 
   @@index([status, updatedAt])
+  @@index([lastMessageAt(sort: Desc), id(sort: Desc)], map: "conversations_last_message_cursor_idx")
   @@map("conversations")
 }
 
 model ConversationMember {
+  /// Active member inbox index conversation_members_account_conversation_idx is managed in 0010 SQL.
   id                String       @id @default(uuid()) @db.Uuid
   conversationId    String       @map("conversation_id") @db.Uuid
   accountId         String       @map("account_id") @db.Uuid
@@ -328,6 +330,7 @@ model Message {
   id             String        @id @default(uuid()) @db.Uuid
   conversationId String        @map("conversation_id") @db.Uuid
   senderId       String        @map("sender_id") @db.Uuid
+  senderPublicId String        @map("sender_public_id") @db.Uuid
   clientMsgId    String        @map("client_msg_id") @db.VarChar(128)
   seq            BigInt
   contentText    String        @map("content_text") @db.Text
@@ -343,6 +346,7 @@ model Message {
 
   @@unique([conversationId, clientMsgId])
   @@unique([conversationId, seq])
+  @@index([conversationId, seq], map: "messages_conversation_seq_inbox_idx")
   @@index([senderId, createdAt])
   @@map("messages")
 }

tests/integration/database.spec.ts

@@ -298,6 +298,7 @@ describe("database authority constraints", () => {
       data: {
         conversationId: conversation.id,
         senderId: author.id,
+        senderPublicId: author.id,
         clientMsgId: "msg",
         seq: 1n,
         contentText: "hello",
@@ -326,6 +327,7 @@ describe("database authority constraints", () => {
         data: {
           conversationId: conversation.id,
           senderId: author.id,
+          senderPublicId: author.id,
           clientMsgId: "duplicate-client-id",
           seq: 1n,
           contentText: "first",
@@ -338,6 +340,7 @@ describe("database authority constraints", () => {
         data: {
           conversationId: conversation.id,
           senderId: author.id,
+          senderPublicId: author.id,
           clientMsgId: "duplicate-client-id",
           seq: 2n,
           contentText: "second",
@@ -358,6 +361,7 @@ describe("database authority constraints", () => {
         data: {
           conversationId: conversation.id,
           senderId: author.id,
+          senderPublicId: author.id,
           clientMsgId: "first-client-id",
           seq: 1n,
           contentText: "first",
@@ -370,6 +374,7 @@ describe("database authority constraints", () => {
         data: {
           conversationId: conversation.id,
           senderId: author.id,
+          senderPublicId: author.id,
           clientMsgId: "second-client-id",
           seq: 1n,
           contentText: "second",
@@ -426,6 +431,7 @@ describe("database authority constraints", () => {
         data: {
           conversationId: conversation.id,
           senderId: author.id,
+          senderPublicId: author.id,
           clientMsgId: "zero",
           seq: 0n,
           contentText: "x",