perf: Redis pub/sub, PgBouncer, optimistic UI for high concurrency

- Add Redis 7 for pub/sub (lobby + chat real-time), rate limiting, caching - Replace SSE DB polling with Redis pub/sub (lobby: instant approval, chat: instant delivery) - Add PgBouncer (transaction mode, 500 client → 25 pool connections) - Chat SSE stream via Redis pub/sub instead of 3s polling - Optimistic UI in ChatPanel (messages appear before server confirms) - Redis-based rate limiter (works across multiple app replicas) - Prisma query optimization (select only needed fields) - Chat message cache in Redis (10s TTL) - Docker Compose: add redis, pgbouncer services with healthchecks - Production: resource limits, 2 app replicas behind Traefik - Update CLAUDE.md, README.md, .env.example, setup.sh Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-03-22 14:17:25 +03:00
parent 3e50c57ee0
commit a42ec96965
21 changed files with 568 additions and 122 deletions
@@ -2,6 +2,8 @@ import { NextResponse } from "next/server";
 import { z } from "zod";
 import { prisma } from "@/lib/prisma";
 import { getSessionFromRequest } from "@/lib/auth-helpers";
+import { redis } from "@/lib/redis";
+import { publishChatMessage } from "@/lib/chat-pubsub";

 const sendMessageSchema = z.object({
  sessionId: z.string().min(1),
@@ -49,6 +51,13 @@ export async function GET(
    return NextResponse.json({ error: "Unauthorized" }, { status: 401 });
  }

+  // Check Redis cache first (10s TTL to reduce DB hits during active chat)
+  const cacheKey = `chat:${roomId}:${cursor || "latest"}:${limit}`;
+  const cached = await redis.get(cacheKey);
+  if (cached) {
+    return NextResponse.json(JSON.parse(cached));
+  }
+
  const messages = await prisma.chatMessage.findMany({
    where: {
      roomId,
@@ -56,13 +65,25 @@ export async function GET(
    },
    orderBy: { createdAt: "desc" },
    take: limit,
+    select: {
+      id: true,
+      sessionId: true,
+      senderName: true,
+      content: true,
+      createdAt: true,
+    },
  });

  const nextCursor = messages.length === limit
    ? messages[messages.length - 1].createdAt.toISOString()
    : null;

-  return NextResponse.json({ messages: messages.reverse(), nextCursor });
+  const result = { messages: messages.reverse(), nextCursor };
+
+  // Cache for 10 seconds
+  await redis.set(cacheKey, JSON.stringify(result), "EX", 10);
+
+  return NextResponse.json(result);
 }

 export async function POST(
@@ -100,5 +121,13 @@ export async function POST(
    },
  });

+  // Publish to Redis for real-time delivery via SSE
+  await publishChatMessage(roomId, {
+    id: message.id,
+    senderName: message.senderName,
+    content: message.content,
+    createdAt: message.createdAt.toISOString(),
+  });
+
  return NextResponse.json(message, { status: 201 });
 }
@@ -0,0 +1,74 @@
+import { prisma } from "@/lib/prisma";
+import { subscribeChatMessages } from "@/lib/chat-pubsub";
+
+export async function GET(
+  req: Request,
+  { params }: { params: Promise<{ roomId: string }> }
+) {
+  const { roomId } = await params;
+  const url = new URL(req.url);
+  const sessionId = url.searchParams.get("sessionId");
+
+  if (!sessionId) {
+    return new Response(JSON.stringify({ error: "sessionId required" }), {
+      status: 400,
+      headers: { "Content-Type": "application/json" },
+    });
+  }
+
+  // Verify participant
+  const participant = await prisma.participantHistory.findFirst({
+    where: { roomId, sessionId, leftAt: null },
+  });
+  if (!participant) {
+    return new Response(JSON.stringify({ error: "Not a participant" }), {
+      status: 403,
+      headers: { "Content-Type": "application/json" },
+    });
+  }
+
+  const stream = new ReadableStream({
+    start(controller) {
+      const encoder = new TextEncoder();
+      const send = (data: Record<string, unknown>) => {
+        try {
+          controller.enqueue(encoder.encode(`data: ${JSON.stringify(data)}\n\n`));
+        } catch { /* stream closed */ }
+      };
+
+      const { unsubscribe } = subscribeChatMessages(roomId, (msg) => {
+        send(msg);
+      });
+
+      // Keep-alive every 30s
+      const keepAlive = setInterval(() => {
+        try {
+          controller.enqueue(encoder.encode(": keepalive\n\n"));
+        } catch { /* stream closed */ }
+      }, 30000);
+
+      // Timeout after 1 hour
+      const timeout = setTimeout(() => cleanup(), 60 * 60 * 1000);
+
+      let cleaned = false;
+      function cleanup() {
+        if (cleaned) return;
+        cleaned = true;
+        clearInterval(keepAlive);
+        clearTimeout(timeout);
+        unsubscribe().catch(() => {});
+        try { controller.close(); } catch { /* already closed */ }
+      }
+
+      req.signal.addEventListener("abort", () => cleanup());
+    },
+  });
+
+  return new Response(stream, {
+    headers: {
+      "Content-Type": "text/event-stream",
+      "Cache-Control": "no-cache",
+      Connection: "keep-alive",
+    },
+  });
+}
@@ -3,6 +3,7 @@ import { z } from "zod";
 import { compare } from "bcryptjs";
 import { prisma } from "@/lib/prisma";
 import { createToken } from "@/lib/livekit";
