diff --git a/apps/desktop/src/components/plugins/PluginWorkbenchHost.vue b/apps/desktop/src/components/plugins/PluginWorkbenchHost.vue index 330854873e..1ccf923209 100644 --- a/apps/desktop/src/components/plugins/PluginWorkbenchHost.vue +++ b/apps/desktop/src/components/plugins/PluginWorkbenchHost.vue @@ -6,6 +6,7 @@ import { isTauriRuntime } from "@/lib/backend/tauriRuntime"; import { copyToClipboard } from "@/lib/common/clipboard"; import { PluginHostBridge, pluginSandboxDocument, type PluginBridgeTheme, type PluginSaveFileRequest, type PluginSaveFileResult, type PluginWorkbenchContext } from "@/lib/plugins/pluginHostBridge"; import { buildPluginEditorAppearance } from "@/lib/plugins/pluginAppearance"; +import { downloadPluginFile, cancelPluginDownload } from "@/lib/plugins/pluginFileDownload"; import type { InstalledPlugin, PluginUiContribution } from "@/types/database"; import { useI18n } from "vue-i18n"; import { useTheme } from "@/composables/useTheme"; @@ -68,6 +69,7 @@ function currentBridgeTheme(): PluginBridgeTheme { } function createBridge() { + bridge?.dispose(); bridge = new PluginHostBridge( props.plugin, props.contribution, @@ -83,6 +85,8 @@ function createBridge() { reopenConnection: (pluginId, connectionId) => useConnectionStore().reopenPluginConnection(connectionId, pluginId), closeTab: () => emit("closeTab"), saveFile: (_pluginId, request, data) => savePluginFile(request, data), + downloadFile: isTauriRuntime() ? downloadPluginFile : undefined, + cancelDownload: isTauriRuntime() ? cancelPluginDownload : undefined, copyText: (_pluginId, text) => copyToClipboard(text), }, appLocale.value, diff --git a/apps/desktop/src/lib/plugins/pluginFileDownload.ts b/apps/desktop/src/lib/plugins/pluginFileDownload.ts new file mode 100644 index 0000000000..858037c3c3 --- /dev/null +++ b/apps/desktop/src/lib/plugins/pluginFileDownload.ts @@ -0,0 +1,19 @@ +import { Channel, invoke } from "@tauri-apps/api/core"; +import type { PluginDownloadRequest, PluginSaveFileResult } from "./pluginHostBridge"; + +export async function downloadPluginFile(pluginId: string, request: PluginDownloadRequest, onProgress: (progress: unknown) => void): Promise { + const channel = new Channel(); + channel.onmessage = onProgress; + const path = await invoke("download_plugin_file", { + pluginId, + downloadId: request.downloadId, + fileName: request.fileName || "download.bin", + params: request.params, + onProgress: channel, + }); + return path === null ? null : { path }; +} + +export async function cancelPluginDownload(pluginId: string, downloadId: string): Promise { + await invoke("cancel_plugin_download", { pluginId, downloadId }); +} diff --git a/apps/desktop/src/lib/plugins/pluginHostBridge.spec.ts b/apps/desktop/src/lib/plugins/pluginHostBridge.spec.ts index dfdac1540c..714da1a171 100644 --- a/apps/desktop/src/lib/plugins/pluginHostBridge.spec.ts +++ b/apps/desktop/src/lib/plugins/pluginHostBridge.spec.ts @@ -14,6 +14,65 @@ const workbench: PluginWorkbenchContribution = { type: "workbench", id: "sample. const resultView: PluginResultViewContribution = { type: "result-view", id: "sample.graph", label: "Graph" }; describe("PluginHostBridge", () => { + it("streams downloads under the owning plugin, scopes cancellation and reports native capability", async () => { + const messages: any[] = []; + const target = { postMessage: (message: unknown) => messages.push(message) } as unknown as Window; + let finish!: (value: null) => void; + const downloadFile = vi.fn((_pluginId, _request, onProgress) => { + onProgress({ downloadId: "download-1", sent: 300 * 1024 * 1024 }); + return new Promise((resolve) => { + finish = resolve; + }); + }); + const cancelDownload = vi.fn().mockResolvedValue(undefined); + const bridge = new PluginHostBridge(plugin(), workbench, {}, () => target, { + invoke: vi.fn(), + notify: vi.fn(), + sendBinary: vi.fn(), + readAsset: vi.fn(), + downloadFile, + cancelDownload, + }); + const send = (id: string, method: string, params: unknown) => bridge.handleWindowMessage({ source: target, data: { source: "dbx-plugin", version: 1, type: "request", id, method, params } } as MessageEvent); + bridge.sendInit(); + expect(messages[0].capabilities.downloadFile).toBe(true); + send("open", "host.downloadFile", { downloadId: "download-1", fileName: "report.bin", params: { uri: "s3://bucket/report.bin" }, pluginId: "other" }); + await vi.waitFor(() => expect(downloadFile).toHaveBeenCalledOnce()); + expect(downloadFile.mock.calls[0][0]).toBe("sample"); + expect(messages.some((message) => message.method === "host.download.progress" && message.params.sent === 300 * 1024 * 1024)).toBe(true); + send("other", "host.cancelDownload", { downloadId: "someone-else" }); + expect(cancelDownload).not.toHaveBeenCalled(); + send("cancel", "host.cancelDownload", { downloadId: "download-1" }); + await vi.waitFor(() => expect(cancelDownload).toHaveBeenCalledWith("sample", "download-1")); + finish(null); + await vi.waitFor(() => expect(messages.some((message) => message.id === "open" && message.result === null)).toBe(true)); + }); + + it("cancels unfinished native downloads when the workbench is disposed", async () => { + const target = { postMessage: vi.fn() } as unknown as Window; + const cancelDownload = vi.fn().mockResolvedValue(undefined); + const bridge = new PluginHostBridge(plugin(), workbench, {}, () => target, { + invoke: vi.fn(), + notify: vi.fn(), + sendBinary: vi.fn(), + readAsset: vi.fn(), + downloadFile: vi.fn(() => new Promise(() => {})), + cancelDownload, + }); + bridge.handleWindowMessage({ source: target, data: { source: "dbx-plugin", version: 1, type: "request", id: "open", method: "host.downloadFile", params: { downloadId: "owned", params: {} } } } as MessageEvent); + bridge.dispose(); + expect(cancelDownload).toHaveBeenCalledWith("sample", "owned"); + }); + + it("does not advertise native downloads on legacy or web hosts", async () => { + const messages: any[] = []; + const target = { postMessage: (message: unknown) => messages.push(message) } as unknown as Window; + const bridge = new PluginHostBridge(plugin(), workbench, {}, () => target, { invoke: vi.fn(), notify: vi.fn(), sendBinary: vi.fn(), readAsset: vi.fn() }); + bridge.sendInit(); + expect(messages[0].capabilities.downloadFile).toBe(false); + bridge.handleWindowMessage({ source: target, data: { source: "dbx-plugin", version: 1, type: "request", id: "open", method: "host.downloadFile", params: {} } } as MessageEvent); + await vi.waitFor(() => expect(messages[1].error).toContain("desktop host")); + }); it("binds backend calls to the owning plugin identity", async () => { const messages: unknown[] = []; const target = { postMessage: (message: unknown) => messages.push(message) } as unknown as Window; diff --git a/apps/desktop/src/lib/plugins/pluginHostBridge.ts b/apps/desktop/src/lib/plugins/pluginHostBridge.ts index 70d538d547..9f540888d0 100644 --- a/apps/desktop/src/lib/plugins/pluginHostBridge.ts +++ b/apps/desktop/src/lib/plugins/pluginHostBridge.ts @@ -44,6 +44,11 @@ export interface PluginSaveFileResult { path: string; } +export interface PluginDownloadRequest extends PluginSaveFileRequest { + downloadId: string; + params: Record; +} + export interface PluginHostBridgeApi { invoke(pluginId: string, method: string, params?: unknown, timeoutMs?: number): Promise; notify(pluginId: string, method: string, params?: unknown): Promise; @@ -56,6 +61,8 @@ export interface PluginHostBridgeApi { closeTab?(): Promise | void; /** Persist plugin bytes through the host's native save dialog. Resolves null when the user cancels. */ saveFile?(pluginId: string, request: PluginSaveFileRequest, data: Uint8Array): Promise; + downloadFile?(pluginId: string, request: PluginDownloadRequest, onProgress: (progress: unknown) => void): Promise; + cancelDownload?(pluginId: string, downloadId: string): Promise; /** Write text to the system clipboard on behalf of the sandboxed plugin iframe. */ copyText?(pluginId: string, text: string): Promise; } @@ -72,6 +79,7 @@ interface PluginRequestMessage { } export class PluginHostBridge { + private downloads = new Set(); private context: PluginWorkbenchContext; private locale: string; private theme?: PluginBridgeTheme; @@ -120,6 +128,8 @@ export class PluginHostBridge { private requestInit(signal: "load" | "ready"): void { if ((this.initSignals.load && this.initSignals.ready) || (signal === "load" && this.initSignals.load)) { + for (const downloadId of this.downloads) void this.api.cancelDownload?.(this.plugin.manifest.id, downloadId).catch(() => undefined); + this.downloads.clear(); this.initGeneration += 1; this.initSignals = { load: false, ready: false }; this.initStarted = false; @@ -148,6 +158,8 @@ export class PluginHostBridge { /** Stop future posts (queued inits after an async reinit) for a torn-down bridge. */ dispose(): void { this.disposed = true; + for (const downloadId of this.downloads) void this.api.cancelDownload?.(this.plugin.manifest.id, downloadId).catch(() => undefined); + this.downloads.clear(); } private disposed = false; @@ -165,6 +177,7 @@ export class PluginHostBridge { locale: this.locale, theme: this.theme ? clonePluginData(this.theme) : undefined, permissions: [...(this.plugin.manifest.permissions || [])], + capabilities: { downloadFile: !!this.api.downloadFile }, context: snapshotPluginWorkbenchContext(this.context), }); } @@ -215,6 +228,27 @@ export class PluginHostBridge { } private async dispatch(method: string, params: unknown, binary?: ArrayBuffer): Promise { + if (method === "host.downloadFile") { + if (!this.api.downloadFile) throw new Error("Streaming file downloads require the desktop host"); + const input = requireRecord(params, "download params"); + const downloadId = requireProtocolName(input.downloadId, "download ID"); + if (this.downloads.size >= 2 || this.downloads.has(downloadId)) throw new Error("Too many active downloads or duplicate download ID"); + const request = { downloadId, fileName: optionalTrimmedString(input.fileName), params: requireRecord(input.params, "download source") }; + this.downloads.add(downloadId); + try { + return await this.api.downloadFile(this.plugin.manifest.id, request, (progress) => { + this.post({ source: HOST_MESSAGE_SOURCE, version: BRIDGE_VERSION, type: "event", method: "host.download.progress", params: progress }); + }); + } finally { + this.downloads.delete(downloadId); + } + } + if (method === "host.cancelDownload") { + const input = requireRecord(params, "cancel download params"); + const downloadId = requireProtocolName(input.downloadId, "download ID"); + if (this.downloads.has(downloadId)) await this.api.cancelDownload?.(this.plugin.manifest.id, downloadId); + return null; + } if (method === "host.getContext") return snapshotPluginWorkbenchContext(this.context); if (method === "backend.invoke") { const input = requireRecord(params, "backend.invoke params"); @@ -451,6 +485,7 @@ export function pluginSdkSource(initialTheme?: PluginBridgeTheme): string { let context; let locale = 'en'; let theme; + let capabilities = {}; let resolveReady; const initialTheme = ${serializedInitialTheme}; const applyTheme = (value) => { @@ -551,6 +586,9 @@ export function pluginSdkSource(initialTheme?: PluginBridgeTheme): string { get context() { return context; }, get locale() { return locale; }, get theme() { return theme; }, + get capabilities() { return capabilities; }, + downloadFile: (options) => request('host.downloadFile', options), + cancelDownload: (downloadId) => request('host.cancelDownload', { downloadId }), request, invoke: (method, params, options = {}) => request('backend.invoke', { method, params, timeoutMs: options.timeoutMs }), stream, @@ -591,6 +629,7 @@ export function pluginSdkSource(initialTheme?: PluginBridgeTheme): string { pending.delete(message.id); if (message.error) handler.reject(new Error(message.error)); else handler.resolve(message.result); } else if (message.type === 'init') { + capabilities = message.capabilities || {}; context = message.context; locale = typeof message.locale === 'string' ? message.locale : 'en'; applyTheme(message.theme); diff --git a/src-tauri/src/commands/mod.rs b/src-tauri/src/commands/mod.rs index 70055f1b2a..56704c3069 100644 --- a/src-tauri/src/commands/mod.rs +++ b/src-tauri/src/commands/mod.rs @@ -38,6 +38,8 @@ pub mod mq_cmd; #[cfg(feature = "mq-admin")] pub mod mqtt_cmd; pub mod nacos_cmd; +pub mod plugin_download; +pub mod plugin_download_file; pub mod plugins; pub mod prompt_template; pub mod query; diff --git a/src-tauri/src/commands/plugin_download.rs b/src-tauri/src/commands/plugin_download.rs new file mode 100644 index 0000000000..debc147a4a --- /dev/null +++ b/src-tauri/src/commands/plugin_download.rs @@ -0,0 +1,119 @@ +use std::collections::HashMap; +use std::io::Write; +use std::sync::{Arc, LazyLock, Mutex}; +use std::time::Duration; + +use base64::Engine; +use serde_json::{json, Value}; +use tauri::{ipc::Channel, AppHandle, State}; +use tauri_plugin_dialog::DialogExt; +use tokio_util::sync::CancellationToken; + +use super::connection::AppState; + +static DOWNLOADS: LazyLock>> = LazyLock::new(Mutex::default); + +#[tauri::command] +pub fn cancel_plugin_download(plugin_id: String, download_id: String) -> Result<(), String> { + if let Some(token) = DOWNLOADS.lock().map_err(|error| error.to_string())?.get(&format!("{plugin_id}:{download_id}")) + { + token.cancel(); + } + Ok(()) +} + +struct DownloadGuard(String); + +impl Drop for DownloadGuard { + fn drop(&mut self) { + if let Ok(mut downloads) = DOWNLOADS.lock() { + downloads.remove(&self.0); + } + } +} + +#[tauri::command] +pub async fn download_plugin_file( + app: AppHandle, + state: State<'_, Arc>, + plugin_id: String, + download_id: String, + file_name: String, + mut params: Value, + on_progress: Channel, +) -> Result, String> { + let download_uuid = uuid::Uuid::parse_str(&download_id).map_err(|error| error.to_string())?; + let input = params.as_object_mut().ok_or("Download params must be an object")?; + input.insert("downloadId".into(), json!(download_id)); + let cancel = CancellationToken::new(); + let key = format!("{plugin_id}:{download_id}"); + { + let mut downloads = DOWNLOADS.lock().map_err(|error| error.to_string())?; + if downloads.contains_key(&key) || downloads.len() >= 8 { + return Err("Too many active downloads or duplicate download ID".into()); + } + downloads.insert(key.clone(), cancel.clone()); + } + let _guard = DownloadGuard(key); + let file_name = file_name.rsplit(['/', '\\']).next().filter(|name| !name.is_empty()).unwrap_or("download.bin"); + let (sender, receiver) = tokio::sync::oneshot::channel(); + app.dialog().file().set_file_name(file_name).save_file(move |path| { + let _ = sender.send(path); + }); + let chosen = receiver.await.map_err(|error| error.to_string())?; + let Some(chosen) = chosen else { return Ok(None) }; + if cancel.is_cancelled() { + return Ok(None); + } + let destination = chosen.into_path().map_err(|error| error.to_string())?; + let mut output = super::plugin_download_file::DownloadFile::new(destination.clone(), &download_uuid.to_string()) + .map_err(|error| error.to_string())?; + let host = state.plugin_host.clone(); + let task = async { + let metadata: Value = host + .invoke(&plugin_id, "filesystem/download/open", params.clone(), None, Some(Duration::from_secs(120))) + .await?; + let total = metadata.get("size").and_then(Value::as_u64).unwrap_or(0); + let mut transferred = 0u64; + loop { + if cancel.is_cancelled() { + return Ok(None); + } + let chunk: Value = host + .invoke(&plugin_id, "filesystem/download/read", params.clone(), None, Some(Duration::from_secs(120))) + .await?; + let encoded = chunk.get("dataBase64").and_then(Value::as_str).ok_or("Invalid download chunk")?; + if encoded.len() > 1_398_104 { + return Err("Plugin download chunk exceeds 1 MiB".to_string()); + } + let bytes = base64::engine::general_purpose::STANDARD.decode(encoded).map_err(|error| error.to_string())?; + let done = chunk.get("done").and_then(Value::as_bool).ok_or("Missing download completion flag")?; + if bytes.is_empty() && !done { + return Err("Plugin download made no progress".into()); + } + output.write_all(&bytes).map_err(|error| error.to_string())?; + transferred += bytes.len() as u64; + on_progress + .send(json!({"downloadId": download_id, "sent": transferred, "total": total})) + .map_err(|error| error.to_string())?; + if done { + break; + } + } + if params.get("archive").and_then(Value::as_bool) != Some(true) && transferred != total { + return Err("Plugin download size does not match object metadata".into()); + } + if cancel.is_cancelled() { + return Ok(None); + } + output.commit().map_err(|error| error.to_string())?; + Ok(Some(destination.to_string_lossy().into_owned())) + }; + let result = tokio::select! { + result = task => result, + _ = cancel.cancelled() => Ok(None), + }; + let _: Result = + host.invoke(&plugin_id, "filesystem/download/close", params, None, Some(Duration::from_secs(10))).await; + result +} diff --git a/src-tauri/src/commands/plugin_download_file.rs b/src-tauri/src/commands/plugin_download_file.rs new file mode 100644 index 0000000000..82a751c085 --- /dev/null +++ b/src-tauri/src/commands/plugin_download_file.rs @@ -0,0 +1,80 @@ +use std::fs::{File, OpenOptions}; +use std::io::{self, Write}; +use std::path::PathBuf; + +pub struct DownloadFile { + destination: PathBuf, + temporary: PathBuf, + file: Option, +} + +impl DownloadFile { + pub fn new(destination: PathBuf, id: &str) -> io::Result { + let name = destination.file_name().ok_or_else(|| io::Error::other("Missing destination filename"))?; + let mut temporary_name = std::ffi::OsString::from("."); + temporary_name.push(name); + temporary_name.push(format!(".{id}.part")); + let temporary = destination.with_file_name(temporary_name); + let file = OpenOptions::new().write(true).create_new(true).open(&temporary)?; + Ok(Self { destination, temporary, file: Some(file) }) + } + + pub fn commit(&mut self) -> io::Result<()> { + if let Some(file) = self.file.take() { + file.sync_all()?; + } + std::fs::rename(&self.temporary, &self.destination) + } +} + +impl Write for DownloadFile { + fn write(&mut self, bytes: &[u8]) -> io::Result { + self.file.as_mut().ok_or_else(|| io::Error::other("Download is closed"))?.write(bytes) + } + + fn flush(&mut self) -> io::Result<()> { + self.file.as_mut().ok_or_else(|| io::Error::other("Download is closed"))?.flush() + } +} + +impl Drop for DownloadFile { + fn drop(&mut self) { + self.file.take(); + let _ = std::fs::remove_file(&self.temporary); + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn cancellation_preserves_existing_destination() { + let directory = tempfile::tempdir().unwrap(); + let destination = directory.path().join("report.bin"); + std::fs::write(&destination, b"original").unwrap(); + { + let mut output = DownloadFile::new(destination.clone(), "cancel").unwrap(); + output.write_all(b"partial").unwrap(); + assert_eq!(std::fs::read(&destination).unwrap(), b"original"); + } + assert_eq!(std::fs::read(&destination).unwrap(), b"original"); + assert_eq!(std::fs::read_dir(directory.path()).unwrap().count(), 1); + } + + #[test] + fn success_replaces_destination_only_after_completion() { + let directory = tempfile::tempdir().unwrap(); + let destination = directory.path().join("报告.bin"); + std::fs::write(&destination, b"original").unwrap(); + let mut output = DownloadFile::new(destination.clone(), "complete").unwrap(); + for _ in 0..300 { + output.write_all(&vec![7; 1024 * 1024]).unwrap(); + } + assert_eq!(std::fs::read(&destination).unwrap(), b"original"); + output.commit().unwrap(); + assert_eq!(std::fs::metadata(&destination).unwrap().len(), 300 * 1024 * 1024); + drop(output); + assert_eq!(std::fs::read_dir(directory.path()).unwrap().count(), 1); + } +} diff --git a/src-tauri/src/lib.rs b/src-tauri/src/lib.rs index bbe1eda750..3f39657711 100644 --- a/src-tauri/src/lib.rs +++ b/src-tauri/src/lib.rs @@ -1890,6 +1890,8 @@ pub fn run() { commands::plugins::list_active_plugins, commands::plugins::stop_plugin, commands::plugins::invoke_plugin, + commands::plugin_download::download_plugin_file, + commands::plugin_download::cancel_plugin_download, commands::plugins::invoke_plugin_connection_action, commands::plugins::notify_plugin, commands::plugins::send_plugin_binary,