From 34d647d6112fd2d7634968a40a59cd290d0662a8 Mon Sep 17 00:00:00 2001
From: hypercross <helios.tnak@gmail.com>
Date: Mon, 6 Jul 2026 15:38:28 +0800
Subject: [PATCH] feat: implement websocket-stream for journal server

Integrate `websocket-stream` to handle MQTT broker upgrades in the
CLI and add granular connection status tracking to the UI.

- Add `websocket-stream` dependency
- Update CLI to use `websocket-stream` for socket upgrades
- Enhance `journalStream` store with `connectionStatus` and
  `connectionError` states
- Add visual connection status indicators to `App` and `JournalPanel`
---
 package-lock.json                       | 67 ++++++++++++++++++++++++-
 package.json                            |  1 +
 src/App.tsx                             | 16 +++++-
 src/cli/journal.ts                      |  7 ++-
 src/components/journal/JournalPanel.tsx | 22 +++++---
 src/components/stores/journalStream.ts  | 15 ++++++
 6 files changed, 116 insertions(+), 12 deletions(-)

diff --git a/package-lock.json b/package-lock.json
index cc5b115..9318c03 100644
--- a/package-lock.json
+++ b/package-lock.json
@@ -29,6 +29,7 @@
         "solid-js": "^1.9.3",
         "three": "^0.183.2",
         "three-3mf-exporter": "^45.1.0",
+        "websocket-stream": "^5.5.2",
         "ws": "^8.21.0",
         "yarn-spinner-loader": "^0.1.0",
         "zod": "^4.4.3"
@@ -3723,6 +3724,12 @@
       "integrity": "sha512-8+9WqebbFzpX9OR+Wa6O29asIogeRMzcGtAINdpMHHyAg10f05aSFVBbcEqGf/PXw1EjAZ+q2/bEBg3DvurK3Q==",
       "license": "Python-2.0"
     },
+    "node_modules/async-limiter": {
+      "version": "1.0.1",
+      "resolved": "https://registry.npmjs.org/async-limiter/-/async-limiter-1.0.1.tgz",
+      "integrity": "sha512-csOlWGAcRFJaI6m+F2WKdnMKr4HhdhFVBk0H/QbJFMCr+uO2kwohwXQPxw/9OCxp05r5ghVBFSyioixx3gfkNQ==",
+      "license": "MIT"
+    },
     "node_modules/asynckit": {
       "version": "0.4.0",
       "resolved": "https://registry.npmjs.org/asynckit/-/asynckit-0.4.0.tgz",
@@ -5260,6 +5267,18 @@
         "node": ">= 0.4"
       }
     },
+    "node_modules/duplexify": {
+      "version": "3.7.1",
+      "resolved": "https://registry.npmjs.org/duplexify/-/duplexify-3.7.1.tgz",
+      "integrity": "sha512-07z8uv2wMyS51kKhD1KsdXJg5WQ6t93RneqRxUHnskXVtlYYkLqM0gqStQZ3pj073g687jPCHrqNfCzawLYh5g==",
+      "license": "MIT",
+      "dependencies": {
+        "end-of-stream": "^1.0.0",
+        "inherits": "^2.0.1",
+        "readable-stream": "^2.0.0",
+        "stream-shift": "^1.0.0"
+      }
+    },
     "node_modules/electron-to-chromium": {
       "version": "1.5.331",
       "resolved": "https://registry.npmjs.org/electron-to-chromium/-/electron-to-chromium-1.5.331.tgz",
@@ -5287,6 +5306,15 @@
       "dev": true,
       "license": "MIT"
     },
