Add live notifications for platform events via SSE

frontend/src/app/api/notifications/[id]/read/route.ts

@@ -0,0 +1,55 @@
+import { markAsRead } from "@/lib/notification-store";
+import { buildNoStoreJson } from "@/lib/api-response";
+import { checkRateLimit } from "@/lib/rate-limit";
+
+export const runtime = "nodejs";
+export const dynamic = "force-dynamic";
+
+type RouteContext = {
+  params: Promise<{ id: string }>;
+};
+
+export async function POST(request: Request, context: RouteContext) {
+  const { response: rateLimitResponse, headers: rateLimitHeaders } =
+    checkRateLimit(request);
+  if (rateLimitResponse) {
+    return rateLimitResponse;
+  }
+
+  const { id } = await context.params;
+
+  let body: unknown;
+  try {
+    body = await request.json();
+  } catch {
+    return buildNoStoreJson(
+      { ok: false, error: "Request body must be valid JSON." },
+      400,
+      rateLimitHeaders,
+    );
+  }
+
+  const userId = String(
+    (body as Record<string, unknown> | null)?.userId ?? "",
+  ).trim();
+
+  if (!userId) {
+    return buildNoStoreJson(
+      { ok: false, error: "userId is required." },
+      400,
+      rateLimitHeaders,
+    );
+  }
+
+  const notification = markAsRead(userId, id);
+
+  if (!notification) {
+    return buildNoStoreJson(
+      { ok: false, error: "Notification not found." },
+      404,
+      rateLimitHeaders,
+    );
+  }
+
+  return buildNoStoreJson({ ok: true, notification }, 200, rateLimitHeaders);
+}

frontend/src/app/api/notifications/read-all/route.ts

@@ -0,0 +1,41 @@
+import { markAllAsRead } from "@/lib/notification-store";
+import { buildNoStoreJson } from "@/lib/api-response";
+import { checkRateLimit } from "@/lib/rate-limit";
+
+export const runtime = "nodejs";
+export const dynamic = "force-dynamic";
+
+export async function POST(request: Request) {
+  const { response: rateLimitResponse, headers: rateLimitHeaders } =
+    checkRateLimit(request);
+  if (rateLimitResponse) {
+    return rateLimitResponse;
+  }
+
+  let body: unknown;
+  try {
+    body = await request.json();
+  } catch {
+    return buildNoStoreJson(
+      { ok: false, error: "Request body must be valid JSON." },
+      400,
+      rateLimitHeaders,
+    );
+  }
+
+  const userId = String(
+    (body as Record<string, unknown> | null)?.userId ?? "",
+  ).trim();
+
+  if (!userId) {
+    return buildNoStoreJson(
+      { ok: false, error: "userId is required." },
+      400,
+      rateLimitHeaders,
+    );
+  }
+
+  const updatedCount = markAllAsRead(userId);
+
+  return buildNoStoreJson({ ok: true, updatedCount }, 200, rateLimitHeaders);
+}

frontend/src/app/api/notifications/route.ts

@@ -0,0 +1,44 @@
+import {
+  getUnreadCount,
+  listNotifications,
+} from "@/lib/notification-store";
+import { buildNoStoreJson } from "@/lib/api-response";
+import { checkRateLimit } from "@/lib/rate-limit";
+
+export const runtime = "nodejs";
+export const dynamic = "force-dynamic";
+
+export async function GET(request: Request) {
+  const { response: rateLimitResponse, headers: rateLimitHeaders } =
+    checkRateLimit(request);
+  if (rateLimitResponse) {
+    return rateLimitResponse;
+  }
+
+  const { searchParams } = new URL(request.url);
+  const userId = (searchParams.get("userId") ?? "").trim();
+
+  if (!userId) {
+    return buildNoStoreJson(
+      {
+        ok: false,
+        error: "userId query parameter is required.",
+      },
+      400,
+      rateLimitHeaders,
+    );
+  }
+
+  const notifications = listNotifications(userId);
+  const unreadCount = getUnreadCount(userId);
+
+  return buildNoStoreJson(
+    {
+      ok: true,
+      notifications,
+      unreadCount,
+    },
+    200,
+    rateLimitHeaders,
+  );
+}

frontend/src/app/api/notifications/stream/route.ts

