From 24c5d04bfa2002ece62c11381e20c2b621e395f5 Mon Sep 17 00:00:00 2001 From: Thomas Date: Fri, 10 Jul 2026 17:16:23 +0200 Subject: [PATCH] fix(beszel): disable request handler cache for stats to make live mode work (#6244) --- packages/api/src/router/widgets/beszel.ts | 67 ++++++++++ .../src/docs/homarr-docs-sitemap.ts | 2 + .../src/beszel/beszel-integration.ts | 120 ++++++++++++++++++ .../integrations/src/beszel/beszel-types.ts | 16 +++ packages/integrations/src/mock/data/beszel.ts | 5 +- packages/request-handler/src/beszel.ts | 3 + .../src/lib/integration-request-handler.ts | 2 + .../src/beszel/_shared/use-live-stats.ts | 63 ++++++++- 8 files changed, 269 insertions(+), 9 deletions(-) diff --git a/packages/api/src/router/widgets/beszel.ts b/packages/api/src/router/widgets/beszel.ts index 797de8609..0273f6896 100644 --- a/packages/api/src/router/widgets/beszel.ts +++ b/packages/api/src/router/widgets/beszel.ts @@ -1,7 +1,10 @@ import { TRPCError } from "@trpc/server"; +import { observable } from "@trpc/server/observable"; import { z } from "zod/v4"; import { createLogger } from "@homarr/core/infrastructure/logs"; +import { createIntegrationAsync } from "@homarr/integrations"; +import type { LiveStatsEvent } from "@homarr/integrations/types"; import { beszelAlertsRequestHandler, beszelStatsRequestHandler, @@ -201,4 +204,68 @@ export const beszelRouter = createTRPCRouter({ }; } }), + + subscribeSystemStats: publicProcedure + .concat(createManyIntegrationMiddleware("query", "beszel", "mock")) + .input( + z.object({ + systemId: z.string(), + }), + ) + .subscription(({ ctx, input }) => { + return observable((emit) => { + const controller = new AbortController(); + let isActive = true; + + void (async () => { + const integration = ctx.integrations[0]; + if (!integration) { + emit.error( + new TRPCError({ code: "BAD_REQUEST", message: "At least one Beszel integrationId is required" }), + ); + return; + } + + try { + const instance = await createIntegrationAsync(integration); + if (typeof instance.subscribeRealtimeMetrics === "function") { + await instance.subscribeRealtimeMetrics( + input.systemId, + (event) => { + if (isActive) emit.next(event); + }, + controller.signal, + ); + } else { + emit.error( + new TRPCError({ + code: "METHOD_NOT_SUPPORTED", + message: `Integration ${integration.kind} does not support realtime metrics`, + }), + ); + } + } catch (error) { + if (isActive) { + emit.error( + error instanceof TRPCError + ? error + : new TRPCError({ + code: "INTERNAL_SERVER_ERROR", + message: error instanceof Error ? error.message : String(error), + }), + ); + } + } + + if (isActive) { + emit.complete(); + } + })(); + + return () => { + isActive = false; + controller.abort(); + }; + }); + }), }); diff --git a/packages/definitions/src/docs/homarr-docs-sitemap.ts b/packages/definitions/src/docs/homarr-docs-sitemap.ts index bd813adc0..2e87554c7 100644 --- a/packages/definitions/src/docs/homarr-docs-sitemap.ts +++ b/packages/definitions/src/docs/homarr-docs-sitemap.ts @@ -200,6 +200,7 @@ export type HomarrDocumentationPath = | "/docs/integrations/tdarr" | "/docs/integrations/technitium-dns" | "/docs/integrations/tracearr" + | "/docs/integrations/traefik" | "/docs/integrations/transmission" | "/docs/integrations/truenas" | "/docs/integrations/umami" @@ -262,6 +263,7 @@ export type HomarrDocumentationPath = | "/docs/widgets/system-resources" | "/docs/widgets/timetable" | "/docs/widgets/tracearr" + | "/docs/widgets/traefik" | "/docs/widgets/umami" | "/docs/widgets/ups" | "/docs/widgets/uptime-kuma" diff --git a/packages/integrations/src/beszel/beszel-integration.ts b/packages/integrations/src/beszel/beszel-integration.ts index 186ed2203..e9c4b35b2 100644 --- a/packages/integrations/src/beszel/beszel-integration.ts +++ b/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 { + 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; + + // 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 | 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 { const response = await input.fetchAsync(this.url("/api/collections/users/auth-with-password"), { method: "POST", diff --git a/packages/integrations/src/beszel/beszel-types.ts b/packages/integrations/src/beszel/beszel-types.ts index e5c6566a5..1abbc6564 100644 --- a/packages/integrations/src/beszel/beszel-types.ts +++ b/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 { page: number; perPage: number; diff --git a/packages/integrations/src/mock/data/beszel.ts b/packages/integrations/src/mock/data/beszel.ts index c6998aa84..a83347553 100644 --- a/packages/integrations/src/mock/data/beszel.ts +++ b/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 { 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((resolve) => { diff --git a/packages/request-handler/src/beszel.ts b/packages/request-handler/src/beszel.ts index 4a60a3a3a..a6619e6ef 100644 --- a/packages/request-handler/src/beszel.ts +++ b/packages/request-handler/src/beszel.ts @@ -122,6 +122,9 @@ export const beszelStatsRequestHandler = createIntegrationRequestHandler< "beszel" | "mock", { systemId: string; timePeriod: string; includeDocker: boolean } >({ + // No cache — the widget polls every 5s for live updates; + // a TTL here would serve stale records between polls. + cacheTtlMs: 0, async requestAsync(integration, input) { const start = performance.now(); const config = timePeriodConfig[input.timePeriod] ?? { type: "1m", perPage: 60 }; diff --git a/packages/request-handler/src/lib/integration-request-handler.ts b/packages/request-handler/src/lib/integration-request-handler.ts index 359d63dbc..e60be1fb8 100644 --- a/packages/request-handler/src/lib/integration-request-handler.ts +++ b/packages/request-handler/src/lib/integration-request-handler.ts @@ -12,6 +12,7 @@ type IntegrationOfKind = Omit> { requestAsync: (integration: IntegrationOfKind, input: TInput) => Promise; + cacheTtlMs?: number; } export const createIntegrationRequestHandler = < @@ -30,6 +31,7 @@ export const createIntegrationRequestHandler = < } return options.requestAsync(integration, input.options); }, + cacheTtlMs: options.cacheTtlMs, }); return { diff --git a/packages/widgets/src/beszel/_shared/use-live-stats.ts b/packages/widgets/src/beszel/_shared/use-live-stats.ts index b07bbbd2c..f8c351bf8 100644 --- a/packages/widgets/src/beszel/_shared/use-live-stats.ts +++ b/packages/widgets/src/beszel/_shared/use-live-stats.ts @@ -1,15 +1,64 @@ "use client"; +import { useCallback, useEffect, useRef, useState } from "react"; + import { clientApi } from "@homarr/api/client"; +import type { BeszelContainerStatsRecord, BeszelSystemStatsRecord } from "@homarr/integrations/types"; + +// 120 records at ~1 per second = 2 minutes of rolling data for charts +const MAX_BUFFER = 120; export const useLiveStats = (integrationIds: string[], systemId: string, enabled: boolean) => { - const { data, error } = clientApi.widget.beszel.getSystemStats.useQuery( - { integrationIds, systemId, timePeriod: "1m" as const, includeDocker: true }, - { enabled: enabled && systemId !== "", refetchInterval: 5_000 }, + const [systemStats, setSystemStats] = useState([]); + const [containerStats, setContainerStats] = useState([]); + const [error, setError] = useState(null); + + // Append to buffer, trimming to MAX_BUFFER + const appendSystemStats = useCallback((record: BeszelSystemStatsRecord) => { + setSystemStats((prev) => { + const next = [...prev, record]; + return next.length > MAX_BUFFER ? next.slice(-MAX_BUFFER) : next; + }); + }, []); + + const appendContainerStats = useCallback((record: BeszelContainerStatsRecord) => { + setContainerStats((prev) => { + const next = [...prev, record]; + return next.length > MAX_BUFFER ? next.slice(-MAX_BUFFER) : next; + }); + }, []); + + clientApi.widget.beszel.subscribeSystemStats.useSubscription( + { integrationIds, systemId }, + { + enabled: enabled && systemId !== "", + onData(event) { + setError(null); + if (event.type === "system_stats") { + appendSystemStats(event.record); + } else { + appendContainerStats(event.record); + } + }, + onError(err) { + setError(err instanceof Error ? err : new Error(String(err))); + }, + }, ); - return { - data: data ? { systemStats: data.systemStats, containerStats: data.containerStats } : null, - error: error ?? (data?.error ? new Error(data.error) : null), - }; + // Reset buffers when system changes or integration changes + const prevKeyRef = useRef(""); + const currentKey = `${integrationIds.join(",")}:${systemId}`; + useEffect(() => { + if (prevKeyRef.current !== currentKey) { + prevKeyRef.current = currentKey; + setSystemStats([]); + setContainerStats([]); + setError(null); + } + }, [currentKey]); + + const data = systemStats.length > 0 || containerStats.length > 0 ? { systemStats, containerStats } : null; + + return { data, error }; };