+    "node_modules/end-of-stream": {
+      "version": "1.4.5",
+      "resolved": "https://registry.npmjs.org/end-of-stream/-/end-of-stream-1.4.5.tgz",
+      "integrity": "sha512-ooEGc6HP26xXq/N+GCGOT0JKCLDGrq2bQUZrQ7gyrJiZANJ/8YDTxTpQBXGMn+WbIQXNVpyWymm7KYVICQnyOg==",
+      "license": "MIT",
+      "dependencies": {
+        "once": "^1.4.0"
+      }
+    },
     "node_modules/enhanced-resolve": {
       "version": "5.20.1",
       "resolved": "https://registry.npmjs.org/enhanced-resolve/-/enhanced-resolve-5.20.1.tgz",
@@ -8810,7 +8838,6 @@
       "version": "1.4.0",
       "resolved": "https://registry.npmjs.org/once/-/once-1.4.0.tgz",
       "integrity": "sha512-lNaJgI+2Q5URQBkccEKHTQOPaXdUxnZZElQTZY0MFUAuaEqe1E+Nyvgdz/aIyNi6Z9MzO5dv1H8n58/GELp3+w==",
-      "dev": true,
       "license": "ISC",
       "dependencies": {
         "wrappy": "1"
@@ -9756,6 +9783,12 @@
         "node": ">= 0.8"
       }
     },
+    "node_modules/stream-shift": {
+      "version": "1.0.3",
+      "resolved": "https://registry.npmjs.org/stream-shift/-/stream-shift-1.0.3.tgz",
+      "integrity": "sha512-76ORR0DO1o1hlKwTbi/DM3EXWGf3ZJYO8cXX5RJwnul2DEg2oyoZyjLNoQM8WsvZiFKCRfC1O0J7iCvie3RZmQ==",
+      "license": "MIT"
+    },
     "node_modules/string_decoder": {
       "version": "1.1.1",
       "resolved": "https://registry.npmjs.org/string_decoder/-/string_decoder-1.1.1.tgz",
@@ -10264,6 +10297,12 @@
         "node": ">=0.8.0"
       }
     },
+    "node_modules/ultron": {
+      "version": "1.1.1",
+      "resolved": "https://registry.npmjs.org/ultron/-/ultron-1.1.1.tgz",
+      "integrity": "sha512-UIEXBNeYmKptWH6z8ZnqTeS8fV74zG0/eRU9VGkpzz+LIJNs8W/zM/L+7ctCkRrgbNnnR0xxw4bKOr0cW0N0Og==",
+      "license": "MIT"
+    },
     "node_modules/undici-types": {
       "version": "6.21.0",
       "resolved": "https://registry.npmjs.org/undici-types/-/undici-types-6.21.0.tgz",
@@ -10547,6 +10586,31 @@
         "node": ">=12"
       }
     },
