perf: WebSocket graph diffs — broadcast only what changed

src/client/hooks/useDocker.ts

@@ -1,8 +1,46 @@
 import { useState, useEffect, useRef, useCallback } from "react";
-import type { Service, Connection, Stats, DockerEvent, LogLine, WSMessage, ActionError, EventLogEntry, NotificationLogEntry } from "../../shared/types";
+import type { Service, Connection, Stats, DockerEvent, LogLine, WSMessage, GraphDiff, ActionError, EventLogEntry, NotificationLogEntry } from "../../shared/types";
 import type { StatsStore } from "./useStatsStore";
 import { arraysEqual, applyProcessing as applyProcessingPure } from "./processing";
 
+function connKey(c: Connection): string {
+  return `${c.from}|${c.to}|${c.network}`;
+}
+
+function applyServicesDiff(current: Service[], diff: GraphDiff): Service[] {
+  let result = [...current];
+  if (diff.servicesRemoved?.length) {
+    const removed = new Set(diff.servicesRemoved);
+    result = result.filter((s) => !removed.has(s.uid));
+  }
+  if (diff.servicesUpdated?.length) {
+    const updated = new Map(diff.servicesUpdated.map((s) => [s.uid, s]));
+    result = result.map((s) => updated.get(s.uid) ?? s);
+  }
+  if (diff.servicesAdded?.length) {
+    const existing = new Set(result.map((s) => s.uid));
+    for (const s of diff.servicesAdded) {
+      if (!existing.has(s.uid)) result.push(s);
+    }
+  }
+  return result;
+}
+
+function applyConnectionsDiff(current: Connection[], diff: GraphDiff): Connection[] {
+  let result = [...current];
+  if (diff.connectionsRemoved?.length) {
+    const removed = new Set(diff.connectionsRemoved);
+    result = result.filter((c) => !removed.has(connKey(c)));
+  }
+  if (diff.connectionsAdded?.length) {
+    const existing = new Set(result.map(connKey));
+    for (const c of diff.connectionsAdded) {
+      if (!existing.has(connKey(c))) result.push(c);
+    }
+  }
+  return result;
+}
+
 export function useDocker(token = "", statsStore?: StatsStore, onPositions?: (pos: Record<string, { x: number; y: number }>) => void) {
   const [services, setServices] = useState<Service[]>([]);
   const [connections, setConnections] = useState<Connection[]>([]);
@@ -86,19 +124,24 @@ export function useDocker(token = "", statsStore?: StatsStore, onPositions?: (po
         }
 
         switch (msg.type as WSMessage["type"]) {
-          case "services": {
-            lastRawServicesRef.current = msg.data as Service[];
-            const incoming = applyProcessing(msg.data as Service[]);
+          case "snapshot": {
+            lastRawServicesRef.current = msg.data.services as Service[];
+            const incoming = applyProcessing(msg.data.services as Service[]);
             setServices((prev) => arraysEqual(prev, incoming) ? prev : incoming);
+            setConnections(msg.data.connections as Connection[]);
             break;
           }
-          case "connections":
-            setConnections((prev) => {
-              if (prev.length === msg.data.length &&
-                  prev.every((c: any, i: number) => c.from === msg.data[i].from && c.to === msg.data[i].to)) return prev;
-              return msg.data;
-            });
+          case "diff": {
+            const diff = msg.data as GraphDiff;
+            const nextRaw = applyServicesDiff(lastRawServicesRef.current, diff);
+            lastRawServicesRef.current = nextRaw;
+            const incoming = applyProcessing(nextRaw);
+            setServices((prev) => arraysEqual(prev, incoming) ? prev : incoming);
+            if (diff.connectionsAdded?.length || diff.connectionsRemoved?.length) {
+              setConnections((prev) => applyConnectionsDiff(prev, diff));
+            }
             break;
+          }
           case "stats": {
             for (const s of msg.data) {
               statsRef.current.set(s.service, s);

src/server/diff.test.ts

@@ -0,0 +1,120 @@
+import { describe, test, expect } from "vitest";
+import { computeGraphDiff, connectionKey } from "./diff";
+import type { Service, Connection } from "../shared/types";
+
+function svc(uid: string, overrides: Partial<Service> = {}): Service {
+  return {
+    id: uid,
+    uid,
+    name: uid.split("/")[1] ?? uid,
+    image: "nginx:latest",
+    state: "running",
+    status: "Up 2 hours",
+    ports: [],
+    networks: ["bridge"],
+    network_ips: {},
+    project: uid.split("/")[0] ?? "proj",
+    compose_file: "/app/docker-compose.yml",
+    env: [],
+    restart_policy: "unless-stopped",
+    memory_limit: 0,
+    cpu_quota: 0,
+    health_status: "",
+    health_log: [],
+    exit_code: 0,
+    restart_count: 0,
+    oom_killed: false,
+    mounts: [],
+    ...overrides,
+  };
+}
+
+function conn(from: string, to: string, network = "bridge"): Connection {
+  return { from, to, network };
+}
+
+describe("computeGraphDiff", () => {
+  test("no-op returns null", () => {
+    const services = [svc("p/a"), svc("p/b")];
+    const connections = [conn("p/a", "p/b")];
+    expect(computeGraphDiff(services, services, connections, connections)).toBeNull();
+  });
+
+  test("add only", () => {
+    const prev = [svc("p/a")];
+    const next = [svc("p/a"), svc("p/b")];
+    const diff = computeGraphDiff(prev, next, [], []);
+    expect(diff).not.toBeNull();
+    expect(diff!.servicesAdded).toHaveLength(1);
+    expect(diff!.servicesAdded![0].uid).toBe("p/b");
+    expect(diff!.servicesRemoved).toBeUndefined();
+    expect(diff!.servicesUpdated).toBeUndefined();
+  });
+
+  test("remove only", () => {
+    const prev = [svc("p/a"), svc("p/b")];
+    const next = [svc("p/a")];
+    const diff = computeGraphDiff(prev, next, [], []);
+    expect(diff!.servicesRemoved).toEqual(["p/b"]);
+    expect(diff!.servicesAdded).toBeUndefined();
+  });
+
+  test("update only — state change", () => {
+    const prev = [svc("p/a", { state: "running" })];
+    const next = [svc("p/a", { state: "exited", exit_code: 1 })];
+    const diff = computeGraphDiff(prev, next, [], []);
+    expect(diff!.servicesUpdated).toHaveLength(1);
+    expect(diff!.servicesUpdated![0].state).toBe("exited");
+    expect(diff!.servicesAdded).toBeUndefined();
+    expect(diff!.servicesRemoved).toBeUndefined();
+  });
+
+  test("update only — restart_count increment", () => {
+    const prev = [svc("p/a", { restart_count: 0 })];
+    const next = [svc("p/a", { restart_count: 1 })];
+    const diff = computeGraphDiff(prev, next, [], []);
+    expect(diff!.servicesUpdated).toHaveLength(1);
+  });
+
+  test("status string change does NOT trigger update", () => {
+    const prev = [svc("p/a", { status: "Up 1 minute" })];
+    const next = [svc("p/a", { status: "Up 2 hours" })];
+    const diff = computeGraphDiff(prev, next, [], []);
+    expect(diff).toBeNull();
+  });
+
+  test("health_log change does NOT trigger update", () => {
+    const prev = [svc("p/a", { health_log: ["healthy at 10:00"] })];
+    const next = [svc("p/a", { health_log: ["healthy at 11:00"] })];
+    const diff = computeGraphDiff(prev, next, [], []);
+    expect(diff).toBeNull();
+  });
+
+  test("mixed: add + remove + update", () => {
+    const prev = [svc("p/a"), svc("p/b"), svc("p/c")];
+    const next = [svc("p/a", { state: "exited" }), svc("p/d")];
+    const diff = computeGraphDiff(prev, next, [], []);
+    expect(diff!.servicesAdded?.map((s) => s.uid)).toEqual(["p/d"]);
+    expect(diff!.servicesRemoved?.sort()).toEqual(["p/b", "p/c"]);
+    expect(diff!.servicesUpdated?.map((s) => s.uid)).toEqual(["p/a"]);
+  });
+
+  test("connection add only", () => {
+    const svcs = [svc("p/a"), svc("p/b")];
+    const diff = computeGraphDiff(svcs, svcs, [], [conn("p/a", "p/b")]);
+    expect(diff!.connectionsAdded).toHaveLength(1);
+    expect(diff!.connectionsRemoved).toBeUndefined();
+  });
+
+  test("connection remove only", () => {
+    const svcs = [svc("p/a"), svc("p/b")];
+    const diff = computeGraphDiff(svcs, svcs, [conn("p/a", "p/b")], []);
+    expect(diff!.connectionsRemoved).toHaveLength(1);
+    expect(diff!.connectionsRemoved![0]).toBe("p/a|p/b|bridge");
+    expect(diff!.connectionsAdded).toBeUndefined();
+  });
+
+  test("connectionKey format", () => {
+    expect(connectionKey(conn("p/a", "p/b", "mynet"))).toBe("p/a|p/b|mynet");
+  });
+});

src/server/diff.ts

@@ -0,0 +1,54 @@
+import type { Service, Connection, GraphDiff } from "../shared/types";
+
+/** Stable fingerprint of a Service for change detection.
+ *  Excludes `status` (verbose uptime string that changes every minute) and
+ *  `health_log` (changes on every health-check interval). */
+function serviceSignature(s: Service): string {
+  return JSON.stringify([
+    s.state, s.image, s.restart_count, s.health_status,
+    s.exit_code, s.oom_killed, s.restart_policy, s.memory_limit, s.cpu_quota,
+    s.ports, s.networks, s.network_ips, s.env, s.mounts, s.compose_file,
+    s.name, s.project,
+  ]);
+}
+
+export function connectionKey(c: Connection): string {
+  return `${c.from}|${c.to}|${c.network}`;
+}
+
+/** Returns a GraphDiff between two graph states, or null when nothing changed. */
+export function computeGraphDiff(
+  prevServices: Service[],
+  nextServices: Service[],
+  prevConnections: Connection[],
+  nextConnections: Connection[],
+): GraphDiff | null {
+  const prevSvcMap = new Map(prevServices.map((s) => [s.uid, s]));
+  const nextSvcMap = new Map(nextServices.map((s) => [s.uid, s]));
+
+  const servicesAdded = nextServices.filter((s) => !prevSvcMap.has(s.uid));
+  const servicesRemoved = prevServices.filter((s) => !nextSvcMap.has(s.uid)).map((s) => s.uid);
+  const servicesUpdated = nextServices.filter((s) => {
+    const prev = prevSvcMap.get(s.uid);
+    return prev !== undefined && serviceSignature(prev) !== serviceSignature(s);
+  });
+
+  const prevConnKeys = new Set(prevConnections.map(connectionKey));
+  const nextConnKeys = new Set(nextConnections.map(connectionKey));
+  const connectionsAdded = nextConnections.filter((c) => !prevConnKeys.has(connectionKey(c)));
+  const connectionsRemoved = prevConnections.filter((c) => !nextConnKeys.has(connectionKey(c))).map(connectionKey);
+
+  const hasChanges =
+    servicesAdded.length > 0 || servicesRemoved.length > 0 || servicesUpdated.length > 0 ||
+    connectionsAdded.length > 0 || connectionsRemoved.length > 0;
+
+  if (!hasChanges) return null;
+
+  const diff: GraphDiff = {};
+  if (servicesAdded.length > 0) diff.servicesAdded = servicesAdded;
+  if (servicesRemoved.length > 0) diff.servicesRemoved = servicesRemoved;
+  if (servicesUpdated.length > 0) diff.servicesUpdated = servicesUpdated;
+  if (connectionsAdded.length > 0) diff.connectionsAdded = connectionsAdded;
+  if (connectionsRemoved.length > 0) diff.connectionsRemoved = connectionsRemoved;
+  return diff;
+}

src/server/index.ts

@@ -6,6 +6,7 @@ import path from "path";
 import fs from "fs";
 import { docker, discoverServices, discoverConnections, getContainerLogs, streamContainerLogs } from "./docker";
 import { pollStats, watchDockerEvents } from "./watcher";
+import { computeGraphDiff } from "./diff";
 import { loadDiscordConfig, saveDiscordConfig, notifyStateChange, notifyResourceAlert, notifyUIAction, notifyActionError, testWebhook, checkDownServices, setNotificationListener } from "./discord";
 import { loadContainerSettings, saveContainerSettings } from "./container-settings";
 import { loadProjectAliases, saveProjectAliases, sanitizeAlias } from "./project-aliases";
@@ -14,7 +15,7 @@ import { getUpdateInfo } from "./update-check";
 import pkg from "../../package.json";
 import { initStatsDB, insertStats, getStatsHistory, getAllServicesStatsHistory } from "./stats-db";
 import { initEventsDB, insertEvent, insertNotification, getEvents, getNotifications, type EventLogEntry, type NotificationLogEntry } from "./events-db";
-import type { Service, Stats, WSMessage, DiscordConfig, ContainerSettings, StatsRange } from "../shared/types";
+import type { Service, Connection, Stats, WSMessage, DiscordConfig, ContainerSettings, StatsRange } from "../shared/types";
 
 /** Directory for persistent data files (SQLite, JSON configs, positions).
  *  Default: ./data subdirectory of cwd. Override via DATA_DIR env var. */
@@ -802,6 +803,28 @@ function cleanupLogStream(ws: WebSocket) {
   }
 }
 
+/** Send full graph + stats snapshot to a newly connected client.
+ *  Uses cached state if available; falls back to a fresh discover on cold start. */
+function sendSnapshot(ws: WebSocket): void {
+  if (lastBroadcastedServices.length > 0) {
+    try {
+      ws.send(JSON.stringify({ type: "snapshot", data: { services: lastBroadcastedServices, connections: lastBroadcastedConnections } }));
+      if (lastStats.length > 0) ws.send(JSON.stringify({ type: "stats", data: lastStats }));
+    } catch {}
+    return;
+  }
+  // Cold start: no data yet — do a fresh discover
+  discoverServices(ALL, PROJECTS).then(async (services) => {
+    const connections = await discoverConnections(services);
+    lastBroadcastedServices = services;
+    lastBroadcastedConnections = connections;
+    try {
+      ws.send(JSON.stringify({ type: "snapshot", data: { services, connections } }));
+      if (lastStats.length > 0) ws.send(JSON.stringify({ type: "stats", data: lastStats }));
+    } catch {}
+  }).catch(() => {});
+}
+
 // ── Docker events ──
 let servicesLock = false;
 let statsLock = false;
@@ -811,18 +834,16 @@ async function refreshServices() {
   servicesLock = true;
   try {
     const services = await discoverServices(ALL, PROJECTS);
-
-    const svcHash = services.map((s) => `${s.uid}:${s.state}`).join("|");
-    if (svcHash !== lastServicesHash) {
-      lastServicesHash = svcHash;
-      broadcast({ type: "services", data: services });
-    }
-
     const connections = await discoverConnections(services);
-    const connHash = connections.map((c) => `${c.from}:${c.to}`).join("|");
-    if (connHash !== lastConnectionsHash) {
-      lastConnectionsHash = connHash;
-      broadcast({ type: "connections", data: connections });
+
+    const diff = computeGraphDiff(
+      lastBroadcastedServices, services,
+      lastBroadcastedConnections, connections,
+    );
+    if (diff) {
+      lastBroadcastedServices = services;
+      lastBroadcastedConnections = connections;
+      broadcast({ type: "diff", data: diff });
     }
 
     // Stats polling is separate — don't block services refresh
@@ -880,8 +901,6 @@ async function refreshStats(services: Service[]) {
 let refreshTimer: ReturnType<typeof setTimeout> | undefined;
 let retryTimer: ReturnType<typeof setTimeout> | undefined;
 function scheduleRefresh() {
-  // Invalidate hash so next refresh always broadcasts (restart: same final state but clients need the update)
-  lastServicesHash = "";
   clearTimeout(refreshTimer);
   clearTimeout(retryTimer);
   refreshTimer = setTimeout(() => {
@@ -892,7 +911,6 @@ function scheduleRefresh() {
 
 // Immediate refresh after action endpoints (container already changed state)
 function immediateRefresh() {
-  lastServicesHash = "";
   clearTimeout(refreshTimer);
   clearTimeout(retryTimer);
   refreshServices();
@@ -920,9 +938,9 @@ watchDockerEvents((event) => {
   } catch {}
 });
 
-// ── Stats polling ──
-let lastServicesHash = "";
-let lastConnectionsHash = "";
+// ── Graph state (used for diff computation and new-client snapshots) ──
+let lastBroadcastedServices: Service[] = [];
+let lastBroadcastedConnections: Connection[] = [];
 /** Last stats snapshot — sent in /api/init so frontend has data immediately
  *  instead of waiting for the next polling cycle (~3s wait). */
 let lastStats: Stats[] = [];
@@ -959,16 +977,7 @@ const server = Bun.serve({
       clients.add(native);
 
       if (!AUTH_TOKEN) {
-        // No auth required — send data immediately
-        discoverServices(ALL, PROJECTS).then(async (services) => {
-          const connections = await discoverConnections(services);
-          const stats = await pollStats(services);
-          try {
-            native.send(JSON.stringify({ type: "services", data: services }));
-            native.send(JSON.stringify({ type: "connections", data: connections }));
-            native.send(JSON.stringify({ type: "stats", data: stats }));
-          } catch {}
-        }).catch(() => {});
+        sendSnapshot(native);
       }
     },
     close(ws) {
@@ -993,16 +1002,7 @@ const server = Bun.serve({
           if (msg.token === AUTH_TOKEN) {
             authenticatedClients.add(native);
             native.send(JSON.stringify({ type: "auth_ok" }));
-            // Send current services/connections/stats immediately
-            discoverServices(ALL, PROJECTS).then(async (services) => {
-              const connections = await discoverConnections(services);
-              const stats = await pollStats(services);
-              try {
-                native.send(JSON.stringify({ type: "services", data: services }));
-                native.send(JSON.stringify({ type: "connections", data: connections }));
-                native.send(JSON.stringify({ type: "stats", data: stats }));
-              } catch {}
-            }).catch(() => {});
+            sendSnapshot(native);
           } else {
             if (AUTH_TOKEN) recordFailedAttempt(wsIp);
             native.send(JSON.stringify({ type: "auth_error" }));

src/shared/types.ts

@@ -165,9 +165,17 @@ export interface NotificationLogEntry {
   message: string;
 }
 
+export interface GraphDiff {
+  servicesAdded?: Service[];
+  servicesRemoved?: string[];      // uids
+  servicesUpdated?: Service[];     // full object (includes uid)
+  connectionsAdded?: Connection[];
+  connectionsRemoved?: string[];   // "from|to|network" keys
+}
+
 export type WSMessage =
-  | { type: "services"; data: Service[] }
-  | { type: "connections"; data: Connection[] }
+  | { type: "snapshot"; data: { services: Service[]; connections: Connection[] } }
+  | { type: "diff"; data: GraphDiff }
   | { type: "stats"; data: Stats[] }
   | { type: "docker_event"; data: DockerEvent }
   | { type: "subscribe_logs"; container: string }