feat: retain application log history (#6633)

* feat: retain application log history

* fix: harden retained log streaming

* fix: harden contextual log focus

* fix: retain multiline log output

packages/api/src/router/log.ts

@@ -11,6 +11,10 @@ import { createTRPCRouter, permissionRequiredProcedure } from "../trpc";
 
 const logger = createLogger({ module: "logRouter" });
 
+type LogSubscriptionEvent =
+  | { type: "history"; messages: LoggerMessage[] }
+  | { type: "message"; message: LoggerMessage };
+
 export const logRouter = createTRPCRouter({
   subscribe: permissionRequiredProcedure
     .requiresPermission("other-view-logs")
@@ -20,15 +24,61 @@ export const logRouter = createTRPCRouter({
       }),
     )
     .subscription(({ input }) => {
-      return observable<LoggerMessage>((emit) => {
-        const unsubscribe = loggingChannel.subscribe((data) => {
-          if (!input.levels.includes(data.level)) return;
-          emit.next(data);
-        });
-        logger.info("A tRPC client has connected to the logging procedure");
+      return observable<LogSubscriptionEvent>((emit) => {
+        let isClosed = false;
+        let unsubscribe: (() => void) | null = null;
+        let isHistoryLoaded = false;
+        const bufferedMessages: LoggerMessage[] = [];
+
+        const onMessage = (message: LoggerMessage) => {
+          if (!input.levels.includes(message.level)) return;
+
+          if (!isHistoryLoaded) {
+            bufferedMessages.push(message);
+            return;
+          }
+
+          emit.next({ type: "message", message });
+        };
+
+        void loggingChannel
+          .subscribeAsync(onMessage)
+          .then(async (unsubscribeFromLogging) => {
+            if (isClosed) {
+              unsubscribeFromLogging();
+              return;
+            }
+
+            unsubscribe = unsubscribeFromLogging;
+            logger.info("A tRPC client has connected to the logging procedure");
+
+            const history = (await loggingChannel.getRecentAsync()).filter((message) =>
+              input.levels.includes(message.level),
+            );
+            if (isClosed) return;
+
+            emit.next({ type: "history", messages: history });
+
+            const historyIds = new Set(history.map((message) => message.id));
+            isHistoryLoaded = true;
+            for (const message of bufferedMessages) {
+              if (!historyIds.has(message.id)) {
+                emit.next({ type: "message", message });
+              }
+            }
+            bufferedMessages.length = 0;
+          })
+          .catch((error: unknown) => {
+            unsubscribe?.();
+            unsubscribe = null;
+            if (!isClosed) {
+              emit.error(error);
+            }
+          });
 
         return () => {
-          unsubscribe();
+          isClosed = true;
+          unsubscribe?.();
         };
       });
     }),

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<LoggerMessage[]>();
+    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();
+  });
+});