feat: direct messages

1:1 conversations (userAId < userBId dedup pair), messages, per-participant read tracking (conversation_reads). Block relationship is enforced on every send. UI: /messages list + /messages/[id] thread (5s poll), MessageButton on profiles, unread-badged nav icon. This completes the social-feature backlog: notifications, rate limiting, blocking, reporting, search/discovery, mentions, DMs, plus fixes for the recipe-visibility 404, follow race, and 2-level comment thread cap found during the earlier audit. Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
2026-07-03 22:24:56 +02:00
parent a51ba85253
commit c3776238c7
16 changed files with 4815 additions and 0 deletions
@@ -0,0 +1,76 @@
+import { NextRequest, NextResponse } from "next/server";
+import { z } from "zod";
+import { db, conversations, conversationReads, messages, eq, asc } from "@epicure/db";
+import { requireSession } from "@/lib/api-auth";
+import { applyRateLimit } from "@/lib/rate-limit";
+import { isParticipant, otherParticipantId } from "@/lib/messaging";
+import { isBlockedEitherWay } from "@/lib/blocks";
+import { randomUUID } from "crypto";
+
+interface RouteContext {
+  params: Promise<{ id: string }>;
+}
+
+const Schema = z.object({ content: z.string().min(1).max(4000) });
+
+export async function GET(_req: NextRequest, { params }: RouteContext) {
+  const { session, response } = await requireSession();
+  if (response) return response;
+  const { id } = await params;
+
+  const conversation = await db.query.conversations.findFirst({ where: eq(conversations.id, id) });
+  if (!conversation || !isParticipant(conversation, session!.user.id)) {
+    return NextResponse.json({ error: "Not found" }, { status: 404 });
+  }
+
+  const rows = await db
+    .select()
+    .from(messages)
+    .where(eq(messages.conversationId, id))
+    .orderBy(asc(messages.createdAt))
+    .limit(200);
+
+  await db
+    .insert(conversationReads)
+    .values({ conversationId: id, userId: session!.user.id, lastReadAt: new Date() })
+    .onConflictDoUpdate({
+      target: [conversationReads.conversationId, conversationReads.userId],
+      set: { lastReadAt: new Date() },
+    });
+
+  return NextResponse.json({ messages: rows });
+}
+
+export async function POST(req: NextRequest, { params }: RouteContext) {
+  const { session, response } = await requireSession();
+  if (response) return response;
+  const { id } = await params;
+
+  const limited = await applyRateLimit(`rl:message:${session!.user.id}`, 30, 60);
+  if (limited) return limited;
+
+  const conversation = await db.query.conversations.findFirst({ where: eq(conversations.id, id) });
+  if (!conversation || !isParticipant(conversation, session!.user.id)) {
+    return NextResponse.json({ error: "Not found" }, { status: 404 });
+  }
+
+  const otherId = otherParticipantId(conversation, session!.user.id);
+  if (await isBlockedEitherWay(session!.user.id, otherId)) {
+    return NextResponse.json({ error: "Not found" }, { status: 404 });
+  }
+
+  const body = await req.json() as unknown;
+  const parsed = Schema.safeParse(body);
+  if (!parsed.success) return NextResponse.json({ error: "Validation error" }, { status: 400 });
+
+  const messageId = randomUUID();
+  await db.insert(messages).values({
+    id: messageId,
+    conversationId: id,
+    senderId: session!.user.id,
+    content: parsed.data.content,
+  });
+  await db.update(conversations).set({ lastMessageAt: new Date() }).where(eq(conversations.id, id));
+
+  return NextResponse.json({ id: messageId }, { status: 201 });
+}
@@ -0,0 +1,99 @@
+import { NextRequest, NextResponse } from "next/server";
+import { z } from "zod";
+import { db, conversations, conversationReads, messages, users, eq, or, desc, and, gt, sql } from "@epicure/db";
+import { requireSession } from "@/lib/api-auth";
+import { applyRateLimit } from "@/lib/rate-limit";
+import { findOrCreateConversation, otherParticipantId } from "@/lib/messaging";
+import { isBlockedEitherWay } from "@/lib/blocks";
+import { randomUUID } from "crypto";
+
+const CreateSchema = z.object({
+  username: z.string().min(1),
+  content: z.string().min(1).max(4000).optional(),
+});
+
+export async function GET() {
+  const { session, response } = await requireSession();
+  if (response) return response;
+
+  const rows = await db
+    .select()
+    .from(conversations)
+    .where(or(eq(conversations.userAId, session!.user.id), eq(conversations.userBId, session!.user.id)))
+    .orderBy(desc(conversations.lastMessageAt))
+    .limit(50);
+
+  const result = await Promise.all(
+    rows.map(async (conv) => {
+      const otherId = otherParticipantId(conv, session!.user.id);
+      const [other, lastMessage, readRow] = await Promise.all([
+        db.query.users.findFirst({
+          where: eq(users.id, otherId),
+          columns: { id: true, name: true, username: true, avatarUrl: true },
+        }),
+        db.query.messages.findFirst({
+          where: eq(messages.conversationId, conv.id),
+          orderBy: (m, { desc: d }) => [d(m.createdAt)],
+        }),
+        db.query.conversationReads.findFirst({
+          where: and(eq(conversationReads.conversationId, conv.id), eq(conversationReads.userId, session!.user.id)),
+        }),
+      ]);
+
+      const unreadRow = await db
+        .select({ total: sql<string>`count(*)` })
+        .from(messages)
+        .where(
+          and(
+            eq(messages.conversationId, conv.id),
+            readRow ? gt(messages.createdAt, readRow.lastReadAt) : sql`true`,
+            sql`${messages.senderId} != ${session!.user.id}`
+          )
+        );
+
+      return {
+        id: conv.id,
+        otherUser: other,
+        lastMessage: lastMessage?.content ?? null,
+        lastMessageAt: conv.lastMessageAt,
+        unreadCount: Number(unreadRow[0]?.total ?? 0),
+      };
+    })
+  );
+
+  return NextResponse.json({ conversations: result });
+}
+
+export async function POST(req: NextRequest) {
+  const { session, response } = await requireSession();
+  if (response) return response;
+
+  const limited = await applyRateLimit(`rl:message:${session!.user.id}`, 30, 60);
+  if (limited) return limited;
+
+  const body = await req.json() as unknown;
+  const parsed = CreateSchema.safeParse(body);
+  if (!parsed.success) return NextResponse.json({ error: "Validation error" }, { status: 400 });
+
+  const target = await db.query.users.findFirst({ where: eq(users.username, parsed.data.username) });
+  if (!target) return NextResponse.json({ error: "Not found" }, { status: 404 });
+  if (target.id === session!.user.id) return NextResponse.json({ error: "Cannot message yourself" }, { status: 400 });
+
+  if (await isBlockedEitherWay(session!.user.id, target.id)) {
+    return NextResponse.json({ error: "Not found" }, { status: 404 });
+  }
+
+  const conversation = await findOrCreateConversation(session!.user.id, target.id);
+
+  if (parsed.data.content) {
+    await db.insert(messages).values({
+      id: randomUUID(),
+      conversationId: conversation.id,
+      senderId: session!.user.id,
+      content: parsed.data.content,
+    });
+    await db.update(conversations).set({ lastMessageAt: new Date() }).where(eq(conversations.id, conversation.id));
+  }
+
+  return NextResponse.json({ conversationId: conversation.id }, { status: 201 });
+}