homeio

1import { NextResponse } from "next/server";
2import { serverEnv } from "@/lib/server/env";
3import {
4  createRequestId,
5  logServerAction,
6  withServerTiming,
7} from "@/lib/server/logging/logger";
8import {
9  getLatestNetworkEvent,
10  subscribeToNetworkEvents,
11} from "@/lib/server/modules/network/events";
12import { toSseChunk } from "@/lib/server/realtime/sse";
13import type { NetworkEvent } from "@/lib/shared/contracts/network";
14import { requireApiSession } from "@/lib/server/modules/auth/api";
15
16export const runtime = "nodejs";
17
18export async function GET(request: Request) {
19  const apiSession = await requireApiSession(request);
20  if (apiSession.response) return apiSession.response;
21  const requestId = createRequestId();
22
23  try {
24    return await withServerTiming(
25      {
26        layer: "api",
27        action: "network.events.stream.get",
28        requestId,
29      },
30      async () => {
31        const encoder = new TextEncoder();
32
33        const stream = new ReadableStream<Uint8Array>({
34          start(controller) {
35            let closed = false;
36            let unsubscribe: (() => void) | null = null;
37
38            const close = () => {
39              if (closed) return;
40              closed = true;
41              if (unsubscribe) unsubscribe();
42              controller.close();
43            };
44
45            const sendHeartbeat = () => {
46              if (closed) return;
47              controller.enqueue(
48                encoder.encode(
49                  toSseChunk("heartbeat", {
50                    timestamp: new Date().toISOString(),
51                  }),
52                ),
53              );
54            };
55
56            const sendEvent = (event: NetworkEvent) => {
57              if (closed) return;
58              controller.enqueue(
59                encoder.encode(toSseChunk(event.type, event, { id: event.timestamp })),
60              );
61            };
62
63            const latest = getLatestNetworkEvent();
64            if (latest) {
65              sendEvent(latest);
66            }
67
68            unsubscribe = subscribeToNetworkEvents(sendEvent);
69
70            const heartbeat = setInterval(sendHeartbeat, serverEnv.SSE_HEARTBEAT_MS);
71            const teardown = () => {
72              clearInterval(heartbeat);
73              close();
74            };
75
76            request.signal.addEventListener("abort", teardown);
77          },
78        });
79
80        return new Response(stream, {
81          headers: {
82            "Content-Type": "text/event-stream",
83            "Cache-Control": "no-cache, no-transform",
84            Connection: "keep-alive",
85            "X-Accel-Buffering": "no",
86          },
87        });
88      },
89    );
90  } catch (error) {
91    logServerAction({
92      level: "error",
93      layer: "api",
94      action: "network.events.stream.get.response",
95      status: "error",
96      requestId,
97      message: "Unable to stream network events",
98      error,
99    });
100
101    return NextResponse.json(
102      {
103        error: "Unable to stream network events",
104        code: "internal_error",
105      },
106      {
107        status: 500,
108      },
109    );
110  }
111}
112