From c9769bf53083147a38c69603fd19933f69d8d06c Mon Sep 17 00:00:00 2001 From: Thomas Camlong Date: Mon, 15 Jun 2026 14:58:47 +0200 Subject: [PATCH] feat(beszel): add SSE subscription endpoint and per-period cache TTL - Add subscribeSystemStats tRPC subscription bridging PocketBase SSE - Add subscribeRealtimeMetrics to BeszelIntegration with SSE parsing - Add error handling: try/catch in subscription, response check in topic POST - Add cacheDurationForInput to request handler for per-period cache TTL - Add mock subscribeRealtimeMetrics with 1s interval for dev testing --- packages/api/src/router/widgets/beszel.ts | 100 ++++++++++++++--- .../src/beszel/beszel-integration.ts | 102 ++++++++++++++++++ packages/integrations/src/mock/data/beszel.ts | 29 +++++ .../integrations/src/mock/mock-integration.ts | 1 + packages/request-handler/src/beszel.ts | 81 ++++++++------ .../lib/cached-integration-request-handler.ts | 8 +- 6 files changed, 272 insertions(+), 49 deletions(-) diff --git a/packages/api/src/router/widgets/beszel.ts b/packages/api/src/router/widgets/beszel.ts index d3b895d88..c1c0fe7c6 100644 --- a/packages/api/src/router/widgets/beszel.ts +++ b/packages/api/src/router/widgets/beszel.ts @@ -1,12 +1,17 @@ import { observable } from "@trpc/server/observable"; import { z } from "zod/v4"; +import { createIntegrationAsync } from "@homarr/integrations"; +import type { BeszelContainerStatsRecord, BeszelSystemStatsRecord } from "@homarr/integrations/types"; import { beszelAlertsRequestHandler, beszelStatsRequestHandler, beszelSystemsRequestHandler, } from "@homarr/request-handler/beszel"; -import type { BeszelAlertsData, BeszelSystemRow } from "@homarr/request-handler/beszel"; +import type { + BeszelAlertsData, + BeszelSystemRow, +} from "@homarr/request-handler/beszel"; import { createManyIntegrationMiddleware } from "../../middlewares/integration"; import { createTRPCRouter, publicProcedure } from "../../trpc"; @@ -17,8 +22,14 @@ export const beszelRouter = createTRPCRouter({ .query(async ({ ctx }) => { const results = await Promise.all( ctx.integrations.map(async (integration) => { - const innerHandler = beszelSystemsRequestHandler.handler(integration, {}); - const { data, timestamp } = await innerHandler.getCachedOrUpdatedDataAsync({ forceUpdate: false }); + const innerHandler = beszelSystemsRequestHandler.handler( + integration, + {}, + ); + const { data, timestamp } = + await innerHandler.getCachedOrUpdatedDataAsync({ + forceUpdate: false, + }); return { integrationId: integration.id, integrationName: integration.name, @@ -34,9 +45,16 @@ export const beszelRouter = createTRPCRouter({ subscribeSystems: publicProcedure .concat(createManyIntegrationMiddleware("query", "beszel", "mock")) .subscription(({ ctx }) => { - return observable<{ integrationId: string; systems: BeszelSystemRow[]; timestamp: Date }>((emit) => { + return observable<{ + integrationId: string; + systems: BeszelSystemRow[]; + timestamp: Date; + }>((emit) => { const unsubscribes = ctx.integrations.map((integration) => { - const innerHandler = beszelSystemsRequestHandler.handler(integration, {}); + const innerHandler = beszelSystemsRequestHandler.handler( + integration, + {}, + ); return innerHandler.subscribe((systems) => { emit.next({ integrationId: integration.id, @@ -62,11 +80,17 @@ export const beszelRouter = createTRPCRouter({ .query(async ({ ctx, input }) => { const results = await Promise.all( ctx.integrations.map(async (integration) => { - const alertsHandler = beszelAlertsRequestHandler.handler(integration, { - includeHistory: input.includeHistory, - maxHistoryItems: input.maxHistoryItems, - }); - const systemsHandler = beszelSystemsRequestHandler.handler(integration, {}); + const alertsHandler = beszelAlertsRequestHandler.handler( + integration, + { + includeHistory: input.includeHistory, + maxHistoryItems: input.maxHistoryItems, + }, + ); + const systemsHandler = beszelSystemsRequestHandler.handler( + integration, + {}, + ); const [alertsResult, systemsResult] = await Promise.all([ alertsHandler.getCachedOrUpdatedDataAsync({ forceUpdate: false }), systemsHandler.getCachedOrUpdatedDataAsync({ forceUpdate: false }), @@ -96,14 +120,22 @@ export const beszelRouter = createTRPCRouter({ }), ) .subscription(({ ctx, input }) => { - return observable<{ integrationId: string; alerts: BeszelAlertsData; timestamp: Date }>((emit) => { + return observable<{ + integrationId: string; + alerts: BeszelAlertsData; + timestamp: Date; + }>((emit) => { const unsubscribes = ctx.integrations.map((integration) => { const innerHandler = beszelAlertsRequestHandler.handler(integration, { includeHistory: input.includeHistory, maxHistoryItems: input.maxHistoryItems, }); return innerHandler.subscribe((data) => { - emit.next({ integrationId: integration.id, alerts: data, timestamp: new Date() }); + emit.next({ + integrationId: integration.id, + alerts: data, + timestamp: new Date(), + }); }); }); return () => { @@ -117,7 +149,7 @@ export const beszelRouter = createTRPCRouter({ .input( z.object({ systemId: z.string(), - timePeriod: z.enum(["1h", "12h", "24h", "1w", "30d"]), + timePeriod: z.enum(["1m", "1h", "12h", "24h", "1w", "30d"]), includeDocker: z.boolean().default(true), }), ) @@ -129,7 +161,47 @@ export const beszelRouter = createTRPCRouter({ timePeriod: input.timePeriod, includeDocker: input.includeDocker, }); - const { data, timestamp } = await innerHandler.getCachedOrUpdatedDataAsync({ forceUpdate: false }); + const { data, timestamp } = + await innerHandler.getCachedOrUpdatedDataAsync({ forceUpdate: false }); return { integrationId: integration.id, ...data, updatedAt: timestamp }; }), + + subscribeSystemStats: publicProcedure + .concat(createManyIntegrationMiddleware("query", "beszel", "mock")) + .input( + z.object({ + systemId: z.string(), + includeDocker: z.boolean().default(true), + }), + ) + .subscription(({ ctx, input }) => { + return observable<{ + stats: BeszelSystemStatsRecord; + containerStats: BeszelContainerStatsRecord | null; + }>((emit) => { + const integration = ctx.integrations[0]; + if (!integration) return () => {}; + + const abortController = new AbortController(); + + void (async () => { + try { + const instance = await createIntegrationAsync(integration); + await instance.subscribeRealtimeMetrics( + input.systemId, + (data) => emit.next(data), + abortController.signal, + ); + } catch (error) { + if (!abortController.signal.aborted) { + emit.error(error instanceof Error ? error : new Error(String(error))); + } + } + })(); + + return () => { + abortController.abort(); + }; + }); + }), }); diff --git a/packages/integrations/src/beszel/beszel-integration.ts b/packages/integrations/src/beszel/beszel-integration.ts index eb646c5dc..d74303164 100644 --- a/packages/integrations/src/beszel/beszel-integration.ts +++ b/packages/integrations/src/beszel/beszel-integration.ts @@ -1,4 +1,5 @@ import { ResponseError } from "@homarr/common/server"; +import { createLogger } from "@homarr/core/infrastructure/logs"; import { fetchWithTrustedCertificatesAsync } from "@homarr/core/infrastructure/http"; import type { IntegrationInput, IntegrationTestingInput } from "../base/integration"; @@ -12,17 +13,21 @@ import type { BeszelAlertHistory, BeszelAuthResponse, BeszelContainer, + BeszelContainerStats, BeszelContainerStatsRecord, BeszelSmartDevice, BeszelSystem, BeszelSystemDetails, BeszelSystemdService, + BeszelSystemStats, BeszelSystemStatsRecord, CreateAlertInput, PocketBaseListResponse, UpdateAlertInput, } from "./beszel-types"; +const logger = createLogger({ module: "beszel-integration" }); + const escapeFilterValue = (value: string) => value.replace(/\\/g, "\\\\").replace(/'/g, "\\'"); interface BeszelSession { @@ -245,6 +250,103 @@ export class BeszelIntegration extends Integration { }); } + public async subscribeRealtimeMetrics( + systemId: string, + onMessage: (data: { stats: BeszelSystemStatsRecord; containerStats: BeszelContainerStatsRecord | null }) => void, + signal: AbortSignal, + ): Promise { + const session = await this.authenticateAsync(); + const sseUrl = this.url("/api/realtime"); + + const sseResponse = await fetchWithTrustedCertificatesAsync(sseUrl, { + headers: { Authorization: session.token }, + signal, + }); + + if (!sseResponse.ok || !sseResponse.body) { + throw new ResponseError(sseResponse); + } + + const reader = sseResponse.body.getReader(); + const decoder = new TextDecoder(); + let buffer = ""; + let clientId: string | null = null; + + const processLine = (line: string) => { + if (!line.startsWith("data:")) return; + const jsonStr = line.slice(5).trim(); + if (!jsonStr) return; + + try { + const parsed = JSON.parse(jsonStr) as Record; + + if (!clientId && typeof parsed.clientId === "string") { + clientId = parsed.clientId; + this.subscribeTopic(clientId, systemId, session.token).catch((err) => { + logger.warn("Failed to subscribe to rt_metrics topic", { error: err }); + }); + return; + } + + if ("stats" in parsed) { + const stats: BeszelSystemStatsRecord = { + id: "", + system: systemId, + stats: parsed.stats as BeszelSystemStats, + type: "1m", + created: new Date().toISOString(), + updated: new Date().toISOString(), + }; + const containerStats: BeszelContainerStatsRecord | null = + parsed.container && Array.isArray(parsed.container) && (parsed.container as BeszelContainerStats[]).length > 0 + ? { + id: "", + system: systemId, + stats: parsed.container as BeszelContainerStats[], + type: "1m", + created: new Date().toISOString(), + updated: new Date().toISOString(), + } + : null; + onMessage({ stats, containerStats }); + } + } catch { + logger.debug("Failed to parse SSE message", { line }); + } + }; + + try { + while (!signal.aborted) { + 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) { + processLine(line); + } + } + } finally { + reader.cancel().catch(() => {}); + } + } + + private async subscribeTopic(clientId: string, systemId: string, token: string): Promise { + const topic = `rt_metrics?options=${JSON.stringify({ query: { system: systemId } })}`; + const response = await fetchWithTrustedCertificatesAsync(this.url("/api/realtime"), { + method: "POST", + headers: { + Authorization: token, + "Content-Type": "application/json", + }, + body: JSON.stringify({ clientId, subscriptions: [topic] }), + }); + if (!response.ok) { + throw new ResponseError(response); + } + } + 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/mock/data/beszel.ts b/packages/integrations/src/mock/data/beszel.ts index 40acc78be..e099f00be 100644 --- a/packages/integrations/src/mock/data/beszel.ts +++ b/packages/integrations/src/mock/data/beszel.ts @@ -330,6 +330,35 @@ export class BeszelMockService { ]; } + public async subscribeRealtimeMetrics( + systemId: string, + onMessage: (data: { stats: BeszelSystemStatsRecord; containerStats: BeszelContainerStatsRecord | null }) => void, + signal: AbortSignal, + ): Promise { + const sys = resolveSystem(systemId); + const emit = () => { + if (signal.aborted) return; + const [stats] = generateSystemStatsSeries(1, sys); + if (!stats) return; + const record: BeszelSystemStatsRecord = { + id: `mock-rt-${Date.now()}`, + system: systemId, + stats, + type: "1m", + created: new Date().toISOString(), + updated: new Date().toISOString(), + }; + onMessage({ stats: record, containerStats: null }); + }; + await new Promise((resolve) => { + const interval = setInterval(emit, 1000); + signal.addEventListener("abort", () => { + clearInterval(interval); + resolve(); + }); + }); + } + public async getAlertHistoryAsync(_systemId?: string, perPage = 10): Promise { const now = Date.now(); const names = ["CPU", "Memory", "Disk", "Temperature", "Bandwidth"]; diff --git a/packages/integrations/src/mock/mock-integration.ts b/packages/integrations/src/mock/mock-integration.ts index 0804267b1..d7eb412cf 100644 --- a/packages/integrations/src/mock/mock-integration.ts +++ b/packages/integrations/src/mock/mock-integration.ts @@ -140,4 +140,5 @@ export class MockIntegration getContainerStatsAsync = MockIntegration.beszel.getContainerStatsAsync.bind(MockIntegration.beszel); getAlertsAsync = MockIntegration.beszel.getAlertsAsync.bind(MockIntegration.beszel); getAlertHistoryAsync = MockIntegration.beszel.getAlertHistoryAsync.bind(MockIntegration.beszel); + subscribeRealtimeMetrics = MockIntegration.beszel.subscribeRealtimeMetrics.bind(MockIntegration.beszel); } diff --git a/packages/request-handler/src/beszel.ts b/packages/request-handler/src/beszel.ts index 053edf52b..f3dfc8f44 100644 --- a/packages/request-handler/src/beszel.ts +++ b/packages/request-handler/src/beszel.ts @@ -24,7 +24,10 @@ export type { BeszelSystemRow } from "@homarr/integrations/types"; * bb=bandwidth(bytes/s), b=bandwidth(Mbps legacy), dt=disk temp, bat=battery, * sv=[running,total] services, h=hostname, m=CPU model, c=cores, ct=threads */ -function mapToSystemRow(system: BeszelSystem, details: BeszelSystemDetails | null): BeszelSystemRow { +function mapToSystemRow( + system: BeszelSystem, + details: BeszelSystemDetails | null, +): BeszelSystemRow { const info = system.info; return { id: system.id, @@ -50,25 +53,28 @@ function mapToSystemRow(system: BeszelSystem, details: BeszelSystemDetails | nul }; } -export const beszelSystemsRequestHandler = createCachedIntegrationRequestHandler< - BeszelSystemRow[], - "beszel" | "mock", - Record ->({ - async requestAsync(integration) { - const instance = await createIntegrationAsync(integration); - const systems = await instance.getSystemsAsync(); - const enriched = await Promise.all( - systems.map(async (system) => { - const details = await instance.getSystemDetailsAsync(system.id).catch(() => null); - return mapToSystemRow(system, details); - }), - ); - return enriched; - }, - cacheDuration: dayjs.duration(5, "seconds"), - queryKey: "beszelSystems", -}); +export const beszelSystemsRequestHandler = + createCachedIntegrationRequestHandler< + BeszelSystemRow[], + "beszel" | "mock", + Record + >({ + async requestAsync(integration) { + const instance = await createIntegrationAsync(integration); + const systems = await instance.getSystemsAsync(); + const enriched = await Promise.all( + systems.map(async (system) => { + const details = await instance + .getSystemDetailsAsync(system.id) + .catch(() => null); + return mapToSystemRow(system, details); + }), + ); + return enriched; + }, + cacheDuration: dayjs.duration(5, "seconds"), + queryKey: "beszelSystems", + }); export interface BeszelAlertsData { alerts: BeszelAlert[]; @@ -83,19 +89,22 @@ export const beszelAlertsRequestHandler = createCachedIntegrationRequestHandler< async requestAsync(integration, input) { const instance = await createIntegrationAsync(integration); const alerts = await instance.getAlertsAsync(); - const history = input.includeHistory ? await instance.getAlertHistoryAsync(undefined, input.maxHistoryItems) : []; + const history = input.includeHistory + ? await instance.getAlertHistoryAsync(undefined, input.maxHistoryItems) + : []; return { alerts, history }; }, cacheDuration: dayjs.duration(15, "seconds"), queryKey: "beszelAlerts", }); -const timePeriodConfig: Record = { - "1h": { type: "1m", perPage: 60 }, - "12h": { type: "10m", perPage: 72 }, - "24h": { type: "20m", perPage: 72 }, - "1w": { type: "120m", perPage: 84 }, - "30d": { type: "480m", perPage: 90 }, +const timePeriodConfig: Record = { + "1m": { type: "1m", perPage: 60, cacheSeconds: 5 }, + "1h": { type: "1m", perPage: 60, cacheSeconds: 60 }, + "12h": { type: "10m", perPage: 72, cacheSeconds: 300 }, + "24h": { type: "20m", perPage: 72, cacheSeconds: 600 }, + "1w": { type: "120m", perPage: 84, cacheSeconds: 1800 }, + "30d": { type: "480m", perPage: 90, cacheSeconds: 3600 }, }; export interface BeszelStatsData { @@ -109,14 +118,24 @@ export const beszelStatsRequestHandler = createCachedIntegrationRequestHandler< { systemId: string; timePeriod: string; includeDocker: boolean } >({ async requestAsync(integration, input) { - const config = timePeriodConfig[input.timePeriod] ?? { type: "1m", perPage: 60 }; + const config = timePeriodConfig[input.timePeriod] ?? { type: "1m", perPage: 60, cacheSeconds: 60 }; const instance = await createIntegrationAsync(integration); - const systemStats = await instance.getSystemStatsAsync(input.systemId, config.type, config.perPage); + const systemStats = await instance.getSystemStatsAsync( + input.systemId, + config.type, + config.perPage, + ); const containerStats = input.includeDocker - ? await instance.getContainerStatsAsync(input.systemId, config.type, config.perPage).catch(() => []) + ? await instance + .getContainerStatsAsync(input.systemId, config.type, config.perPage) + .catch(() => []) : []; return { systemStats, containerStats }; }, - cacheDuration: dayjs.duration(30, "seconds"), + cacheDuration: dayjs.duration(60, "seconds"), queryKey: "beszelStats", + cacheDurationForInput(input) { + const config = timePeriodConfig[input.timePeriod]; + return config ? dayjs.duration(config.cacheSeconds, "seconds") : undefined; + }, }); diff --git a/packages/request-handler/src/lib/cached-integration-request-handler.ts b/packages/request-handler/src/lib/cached-integration-request-handler.ts index 3cae8df0f..8c26fcbe6 100644 --- a/packages/request-handler/src/lib/cached-integration-request-handler.ts +++ b/packages/request-handler/src/lib/cached-integration-request-handler.ts @@ -14,10 +14,10 @@ type IntegrationOfKind = Omit> { - // Unique key for this request handler queryKey: string; requestAsync: (integration: IntegrationOfKind, input: TInput) => Promise; cacheDuration: Duration; + cacheDurationForInput?: (input: TInput) => Duration | undefined; } export const createCachedIntegrationRequestHandler = < @@ -34,9 +34,9 @@ export const createCachedIntegrationRequestHandler = < requestAsync: async (input: { options: TInput; integration: IntegrationOfKind }) => { return await options.requestAsync(input.integration, input.options); }, - cacheDuration: options.cacheDuration, - createRedisChannel(input, options) { - return createIntegrationOptionsChannel(input.integration.id, options.queryKey, input.options); + cacheDuration: options.cacheDurationForInput?.(itemOptions) ?? options.cacheDuration, + createRedisChannel(input, handlerOptions) { + return createIntegrationOptionsChannel(input.integration.id, handlerOptions.queryKey, input.options); }, }).handler({ options: itemOptions, integration }), };