+import { checkRateLimit } from "@/lib/rate-limit";

 const joinSchema = z.object({
  displayName: z.string().min(1).max(100),
@@ -11,11 +12,6 @@ const joinSchema = z.object({
  sessionFingerprint: z.string().min(1),
 });

-// CRITICAL-10: In-memory rate limiter for PIN attempts
-const pinAttempts = new Map<string, { count: number; resetTime: number }>();
-const PIN_MAX_ATTEMPTS = 5;
-const PIN_WINDOW_MS = 60_000; // 1 minute
-
 export async function POST(
  req: Request,
  { params }: { params: Promise<{ roomId: string }> }
@@ -63,22 +59,14 @@ export async function POST(
      return NextResponse.json({ error: "PIN required" }, { status: 401 });
    }

-    // Rate limit PIN attempts by roomId + IP
-    const ip = req.headers.get("x-forwarded-for")?.split(",")[0]?.trim() || "unknown";
-    const rateLimitKey = `${roomId}:${ip}`;
-    const now = Date.now();
-    const attempt = pinAttempts.get(rateLimitKey);
-
-    if (attempt && now < attempt.resetTime) {
-      if (attempt.count >= PIN_MAX_ATTEMPTS) {
-        return NextResponse.json(
-          { error: "Too many PIN attempts. Try again later." },
-          { status: 429 }
-        );
-      }
-      attempt.count++;
-    } else {
-      pinAttempts.set(rateLimitKey, { count: 1, resetTime: now + PIN_WINDOW_MS });
+    // Rate limit PIN attempts by roomId + IP (Redis-based)
+    const clientIp = req.headers.get("x-forwarded-for")?.split(",")[0]?.trim() || "unknown";
+    const { allowed } = await checkRateLimit(`pin:${roomId}:${clientIp}`, 5, 60);
+    if (!allowed) {
+      return NextResponse.json(
+        { error: "Too many PIN attempts. Try again later." },
+        { status: 429 }
+      );
    }

    const valid = await compare(pin, room.pinHash);
@@ -3,6 +3,7 @@ import { z } from "zod";
 import { prisma } from "@/lib/prisma";
 import { getSessionFromRequest } from "@/lib/auth-helpers";
 import { createToken } from "@/lib/livekit";
+import { publishLobbyEvent } from "@/lib/lobby-pubsub";

 const lobbyActionSchema = z.object({
  lobbyEntryId: z.string().min(1),
@@ -74,6 +75,7 @@ export async function POST(
      where: { id: lobbyEntryId },
      data: { status: "REJECTED" },
    });
+    await publishLobbyEvent(roomId, entry.sessionId, { status: "REJECTED" });
    return NextResponse.json({ status: "rejected" });
  }

@@ -105,9 +107,11 @@ export async function POST(
    }),
  ]);

-  // Store token temporarily in lobbyEntry metadata — we use the updatedAt as a signal
-  // The SSE endpoint will query the entry status and generate a fresh token
-  // We store it via a simple in-memory approach or re-generate in SSE
-  // For simplicity, we return it here too (host can relay it)
+  // Publish to Redis so the SSE stream picks it up instantly
+  await publishLobbyEvent(roomId, entry.sessionId, {
+    status: "APPROVED",
+    token,
+  });
+
  return NextResponse.json({ status: "approved", token });
 }
@@ -1,5 +1,6 @@
 import { prisma } from "@/lib/prisma";
 import { createToken } from "@/lib/livekit";
+import { subscribeLobbyEvents } from "@/lib/lobby-pubsub";

 export async function GET(
  req: Request,
@@ -16,89 +17,103 @@ export async function GET(
    });
  }

