feat(realtime): add channel presence

src/app/api/realtime/presence/route.ts

@@ -0,0 +1,38 @@
+import { NextRequest, NextResponse } from "next/server";
+import config from "@payload-config";
+import { getPayload } from "payload";
+import { removePresence, touchPresence } from "@/lib/realtime/presence";
+
+export const dynamic = "force-dynamic";
+export const runtime = "nodejs";
+
+async function authenticate(req: NextRequest) {
+  const payload = await getPayload({ config: await config });
+  return payload.auth({ headers: req.headers, canSetHeaders: false });
+}
+
+function channelFrom(req: NextRequest): string | null {
+  const channel = req.nextUrl.searchParams.get("channel")?.trim();
+  return channel && /^[a-z0-9:_-]{1,80}$/i.test(channel) ? channel : null;
+}
+
+export async function POST(req: NextRequest) {
+  const channel = channelFrom(req);
+  if (!channel) return NextResponse.json({ ok: false, error: "Invalid channel." }, { status: 400 });
+  const { user } = await authenticate(req);
+  if (!user) return NextResponse.json({ ok: false }, { status: 401 });
+  touchPresence(channel, {
+    id: user.id as number,
+    name: user.payloadDisplayName || user.username || `Pilot #${user.id}`,
+  });
+  return NextResponse.json({ ok: true });
+}
+
+export async function DELETE(req: NextRequest) {
+  const channel = channelFrom(req);
+  if (!channel) return NextResponse.json({ ok: false, error: "Invalid channel." }, { status: 400 });
+  const { user } = await authenticate(req);
+  if (!user) return NextResponse.json({ ok: false }, { status: 401 });
+  removePresence(channel, user.id as number);
+  return NextResponse.json({ ok: true });
+}

src/app/api/realtime/route.ts

@@ -1,11 +1,13 @@
 import { NextRequest } from "next/server";
 import { type SSEEventClient, subscribe, unsubscribe } from "@/lib/realtime/bus";
