Implement log streaming with subscription-based approach instead of polling

Co-authored-by: ajnart <49837342+ajnart@users.noreply.github.com>

packages/api/src/router/docker/docker-router.ts

@@ -4,7 +4,11 @@ import { z } from "zod/v4";
 
 import type { Container, ContainerState, Docker, Port } from "@homarr/docker";
 import { DockerSingleton } from "@homarr/docker";
-import { dockerContainersRequestHandler, getContainerLogsAsync } from "@homarr/request-handler/docker";
+import {
+  dockerContainersRequestHandler,
+  getContainerLogsAsync,
+  streamContainerLogsAsync,
+} from "@homarr/request-handler/docker";
 
 import { dockerMiddleware } from "../../middlewares/docker";
 import { createTRPCRouter, permissionRequiredProcedure } from "../../trpc";
@@ -127,6 +131,51 @@ export const dockerRouter = createTRPCRouter({
         logs,
       };
     }),
+  subscribeLogs: permissionRequiredProcedure
+    .requiresPermission("admin")
+    .concat(dockerMiddleware())
+    .input(
+      z.object({
+        id: z.string(),
+        tail: z.number().min(1).max(1000).optional(),
+      }),
+    )
+    .subscription(({ input }) => {
+      return observable<string>((emit) => {
+        let cleanup: (() => void) | undefined;
+
+        streamContainerLogsAsync(
+          input.id,
+          input.tail ?? 200,
+          (data) => {
+            emit.next(data);
+          },
+          (err) => {
+            emit.error(
+              new TRPCError({
+                code: "INTERNAL_SERVER_ERROR",
+                message: err.message,
+              }),
+            );
+          },
+        )
+          .then((cleanupFn) => {
+            cleanup = cleanupFn;
+          })
+          .catch((err) => {
+            emit.error(
+              new TRPCError({
+                code: "NOT_FOUND",
+                message: err instanceof Error ? err.message : "Container not found",
+              }),
+            );
+          });
+
+        return () => {
+          cleanup?.();
+        };
+      });
+    }),
 });
 
 const getContainerOrDefaultAsync = async (instance: Docker, id: string) => {

packages/modals-collection/src/docker/docker-logs-modal.tsx

@@ -1,6 +1,6 @@
 "use client";
 
-import { useEffect, useRef } from "react";
+import { useEffect, useRef, useState } from "react";
 import { Box, Group, Loader, ScrollArea, Stack, Text } from "@mantine/core";
 
 import { clientApi } from "@homarr/api/client";
@@ -16,41 +16,79 @@ interface DockerLogsModalInnerProps {
 export const DockerLogsModal = createModal<DockerLogsModalInnerProps>(({ innerProps }) => {
   const t = useI18n();
   const viewportRef = useRef<HTMLDivElement | null>(null);
-  const didInitialScrollRef = useRef(false);
+  const [logs, setLogs] = useState<string>("");
+  const [isLoading, setIsLoading] = useState(true);
+  const [error, setError] = useState<string | null>(null);
+  const shouldScrollRef = useRef(true);
 
-  const { data, isPending, isError } = clientApi.docker.logs.useQuery(
-    { id: innerProps.id, tail: innerProps.tail },
+  // Get initial logs
+  const { data: initialData } = clientApi.docker.logs.useQuery(
+    { id: innerProps.id, tail: innerProps.tail ?? 200 },
     {
       refetchOnWindowFocus: false,
       retry: 0,
-      refetchInterval: 1000,
-      refetchIntervalInBackground: true,
     },
   );
 
+  // Subscribe to streaming logs
+  clientApi.docker.subscribeLogs.useSubscription(
+    { id: innerProps.id, tail: innerProps.tail ?? 200 },
+    {
+      onData(data) {
+        setLogs((prev) => prev + data);
+        setIsLoading(false);
+      },
+      onError(err) {
+        setError(err.message);
+        setIsLoading(false);
+      },
+    },
+  );
+
+  // Initialize with static logs
   useEffect(() => {
-    if (didInitialScrollRef.current) return;
-    if (!data?.logs) return;
+    if (initialData?.logs) {
+      setLogs(initialData.logs);
+      setIsLoading(false);
+    }
+  }, [initialData]);
+
+  // Auto-scroll to bottom when new logs arrive
+  useEffect(() => {
+    if (!shouldScrollRef.current) return;
     const viewport = viewportRef.current;
     if (!viewport) return;
-    viewport.scrollTop = viewport.scrollHeight;
-    didInitialScrollRef.current = true;
-  }, [data?.logs]);
+    
+    // Check if user has scrolled up
+    const isAtBottom = viewport.scrollHeight - viewport.scrollTop - viewport.clientHeight < 100;
+    if (isAtBottom) {
+      viewport.scrollTop = viewport.scrollHeight;
+    }
+  }, [logs]);
+
+  // Handle manual scroll to determine auto-scroll behavior
+  const handleScroll = () => {
+    const viewport = viewportRef.current;
+    if (!viewport) return;
+    
+    const isAtBottom = viewport.scrollHeight - viewport.scrollTop - viewport.clientHeight < 100;
+    shouldScrollRef.current = isAtBottom;
+  };
 
   return (
     <Box>
       <Stack gap="sm">
-        {isPending ? (
+        {isLoading ? (
           <Group justify="center" py="md">
             <Loader size="sm" />
           </Group>
-        ) : isError ? (
+        ) : error ? (
           <Text c="red" size="sm">
-            Failed to load logs
+            {error}
           </Text>
         ) : (
-          <ScrollArea h="75vh" viewportRef={viewportRef}>
-            {data?.logs ? (
+          <ScrollArea h="75vh" viewportRef={viewportRef} onScrollPositionChange={handleScroll}>
+            {logs ? (
               <pre
                 style={{
                   margin: 0,
@@ -58,7 +96,7 @@ export const DockerLogsModal = createModal<DockerLogsModalInnerProps>(({ innerPr
                   overflowWrap: "anywhere",
                 }}
               >
-                <code>{data.logs}</code>
+                <code>{logs}</code>
               </pre>
             ) : (
               <Text size="sm" c="dimmed">

packages/request-handler/src/docker.ts

@@ -57,6 +57,57 @@ export const getContainerLogsAsync = async (id: string, tail = 200) => {
   return decodeDockerLogs(rawLogs);
 };
 
+export const streamContainerLogsAsync = async (
+  id: string,
+  tail: number,
+  onData: (data: string) => void,
+  onError: (err: Error) => void,
+) => {
+  const container = await findContainerByIdAsync(id);
+  if (!container) {
+    onError(new Error("Container not found"));
+    return () => undefined;
+  }
+
+  const stream = await container.logs({
+    tail,
+    stdout: true,
+    stderr: true,
+    follow: true,
+  });
+
+  // Docker multiplexed stream processing
+  let buffer = Buffer.alloc(0);
+
+  const processChunk = (chunk: Buffer) => {
+    buffer = Buffer.concat([buffer, chunk]);
+
+    while (buffer.length >= 8) {
+      const length = buffer.readUInt32BE(4);
+      const fullMessageSize = 8 + length;
+
+      if (buffer.length < fullMessageSize) {
+        // Wait for more data
+        break;
+      }
+
+      const payload = buffer.subarray(8, fullMessageSize);
+      onData(payload.toString("utf-8"));
+
+      buffer = buffer.subarray(fullMessageSize);
+    }
+  };
+
+  stream.on("data", processChunk);
+  stream.on("error", onError);
+
+  return () => {
+    stream.removeListener("data", processChunk);
+    stream.removeListener("error", onError);
+    stream.destroy();
+  };
+};
+
 const decodeDockerLogs = (logs: Buffer | string) => {
   if (typeof logs === "string") {
     return logs;