@@ -0,0 +1,89 @@
+import {
+  getUnreadCount,
+  listNotifications,
+  subscribe,
+} from "@/lib/notification-store";
+import { checkRateLimit } from "@/lib/rate-limit";
+
+export const runtime = "nodejs";
+export const dynamic = "force-dynamic";
+
+const HEARTBEAT_INTERVAL_MS = 25_000;
+
+function sseMessage(event: string, data: unknown): string {
+  return `event: ${event}\ndata: ${JSON.stringify(data)}\n\n`;
+}
+
+/**
+ * Server-Sent Events stream of live notifications for a user.
+ * Emits an initial `snapshot` with the current list + unread count, then a
+ * `notification` event per newly created record, plus periodic `ping`
+ * heartbeats to keep intermediary proxies from closing the connection.
+ */
+export async function GET(request: Request) {
+  const { response: rateLimitResponse, headers: rateLimitHeaders } =
+    checkRateLimit(request);
+  if (rateLimitResponse) {
+    return rateLimitResponse;
+  }
+
+  const { searchParams } = new URL(request.url);
+  const userId = (searchParams.get("userId") ?? "").trim();
+
+  if (!userId) {
+    return new Response(
+      JSON.stringify({ ok: false, error: "userId query parameter is required." }),
+      { status: 400, headers: { "Content-Type": "application/json" } },
+    );
+  }
+
+  const encoder = new TextEncoder();
+  let unsubscribe: () => void = () => {};
+  let heartbeat: ReturnType<typeof setInterval> | undefined;
+
+  const stream = new ReadableStream({
+    start(controller) {
+      controller.enqueue(
+        encoder.encode(
+          sseMessage("snapshot", {
+            notifications: listNotifications(userId),
+            unreadCount: getUnreadCount(userId),
+          }),
+        ),
+      );
+
+      unsubscribe = subscribe(userId, (record) => {
+        controller.enqueue(
+          encoder.encode(
+            sseMessage("notification", {
+              notification: record,
+              unreadCount: getUnreadCount(userId),
+            }),
+          ),
+        );
+      });
+
+      heartbeat = setInterval(() => {
+        try {
+          controller.enqueue(encoder.encode(": ping\n\n"));
+        } catch {
+          // Controller already closed; the interval is cleared on cancel().
+        }
+      }, HEARTBEAT_INTERVAL_MS);
+    },
+    cancel() {
+      unsubscribe();
+      if (heartbeat) clearInterval(heartbeat);
+    },
+  });
+
+  return new Response(stream, {
+    headers: {
+      ...rateLimitHeaders,
+      "Content-Type": "text/event-stream",
+      "Cache-Control": "no-cache, no-transform",
+      Connection: "keep-alive",
+      "X-Accel-Buffering": "no",
+    },
+  });
+}

frontend/src/app/api/tasks/[taskId]/comments/route.ts

@@ -0,0 +1,53 @@
+import { addComment } from "@/lib/task-workflow";
+import { buildNoStoreJson } from "@/lib/api-response";
+import { checkRateLimit } from "@/lib/rate-limit";
+
+export const runtime = "nodejs";
+export const dynamic = "force-dynamic";
+
+type RouteContext = {
+  params: Promise<{ taskId: string }>;
+};
+
+export async function POST(request: Request, context: RouteContext) {
+  const { response: rateLimitResponse, headers: rateLimitHeaders } =
+    checkRateLimit(request);
+  if (rateLimitResponse) {
+    return rateLimitResponse;
+  }
+
+  const { taskId } = await context.params;
+
+  let body: unknown;
+  try {
+    body = await request.json();
+  } catch {
+    return buildNoStoreJson(
+      { ok: false, error: "Request body must be valid JSON." },
+      400,
+      rateLimitHeaders,
+    );
+  }
+
+  const payload = (body ?? {}) as Record<string, unknown>;
+  const submissionId = payload.submissionId
+    ? String(payload.submissionId)
+    : undefined;
+
+  const result = addComment({
+    taskId,
+    submissionId,
+    author: String(payload.author ?? ""),
+    message: String(payload.message ?? ""),
+  });
+
+  if (!result.ok) {
+    return buildNoStoreJson(
+      { ok: false, error: result.error, details: result.details },
+      result.status,
+      rateLimitHeaders,
+    );
+  }
+
+  return buildNoStoreJson({ ok: true, comment: result.comment }, 201, rateLimitHeaders);
+}

frontend/src/app/api/tasks/[taskId]/submissions/[submissionId]/approve/route.ts

@@ -0,0 +1,57 @@
+import { approveSubmission } from "@/lib/task-workflow";
+import { buildNoStoreJson } from "@/lib/api-response";
+import { checkRateLimit } from "@/lib/rate-limit";
+
+export const runtime = "nodejs";
+export const dynamic = "force-dynamic";
+
+type RouteContext = {
+  params: Promise<{ taskId: string; submissionId: string }>;
+};
+
+export async function POST(request: Request, context: RouteContext) {
+  const { response: rateLimitResponse, headers: rateLimitHeaders } =
+    checkRateLimit(request);
+  if (rateLimitResponse) {
+    return rateLimitResponse;
+  }
+
+  const { taskId, submissionId } = await context.params;
+
+  let body: unknown;
+  try {
+    body = await request.json();
+  } catch {
+    return buildNoStoreJson(
+      { ok: false, error: "Request body must be valid JSON." },
+      400,
+      rateLimitHeaders,
+    );
+  }
+
+  const actor = String((body as Record<string, unknown> | null)?.actor ?? "").trim();
+
+  if (!actor) {
+    return buildNoStoreJson(
+      { ok: false, error: "actor is required." },
+      400,
+      rateLimitHeaders,
+    );
+  }
+
+  const result = approveSubmission(taskId, submissionId, actor);
+
+  if (!result.ok) {
+    return buildNoStoreJson(
+      { ok: false, error: result.error, details: result.details },
+      result.status,
+      rateLimitHeaders,
+    );
+  }
+
+  return buildNoStoreJson(
+    { ok: true, task: result.task, submission: result.submission },
+    200,
+    rateLimitHeaders,
+  );
+}