diff --git a/src/bun/index.ts b/src/bun/index.ts index e7f69bb..d238a40 100644 --- a/src/bun/index.ts +++ b/src/bun/index.ts @@ -2,6 +2,7 @@ import { BrowserWindow, Updater } from "electrobun/bun"; import { BRANDING } from "../shared/branding"; import { SessionRegistry } from "./fs/registry"; import { createAppRPC } from "./rpc-handlers"; +import { TransferEngine } from "./transfers/engine"; const PROD_URL = "views://app/index.html"; @@ -24,7 +25,8 @@ async function getMainViewUrl(): Promise { } const registry = new SessionRegistry(); -const rpc = createAppRPC(registry); +const engine = new TransferEngine(registry); +const rpc = createAppRPC(registry, engine); const url = await getMainViewUrl(); new BrowserWindow({ @@ -39,4 +41,6 @@ new BrowserWindow({ rpc, }); +engine.onUpdate = (snapshot) => rpc.send.transferUpdate(snapshot); + console.log(`${BRANDING.appName} started (${url})`); diff --git a/src/bun/rpc-handlers.ts b/src/bun/rpc-handlers.ts index cef2d12..b3ec9c7 100644 --- a/src/bun/rpc-handlers.ts +++ b/src/bun/rpc-handlers.ts @@ -2,6 +2,7 @@ import { BrowserView } from "electrobun/bun"; import type { AppRPC } from "../shared/rpc"; import type { Result } from "../shared/types"; import type { SessionRegistry } from "./fs/registry"; +import type { TransferEngine } from "./transfers/engine"; import { removeRecursive } from "./transfers/walk"; // SFTP ops can be slow (big listings, recursive deletes) — the 1s electrobun @@ -18,7 +19,7 @@ async function wrap(fn: () => Promise | T): Promise({ maxRequestTime: MAX_REQUEST_TIME_MS, handlers: { @@ -70,10 +71,21 @@ export function createAppRPC(registry: SessionRegistry) { // P4: SFTP sessions connect: async () => NOT_YET, disconnect: async () => NOT_YET, - // P3: transfer engine - enqueueTransfers: async () => NOT_YET, - cancelTransfer: async () => NOT_YET, - dismissTransfer: async () => NOT_YET, + + enqueueTransfers: (params) => + wrap(async () => ({ jobIds: engine.enqueue(params) })), + + cancelTransfer: ({ jobId }) => + wrap(async () => { + engine.cancel(jobId); + return {}; + }), + + dismissTransfer: ({ jobId }) => + wrap(async () => { + engine.dismiss(jobId); + return {}; + }), // P4: connection store getConnections: async () => NOT_YET, upsertConnection: async () => NOT_YET, diff --git a/src/bun/transfers/engine.ts b/src/bun/transfers/engine.ts new file mode 100644 index 0000000..fe1de61 --- /dev/null +++ b/src/bun/transfers/engine.ts @@ -0,0 +1,227 @@ +import { pipeline } from "node:stream/promises"; +import type { Readable } from "node:stream"; +import type { EntryType, TransferSnapshot, TransferState } from "../../shared/types"; +import type { FileSystemProvider } from "../fs/provider"; +import type { SessionRegistry } from "../fs/registry"; +import { walkTree } from "./walk"; + +// One at a time keeps remote servers happy; bump for parallel jobs. +const TRANSFER_CONCURRENCY = 1; +const PROGRESS_INTERVAL_MS = 100; + +interface Job { + snapshot: TransferSnapshot; + type: EntryType; + cancelled: boolean; + currentStream: Readable | null; +} + +const TERMINAL: TransferState[] = ["completed", "failed", "cancelled"]; + +async function ensureDir(fs: FileSystemProvider, path: string): Promise { + try { + const st = await fs.stat(path); + if (st.type === "dir") return; + throw new Error(`"${path}" exists and is not a directory`); + } catch { + await fs.mkdir(path); + } +} + +/** + * FIFO queue of copy jobs. Because both providers speak node streams, one + * pipeline() uniformly covers local->local, upload, download and + * remote->remote (piped through this process). + */ +export class TransferEngine { + onUpdate: (snapshot: TransferSnapshot) => void = () => {}; + + private jobs = new Map(); + private queue: Job[] = []; + private running = 0; + private counter = 0; + private lastEmit = new Map(); + + constructor(private registry: SessionRegistry) {} + + enqueue(params: { + srcSessionId: string; + items: { path: string; type: EntryType }[]; + dstSessionId: string; + dstDir: string; + }): string[] { + const src = this.registry.get(params.srcSessionId); + const jobIds: string[] = []; + for (const item of params.items) { + const id = `job-${++this.counter}`; + const job: Job = { + type: item.type, + cancelled: false, + currentStream: null, + snapshot: { + id, + name: src.basename(item.path), + srcSessionId: params.srcSessionId, + srcPath: item.path, + dstSessionId: params.dstSessionId, + dstDir: params.dstDir, + state: "queued", + totalFiles: 0, + totalBytes: 0, + doneFiles: 0, + doneBytes: 0, + currentFile: "", + currentFileBytes: 0, + currentFileTotal: 0, + }, + }; + this.jobs.set(id, job); + this.queue.push(job); + jobIds.push(id); + this.emit(job, true); + } + this.pump(); + return jobIds; + } + + cancel(jobId: string): void { + const job = this.jobs.get(jobId); + if (!job || TERMINAL.includes(job.snapshot.state)) return; + job.cancelled = true; + if (job.snapshot.state === "queued") { + this.queue = this.queue.filter((j) => j !== job); + this.finish(job, "cancelled"); + } else { + job.currentStream?.destroy(new Error("transfer cancelled")); + } + } + + dismiss(jobId: string): void { + const job = this.jobs.get(jobId); + if (job && TERMINAL.includes(job.snapshot.state)) this.jobs.delete(jobId); + } + + /** Fail every non-terminal job that touches a dead session. */ + failSession(sessionId: string, reason: string): void { + for (const job of this.jobs.values()) { + const { srcSessionId, dstSessionId, state } = job.snapshot; + if (TERMINAL.includes(state)) continue; + if (srcSessionId !== sessionId && dstSessionId !== sessionId) continue; + job.cancelled = true; + job.snapshot.error = reason; + if (state === "queued") { + this.queue = this.queue.filter((j) => j !== job); + this.finish(job, "failed"); + } else { + job.currentStream?.destroy(new Error(reason)); + } + } + } + + private pump(): void { + while (this.running < TRANSFER_CONCURRENCY && this.queue.length) { + const job = this.queue.shift()!; + this.running++; + this.run(job) + .catch(() => {}) + .finally(() => { + this.running--; + this.pump(); + }); + } + } + + private async run(job: Job): Promise { + const snap = job.snapshot; + try { + const src = this.registry.get(snap.srcSessionId); + const dst = this.registry.get(snap.dstSessionId); + + snap.state = "scanning"; + this.emit(job, true); + + // collect the work list + let files: { srcPath: string; dstPath: string; size: number; name: string }[]; + if (job.type === "dir") { + const dstRoot = dst.join(snap.dstDir, snap.name); + this.guardAgainstSelfCopy(snap, dstRoot); + const tree = await walkTree(src, snap.srcPath); + for (const segments of tree.dirs) { + if (job.cancelled) throw new Error("transfer cancelled"); + await ensureDir(dst, dst.join(dstRoot, ...segments)); + } + files = tree.files.map((f) => ({ + srcPath: f.entry.path, + dstPath: dst.join(dstRoot, ...f.dirSegments, f.entry.name), + size: f.entry.size, + name: f.entry.name, + })); + } else { + const dstPath = dst.join(snap.dstDir, snap.name); + this.guardAgainstSelfCopy(snap, dstPath); + const st = await src.stat(snap.srcPath); + files = [{ srcPath: snap.srcPath, dstPath, size: st.size, name: snap.name }]; + } + + snap.totalFiles = files.length; + snap.totalBytes = files.reduce((sum, f) => sum + f.size, 0); + snap.state = "running"; + this.emit(job, true); + + for (const file of files) { + if (job.cancelled) throw new Error("transfer cancelled"); + snap.currentFile = file.name; + snap.currentFileBytes = 0; + snap.currentFileTotal = file.size; + this.emit(job, true); + + const readStream = src.createReadStream(file.srcPath); + job.currentStream = readStream; + readStream.on("data", (chunk: Buffer | string) => { + const n = typeof chunk === "string" ? Buffer.byteLength(chunk) : chunk.length; + snap.currentFileBytes += n; + snap.doneBytes += n; + this.emit(job); + }); + await pipeline(readStream, dst.createWriteStream(file.dstPath)); + job.currentStream = null; + snap.doneFiles++; + this.emit(job, true); + } + + this.finish(job, "completed"); + } catch (error) { + job.currentStream = null; + if (!job.cancelled && !snap.error) { + snap.error = error instanceof Error ? error.message : String(error); + } + this.finish(job, job.cancelled && !snap.error ? "cancelled" : "failed"); + } + } + + /** Copying onto itself truncates the source; into a descendant never ends. */ + private guardAgainstSelfCopy(snap: TransferSnapshot, dstPath: string): void { + if (snap.srcSessionId !== snap.dstSessionId) return; + if (dstPath === snap.srcPath) { + throw new Error("Source and destination are the same"); + } + if (dstPath.startsWith(snap.srcPath.endsWith("/") ? snap.srcPath : `${snap.srcPath}/`)) { + throw new Error("Cannot copy a folder into itself"); + } + } + + private finish(job: Job, state: TransferState): void { + if (TERMINAL.includes(job.snapshot.state)) return; + job.snapshot.state = state; + this.emit(job, true); + } + + private emit(job: Job, force = false): void { + const now = Date.now(); + if (!force && now - (this.lastEmit.get(job.snapshot.id) ?? 0) < PROGRESS_INTERVAL_MS) { + return; + } + this.lastEmit.set(job.snapshot.id, now); + this.onUpdate({ ...job.snapshot }); + } +} diff --git a/ui/src/app/app.html b/ui/src/app/app.html index e7fb0c7..eddd6a6 100644 --- a/ui/src/app/app.html +++ b/ui/src/app/app.html @@ -20,8 +20,17 @@ + +