Pandore/homarr
Public Access
1
0 Fork 0
Code Actions Activity

feat(beszel): add SSE subscription endpoint and per-period cache TTL

Browse Source 
- 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
This commit is contained in:
Thomas Camlong committed 2026-06-15 14:58:47 +02:00
1 parent 5035c26dbd
commit c9769bf530
6 files changed
+272 -49
Download Patch File Download Diff File Expand all files Collapse all files

No files matched your search

packages/api/src/router/widgets/beszel.ts
+86 -14
View File
@@ -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();
};
});
}),
});
packages/integrations/src/beszel/beszel-integration.ts
+102
View File
@@ -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<void> {
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<string, unknown>;
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<void> {
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<TestingResult> {
const response = await input.fetchAsync(this.url("/api/collections/users/auth-with-password"), {
method: "POST",
packages/integrations/src/mock/data/beszel.ts
+29
View File
@@ -330,6 +330,35 @@ export class BeszelMockService {
];
}
public async subscribeRealtimeMetrics(
systemId: string,
onMessage: (data: { stats: BeszelSystemStatsRecord; containerStats: BeszelContainerStatsRecord | null }) => void,
signal: AbortSignal,
): Promise<void> {
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<void>((resolve) => {
const interval = setInterval(emit, 1000);
signal.addEventListener("abort", () => {
clearInterval(interval);
resolve();
});
});
}
public async getAlertHistoryAsync(_systemId?: string, perPage = 10): Promise<BeszelAlertHistory[]> {
const now = Date.now();
const names = ["CPU", "Memory", "Disk", "Temperature", "Bandwidth"];
packages/integrations/src/mock/mock-integration.ts
+1
View File
@@ -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);
}
packages/request-handler/src/beszel.ts
+50 -31
View File
@@ -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<string, never>
>({
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<string, never>
>({
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<string, { type: string; perPage: number }> = {
"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<string, { type: string; perPage: number; cacheSeconds: number }> = {
"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;
},
});
packages/request-handler/src/lib/cached-integration-request-handler.ts
+4 -4
View File
@@ -14,10 +14,10 @@ type IntegrationOfKind<TKind extends IntegrationKind> = Omit<Integration, "kind"
};
interface Options<TData, TKind extends IntegrationKind, TInput extends Record<string, unknown>> {
// Unique key for this request handler
queryKey: string;
requestAsync: (integration: IntegrationOfKind<TKind>, input: TInput) => Promise<TData>;
cacheDuration: Duration;
cacheDurationForInput?: (input: TInput) => Duration | undefined;
}
export const createCachedIntegrationRequestHandler = <
@@ -34,9 +34,9 @@ export const createCachedIntegrationRequestHandler = <
requestAsync: async (input: { options: TInput; integration: IntegrationOfKind<TKind> }) => {
return await options.requestAsync(input.integration, input.options);
},
cacheDuration: options.cacheDuration,
createRedisChannel(input, options) {
return createIntegrationOptionsChannel<TData>(input.integration.id, options.queryKey, input.options);
cacheDuration: options.cacheDurationForInput?.(itemOptions) ?? options.cacheDuration,
createRedisChannel(input, handlerOptions) {
return createIntegrationOptionsChannel<TData>(input.integration.id, handlerOptions.queryKey, input.options);
},
}).handler({ options: itemOptions, integration }),
};