feat(chat): add ephemeral store-and-forward messaging

.env.example

@@ -26,3 +26,14 @@ JUDGE_BASE_URL="http://localhost:2000/api/v2"
 JUDGE_API_KEY="replace-with-proxy-api-key"
 PISTON_PYTHON_VERSION="3.10.0"
 PISTON_CPP_VERSION="10.2.0"
+
+# Ephemeral chat. Run `npm run chat:gateway` on a persistent host.
+CHAT_REALTIME_TOKEN_SECRET="replace-with-a-random-secret"
+CHAT_REALTIME_PUBLISH_SECRET="replace-with-a-different-random-secret"
+CHAT_REALTIME_INTERNAL_URL="http://localhost:8787"
+NEXT_PUBLIC_CHAT_WS_URL="ws://localhost:8787/socket"
+CHAT_ALLOWED_ORIGINS="http://localhost:3000"
+CHAT_GATEWAY_PORT="8787"
+
+# Vercel sends this bearer secret to authenticated cron routes.
+CRON_SECRET="replace-with-a-random-secret"

app/(auth)/join/page.tsx

@@ -53,6 +53,9 @@ export default async function JoinPage() {
               >
                 Continue as applicant (dev)
               </a>
+              <a className="secondary-button" href="/dev-login?role=active">
+                Continue as member (dev)
+              </a>
               <a className="secondary-button" href="/dev-login?role=admin">
                 Continue as admin (dev)
               </a>

app/account-bar.tsx

@@ -5,6 +5,7 @@ import {
   relativeTimeFromNow,
   unreadNotificationCount,
 } from "../lib/notifications";
+import MessageIndicator from "./message-indicator";
 import NotificationBell, { type NotificationItem } from "./notification-bell";
 import SiteHeader from "./site-header";
 
@@ -46,6 +47,7 @@ export default async function AccountBar({
       >
         <button type="submit">Sign out</button>
       </form>
+      <MessageIndicator userId={session.user.id} />
       <NotificationBell items={notificationItems} unreadCount={unreadCount} />
     </SiteHeader>
   );

app/api/chat/ack/route.ts

@@ -0,0 +1,19 @@
+import { NextResponse } from "next/server";
+import { chatAckSchema } from "../../../../lib/chat";
+import { activeChatUser } from "../../../../lib/chat-auth";
+import { chatErrorResponse, unauthorizedChatResponse } from "../../../../lib/chat-http";
+import { acknowledgeChatMessages } from "../../../../lib/chat-service";
+
+export async function POST(request: Request) {
+  const user = await activeChatUser();
+  if (!user) return unauthorizedChatResponse();
+
+  try {
+    const parsed = chatAckSchema.safeParse(await request.json());
+    if (!parsed.success) return NextResponse.json({ error: "INVALID_ACK" }, { status: 400 });
+
+    return NextResponse.json(await acknowledgeChatMessages(user.id, parsed.data.messageIds));
+  } catch (error) {
+    return chatErrorResponse(error);
+  }
+}

app/api/chat/conversations/route.ts

@@ -0,0 +1,51 @@
+import { ChatConversationType } from "@/prisma-client";
+import { NextResponse } from "next/server";
+import {
+  createDirectConversationSchema,
+  createGroupConversationSchema,
+} from "../../../../lib/chat";
+import { activeChatUser } from "../../../../lib/chat-auth";
+import { chatErrorResponse, unauthorizedChatResponse } from "../../../../lib/chat-http";
+import {
+  createDirectConversation,
+  createGroupConversation,
+  listChatConversations,
+} from "../../../../lib/chat-service";
+
+export const dynamic = "force-dynamic";
+
+export async function GET() {
+  const user = await activeChatUser();
+  if (!user) return unauthorizedChatResponse();
+  return NextResponse.json({ conversations: await listChatConversations(user.id) });
+}
+
+export async function POST(request: Request) {
+  const user = await activeChatUser();
+  if (!user) return unauthorizedChatResponse();
+
+  try {
+    const body = (await request.json()) as { type?: unknown };
+    if (body.type === ChatConversationType.DIRECT) {
+      const parsed = createDirectConversationSchema.safeParse(body);
+      if (!parsed.success) return NextResponse.json({ error: "INVALID_DIRECT" }, { status: 400 });
+      const conversation = await createDirectConversation(user.id, parsed.data.recipientId);
+      return NextResponse.json({ conversationId: conversation.id });
+    }
+
+    if (body.type === ChatConversationType.GROUP) {
+      const parsed = createGroupConversationSchema.safeParse(body);
+      if (!parsed.success) return NextResponse.json({ error: "INVALID_GROUP" }, { status: 400 });
+      const conversation = await createGroupConversation(
+        user.id,
+        parsed.data.title,
+        parsed.data.memberIds,
+      );
+      return NextResponse.json({ conversationId: conversation.id }, { status: 201 });
+    }
+
+    return NextResponse.json({ error: "INVALID_CONVERSATION_TYPE" }, { status: 400 });
+  } catch (error) {
+    return chatErrorResponse(error);
+  }
+}

app/api/chat/inbox/route.ts

@@ -0,0 +1,18 @@
+import { NextResponse } from "next/server";
+import { activeChatUser } from "../../../../lib/chat-auth";
+import { chatErrorResponse, unauthorizedChatResponse } from "../../../../lib/chat-http";
+import { listPendingChatMessages } from "../../../../lib/chat-service";
+
+export const dynamic = "force-dynamic";
+
+export async function GET(request: Request) {
+  const user = await activeChatUser();
+  if (!user) return unauthorizedChatResponse();
+
+  try {
+    const cursor = new URL(request.url).searchParams.get("cursor")?.trim() || undefined;
+    return NextResponse.json(await listPendingChatMessages(user.id, cursor));
+  } catch (error) {
+    return chatErrorResponse(error);
+  }
+}

app/api/chat/messages/route.ts

@@ -0,0 +1,37 @@
+import { NextResponse } from "next/server";
+import { sendChatMessageSchema } from "../../../../lib/chat";
+import { activeChatUser } from "../../../../lib/chat-auth";
+import { chatErrorResponse, unauthorizedChatResponse } from "../../../../lib/chat-http";
+import { publishChatEnvelope } from "../../../../lib/chat-realtime";
+import { enqueueChatMessage } from "../../../../lib/chat-service";
+
+export async function POST(request: Request) {
+  const user = await activeChatUser();
+  if (!user) return unauthorizedChatResponse();
+
+  try {
+    const parsed = sendChatMessageSchema.safeParse(await request.json());
+    if (!parsed.success) {
+      return NextResponse.json({ error: "INVALID_MESSAGE" }, { status: 400 });
+    }
+
+    const result = await enqueueChatMessage({ ...parsed.data, senderId: user.id });
+    const realtime =
+      result.envelope && result.recipientIds.length > 0
+        ? await publishChatEnvelope(result.recipientIds, result.envelope)
+        : { attempted: 0, delivered: 0, unavailable: false };
+
+    return NextResponse.json(
+      {
+        messageId: result.messageId,
+        conversationId: result.conversationId,
+        createdAt: result.createdAt,
+        duplicate: result.duplicate,
+        realtime,
+      },
+      { status: result.duplicate ? 200 : 202 },
+    );
+  } catch (error) {
+    return chatErrorResponse(error);
+  }
+}

app/api/chat/read/route.ts

@@ -0,0 +1,19 @@
+import { NextResponse } from "next/server";
+import { markChatReadSchema } from "../../../../lib/chat";
+import { activeChatUser } from "../../../../lib/chat-auth";
+import { chatErrorResponse, unauthorizedChatResponse } from "../../../../lib/chat-http";
+import { markChatRead } from "../../../../lib/chat-service";
+
+export async function POST(request: Request) {
+  const user = await activeChatUser();
+  if (!user) return unauthorizedChatResponse();
+
+  try {
+    const parsed = markChatReadSchema.safeParse(await request.json());
+    if (!parsed.success) return NextResponse.json({ error: "INVALID_READ" }, { status: 400 });
+    await markChatRead(user.id, parsed.data.conversationId);
+    return NextResponse.json({ ok: true });
+  } catch (error) {
+    return chatErrorResponse(error);
+  }
+}

app/api/chat/report/route.ts

@@ -0,0 +1,19 @@
+import { NextResponse } from "next/server";
+import { chatReportSchema } from "../../../../lib/chat";
+import { activeChatUser } from "../../../../lib/chat-auth";
+import { chatErrorResponse, unauthorizedChatResponse } from "../../../../lib/chat-http";
+import { reportChatMessage } from "../../../../lib/chat-service";
+
+export async function POST(request: Request) {
+  const user = await activeChatUser();
+  if (!user) return unauthorizedChatResponse();
+
+  try {
+    const parsed = chatReportSchema.safeParse(await request.json());
+    if (!parsed.success) return NextResponse.json({ error: "INVALID_REPORT" }, { status: 400 });
+    const report = await reportChatMessage({ ...parsed.data, reporterId: user.id });
+    return NextResponse.json({ reportId: report.id }, { status: 201 });
+  } catch (error) {
+    return chatErrorResponse(error);
+  }
+}

app/api/chat/token/route.ts

@@ -0,0 +1,14 @@
+import { NextResponse } from "next/server";
+import { activeChatUser } from "../../../../lib/chat-auth";
+import { unauthorizedChatResponse } from "../../../../lib/chat-http";
+import { createChatSocketToken } from "../../../../lib/chat-realtime";
+
+export const dynamic = "force-dynamic";
+
+export async function POST() {
+  const user = await activeChatUser();
+  if (!user) return unauthorizedChatResponse();
+
+  const token = await createChatSocketToken(user.id);
+  return NextResponse.json({ token, socketUrl: process.env.NEXT_PUBLIC_CHAT_WS_URL ?? null });
+}

app/api/cron/chat-cleanup/route.ts

@@ -0,0 +1,27 @@
+import { timingSafeEqual } from "node:crypto";
+import { NextResponse } from "next/server";
+import { cleanupExpiredChatData } from "../../../../lib/chat-service";
+
+function authorized(request: Request) {
+  const expected = process.env.CRON_SECRET;
+  const authorization = request.headers.get("authorization");
+  const provided = authorization?.startsWith("Bearer ") ? authorization.slice(7) : "";
+
+  if (!expected) return false;
+  const expectedBuffer = Buffer.from(expected);
+  const providedBuffer = Buffer.from(provided);
+  return (
+    expectedBuffer.length === providedBuffer.length &&
+    timingSafeEqual(expectedBuffer, providedBuffer)
+  );
+}
+
+export async function GET(request: Request) {
+  if (!authorized(request)) {
+    return NextResponse.json({ error: "Unauthorized" }, { status: 401 });
+  }
+
+  const result = await cleanupExpiredChatData();
+  console.info("Chat cleanup completed", result);
+  return NextResponse.json({ ok: true, ...result });
+}

app/api/health/route.ts

@@ -17,6 +17,12 @@ export async function GET() {
     googleClientSecret: isConfigured(process.env.AUTH_GOOGLE_SECRET),
     judgeBaseUrl: process.env.JUDGE_PROVIDER === "fake" || isConfigured(process.env.JUDGE_BASE_URL),
     judgeApiKey: process.env.JUDGE_PROVIDER === "fake" || isConfigured(process.env.JUDGE_API_KEY),
+    chatRealtime:
+      process.env.NODE_ENV !== "production" ||
+      (isConfigured(process.env.CHAT_REALTIME_TOKEN_SECRET) &&
+        isConfigured(process.env.CHAT_REALTIME_PUBLISH_SECRET) &&
+        isConfigured(process.env.CHAT_REALTIME_INTERNAL_URL) &&
+        isConfigured(process.env.NEXT_PUBLIC_CHAT_WS_URL)),
   };
 
   let database = false;
@@ -44,6 +50,8 @@ export async function GET() {
           prisma.problem.count(),
           prisma.testCase.count(),
           prisma.submission.count(),
+          prisma.chatConversation.count(),
+          prisma.chatDelivery.count(),
         ]);
         schema = true;
       } catch (error) {

app/dev-login/route.ts

@@ -36,10 +36,11 @@ export async function GET(request: Request) {
     return NextResponse.json({ error: "Not found" }, { status: 404 });
   }
 
+  const requestedRole = new URL(request.url).searchParams.get("role");
   const role: LocalDevRole =
-    new URL(request.url).searchParams.get("role") === "admin" ? "admin" : "member";
+    requestedRole === "admin" ? "admin" : requestedRole === "active" ? "active" : "member";
 
   await signIn(localDevProviderId(role), {
-    redirectTo: role === "admin" ? "/admin/cohort" : "/apply",
+    redirectTo: role === "admin" ? "/admin/cohort" : role === "active" ? "/dashboard" : "/apply",
   });
 }