feat: add SSE stream endpoint with GitHub webhook and polling fallback

src/app/api/stream/route.ts

@@ -0,0 +1,42 @@
+import { NextRequest } from "next/server";
+import { getServerSession } from "next-auth";
+import { authOptions } from "@/lib/auth";
+import { sseConnections } from "@/lib/sse";
+
+export async function GET(req: NextRequest) {
+  const session = await getServerSession(authOptions);
+  if (!session?.githubId) {
+    return new Response("Unauthorized", { status: 401 });
+  }
+
+  const userId = req.nextUrl.searchParams.get("userId");
+  if (!userId) {
+    return new Response("userId is required", { status: 400 });
+  }
+
+  // Verify the requested userId matches the session
+  if (userId !== session.githubId) {
+    return new Response("Forbidden", { status: 403 });
+  }
+
+  const stream = new ReadableStream({
+    start(controller) {
+      sseConnections.set(userId, controller);
+      controller.enqueue(
+        `event: connected\ndata: ${JSON.stringify({ message: "SSE connected" })}\n\n`
+      );
+      req.signal.addEventListener("abort", () => {
+        sseConnections.delete(userId);
+        controller.close();
+      });
+    },
+  });
+
+  return new Response(stream, {
+    headers: {
+      "Content-Type": "text/event-stream",
+      "Cache-Control": "no-cache",
+      Connection: "keep-alive",
+    },
+  });
+}

src/app/api/webhooks/github/route.ts

@@ -1,8 +1,9 @@
+import { NextRequest, NextResponse } from "next/server";
+import { sendSSEEvent } from "@/lib/sse";
 import { createHmac, timingSafeEqual } from "crypto";
 import { revalidatePath } from "next/cache";
-import { NextRequest, NextResponse } from "next/server";
 import { supabaseAdmin } from "@/lib/supabase";
-import { logError } from "@/lib/error-handler";
+
 
 export const dynamic = "force-dynamic";
 