+import { getPresence } from "@/lib/realtime/presence";
 
 export const dynamic = "force-dynamic";
 export const runtime = "nodejs";
 
 export async function GET(req: NextRequest) {
   const encoder = new TextEncoder();
+  const channel = req.nextUrl.searchParams.get("channel");
 
   let client: SSEEventClient | null = null;
   let heartbeat: ReturnType<typeof setInterval> | null = null;
@@ -21,10 +23,14 @@ export async function GET(req: NextRequest) {
     start(controller) {
       client = {
         send: (data) => controller.enqueue(encoder.encode(data)),
+        events: channel ? new Set(["presence"]) : undefined,
       };
       subscribe(client);
 
       controller.enqueue(encoder.encode(": connected\n\n"));
+      if (channel) {
+        controller.enqueue(encoder.encode(`event: presence\ndata: ${JSON.stringify({ channel, action: "snapshot", participants: getPresence(channel) })}\n\n`));
+      }
 
       heartbeat = setInterval(() => {
         try {

src/components/frontend/flappy/FlappyArena.tsx

@@ -1,7 +1,8 @@
 "use client";
 
-import { useState } from "react";
+import { useCallback, useState } from "react";
 import type { FlappyStats } from "@/app/(frontend)/flappy/actions";
+import { useRealtimePresence } from "@/hooks/useRealtimePresence";
 import { FlappyBird } from "./FlappyBird";
 import { FlappyLeaderboard } from "./FlappyLeaderboard";
 
@@ -14,10 +15,16 @@ export function FlappyArena({
 }) {
   const [refreshKey, setRefreshKey] = useState(0);
 
+  const [playing, setPlaying] = useState(false);
+  const activePilots = useRealtimePresence("flappy", playing, true);
+  const onPhaseChange = useCallback((phase: "ready" | "playing" | "over") => {
+    setPlaying(phase === "playing");
+  }, []);
+
   return (
     <div className="flex flex-col items-center gap-6">
-      <FlappyBird initialStats={initialStats} onGameRecorded={() => setRefreshKey((k) => k + 1)} />
-      <FlappyLeaderboard refreshKey={refreshKey} currentUserId={currentUserId} />
+      <FlappyBird initialStats={initialStats} onGameRecorded={() => setRefreshKey((k) => k + 1)} onPhaseChange={onPhaseChange} />
+      <FlappyLeaderboard refreshKey={refreshKey} currentUserId={currentUserId} activePilots={activePilots} />
     </div>
   );
 }

src/components/frontend/flappy/FlappyBird.tsx

@@ -324,9 +324,11 @@ function birdHitsPipe(birdX: number, birdY: number, pipe: Pipe): boolean {
 export function FlappyBird({
   initialStats: serverStats,
   onGameRecorded,
+  onPhaseChange,
 }: {
   initialStats: FlappyStats;
   onGameRecorded?: () => void;
+  onPhaseChange?: (phase: Phase) => void;
 }) {
   const canvasRef = useRef<HTMLCanvasElement | null>(null);
   const wrapperRef = useRef<HTMLDivElement | null>(null);
@@ -358,7 +360,8 @@ export function FlappyBird({
   const setPhaseSafe = useCallback((p: Phase) => {
     phaseRef.current = p;
     setPhase(p);
-  }, []);
+    onPhaseChange?.(p);
+  }, [onPhaseChange]);
 
   const reset = useCallback(() => {
     const g = gameRef.current;

src/components/frontend/flappy/FlappyLeaderboard.tsx

@@ -3,6 +3,7 @@
 import { useEffect, useState } from "react";
 import { Trophy } from "lucide-react";
 import { cn } from "@/lib/utils";
+import type { PresenceParticipant } from "@/hooks/useRealtimePresence";
 
 interface ApiUser {
   id: number;
@@ -40,6 +41,7 @@ interface LeaderboardRow {
 interface FlappyLeaderboardProps {
   refreshKey?: number;
   currentUserId?: number | string;
+  activePilots?: PresenceParticipant[];
 }
 
 function rankClass(index: number) {
@@ -49,7 +51,7 @@ function rankClass(index: number) {
   return "text-muted-foreground";
 }
 
-export function FlappyLeaderboard({ refreshKey = 0, currentUserId }: FlappyLeaderboardProps) {
+export function FlappyLeaderboard({ refreshKey = 0, currentUserId, activePilots = [] }: FlappyLeaderboardProps) {
   const [rows, setRows] = useState<LeaderboardRow[] | null>(null);
   const [failed, setFailed] = useState(false);
 
@@ -105,6 +107,13 @@ export function FlappyLeaderboard({ refreshKey = 0, currentUserId }: FlappyLeade
         <h2 className="text-sm font-semibold">Flight Simulator Leaderboard</h2>
         <span className="ml-auto text-xs text-muted-foreground">Top 10 · best run (bird-free)</span>
       </div>
+      <div className="border-b border-border px-4 py-2 text-xs text-muted-foreground" aria-live="polite">
+        <span className="font-medium text-emerald-400">{activePilots.length}</span>{" "}
+        {activePilots.length === 1 ? "pilot is" : "pilots are"} flying now
+        {activePilots.length > 0 && (
+          <span className="ml-1 text-foreground">({activePilots.map((pilot) => pilot.name).join(", ")})</span>
+        )}
+      </div>
 
       {failed ? (
         <p className="px-4 py-4 text-sm text-muted-foreground">Could not load the leaderboard.</p>
@@ -142,6 +151,9 @@ export function FlappyLeaderboard({ refreshKey = 0, currentUserId }: FlappyLeade
                     {row.isCurrentUser && (
                       <span className="shrink-0 text-xs text-muted-foreground">(you)</span>
                     )}
+                    {activePilots.some((pilot) => pilot.id === row.id) && (
+                      <span className="size-2 rounded-full bg-emerald-400" title="Playing now" aria-label="Playing now" />
+                    )}
                   </span>
                 </td>
                 <td className="px-2 py-2 text-right tabular-nums text-muted-foreground">

src/hooks/useRealtimePresence.ts

@@ -0,0 +1,69 @@
+"use client";
+
+import { useEffect, useState } from "react";
+
+export interface PresenceParticipant {
+  id: number;
+  name: string;
+  lastSeen: string;
+}
+
+interface PresenceEvent {
+  channel?: string;
+  action?: "snapshot" | "join" | "update" | "leave";
+  participants?: PresenceParticipant[];
+  participant?: PresenceParticipant;
+  participantId?: number;
+}
+
+const HEARTBEAT_MS = 15_000;
+
+export function useRealtimePresence(channel: string, active: boolean, observe = true): PresenceParticipant[] {
+  const [participants, setParticipants] = useState<PresenceParticipant[]>([]);
+
+  useEffect(() => {
+    if (!observe) {
+      setParticipants([]);
+      return;
+    }
+
+    let cancelled = false;
+    const endpoint = `/api/realtime/presence?channel=${encodeURIComponent(channel)}`;
+    const send = (method: "POST" | "DELETE") => {
+      void fetch(endpoint, { method, keepalive: method === "DELETE" }).catch(() => undefined);
+    };
+    const source = new EventSource(`/api/realtime?channel=${encodeURIComponent(channel)}`);
+    const onPresence = (event: MessageEvent<string>) => {
+      if (cancelled) return;
+      let detail: PresenceEvent;
+      try {
+        detail = JSON.parse(event.data) as PresenceEvent;
+      } catch {
+        return;
+      }
+      if (detail.action === "snapshot") setParticipants(detail.participants ?? []);
+      if ((detail.action === "join" || detail.action === "update") && detail.participant) {
+        setParticipants((current) => [
+          ...current.filter((item) => item.id !== detail.participant!.id),
+          detail.participant!,
+        ]);
+      }
+      if (detail.action === "leave" && detail.participantId !== undefined) {
+        setParticipants((current) => current.filter((item) => item.id !== detail.participantId));
+      }
+    };
+    source.addEventListener("presence", onPresence);
+    if (active) send("POST");
+    const heartbeat = active ? window.setInterval(() => send("POST"), HEARTBEAT_MS) : undefined;
+
+    return () => {
+      cancelled = true;
+      if (heartbeat) window.clearInterval(heartbeat);
+      source.removeEventListener("presence", onPresence);
+      source.close();
+      if (active) send("DELETE");
+    };
+  }, [channel, active, observe]);
+
+  return participants;
+}

src/lib/realtime/bus.ts

@@ -1,5 +1,6 @@
 export interface SSEEventClient {
   send: (data: string) => void;
+  events?: Set<string>;
 }
 
 const clients = new Set<SSEEventClient>();
@@ -20,6 +21,7 @@ export function clientCount(): number {
 export function broadcast(event: string, data: unknown): number {
   const payload = `event: ${event}\ndata: ${JSON.stringify(data)}\n\n`;
   for (const client of clients) {
+    if (client.events && !client.events.has(event)) continue;
     try {
       client.send(payload);
     } catch {

src/lib/realtime/presence.ts

@@ -0,0 +1,56 @@
+import { broadcast } from "./bus";
+
+export interface PresenceParticipant {
+  id: number;
+  name: string;
+  lastSeen: string;
+}
+
+interface PresenceRecord extends PresenceParticipant {
+  expiresAt: number;
+}
+
+export const PRESENCE_TTL_MS = 45_000;
+const channels = new Map<string, Map<number, PresenceRecord>>();
+
+function prune(channel: string, now = Date.now()): void {
+  const entries = channels.get(channel);
+  if (!entries) return;
+  for (const [id, entry] of entries) {
+    if (entry.expiresAt <= now) {
+      entries.delete(id);
+      broadcast("presence", { channel, action: "leave", participantId: id });
+    }
+  }
+  if (entries.size === 0) channels.delete(channel);
+}
+
+export function getPresence(channel: string): PresenceParticipant[] {
+  prune(channel);
+  return [...(channels.get(channel)?.values() ?? [])].map(({ id, name, lastSeen }) => ({ id, name, lastSeen }));
+}
+
+export function touchPresence(channel: string, participant: Omit<PresenceParticipant, "lastSeen">): void {
+  const now = Date.now();
+  prune(channel, now);
+  let entries = channels.get(channel);
+  if (!entries) {
+    entries = new Map();
+    channels.set(channel, entries);
+  }
+  const wasPresent = entries.has(participant.id);
+  const record = { ...participant, lastSeen: new Date(now).toISOString(), expiresAt: now + PRESENCE_TTL_MS };
+  entries.set(participant.id, record);
+  broadcast("presence", {
+    channel,
+    action: wasPresent ? "update" : "join",
+    participant: { id: record.id, name: record.name, lastSeen: record.lastSeen },
+  });
+}
+
+export function removePresence(channel: string, id: number): void {
+  const entries = channels.get(channel);
+  if (!entries?.delete(id)) return;
+  if (entries.size === 0) channels.delete(channel);
+  broadcast("presence", { channel, action: "leave", participantId: id });
+}