fix(beszel): disable request handler cache for stats to make live mode work (#6244)

packages/integrations/src/beszel/beszel-integration.ts

@@ -20,6 +20,7 @@ import type {
   BeszelSystemdService,
   BeszelSystemStatsRecord,
   CreateAlertInput,
+  LiveStatsEvent,
   PocketBaseListResponse,
   UpdateAlertInput,
 } from "./beszel-types";
@@ -284,6 +285,125 @@ export class BeszelIntegration extends Integration {
     });
   }
 
+  /**
+   * Subscribe to real-time system and container stats via PocketBase SSE.
+   * Connects to the PocketBase Realtime API and parses incoming SSE events,
+   * filtering by systemId. This provides live-updating metrics (≤1s latency
+   * from agent) instead of polling the REST API every 5s.
+   */
+  public async subscribeRealtimeMetrics(
+    systemId: string,
+    onMessage: (event: LiveStatsEvent) => void,
+    signal: AbortSignal,
+  ): Promise<void> {
+    const session = await this.authenticateAsync();
+    const realtimeUrl = this.url("/api/realtime");
+
+    logger.debug("Opening Beszel SSE connection for realtime metrics", {
+      integrationId: this.integration.id,
+      systemId,
+      url: realtimeUrl.pathname,
+    });
+
+    const response = await fetchWithTrustedCertificatesAsync(realtimeUrl, {
+      headers: { Authorization: session.token },
+      signal,
+    });
+
+    if (!response.ok) {
+      logger.warn("Beszel SSE connection failed", {
+        integrationId: this.integration.id,
+        systemId,
+        status: response.status,
+      });
+      throw new ResponseError(response);
+    }
+
+    const reader = response.body?.getReader();
+    if (!reader) throw new Error("No response body in SSE stream");
+
+    const decoder = new TextDecoder();
+    let buffer = "";
+    let subscribed = false;
+
+    const processLine = async (line: string) => {
+      if (!line.startsWith("data: ")) return;
+
+      try {
+        const parsed = JSON.parse(line.slice(6)) as Record<string, unknown>;
+
+        // PB_CONNECT delivers the clientId — use it to POST our subscriptions
+        if (!subscribed && typeof parsed.clientId === "string") {
+          subscribed = true;
+          const subscribeResponse = await fetchWithTrustedCertificatesAsync(realtimeUrl, {
+            method: "POST",
+            headers: {
+              Authorization: session.token,
+              "Content-Type": "application/json",
+            },
+            body: JSON.stringify({
+              clientId: parsed.clientId,
+              subscriptions: ["system_stats", "container_stats"],
+            }),
+            signal,
+          });
+
+          if (!subscribeResponse.ok) {
+            logger.warn("Beszel SSE subscription POST failed", {
+              integrationId: this.integration.id,
+              systemId,
+              status: subscribeResponse.status,
+            });
+            throw new ResponseError(subscribeResponse);
+          }
+
+          logger.debug("Beszel SSE subscribed to collections", {
+            integrationId: this.integration.id,
+            systemId,
+            clientId: parsed.clientId,
+          });
+          return;
+        }
+
+        const record = parsed.record as Record<string, unknown> | undefined;
+        if (!record || record.system !== systemId) return;
+
+        if (Array.isArray(record.stats)) {
+          onMessage({
+            type: "container_stats",
+            record: record as unknown as BeszelContainerStatsRecord,
+          });
+        } else if (record.stats && typeof record.stats === "object") {
+          onMessage({
+            type: "system_stats",
+            record: record as unknown as BeszelSystemStatsRecord,
+          });
+        }
+      } catch (error) {
+        if (error instanceof ResponseError) throw error;
+        // skip malformed SSE data lines
+      }
+    };
+
+    try {
+      while (true) {
+        const { done, value } = await reader.read();
+        if (done) break;
+
+        buffer += decoder.decode(value, { stream: true });
+        const lines = buffer.split("\n");
+        buffer = lines.pop() ?? "";
+
+        for (const line of lines) {
+          await processLine(line);
+        }
+      }
+    } finally {
+      reader.releaseLock();
+      logger.debug("Beszel SSE connection closed", { integrationId: this.integration.id, systemId });
+    }
+  }
+
   protected async testingAsync(input: IntegrationTestingInput): Promise<TestingResult> {
     const response = await input.fetchAsync(this.url("/api/collections/users/auth-with-password"), {
       method: "POST",

packages/integrations/src/beszel/beszel-types.ts

@@ -343,6 +343,22 @@ export interface UpdateAlertInput {
   min?: number;
 }
 
+/**
+ * Live SSE subscription event types for real-time Beszel metrics.
+ * Discriminated union — clients accumulate each type into its own buffer.
+ */
+export interface LiveSystemStatsEvent {
+  type: "system_stats";
+  record: BeszelSystemStatsRecord;
+}
+
+export interface LiveContainerStatsEvent {
+  type: "container_stats";
+  record: BeszelContainerStatsRecord;
+}
+
+export type LiveStatsEvent = LiveSystemStatsEvent | LiveContainerStatsEvent;
+
 export interface PocketBaseListResponse<T> {
   page: number;
   perPage: number;

packages/integrations/src/mock/data/beszel.ts

@@ -7,6 +7,7 @@ import type {
   BeszelSystemDetails,
   BeszelSystemStats,
   BeszelSystemStatsRecord,
+  LiveStatsEvent,
 } from "../../beszel/beszel-types";
 
 const GB = 1024 * 1024 * 1024;
@@ -332,7 +333,7 @@ export class BeszelMockService {
 
   public async subscribeRealtimeMetrics(
     systemId: string,
-    onMessage: (data: { stats: BeszelSystemStatsRecord; containerStats: BeszelContainerStatsRecord | null }) => void,
+    onMessage: (event: LiveStatsEvent) => void,
     signal: AbortSignal,
   ): Promise<void> {
     const sys = resolveSystem(systemId);
@@ -348,7 +349,7 @@ export class BeszelMockService {
         created: new Date().toISOString(),
         updated: new Date().toISOString(),
       };
-      onMessage({ stats: record, containerStats: null });
+      onMessage({ type: "system_stats", record });
     };
     if (signal.aborted) return;
     await new Promise<void>((resolve) => {