Merge remote-tracking branch 'origin/dev-2.9.0' into dev-2.9.0

plugins/file-manager/src/backend/download-routes.ts

@@ -1,6 +1,7 @@
 import type { Express } from "express";
 import type { PluginContext } from "@termix/plugin-sdk/backend";
 import { getMimeType } from "./utils.js";
+import { createDownloadStream } from "./download-stream.js";
 import { execChannel, getSessionSftp, type SSHSession } from "./session.js";
 
 type FileDownloadRoutesDeps = {
@@ -296,9 +297,12 @@ export function registerFileDownloadRoutes(
 
       res.setHeader("Content-Length", String(stats.size));
 
-      const readStream = sftp.createReadStream(filePath);
+      if (res.destroyed) return;
+      const readStream = createDownloadStream(sftp, filePath, stats.size);
+      res.on("close", () => readStream.destroy());
       readStream.on("error", (err) => {
         if (!res.headersSent) {
+          res.removeHeader("Content-Length");
           res.status(500).json({ error: `Download failed: ${err.message}` });
         } else {
           res.destroy();

plugins/file-manager/src/backend/download-stream.ts

@@ -0,0 +1,68 @@
+import { Readable } from "node:stream";
+import type { SFTPWrapper } from "ssh2";
+import {
+  promisifySftpClose,
+  promisifySftpOpen,
+  SFTP_OPEN_READ,
+} from "./sftp-promisify.js";
+
+const CHUNK_SIZE = 32 * 1024;
+const CONCURRENCY = 8;
+
+/** Bounded read-ahead; batches retain file order even if replies arrive out of order. */
+export function createDownloadStream(
+  sftp: SFTPWrapper,
+  path: string,
+  size: number,
+): Readable {
+  const stream = Readable.from(readFile(), {
+    objectMode: false,
+    highWaterMark: CHUNK_SIZE,
+  });
+  return stream;
+
+  async function readChunk(handle: Buffer, position: number, length: number) {
+    const buffer = Buffer.allocUnsafe(length);
+    let offset = 0;
+    while (offset < length && !stream.destroyed) {
+      const bytesRead = await new Promise<number>((resolve, reject) => {
+        sftp.read(
+          handle,
+          buffer,
+          offset,
+          length - offset,
+          position + offset,
+          (err, bytes) => (err ? reject(err) : resolve(bytes)),
+        );
+      });
+      if (bytesRead === 0)
+        throw new Error("File ended before the advertised size");
+      offset += bytesRead;
+    }
+    return buffer;
+  }
+
+  async function* readFile() {
+    if (stream.destroyed) return;
+    const handle = await promisifySftpOpen(sftp, path, SFTP_OPEN_READ, 0o666);
+    try {
+      for (let position = 0; position < size && !stream.destroyed;) {
+        const reads: Promise<Buffer>[] = [];
+        for (let i = 0; i < CONCURRENCY && position < size; i++) {
+          const length = Math.min(CHUNK_SIZE, size - position);
+          reads.push(readChunk(handle, position, length));
+          position += length;
+        }
+        // Settle all reads before closing the shared handle after an error.
+        const results = await Promise.allSettled(reads);
+        if (stream.destroyed) return;
+        for (const result of results) {
+          if (result.status === "rejected") throw result.reason;
+          yield result.value;
+        }
+      }
+    } finally {
+      await promisifySftpClose(sftp, handle);
+    }
+  }
+}

plugins/file-manager/src/frontend/FileManagerGrid.tsx

@@ -458,6 +458,9 @@ export function FileManagerGrid({
   };
 
   const handleFileDrop = (e: React.DragEvent, targetFile: FileItem) => {
+    // OS drops belong to the pane's upload handler, even over an existing row.
+    if (dragState.type !== "internal" && !isLocalFilesDrag(e.dataTransfer))
+      return;
     e.preventDefault();
     e.stopPropagation();
 

plugins/file-manager/tests/backend/download-routes.test.ts

@@ -0,0 +1,124 @@
+import { once } from "node:events";
+import { PassThrough } from "node:stream";
+import type { Express, Request, Response } from "express";
+import type { PluginContext } from "@termix/plugin-sdk/backend";
+import { describe, expect, it, vi } from "vitest";
+import { registerFileDownloadRoutes } from "../../src/backend/download-routes.js";
+import type { SSHSession } from "../../src/backend/session.js";
+
+function setup(remote: object, owned = true) {
+  const routes = new Map<
+    string,
+    (req: Request, res: Response) => Promise<unknown>
+  >();
+  registerFileDownloadRoutes(
+    { post: (path, handler) => routes.set(path, handler) } as Express,
+    {
+      ctx: {
+        currentActor: () => "user",
+        log: { info: vi.fn(), warn: vi.fn(), error: vi.fn() },
+      } as unknown as PluginContext,
+      sshSessions: { s: { isConnected: true, sftp: remote } as SSHSession },
+      scheduleSessionCleanup: vi.fn(),
+      verifySessionOwnership: () => owned,
+    },
+  );
+  const headers = new Map<string, string>();
+  const res = Object.assign(new PassThrough(), {
+    headersSent: false,
+    setHeader: (name: string, value: string) => headers.set(name, value),
+    removeHeader: (name: string) => headers.delete(name),
+    status: vi.fn().mockReturnThis(),
+    json: vi.fn(),
+  });
+  const start = () =>
+    routes.get("/downloadFileStream")!(
+      { body: { sessionId: "s", path: "/file" } } as Request,
+      res as unknown as Response,
+    );
+  return { res, start, headers };
+}
+
+function source() {
+  return {
+    stat: vi.fn((_path, cb) => cb(undefined, { size: 5, isFile: () => true })),
+    open: vi.fn((_path, _flags, _mode, cb) =>
+      cb(undefined, Buffer.from("handle")),
+    ),
+    read: vi.fn((_handle, buffer, offset, _length, _position, cb) => {
+      Buffer.from("hello").copy(buffer, offset);
+      cb(undefined, 5);
+    }),
+    close: vi.fn((_handle, cb) => cb()),
+  };
+}
+
+describe("stream download route", () => {
+  it("streams the file content with its advertised length", async () => {
+    const remote = source();
+    const { res, start, headers } = setup(remote);
+    await start();
+    const chunks: Buffer[] = [];
+    for await (const chunk of res) chunks.push(chunk);
+    expect(Buffer.concat(chunks).toString()).toBe("hello");
+    expect(headers.get("Content-Length")).toBe("5");
+    expect(remote.close).toHaveBeenCalledOnce();
+  });
+
+  it("does not open a file after the HTTP client disconnected during stat", async () => {
+    const remote = source();
+    let finishStat!: () => void;
+    remote.stat.mockImplementation((_path, cb) => {
+      finishStat = () => cb(undefined, { size: 5, isFile: () => true });
+    });
+    const { res, start } = setup(remote);
+    const started = start();
+    await Promise.resolve();
+    res.destroy();
+    finishStat();
+    await started;
+    expect(remote.open).not.toHaveBeenCalled();
+  });
+
+  it("closes the SFTP handle after HTTP disconnect during a read", async () => {
+    const remote = source();
+    let finishRead!: () => void;
+    remote.read.mockImplementation(
+      (_h, buffer, offset, length, position, cb) => {
+        finishRead = () => cb(undefined, 0);
+      },
+    );
+    const { res, start } = setup(remote);
+    await start();
+    await new Promise<void>((resolve) => setImmediate(resolve));
+    res.destroy();
+    await once(res, "close");
+    finishRead();
+    await vi.waitFor(() => expect(remote.close).toHaveBeenCalledOnce());
+    expect(remote.read).toHaveBeenCalledOnce();
+  });
+
+  it("removes the file length before returning an early stream error as JSON", async () => {
+    const remote = source();
+    remote.open.mockImplementation((_p, _f, _m, cb) =>
+      cb(new Error("permission denied"), undefined),
+    );
+    const { res, start, headers } = setup(remote);
+    await start();
+    await vi.waitFor(() => expect(res.status).toHaveBeenCalledWith(500));
+    expect(headers.has("Content-Length")).toBe(false);
+    expect(res.json).toHaveBeenCalledWith({
+      error: "Download failed: permission denied",
+    });
+    res.destroy();
+  });
+
+  it("refuses another user's SSH session before reading a file", async () => {
+    const remote = source();
+    const { res, start } = setup(remote, false);
+    await start();
+    expect(res.status).toHaveBeenCalledWith(403);
+    expect(remote.stat).not.toHaveBeenCalled();
+    res.destroy();
+  });
+});

plugins/file-manager/tests/backend/download-stream-sftp.test.ts

@@ -0,0 +1,74 @@
+import { once } from "node:events";
+import type { AddressInfo } from "node:net";
+import { Client, Server, utils, type SFTPWrapper } from "ssh2";
+import { expect, it } from "vitest";
+import { createDownloadStream } from "../../src/backend/download-stream.js";
+
+it("overlaps real SFTP reads while retaining the serial stream's byte content", async () => {
+  const data = Buffer.alloc(1024 * 1024 + 71);
+  for (let i = 0; i < data.length; i++) data[i] = (i * 7 + (i >>> 10)) % 256;
+  let pending = 0;
+  let peak = 0;
+  let closed = 0;
+  const server = new Server(
+    {
+      hostKeys: [utils.generateKeyPairSync("rsa", { bits: 2048 }).private],
+    },
+    (client) => {
+      client.on("authentication", (ctx) => ctx.accept());
+      client.on("ready", () =>
+        client.on("session", (accept) => {
+          accept().on("sftp", (acceptSftp) => {
+            const sftp = acceptSftp();
+            sftp.on("OPEN", (id) => sftp.handle(id, Buffer.from("file")));
+            sftp.on("READ", (id, _handle, offset, length) => {
+              peak = Math.max(peak, ++pending);
+              setTimeout(() => {
+                pending--;
+                if (offset >= data.length) sftp.status(id, 1);
+                else sftp.data(id, data.subarray(offset, offset + length));
+              }, 10);
+            });
+            sftp.on("CLOSE", (id) => {
+              closed++;
+              sftp.status(id, 0);
+            });
+          });
+        }),
+      );
+    },
+  );
+  const client = new Client();
+  try {
+    await new Promise<void>((resolve) =>
+      server.listen(0, "127.0.0.1", resolve),
+    );
+    client.connect({
+      host: "127.0.0.1",
+      port: (server.address() as AddressInfo).port,
+      username: "test",
+    });
+    await once(client, "ready");
+    const sftp = await new Promise<SFTPWrapper>((resolve, reject) =>
+      client.sftp((err, channel) => (err ? reject(err) : resolve(channel))),
+    );
+    const serial: Buffer[] = [];
+    for await (const chunk of sftp.createReadStream("/file"))
+      serial.push(chunk);
+    expect(Buffer.concat(serial)).toEqual(data);
+    expect(peak).toBe(1);
+    peak = 0;
+    const pipelined: Buffer[] = [];
+    for await (const chunk of createDownloadStream(sftp, "/file", data.length))
+      pipelined.push(chunk);
+    expect(Buffer.concat(pipelined)).toEqual(data);
+    expect(peak).toBeGreaterThan(1);
+    expect(peak).toBeLessThanOrEqual(8);
+    expect(closed).toBe(2);
+  } finally {
+    const closedClient = once(client, "close");
+    client.destroy();
+    await closedClient;
+    await new Promise<void>((resolve) => server.close(() => resolve()));
+  }
+}, 15000);

plugins/file-manager/tests/backend/download-stream.test.ts

@@ -0,0 +1,133 @@
+import { once } from "node:events";
+import { setImmediate as nextTurn } from "node:timers/promises";
+import type { SFTPWrapper } from "ssh2";
+import { describe, expect, it, vi } from "vitest";
+import { createDownloadStream } from "../../src/backend/download-stream.js";
+
+function source(
+  data: Buffer,
+  options: { shortRead?: number; failAt?: number } = {},
+) {
+  let pending = 0;
+  let peak = 0;
+  const closedWithPending: number[] = [];
+  const sftp = {
+    open: vi.fn((_path, _flags, _mode, cb) =>
+      cb(undefined, Buffer.from("handle")),
+    ),
+    close: vi.fn((_handle, cb) => {
+      closedWithPending.push(pending);
+      cb();
+    }),
+    read: vi.fn((_handle, buffer, offset, length, position, cb) => {
+      peak = Math.max(peak, ++pending);
+      // Later file offsets can finish first.
+      setTimeout(
+        () => {
+          pending--;
+          if (position === options.failAt) return cb(new Error("read failed"));
+          const bytes = Math.min(
+            length,
+            options.shortRead ?? length,
+            Math.max(0, data.length - position),
+          );
+          if (bytes > 0) data.copy(buffer, offset, position, position + bytes);
+          cb(undefined, bytes);
+        },
+        position === 0 ? 10 : 0,
+      );
+    }),
+  };
+  return {
+    sftp: sftp as unknown as SFTPWrapper,
+    close: sftp.close,
+    read: sftp.read,
+    peak: () => peak,
+    closedWithPending,
+  };
+}
+
+async function contents(stream: ReturnType<typeof createDownloadStream>) {
+  const buffers: Buffer[] = [];
+  for await (const buffer of stream) buffers.push(buffer);
+  return Buffer.concat(buffers);
+}
+
+describe("SFTP download stream", () => {
+  it("pipelines bounded reads and preserves bytes despite out-of-order replies", async () => {
+    const data = Buffer.alloc(1024 * 1024 + 123);
+    for (let i = 0; i < data.length; i++) data[i] = (i * 13 + (i >>> 8)) % 256;
+    const remote = source(data);
+    expect(
+      await contents(createDownloadStream(remote.sftp, "/file", data.length)),
+    ).toEqual(data);
+    expect(remote.peak()).toBe(8);
+    expect(remote.closedWithPending).toEqual([0]);
+  });
+
+  it("fills short server reads without losing or repeating bytes", async () => {
+    const data = Buffer.from("short reads ".repeat(10000));
+    const remote = source(data, { shortRead: 4096 });
+    expect(
+      await contents(createDownloadStream(remote.sftp, "/file", data.length)),
+    ).toEqual(data);
+    expect(remote.closedWithPending).toEqual([0]);
+  });
+
+  it("handles empty files without issuing reads", async () => {
+    const remote = source(Buffer.alloc(0));
+    expect(
+      await contents(createDownloadStream(remote.sftp, "/empty", 0)),
+    ).toEqual(Buffer.alloc(0));
+    expect(remote.read).not.toHaveBeenCalled();
+    expect(remote.close).toHaveBeenCalledOnce();
+  });
+
+  it("fails on premature EOF instead of returning uninitialized bytes", async () => {
+    const remote = source(Buffer.from("truncated"));
+    await expect(
+      contents(createDownloadStream(remote.sftp, "/file", 100000)),
+    ).rejects.toThrow("File ended");
+    expect(remote.closedWithPending).toEqual([0]);
+  });
+
+  it("settles outstanding reads before closing on a read failure", async () => {
+    const remote = source(Buffer.alloc(1024 * 1024), { failAt: 32768 });
+    await expect(
+      contents(createDownloadStream(remote.sftp, "/file", 1024 * 1024)),
+    ).rejects.toThrow("read failed");
+    expect(remote.read).toHaveBeenCalledTimes(8);
+    expect(remote.closedWithPending).toEqual([0]);
+  });
+
+  it("stops scheduling reads and closes the handle when destroyed mid-batch", async () => {
+    const remote = source(Buffer.alloc(1024 * 1024), { shortRead: 4096 });
+    const replies: Array<() => void> = [];
+    remote.read.mockImplementation(
+      (_handle, _buffer, _offset, _length, _position, cb) => {
+        replies.push(() => cb(undefined, 4096));
+      },
+    );
+    const stream = createDownloadStream(remote.sftp, "/file", 1024 * 1024);
+    stream.resume();
+    await nextTurn();
+    const closed = once(stream, "close");
+    stream.destroy();
+    for (const reply of replies) reply();
+    await closed;
+    expect(remote.read).toHaveBeenCalledTimes(8);
+    expect(remote.closedWithPending).toEqual([0]);
+  });
+
+  it("bounds read-ahead while the consumer is paused", async () => {
+    const remote = source(Buffer.alloc(8 * 1024 * 1024));
+    const stream = createDownloadStream(remote.sftp, "/file", 8 * 1024 * 1024);
+    stream.read(0);
+    await once(stream, "readable");
+    expect(remote.read).toHaveBeenCalledTimes(8);
+    const closed = once(stream, "close");
+    stream.destroy();
+    await closed;
+    expect(remote.closedWithPending).toEqual([0]);
+  });
+});

plugins/file-manager/tests/frontend/FileManagerGrid.test.tsx

@@ -1,4 +1,4 @@
-import { afterEach, describe, expect, it, vi } from "vitest";
+import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
 import { cleanup, fireEvent, render, screen } from "@testing-library/react";
 import "@testing-library/jest-dom/vitest";
 import { FileManagerGrid } from "../../src/frontend/FileManagerGrid";
@@ -44,6 +44,7 @@ const props = {
 afterEach(() => {
   cleanup();
   vi.restoreAllMocks();
+  vi.unstubAllGlobals();
 });
 
 describe("remote file rename", () => {
@@ -89,3 +90,52 @@ describe("remote file rename", () => {
     expect(screen.getByText(file.modified!)).toBeInTheDocument();
   });
 });
+
+describe("external file drops", () => {
+  beforeEach(() => {
+    vi.stubGlobal(
+      "ResizeObserver",
+      class {
+        observe() {}
+        disconnect() {}
+      },
+    );
+  });
+  it.each(["list", "grid"] as const)(
+    "uploads dropped folders over a file in %s view",
+    (viewMode) => {
+      const onUploadItems = vi.fn();
+      render(
+        <FileManagerGrid
+          {...props}
+          viewMode={viewMode}
+          onUploadItems={onUploadItems}
+        />,
+      );
+      const folder = {
+        name: "logs",
+        isDirectory: true,
+        isFile: false,
+      } as FileSystemEntry;
+      fireEvent.drop(screen.getByText(file.name), {
+        dataTransfer: {
+          types: ["Files"],
+          files: [],
+          getData: () => "",
+          items: [{ webkitGetAsEntry: () => folder }],
+        },
+      });
+      expect(onUploadItems).toHaveBeenCalledExactlyOnceWith([folder]);
+    },
+  );
+
+  it("uploads a plain OS file dropped over a row", () => {
+    const onUpload = vi.fn();
+    render(<FileManagerGrid {...props} onUpload={onUpload} />);
+    const files = [new File(["log"], "app.log")];
+    fireEvent.drop(screen.getByText(file.name), {
+      dataTransfer: { types: ["Files"], files, items: [], getData: () => "" },
+    });
+    expect(onUpload).toHaveBeenCalledExactlyOnceWith(files);
+  });
+});