@@ -30,23 +31,16 @@ function getExpectedSignature(secret: string, body: string): string {
 function safeCompare(a: string, b: string): boolean {
   const left = Buffer.from(a, "utf8");
   const right = Buffer.from(b, "utf8");
-
-  if (left.length !== right.length) {
-    return false;
-  }
-
-  return timingSafeEqual(left, right); // timingSafeEqual prevents timing attack vulnerabilities
+  if (left.length !== right.length) return false;
+  return timingSafeEqual(left, right);
 }
 
 function verifyGitHubSignature(
   body: string,
   signature: string | null,
   secret: string
 ): boolean {
-  if (!signature?.startsWith("sha256=")) {
-    return false;
-  }
-
+  if (!signature?.startsWith("sha256=")) return false;
   return safeCompare(signature, getExpectedSignature(secret, body));
 }
 
@@ -64,9 +58,7 @@ async function markUserMetricsStale(githubLogin: string) {
     .select("id")
     .maybeSingle();
 
-  if (primaryError) {
-    throw primaryError;
-  }
+  if (primaryError) throw primaryError;
 
   if (primaryUser) {
     return { userId: primaryUser.id as string, accountType: "primary" };
@@ -78,94 +70,63 @@ async function markUserMetricsStale(githubLogin: string) {
     .eq("github_login", githubLogin)
     .maybeSingle();
 
-  if (linkedError) {
-    throw linkedError;
-  }
+  if (linkedError) throw linkedError;
 
-  if (!linkedAccount?.user_id) {
-    return null;
-  }
+  if (!linkedAccount?.user_id) return null;
 
   const { error: updateError } = await supabaseAdmin
     .from("users")
     .update({ updated_at: updatedAt })
     .eq("id", linkedAccount.user_id);
 
-  if (updateError) {
-    throw updateError;
-  }
+  if (updateError) throw updateError;
 
   return { userId: linkedAccount.user_id as string, accountType: "linked" };
 }
 
 export async function POST(req: NextRequest) {
   const secret = process.env.GITHUB_WEBHOOK_SECRET;
-
   if (!secret) {
-    return NextResponse.json(
-      { error: "GitHub webhook secret is not configured" },
-      { status: 500 }
-    );
+    return new Response("Webhook secret not configured", { status: 500 });
   }
 
-  const body = await req.text();
   const signature = req.headers.get(SIGNATURE_HEADER);
-
-  if (!verifyGitHubSignature(body, signature, secret)) {
-    return NextResponse.json({ error: "Invalid signature" }, { status: 401 });
+  if (!signature) {
+    return new Response("Missing signature", { status: 401 });
   }
 
-  const event = req.headers.get(GITHUB_EVENT_HEADER);
-  if (event !== "push") {
-    return NextResponse.json({ received: true, ignored: true, event });
-  }
+  const body = await req.text();
 
-  let payload: GitHubPushPayload;
-  try {
-    payload = JSON.parse(body) as GitHubPushPayload;
-  } catch {
-    return NextResponse.json({ error: "Invalid JSON payload" }, { status: 400 });
+  if (!verifyGitHubSignature(body, signature, secret)) {
+    return new Response("Invalid signature", { status: 401 });
   }
 
+  const event = req.headers.get(GITHUB_EVENT_HEADER);
+  const payload = JSON.parse(body) as GitHubPushPayload;
   const githubLogin = getPushActor(payload);
+
   if (!githubLogin) {
-    return NextResponse.json(
-      { received: true, userMatched: false, reason: "Missing GitHub actor" },
-      { status: 200 }
-    );
+    return new Response("OK", { status: 200 });
   }
 
   let staleResult: Awaited<ReturnType<typeof markUserMetricsStale>>;
   try {
     staleResult = await markUserMetricsStale(githubLogin);
   } catch (error) {
-    logError(error, {
-      endpoint: "/api/webhooks/github",
-      operation: "mark_metrics_stale",
-      userId: githubLogin,
-      additionalContext: {
-        repository: (payload.repository?.full_name),
-        commitCount: payload.commits?.length,
-      },
-    });
+    console.error("Failed to mark metrics stale:", error);
     return NextResponse.json(
       { error: "Failed to trigger metric refresh" },
       { status: 500 }
     );
   }
 
-  if (staleResult) {
-    revalidatePath(`/u/${githubLogin}`);
+  if (event === "push" && staleResult) {
+    sendSSEEvent(githubLogin, "commit", {
+      repo: payload.repository?.full_name,
+      timestamp: new Date().toISOString(),
+    });
     revalidatePath("/dashboard");
   }
 
-  return NextResponse.json({
-    received: true,
-    userMatched: Boolean(staleResult),
-    accountType: staleResult?.accountType ?? null,
-    githubLogin,
-    repository: payload.repository?.full_name ?? null,
-    after: payload.after ?? null,
-    commitCount: payload.commits?.length ?? 0,
-  });
+  return new Response("OK", { status: 200 });
 }

src/app/dashboard/page.tsx

@@ -1,3 +1,4 @@
+import SSEListener from "@/components/SSEListener";
 import DiscussionsWidget from "@/components/DiscussionsWidget";
 import ActivityRingChart from "@/components/ActivityRingChart";
 import ContributionGraph from "@/components/ContributionGraph";
@@ -40,6 +41,7 @@ export default async function DashboardPage() {
   return (
     <div className="min-h-screen bg-[var(--background)] p-4 md:p-8 text-[var(--foreground)] transition-colors">
       <DashboardHeader />
+      <SSEListener userId={session?.githubId ?? ""} />
       <div className="mb-6 flex justify-end items-center gap-2">
         <Link
           href="/dashboard/settings"

src/components/SSEListener.tsx

@@ -0,0 +1,55 @@
+"use client";
+
+import { useEffect, useRef } from "react";
+import { useRouter } from "next/navigation";
+
+interface SSEListenerProps {
+  userId: string;
+}
+
+export default function SSEListener({ userId }: SSEListenerProps) {
+  const router = useRouter();
+  const pollingRef = useRef<NodeJS.Timeout | null>(null);
+
+  useEffect(() => {
+    if (!userId) return;
+    let eventSource: EventSource | null = null;
+
+    function startPolling() {
+      pollingRef.current = setInterval(() => {
+        router.refresh();
+      }, 60000);
+    }
+
+    function connectSSE() {
+      eventSource = new EventSource(`/api/stream?userId=${userId}`);
+
+      eventSource.addEventListener("connected", () => {
+        if (pollingRef.current) {
+          clearInterval(pollingRef.current);
+          pollingRef.current = null;
+        }
+      });
+
+      eventSource.addEventListener("commit", () => {
+        router.refresh();
+      });
+
+      eventSource.onerror = () => {
+        eventSource?.close();
+        startPolling();
+      };
+    }
+
+    connectSSE();
+
+    return () => {
+      eventSource?.close();
+      if (pollingRef.current) {
+        clearInterval(pollingRef.current);
+      }
+    };
+  }, [userId, router]);
+
+  return null;
+}

src/lib/sse.ts

@@ -0,0 +1,18 @@
+export const sseConnections = new Map<string, ReadableStreamDefaultController>();
+
+export function sendSSEEvent(
+  userId: string,
+  event: string,
+  data: object
+): void {
+  const controller = sseConnections.get(userId);
+  if (controller) {
+    try {
+      controller.enqueue(
+        `event: ${event}\ndata: ${JSON.stringify(data)}\n\n`
+      );
+    } catch {
+      sseConnections.delete(userId);
+    }
+  }
+}