diff --git a/apps/nextjs/src/app/[locale]/manage/tools/logs/client.tsx b/apps/nextjs/src/app/[locale]/manage/tools/logs/client.tsx index f25e72cbb..aa3c8da4c 100644 --- a/apps/nextjs/src/app/[locale]/manage/tools/logs/client.tsx +++ b/apps/nextjs/src/app/[locale]/manage/tools/logs/client.tsx @@ -1,15 +1,14 @@ "use client"; -import type { ComponentType } from "react"; import dynamic from "next/dynamic"; interface ClientSideTerminalComponentProps { focusTimestamp?: number; } -export const ClientSideTerminalComponent = dynamic( +export const ClientSideTerminalComponent = dynamic( () => import("./terminal").then(({ TerminalComponent }) => TerminalComponent), { ssr: false, }, -) as ComponentType; +); diff --git a/apps/nextjs/src/app/[locale]/manage/tools/logs/page.tsx b/apps/nextjs/src/app/[locale]/manage/tools/logs/page.tsx index 75fe27119..8c02cd39b 100644 --- a/apps/nextjs/src/app/[locale]/manage/tools/logs/page.tsx +++ b/apps/nextjs/src/app/[locale]/manage/tools/logs/page.tsx @@ -39,14 +39,9 @@ export default async function LogsManagementPage({ searchParams }: LogsManagemen notFound(); } - const { focus } = await searchParams; - let focusTimestamp: number | undefined; - if (focus !== undefined) { - const parsedFocusTimestamp = Number(focus); - if (Number.isSafeInteger(parsedFocusTimestamp) && parsedFocusTimestamp > 0) { - focusTimestamp = parsedFocusTimestamp; - } - } + const parsedFocusTimestamp = Number((await searchParams).focus); + const focusTimestamp = + Number.isSafeInteger(parsedFocusTimestamp) && parsedFocusTimestamp > 0 ? parsedFocusTimestamp : undefined; return ( diff --git a/apps/nextjs/src/app/[locale]/manage/tools/logs/terminal.tsx b/apps/nextjs/src/app/[locale]/manage/tools/logs/terminal.tsx index 45650cce4..3ae54239b 100644 --- a/apps/nextjs/src/app/[locale]/manage/tools/logs/terminal.tsx +++ b/apps/nextjs/src/app/[locale]/manage/tools/logs/terminal.tsx @@ -17,37 +17,22 @@ import classes from "./terminal.module.css"; const ALL_LOG_LEVELS = [...logLevels]; const TERMINAL_SCROLLBACK_LINES = LOG_HISTORY_MAX_ENTRIES * 4; -interface FocusMarker { - index: number; - text: string; -} - const getFocusMarker = ( messages: LoggerMessage[], visibleMessages: LoggerMessage[], focusTimestamp: number | undefined, markerText: string, expiredMarkerText: string, -): FocusMarker | null => { +) => { if (focusTimestamp === undefined) return null; - let oldestTimestamp = Number.POSITIVE_INFINITY; - for (const message of messages) { - oldestTimestamp = Math.min(oldestTimestamp, message.timestamp.getTime()); - } - if (oldestTimestamp === Number.POSITIVE_INFINITY || focusTimestamp < oldestTimestamp) { + const oldestTimestamp = messages[0]?.timestamp.getTime(); + if (oldestTimestamp === undefined || focusTimestamp < oldestTimestamp) { return { index: 0, text: expiredMarkerText }; } - let markerIndex = 0; - for (const [index, message] of visibleMessages.entries()) { - if (message.timestamp.getTime() <= focusTimestamp) { - markerIndex = index + 1; - } - } - return { - index: markerIndex, + index: visibleMessages.findLastIndex((message) => message.timestamp.getTime() <= focusTimestamp) + 1, text: markerText, }; }; @@ -104,10 +89,7 @@ const renderMessages = ({ output.push(`\x1b[33m--- ${focusMarker.text} ---\x1b[0m`); } - terminal.reset(); - if (output.length === 0) return; - - terminal.write(`${output.join("\r\n")}\r\n`, () => { + terminal.write(`\x1bc${output.join("\r\n")}${output.length > 0 ? "\r\n" : ""}`, () => { if (!focusMarker) { terminal.scrollToBottom(); return; @@ -134,8 +116,8 @@ export const TerminalComponent = ({ focusTimestamp }: TerminalComponentProps) => const messagesRef = useRef([]); const initialFontSizeRef = useRef(fontSize); const activeLevelsRef = useRef(activeLevels); - const markerTextRef = useRef(t("log.context.widgetError")); - const expiredMarkerTextRef = useRef(t("log.context.expiredWidgetError")); + const markerTextRef = useRef(""); + const expiredMarkerTextRef = useRef(""); const focusTimestampRef = useRef(focusTimestamp); activeLevelsRef.current = activeLevels; diff --git a/packages/api/src/router/test/log.spec.ts b/packages/api/src/router/test/log.spec.ts new file mode 100644 index 000000000..9b9d3f6b1 --- /dev/null +++ b/packages/api/src/router/test/log.spec.ts @@ -0,0 +1,79 @@ +import { describe, expect, test, vi } from "vitest"; + +import type { Session } from "@homarr/auth"; +import type { LoggerMessage } from "@homarr/redis"; + +import { logRouter } from "../log"; + +const mocks = vi.hoisted(() => ({ + getRecentAsync: vi.fn(), + subscribeAsync: vi.fn(), +})); + +vi.mock("@homarr/redis", () => ({ + loggingChannel: { + getRecentAsync: mocks.getRecentAsync, + subscribeAsync: mocks.subscribeAsync, + }, +})); + +const session = { + user: { + id: "log-viewer", + permissions: ["other-view-logs"], + colorScheme: "light", + }, + expires: new Date(0).toISOString(), +} satisfies Session; + +const createMessage = (id: string, timestamp: number): LoggerMessage => ({ + id, + timestamp: new Date(timestamp), + level: "info", + message: id, +}); + +describe("log.subscribe", () => { + test("emits history before buffered live messages without duplicates", async () => { + const historyDeferred = Promise.withResolvers(); + let onMessage: ((message: LoggerMessage) => void) | undefined; + const unsubscribe = vi.fn(); + + mocks.subscribeAsync.mockImplementation(async (callback: (message: LoggerMessage) => void) => { + onMessage = callback; + return unsubscribe; + }); + mocks.getRecentAsync.mockReturnValue(historyDeferred.promise); + + const caller = logRouter.createCaller({ + db: null as never, + deviceType: undefined, + session, + }); + const subscription = await caller.subscribe({ levels: ["info"] }); + const events: object[] = []; + const observer = subscription.subscribe({ + next(event) { + events.push(event); + }, + }); + + await vi.waitFor(() => expect(mocks.getRecentAsync).toHaveBeenCalledOnce()); + + const retainedMessage = createMessage("retained", 1); + const liveMessage = createMessage("live", 2); + onMessage?.(retainedMessage); + onMessage?.(liveMessage); + historyDeferred.resolve([retainedMessage]); + + await vi.waitFor(() => + expect(events).toEqual([ + { type: "history", messages: [retainedMessage] }, + { type: "message", message: liveMessage }, + ]), + ); + + observer.unsubscribe(); + expect(unsubscribe).toHaveBeenCalledOnce(); + }); +}); diff --git a/packages/core/src/infrastructure/logs/constants.ts b/packages/core/src/infrastructure/logs/constants.ts index 46e9e85ec..8edd63f00 100644 --- a/packages/core/src/infrastructure/logs/constants.ts +++ b/packages/core/src/infrastructure/logs/constants.ts @@ -1,12 +1,15 @@ +import z from "zod/v4"; + export const logLevels = ["error", "warn", "info", "debug"] as const; export type LogLevel = (typeof logLevels)[number]; -export interface LoggerMessage { - id: string; - timestamp: Date; - message: string; - level: LogLevel; -} +export const loggerMessageSchema = z.object({ + id: z.string(), + timestamp: z.date(), + message: z.string(), + level: z.enum(logLevels), +}); +export type LoggerMessage = z.infer; export const LOG_HISTORY_MAX_ENTRIES = 5_000; export const LOG_HISTORY_KEY = "list:logging:history"; diff --git a/packages/core/src/infrastructure/logs/transports/redis-transport.ts b/packages/core/src/infrastructure/logs/transports/redis-transport.ts index 66b67e29d..4e8f568a2 100644 --- a/packages/core/src/infrastructure/logs/transports/redis-transport.ts +++ b/packages/core/src/infrastructure/logs/transports/redis-transport.ts @@ -25,9 +25,6 @@ export class RedisTransport extends Transport { this.emit("logged", info); }); - // Is only initialized here because it did not work when initialized in the constructor or outside the class - this.redis ??= createRedisClient(); - const message: LoggerMessage = { id: randomUUID(), timestamp: new Date(), @@ -36,15 +33,24 @@ export class RedisTransport extends Transport { }; const serializedMessage = stringify(message); - this.redis - .multi() - .lpush(LOG_HISTORY_KEY, serializedMessage) - .ltrim(LOG_HISTORY_KEY, 0, LOG_HISTORY_MAX_ENTRIES - 1) - .publish(LOG_PUBLISH_CHANNEL, serializedMessage) - .exec() - .catch(() => { - // Ignore errors + callback(); + + void Promise.resolve() + .then(() => { + // Is only initialized here because it did not work when initialized in the constructor or outside the class + if (!this.redis) { + this.redis = createRedisClient(); + this.redis.options.maxRetriesPerRequest = 1; + this.redis.on("error", () => undefined); + } + + return this.redis + .multi() + .lpush(LOG_HISTORY_KEY, serializedMessage) + .ltrim(LOG_HISTORY_KEY, 0, LOG_HISTORY_MAX_ENTRIES - 1) + .publish(LOG_PUBLISH_CHANNEL, serializedMessage) + .exec(); }) - .finally(callback); + .catch(() => undefined); } } diff --git a/packages/redis/src/index.ts b/packages/redis/src/index.ts index 1891f1f50..857f87c4f 100644 --- a/packages/redis/src/index.ts +++ b/packages/redis/src/index.ts @@ -1,9 +1,15 @@ import { parse } from "superjson"; import type { LoggerMessage } from "@homarr/core/infrastructure/logs/constants"; -import { LOG_HISTORY_KEY, LOG_HISTORY_MAX_ENTRIES } from "@homarr/core/infrastructure/logs/constants"; +import { + loggerMessageSchema, + LOG_HISTORY_KEY, + LOG_HISTORY_MAX_ENTRIES, + LOG_PUBLISH_CHANNEL, +} from "@homarr/core/infrastructure/logs/constants"; import { createListChannel, createSubPubChannel } from "./lib/channel"; +import { ChannelSubscriptionTracker } from "./lib/channel-subscription-tracker"; import { createRedisConnection } from "./lib/connection"; export { @@ -21,14 +27,29 @@ export const pingUrlChannel = createListChannel("ping-url"); export type { LoggerMessage } from "@homarr/core/infrastructure/logs/constants"; -const liveLoggingChannel = createSubPubChannel("logging", { persist: false }); const loggingHistoryClient = createRedisConnection(); +const parseLoggerMessage = (message: string): LoggerMessage | null => { + try { + const result = loggerMessageSchema.safeParse(parse(message)); + return result.success ? result.data : null; + } catch { + return null; + } +}; + export const loggingChannel = { - subscribe: liveLoggingChannel.subscribe, - subscribeAsync: liveLoggingChannel.subscribeAsync, + subscribeAsync: (callback: (message: LoggerMessage) => void) => + ChannelSubscriptionTracker.subscribeAsync(LOG_PUBLISH_CHANNEL, (message) => { + const parsedMessage = parseLoggerMessage(message); + if (parsedMessage) callback(parsedMessage); + }), getRecentAsync: async () => { - const messages = await loggingHistoryClient.lrange(LOG_HISTORY_KEY, 0, LOG_HISTORY_MAX_ENTRIES - 1); - return messages.map((message) => parse(message)).toReversed(); + const messages = + (await loggingHistoryClient?.lrange(LOG_HISTORY_KEY, 0, LOG_HISTORY_MAX_ENTRIES - 1).catch(() => [])) ?? []; + return messages + .map(parseLoggerMessage) + .filter((message): message is LoggerMessage => message !== null) + .toReversed(); }, }; diff --git a/packages/redis/src/lib/channel-subscription-tracker.ts b/packages/redis/src/lib/channel-subscription-tracker.ts index 6a060cb9a..fb70633a2 100644 --- a/packages/redis/src/lib/channel-subscription-tracker.ts +++ b/packages/redis/src/lib/channel-subscription-tracker.ts @@ -58,14 +58,18 @@ export class ChannelSubscriptionTracker { const channelSubscriptions = this.subscriptions.get(channelName) ?? new Map(); const id = randomUUID(); - let ready = this.subscriptionReadiness.get(channelName) ?? Promise.resolve(); + let ready = this.subscriptionReadiness.get(channelName); - // If there are no subscriptions to the channel, subscribe to it - if (channelSubscriptions.size === 0) { + // Subscribe when this channel has no active or pending Redis subscription. + if (!ready) { logger.debug("Subscribing to redis channel", { channel: channelName }); ready = this.redis.subscribe(channelName).then(() => undefined); this.subscriptionReadiness.set(channelName, ready); - void ready.catch(() => undefined); + void ready.catch(() => { + if (this.subscriptionReadiness.get(channelName) === ready) { + this.subscriptionReadiness.delete(channelName); + } + }); } logger.debug("Adding redis channel callback", { channel: channelName, id }); diff --git a/packages/redis/src/lib/channel.ts b/packages/redis/src/lib/channel.ts index 2228c32b0..4731ccfa1 100644 --- a/packages/redis/src/lib/channel.ts +++ b/packages/redis/src/lib/channel.ts @@ -1,4 +1,4 @@ -import superjson, { parse } from "superjson"; +import superjson from "superjson"; import { createId } from "@homarr/common"; @@ -8,12 +8,6 @@ import { createRedisConnection } from "./connection"; const publisher = createRedisConnection(); const lastDataClient = createRedisConnection(); -const deserializeCallback = - (callback: (data: TData) => void) => - (message: string) => { - callback(parse(message)); - }; - /** * Creates a new pub/sub channel. * @param name name of the channel @@ -35,19 +29,9 @@ export const createSubPubChannel = (name: string, { persist }: { persist: } }); } - return ChannelSubscriptionTracker.subscribe(channelName, deserializeCallback(callback)); - }, - /** - * Subscribes to the channel and resolves once Redis confirms the subscription. - */ - subscribeAsync: async (callback: (data: TData) => void) => { - if (persist) { - const data = await lastDataClient.get(lastChannelName); - if (data) { - callback(superjson.parse(data)); - } - } - return await ChannelSubscriptionTracker.subscribeAsync(channelName, deserializeCallback(callback)); + return ChannelSubscriptionTracker.subscribe(channelName, (message) => { + callback(superjson.parse(message)); + }); }, /** * Publish data to the channel with last data saved. diff --git a/packages/widgets/src/errors/base-component.tsx b/packages/widgets/src/errors/base-component.tsx index 96a6e85c3..a42365f52 100644 --- a/packages/widgets/src/errors/base-component.tsx +++ b/packages/widgets/src/errors/base-component.tsx @@ -19,14 +19,13 @@ export const BaseWidgetError = (props: BaseWidgetErrorProps) => { const t = useI18n(); const { data: session } = useSession(); const [errorTimestamp] = useState(Date.now); - const canViewLogs = session?.user.permissions.includes("other-view-logs") ?? false; return ( {translateIfNecessary(t, props.message)} - {props.showLogsLink && canViewLogs && ( + {props.showLogsLink && session?.user.permissions.includes("other-view-logs") && ( ({ useI18n: () => (key: string) => key, })); +vi.mock("@homarr/auth/client", () => ({ + useSession: () => ({ data: null }), +})); + const matchMedia = (query: string) => ({ matches: false, media: query,