From 05534fce3278530b48a8081f1afa47dfe98f8cfd Mon Sep 17 00:00:00 2001
From: t8y2 <1156263951@qq.com>
Date: Mon, 14 Sep 2026 13:36:11 +0800
Subject: [PATCH] feat(plugin): add shared stream bridge

---
 .../src/lib/plugins/pluginHostBridge.spec.ts  |  1 +
 .../src/lib/plugins/pluginHostBridge.ts       | 51 +++++++++++++++++++
 plugins/sdk/dev-host/browser-bridge.mjs       | 51 +++++++++++++++++++
 3 files changed, 103 insertions(+)

diff --git a/apps/desktop/src/lib/plugins/pluginHostBridge.spec.ts b/apps/desktop/src/lib/plugins/pluginHostBridge.spec.ts
index d562f59f17..c851a41fef 100644
--- a/apps/desktop/src/lib/plugins/pluginHostBridge.spec.ts
+++ b/apps/desktop/src/lib/plugins/pluginHostBridge.spec.ts
@@ -325,6 +325,7 @@ describe("PluginHostBridge", () => {
 
     const closed = pluginSandboxDocument("<html><head></head><body></body></html>", ["host.events"]);
     expect(closed).toContain("connect-src 'none';");
+    expect(closed).toContain("host.stream.chunk");
   });
 
   it("sends the theme in init and pushes theme updates through env messages", () => {
diff --git a/apps/desktop/src/lib/plugins/pluginHostBridge.ts b/apps/desktop/src/lib/plugins/pluginHostBridge.ts
index 9b59cc4d85..fd08349d29 100644
--- a/apps/desktop/src/lib/plugins/pluginHostBridge.ts
+++ b/apps/desktop/src/lib/plugins/pluginHostBridge.ts
@@ -342,6 +342,56 @@ function pluginSdkSource(): string {
       for (const byte of bytes) binary += String.fromCharCode(byte);
       return btoa(binary);
     };
+    const stream = async (method, params = {}, options = {}) => {
+      const streamId = options.streamId || (globalThis.crypto?.randomUUID?.() || 'stream-' + Date.now() + '-' + (++sequence));
+      const closeMethod = options.closeMethod || 'filesystem/stream/close';
+      let removeListener;
+      let closeRequested = false;
+      let resolveOpen;
+      let rejectOpen;
+      const metadata = {};
+      const opened = new Promise((resolve, reject) => { resolveOpen = resolve; rejectOpen = reject; });
+      const readable = new ReadableStream({
+        start(controller) {
+          const onEvent = (message) => {
+            if (message?.method !== 'host.stream.chunk' && message?.method !== 'host.stream.end' && message?.method !== 'host.stream.error') return;
+            const event = message.params || {};
+            if (event.streamId !== streamId) return;
+            if (message.method === 'host.stream.chunk') {
+              try { controller.enqueue(decode(event.dataBase64 || '')); } catch (error) { controller.error(error); }
+              return;
+            }
+            removeListener?.();
+          removeListener = undefined;
+          if (message.method === 'host.stream.error') {
+              const error = new Error(event.message || 'Plugin stream failed');
+              rejectOpen(error);
+              controller.error(error);
+          } else {
+            Object.assign(metadata, event);
+            controller.close();
+            }
+          };
+          removeListener = () => listeners.event.delete(onEvent);
+          listeners.event.add(onEvent);
+          request('backend.invoke', { method, params: { ...(params || {}), streamId }, timeoutMs: options.timeoutMs }).then(resolveOpen, (error) => {
+            removeListener?.();
+            removeListener = undefined;
+            rejectOpen(error);
+            controller.error(error);
+          });
+        },
+        cancel() {
+          removeListener?.();
+          removeListener = undefined;
+          if (closeRequested) return undefined;
+          closeRequested = true;
+          return request('backend.invoke', { method: closeMethod, params: { streamId } }).catch(() => undefined);
+        },
+      });
+      Object.assign(metadata, await opened);
+      return { stream: readable, metadata };
+    };
     window.dbxPlugin = Object.freeze({
       ready,
       get context() { return context; },
@@ -349,6 +399,7 @@ function pluginSdkSource(): string {
       get theme() { return theme; },
       request,
       invoke: (method, params, options = {}) => request('backend.invoke', { method, params, timeoutMs: options.timeoutMs }),
+      stream,
       notify: (method, params) => request('backend.notify', { method, params }),
       sendBinary: (channel, data) => {
         if (typeof data === 'string') return request('backend.sendBinary', { channel, dataBase64: data });
diff --git a/plugins/sdk/dev-host/browser-bridge.mjs b/plugins/sdk/dev-host/browser-bridge.mjs
index e60a2209db..e670a42837 100644
--- a/plugins/sdk/dev-host/browser-bridge.mjs
+++ b/plugins/sdk/dev-host/browser-bridge.mjs
@@ -21,6 +21,56 @@ function installBridge(channel) {
     return btoa(result);
   };
   const decode = (value) => Uint8Array.from(atob(value), (c) => c.charCodeAt(0));
+  const stream = async (method, params = {}, options = {}) => {
+    const streamId = options.streamId || (globalThis.crypto?.randomUUID?.() || "stream-" + Date.now() + "-" + (++sequence));
+    const closeMethod = options.closeMethod || "filesystem/stream/close";
+    let removeListener;
+    let closeRequested = false;
+    let resolveOpen;
+    let rejectOpen;
+    const metadata = {};
+    const opened = new Promise((resolve, reject) => { resolveOpen = resolve; rejectOpen = reject; });
+    const readable = new ReadableStream({
+      start(controller) {
+        const onEvent = (message) => {
+          if (message?.method !== "host.stream.chunk" && message?.method !== "host.stream.end" && message?.method !== "host.stream.error") return;
+          const event = message.params || {};
+          if (event.streamId !== streamId) return;
+          if (message.method === "host.stream.chunk") {
+            try { controller.enqueue(decode(event.dataBase64 || "")); } catch (error) { controller.error(error); }
+            return;
+          }
+          removeListener?.();
+          removeListener = undefined;
+          if (message.method === "host.stream.error") {
+            const error = new Error(event.message || "Plugin stream failed");
+            rejectOpen(error);
+            controller.error(error);
+          } else {
+            Object.assign(metadata, event);
+            controller.close();
+          }
+        };
+        removeListener = () => listeners.event.delete(onEvent);
+        listeners.event.add(onEvent);
+        request("backend.invoke", { method, params: { ...(params || {}), streamId }, timeoutMs: options.timeoutMs }).then(resolveOpen, (error) => {
+          removeListener?.();
+          removeListener = undefined;
+          rejectOpen(error);
+          controller.error(error);
+        });
+      },
+      cancel() {
+        removeListener?.();
+        removeListener = undefined;
+        if (closeRequested) return undefined;
+        closeRequested = true;
+        return request("backend.invoke", { method: closeMethod, params: { streamId } }).catch(() => undefined);
+      },
+    });
+    Object.assign(metadata, await opened);
+    return { stream: readable, metadata };
+  };
   const request = (method, params) =>
     new Promise((resolve, reject) => {
       const id = ++sequence;
@@ -56,6 +106,7 @@ function installBridge(channel) {
     },
     request,
     invoke: (method, params, options = {}) => request("backend.invoke", { method, params, timeoutMs: options.timeoutMs }),
+    stream,
     notify: (method, params) => request("backend.notify", { method, params }),
     sendBinary: (channel, data) => request("backend.sendBinary", { channel, dataBase64: typeof data === "string" ? data : encode(data) }),
     readAsset: (path) => request("ui.readAsset", { path }),