homeio

1import { serverEnv } from "@/lib/server/env";
2import {
3  createRequestId,
4  logServerAction,
5  withServerTiming,
6} from "@/lib/server/logging/logger";
7import { getAllContainersStats } from "@/lib/server/modules/docker/stats";
8import { toSseChunk } from "@/lib/server/realtime/sse";
9import type { DockerStatsPayload } from "@/lib/shared/contracts/docker";
10import { requireApiSession } from "@/lib/server/modules/auth/api";
11
12export const runtime = "nodejs";
13
14/** Soft cap on concurrent SSE clients for the Docker stats stream. */
15const MAX_DOCKER_STATS_CONNECTIONS = 10;
16let activeDockerStatsConnections = 0;
17
18/**
19 * GET /api/v1/docker/stats/stream
20 *
21 * Server-Sent Events stream for real-time Docker container stats.
22 * Returns 429 when more than MAX_DOCKER_STATS_CONNECTIONS clients are connected.
23 */
24export async function GET(request: Request) {
25  const apiSession = await requireApiSession(request);
26  if (apiSession.response) return apiSession.response;
27  if (activeDockerStatsConnections >= MAX_DOCKER_STATS_CONNECTIONS) {
28    logServerAction({
29      level: "warn",
30      layer: "api",
31      action: "docker.stats.stream.rate-limit",
32      status: "error",
33      message: `Docker stats SSE connection rejected — limit of ${MAX_DOCKER_STATS_CONNECTIONS} reached`,
34    });
35    return new Response(
36      JSON.stringify({ error: "Too many SSE connections", code: "rate_limited" }),
37      {
38        status: 429,
39        headers: { "Content-Type": "application/json" },
40      },
41    );
42  }
43
44  const encoder = new TextEncoder();
45  const requestId = createRequestId();
46  activeDockerStatsConnections++;
47
48  return withServerTiming(
49    {
50      layer: "api",
51      action: "docker.stats.stream.connect",
52      requestId,
53    },
54    async () => {
55      const stream = new ReadableStream<Uint8Array>({
56        start(controller) {
57          let closed = false;
58          const connectedAt = performance.now();
59
60          const pushStats = async () => {
61            if (closed) return;
62
63            try {
64              const result = await getAllContainersStats();
65              if (closed) return;
66
67              const payload: DockerStatsPayload = result;
68
69              controller.enqueue(
70                encoder.encode(
71                  toSseChunk("stats.updated", payload, {
72                    id: Date.now().toString(),
73                  }),
74                ),
75              );
76            } catch (error) {
77              logServerAction({
78                level: "error",
79                layer: "realtime",
80                action: "docker.stats.stream.push",
81                status: "error",
82                requestId,
83                error,
84              });
85            }
86          };
87
88          void pushStats();
89
90          // Update stats at the configured interval
91          const statsInterval = setInterval(
92            () => {
93              void pushStats();
94            },
95            serverEnv.METRICS_PUBLISH_INTERVAL_MS,
96          );
97
98          const heartbeatInterval = setInterval(() => {
99            if (closed) return;
100
101            controller.enqueue(
102              encoder.encode(
103                toSseChunk("heartbeat", { timestamp: new Date().toISOString() }),
104              ),
105            );
106          }, serverEnv.SSE_HEARTBEAT_MS);
107
108          const close = () => {
109            if (closed) return;
110            closed = true;
111            clearInterval(statsInterval);
112            clearInterval(heartbeatInterval);
113            activeDockerStatsConnections--;
114            controller.close();
115
116            logServerAction({
117              layer: "realtime",
118              action: "docker.stats.stream.disconnect",
119              status: "success",
120              requestId,
121              durationMs: Number((performance.now() - connectedAt).toFixed(2)),
122            });
123          };
124
125          request.signal.addEventListener("abort", close);
126        },
127      });
128
129      return new Response(stream, {
130        headers: {
131          "Content-Type": "text/event-stream",
132          "Cache-Control": "no-cache, no-transform",
133          Connection: "keep-alive",
134          "X-Accel-Buffering": "no",
135        },
136      });
137    },
138  );
139}
140