+    "node_modules/websocket-stream": {
+      "version": "5.5.2",
+      "resolved": "https://registry.npmjs.org/websocket-stream/-/websocket-stream-5.5.2.tgz",
+      "integrity": "sha512-8z49MKIHbGk3C4HtuHWDtYX8mYej1wWabjthC/RupM9ngeukU4IWoM46dgth1UOS/T4/IqgEdCDJuMe2039OQQ==",
+      "license": "BSD-2-Clause",
+      "dependencies": {
+        "duplexify": "^3.5.1",
+        "inherits": "^2.0.1",
+        "readable-stream": "^2.3.3",
+        "safe-buffer": "^5.1.2",
+        "ws": "^3.2.0",
+        "xtend": "^4.0.0"
+      }
+    },
+    "node_modules/websocket-stream/node_modules/ws": {
+      "version": "3.3.3",
+      "resolved": "https://registry.npmjs.org/ws/-/ws-3.3.3.tgz",
+      "integrity": "sha512-nnWLa/NwZSt4KQJu51MYlCcSQ5g7INpOrOMt4XV8j4dqTXdmlUmSHQ8/oLC069ckre0fRsgfvsKwbTdtKLCDkA==",
+      "license": "MIT",
+      "dependencies": {
+        "async-limiter": "~1.0.0",
+        "safe-buffer": "~5.1.0",
+        "ultron": "~1.1.0"
+      }
+    },
     "node_modules/whatwg-encoding": {
       "version": "2.0.0",
       "resolved": "https://registry.npmjs.org/whatwg-encoding/-/whatwg-encoding-2.0.0.tgz",
@@ -10677,7 +10741,6 @@
       "version": "1.0.2",
       "resolved": "https://registry.npmjs.org/wrappy/-/wrappy-1.0.2.tgz",
       "integrity": "sha512-l4Sp/DRseor9wL6EvV2+TuQn63dMkPjZ/sp9XkghTEbV9KlPS1xUsZ3u7/IQO4wxtcFB4bgpQPRcR3QCvezPcQ==",
-      "dev": true,
       "license": "ISC"
     },
     "node_modules/write-file-atomic": {
diff --git a/package.json b/package.json
index 3896d16..4181d4b 100644
--- a/package.json
+++ b/package.json
@@ -52,6 +52,7 @@
     "solid-js": "^1.9.3",
     "three": "^0.183.2",
     "three-3mf-exporter": "^45.1.0",
+    "websocket-stream": "^5.5.2",
     "ws": "^8.21.0",
     "yarn-spinner-loader": "^0.1.0",
     "zod": "^4.4.3"
diff --git a/src/App.tsx b/src/App.tsx
index b45c644..13bc11c 100644
--- a/src/App.tsx
+++ b/src/App.tsx
@@ -1,5 +1,6 @@
 import { Component, createEffect, createMemo, createSignal } from "solid-js";
 import { useLocation } from "@solidjs/router";
+import { useJournalStream } from "./components/stores/journalStream";
 
 // 导入组件以注册自定义元素
 import "./components";
@@ -15,6 +16,7 @@ import { JournalPanel } from "./components/journal";
 
 const App: Component = () => {
   const location = useLocation();
+  const stream = useJournalStream();
   const [isSidebarOpen, setIsSidebarOpen] = createSignal(false);
   const [isDocOpen, setIsDocOpen] = createSignal(false);
   const [isDataSourceOpen, setIsDataSourceOpen] = createSignal(false);
@@ -84,13 +86,23 @@ const App: Component = () => {
           <div class="flex-1" />
           <button
             onClick={() => setIsJournalOpen((v) => !v)}
-            class={`w-8 h-8 rounded flex items-center justify-center text-sm ${
+            class={`relative w-8 h-8 rounded flex items-center justify-center text-sm ${
               isJournalOpen()
                 ? "bg-blue-100 text-blue-600"
                 : "text-gray-500 hover:bg-gray-100"
             }`}
-            title="Journal"
+            title={`Journal (${stream.connectionStatus})`}
           >
+            <span
+              class="absolute top-0.5 right-0.5 w-2 h-2 rounded-full border border-white"
+              classList={{
+                "bg-gray-400": stream.connectionStatus === "disconnected",
+                "bg-yellow-400 animate-pulse":
+                  stream.connectionStatus === "connecting",
+                "bg-green-500": stream.connectionStatus === "connected",
+                "bg-red-500": stream.connectionStatus === "error",
+              }}
+            />
             📋
           </button>
           <button
diff --git a/src/cli/journal.ts b/src/cli/journal.ts
index 804a9d3..1b59c0a 100644
--- a/src/cli/journal.ts
+++ b/src/cli/journal.ts
@@ -20,6 +20,7 @@ import {
 } from "fs";
 import type { Server as HttpServer } from "http";
 import type { AedesPublishPacket } from "aedes";
+import websocketStream from "websocket-stream";
 
 // ---------------------------------------------------------------------------
 // Types
@@ -106,7 +107,11 @@ export async function createJournalServer(
 
   httpServer.on("upgrade", (req, socket, head) => {
     wsServer.handleUpgrade(req, socket, head, (ws) => {
-      broker.handle(ws as unknown as import("net").Socket);
+      // websocket-stream's types reference an older ws version; cast works.
+      const stream = websocketStream(
+        ws as unknown as Parameters<typeof websocketStream>[0],
+      );
+      broker.handle(stream);
     });
   });
 
diff --git a/src/components/journal/JournalPanel.tsx b/src/components/journal/JournalPanel.tsx
index 9ffdc73..e19d9e7 100644
--- a/src/components/journal/JournalPanel.tsx
+++ b/src/components/journal/JournalPanel.tsx
@@ -38,12 +38,17 @@ export const JournalPanel: Component<JournalPanelProps> = (props) => {
         <div class="flex items-center justify-between px-3 py-2 border-b border-gray-200">
           <div class="flex items-center gap-2">
             <h2 class="text-sm font-semibold text-gray-700">Journal</h2>
-            <Show when={stream.connected}>
-              <span
-                class="w-2 h-2 rounded-full bg-green-500"
-                title="Connected"
-              />
-            </Show>
+            <span
+              class="w-2 h-2 rounded-full"
+              classList={{
+                "bg-gray-400": stream.connectionStatus === "disconnected",
+                "bg-yellow-400 animate-pulse":
+                  stream.connectionStatus === "connecting",
+                "bg-green-500": stream.connectionStatus === "connected",
+                "bg-red-500": stream.connectionStatus === "error",
+              }}
+              title={stream.connectionStatus}
+            />
           </div>
           <button
             onClick={props.onClose}
@@ -104,7 +109,10 @@ const ConnectDialog: Component = () => {
       await hydrateFromServer(sessionId);
       await connectStream(sessionId, brokerUrl);
     } catch (e) {
-      setError(e instanceof Error ? e.message : "Connection failed");
+      const msg = e instanceof Error ? e.message : "Connection failed";
+      setError(msg);
+      // If connectStream threw, the error handler also sets connectionStatus to error.
+      // Make sure the error message is captured.
     } finally {
       setConnecting(false);
     }
diff --git a/src/components/stores/journalStream.ts b/src/components/stores/journalStream.ts
index bb44ad4..ec74d96 100644
--- a/src/components/stores/journalStream.ts
+++ b/src/components/stores/journalStream.ts
@@ -33,6 +33,10 @@ export interface JournalStreamState {
   revealedPaths: Set<string>;
   /** MQTT connection status */
   connected: boolean;
+  /** Granular connection state for UI indicators */
+  connectionStatus: "disconnected" | "connecting" | "connected" | "error";
+  /** Last connection error message, if any */
+  connectionError: string | null;
   /** This client's identity */
   myName: string;
   /** Broker URL, set after connect */
@@ -84,6 +88,8 @@ const [state, setState] = createStore<JournalStreamState>({
   senderSeq: {},
   revealedPaths: new Set(),
   connected: false,
+  connectionStatus: "disconnected",
+  connectionError: null,
   myName: persisted.myName,
   brokerUrl: persisted.brokerUrl,
 });
@@ -189,6 +195,9 @@ export async function connectStream(
 ): Promise<void> {
   const { default: mqtt } = await import("mqtt");
 
+  setState("connectionStatus", "connecting");
+  setState("connectionError", null);
+
   const client = mqtt.connect(brokerUrl, {
     clientId: `${state.myName}-${Date.now()}`,
     protocol: brokerUrl.startsWith("wss") ? "wss" : "ws",
@@ -200,6 +209,8 @@ export async function connectStream(
   client.on("connect", () => {
     _mqttConnected = true;
     setState("connected", true);
+    setState("connectionStatus", "connected");
+    setState("connectionError", null);
     setState("brokerUrl", brokerUrl);
 
     // Persist connection info for next time
@@ -251,10 +262,13 @@ export async function connectStream(
   client.on("close", () => {
     _mqttConnected = false;
     setState("connected", false);
+    setState("connectionStatus", "disconnected");
   });
 
   client.on("error", (err) => {
     console.error("[stream] mqtt error:", err);
+    setState("connectionStatus", "error");
+    setState("connectionError", err.message);
   });
 }
 
@@ -416,6 +430,7 @@ export function disconnectStream(): void {
     _mqttConnected = false;
   }
   setState("connected", false);
+  setState("connectionStatus", "disconnected");
 }
 
 // ---------------------------------------------------------------------------