-  const room = await prisma.room.findUnique({ where: { id: roomId } });
-  if (!room) {
-    return new Response(JSON.stringify({ error: "Room not found" }), {
-      status: 404,
-      headers: { "Content-Type": "application/json" },
-    });
-  }
-
-  // CRITICAL-8: Verify sessionId matches an existing LobbyEntry for this room
+  // Verify sessionId has a LobbyEntry for this room
  const existingEntry = await prisma.lobbyEntry.findFirst({
    where: { roomId, sessionId },
+    orderBy: { createdAt: "desc" },
  });
  if (!existingEntry) {
-    return new Response(JSON.stringify({ error: "No lobby entry found for this session" }), {
-      status: 403,
-      headers: { "Content-Type": "application/json" },
+    return new Response(
+      JSON.stringify({ error: "No lobby entry found for this session" }),
+      { status: 403, headers: { "Content-Type": "application/json" } }
+    );
+  }
+
+  // Check if already approved/rejected
+  if (existingEntry.status === "APPROVED") {
+    const freshRoom = await prisma.room.findUnique({ where: { id: roomId } });
+    if (!freshRoom?.livekitRoomName) {
+      return new Response(
+        JSON.stringify({ error: "Room not started yet" }),
+        { status: 400, headers: { "Content-Type": "application/json" } }
+      );
+    }
+    const canPublish = !freshRoom.webinarMode;
+    const token = await createToken(sessionId, freshRoom.livekitRoomName, {
+      name: existingEntry.displayName,
+      canPublish,
+      canSubscribe: true,
+      canPublishData: true,
+    });
+    const body = `data: ${JSON.stringify({ status: "APPROVED", token })}\n\n`;
+    return new Response(body, {
+      headers: {
+        "Content-Type": "text/event-stream",
+        "Cache-Control": "no-cache",
+        Connection: "keep-alive",
+      },
    });
  }

+  if (existingEntry.status === "REJECTED") {
+    const body = `data: ${JSON.stringify({ status: "REJECTED" })}\n\n`;
+    return new Response(body, {
+      headers: {
+        "Content-Type": "text/event-stream",
+        "Cache-Control": "no-cache",
+        Connection: "keep-alive",
+      },
+    });
+  }
+
+  // PENDING — subscribe to Redis channel and wait for event
  const stream = new ReadableStream({
-    async start(controller) {
+    start(controller) {
      const encoder = new TextEncoder();
      const send = (data: Record<string, unknown>) => {
-        controller.enqueue(encoder.encode(`data: ${JSON.stringify(data)}\n\n`));
+        try {
+          controller.enqueue(encoder.encode(`data: ${JSON.stringify(data)}\n\n`));
+        } catch {
+          /* stream may be closed */
+        }
      };

-      let resolved = false;
-      const maxPolls = 150; // 5 minutes at 2s intervals
+      // Send initial PENDING status
+      send({ status: "PENDING" });

-      for (let i = 0; i < maxPolls && !resolved; i++) {
-        try {
-          const entry = await prisma.lobbyEntry.findFirst({
-            where: { roomId, sessionId },
-            orderBy: { createdAt: "desc" },
-          });
-
-          if (!entry) {
-            send({ status: "error", message: "Lobby entry not found" });
-            resolved = true;
-            break;
-          }
-
-          if (entry.status === "APPROVED") {
-            // Re-fetch room to get fresh livekitRoomName (may have been started after lobby entry)
-            const freshRoom = await prisma.room.findUnique({ where: { id: roomId } });
-            if (!freshRoom?.livekitRoomName) {
-              send({ status: "error", message: "Room not started yet" });
-              resolved = true;
-              break;
-            }
-            const canPublish = !freshRoom.webinarMode;
-            const token = await createToken(sessionId, freshRoom.livekitRoomName, {
-              name: entry.displayName,
-              canPublish,
-              canSubscribe: true,
-              canPublishData: true,
-            });
-            send({ status: "APPROVED", token });
-            resolved = true;
-            break;
-          }
-
-          if (entry.status === "REJECTED") {
-            send({ status: "REJECTED" });
-            resolved = true;
-            break;
-          }
-
-          send({ status: "PENDING" });
-        } catch {
-          send({ status: "error", message: "Internal error" });
-          resolved = true;
-          break;
+      // Subscribe to Redis pub/sub
+      const { unsubscribe } = subscribeLobbyEvents(
+        roomId,
+        sessionId,
+        (event) => {
+          send(event);
+          cleanup();
        }
+      );

-        await new Promise((r) => setTimeout(r, 2000));
-      }
-
-      if (!resolved) {
+      // Timeout after 5 minutes
+      const timeout = setTimeout(() => {
        send({ status: "timeout" });
+        cleanup();
+      }, 5 * 60 * 1000);
+
+      let cleaned = false;
+      function cleanup() {
+        if (cleaned) return;
+        cleaned = true;
+        clearTimeout(timeout);
+        unsubscribe().catch(() => {});
+        try {
+          controller.close();
+        } catch {
+          /* already closed */
+        }
      }

-      controller.close();
+      // Handle client disconnect via AbortSignal
+      req.signal.addEventListener("abort", () => {
+        cleanup();
+      });
    },
  });

@@ -59,6 +59,18 @@ export async function GET(req: Request) {
  const rooms = await prisma.room.findMany({
    where,
    orderBy: { createdAt: "desc" },
+    select: {
+      id: true,
+      name: true,
+      code: true,
+      status: true,
+      lobbyEnabled: true,
+      webinarMode: true,
+      isLocked: true,
+      createdAt: true,
+      startedAt: true,
+      endedAt: true,
+    },
  });

  return NextResponse.json(rooms);