feat(server): chunked import preview endpoint and resumable upload in the Import page and CLI (#11224)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - The previous pull request added server-side chunked resumable import transfers; without clients, large imports still ride the single fragile upload > - The Import page and the CLI need to slice large packages, upload parts with retry and progress, resume after interruptions, and preview before applying > - Preview is the missing server piece: the browser flow is preview-then-import, so a completed spool must be previewable without re-uploading > - This pull request adds the transfer preview endpoint, switches the Import page to the chunked path for zips over 48 MB, and teaches the CLI the same for oversized local imports > - The benefit is that large imports get progress, per-part retry, and resume in both clients, while small imports keep the exact single-shot path they have today ## Linked Issues or Issue Description **What happened?** With only the server transfer routes in place, users still upload large company packages as one request from the Import page and the CLI: no progress indication, no retry below the whole file, and no resume after a dropped connection or refresh. The preview-then-import flow also cannot run against an uploaded transfer, forcing a second full upload. **Expected behavior** A large package uploads once as verified parts with visible progress; preview and import both run against the uploaded spool; an interrupted upload resumes with only the missing parts re-sent; packages at or below 48 MB behave exactly as before. **Steps to reproduce** 1. Select a 500 MB zip on the Import page over an unreliable connection. 2. Watch the single upload fail near the end and restart from zero, twice — once for preview, once for import. 3. Same story headless via the CLI. ## What Changed - Server: `POST /import/transfers/:id/preview` runs the existing preview logic against the completed spool (shared assembly + whole-file verification helper with apply); preview neither completes the run nor deletes the spool, so the subsequent apply reuses it. Missing parts respond with the missing list. - UI: zips over 48 MB take the chunked path in both preview and import — the file is sliced into 32 MB parts hashed with WebCrypto (single ArrayBuffer, no second copy), the transfer is created or resumed (the create response's missing-parts list drives what uploads), parts upload sequentially with three attempts each and visible progress, then transfer preview/apply replace the multipart calls. The existing preview pane, collision handling, adapter overrides, and async job polling are unchanged; ≤ 48 MB keeps the single-shot path. - CLI: oversized local `.zip` or folder imports zip/slice/hash with node crypto, upload with resume and per-part retry and progress lines, and use transfer preview/apply. Small packages keep the inline path byte-identical. - Failure honesty: adapters/API errors fail open to existing behavior; a part failing all attempts surfaces a durable error panel with resume intact. ## Verification - Server: preview-then-apply on one spool (run stays open, spool intact, then apply completes), preview with missing parts rejected — added to the transfer route suite (embedded Postgres). - UI suite: large file takes the chunked path (manifest shape, part uploads, progress, apply on a resumed transfer, single-shot endpoints never called), small file stays single-shot, part failure after three attempts surfaces the error panel without running preview, resume re-uploads only the missing part. - CLI: manifest slicing/hashing, threshold behavior for zip and folder sources, folder-zip round-trip through the real zip reader, upload resume/retry/exhaustion/already-completed, full-command chunked and small-zip inline flows. - Server, ui, cli typechecks clean. Exact counts in the PR checks. ## Risks - The 48 MB threshold only routes between two verified paths; behavior below it is untouched. - Chunked CLI imports use the board-scoped transfer routes, so oversized CLI imports need board credentials (agent tokens keep the agent-safe small-file path). No privilege change — board actors already had the generic routes — but the two size regimes differ semantically; called out for review. - A CLI dry-run over the threshold uploads parts before previewing; the spool persists (24 h sweep) and a later apply resumes without re-upload — inherent to preview-against-spool. Stacked on #11223 — merge that first; this PR then shows only the preview endpoint and client changes. ## Model Used - Claude Fable 5 (`claude-fable-5`) via Claude Code CLI, with extended thinking and tool use (multi-agent implementation with independent verification). ## Checklist - [x] I have included a thinking path that traces from project context to this change - [x] I have specified the model used (with version and capability details) - [x] I have checked ROADMAP.md and confirmed this PR does not duplicate planned core work - [x] I have searched GitHub for duplicate or related PRs and linked them above - [x] I have either (a) linked existing issues with `Fixes: #` / `Closes #` / `Refs #` OR (b) described the issue in-PR following the relevant issue template - [x] I have not referenced internal/instance-local Paperclip issues or links (only public GitHub `#NNN` / `github.com/paperclipai/paperclip` URLs) - [x] My branch name describes the change (e.g. `docs/...`, `fix/...`) and contains no internal Paperclip ticket id or instance-derived details - [x] I have run tests locally and they pass - [x] I have added or updated tests where applicable - [x] I have updated relevant documentation to reflect my changes - [x] I have considered and documented any risks above - [ ] All Paperclip CI gates are green - [ ] Greptile is 5/5 with no open P2s, recommendations, or follow-ups - [x] I will address all Greptile and reviewer comments before requesting merge
This commit is contained in:
Devin Foley
GitHub
parent
8f478242f1
commit
0a95ada1be
|
|
@ -0,0 +1,494 @@
|
|||
import { createHash } from "node:crypto";
|
||||
import { deflateRawSync } from "node:zlib";
|
||||
import { mkdir, mkdtemp, rm, writeFile } from "node:fs/promises";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
import { Command } from "commander";
|
||||
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import { readZipArchive } from "@paperclipai/shared/portability-zip";
|
||||
import {
|
||||
CHUNKED_IMPORT_THRESHOLD_BYTES,
|
||||
IMPORT_TRANSFER_PART_SIZE_BYTES,
|
||||
buildImportTransferManifest,
|
||||
registerCompanyCommands,
|
||||
EXISTING_COMPANY_CHUNKED_IMPORT_THRESHOLD_BYTES,
|
||||
resolveChunkedImportZip,
|
||||
uploadCompanyImportTransfer,
|
||||
} from "../commands/client/company.js";
|
||||
import { createStoredZipArchive } from "./helpers/zip.js";
|
||||
|
||||
const ORIGINAL_ENV = { ...process.env };
|
||||
|
||||
const tempDirs: string[] = [];
|
||||
|
||||
async function makeTempDir(): Promise<string> {
|
||||
const dir = await mkdtemp(path.join(os.tmpdir(), "paperclip-company-import-transfer-"));
|
||||
tempDirs.push(dir);
|
||||
return dir;
|
||||
}
|
||||
|
||||
function sha256Hex(bytes: Uint8Array): string {
|
||||
return createHash("sha256").update(bytes).digest("hex");
|
||||
}
|
||||
|
||||
/** Two-part fixture: one full 32 MB part plus a short tail. */
|
||||
function buildTwoPartZipBytes(): Uint8Array {
|
||||
const bytes = new Uint8Array(IMPORT_TRANSFER_PART_SIZE_BYTES + 3);
|
||||
bytes.set([1, 2, 3], IMPORT_TRANSFER_PART_SIZE_BYTES);
|
||||
return bytes;
|
||||
}
|
||||
|
||||
afterEach(async () => {
|
||||
for (const dir of tempDirs.splice(0)) {
|
||||
await rm(dir, { recursive: true, force: true });
|
||||
}
|
||||
});
|
||||
|
||||
describe("buildImportTransferManifest", () => {
|
||||
it("declares the whole zip and its 32 MB byte-range parts with content hashes", () => {
|
||||
const zipBytes = buildTwoPartZipBytes();
|
||||
const manifest = buildImportTransferManifest(zipBytes);
|
||||
|
||||
expect(manifest.totalBytes).toBe(zipBytes.length);
|
||||
expect(manifest.partSizeBytes).toBe(IMPORT_TRANSFER_PART_SIZE_BYTES);
|
||||
expect(manifest.zipSha256).toBe(sha256Hex(zipBytes));
|
||||
expect(manifest.parts).toEqual([
|
||||
{
|
||||
index: 0,
|
||||
byteSize: IMPORT_TRANSFER_PART_SIZE_BYTES,
|
||||
sha256: sha256Hex(zipBytes.subarray(0, IMPORT_TRANSFER_PART_SIZE_BYTES)),
|
||||
},
|
||||
{
|
||||
index: 1,
|
||||
byteSize: 3,
|
||||
sha256: sha256Hex(zipBytes.subarray(IMPORT_TRANSFER_PART_SIZE_BYTES)),
|
||||
},
|
||||
]);
|
||||
});
|
||||
});
|
||||
|
||||
// Minimal single-entry DEFLATE zip, byte-compatible with the shared reader —
|
||||
// the stored-zip helper cannot model a small-compressed/large-inflated entry.
|
||||
function buildDeflateZip(entryPath: string, text: string): Uint8Array {
|
||||
const raw = Buffer.from(text, "utf8");
|
||||
const body = deflateRawSync(raw);
|
||||
const name = Buffer.from(entryPath, "utf8");
|
||||
let crc = 0xffffffff;
|
||||
for (const byte of raw) {
|
||||
crc ^= byte;
|
||||
for (let bit = 0; bit < 8; bit += 1) crc = (crc & 1) === 1 ? (crc >>> 1) ^ 0xedb88320 : crc >>> 1;
|
||||
}
|
||||
crc = (crc ^ 0xffffffff) >>> 0;
|
||||
const local = Buffer.alloc(30 + name.length);
|
||||
local.writeUInt32LE(0x04034b50, 0);
|
||||
local.writeUInt16LE(20, 4);
|
||||
local.writeUInt16LE(0x0800, 6);
|
||||
local.writeUInt16LE(8, 8);
|
||||
local.writeUInt32LE(crc, 14);
|
||||
local.writeUInt32LE(body.length, 18);
|
||||
local.writeUInt32LE(raw.length, 22);
|
||||
local.writeUInt16LE(name.length, 26);
|
||||
name.copy(local, 30);
|
||||
const central = Buffer.alloc(46 + name.length);
|
||||
central.writeUInt32LE(0x02014b50, 0);
|
||||
central.writeUInt16LE(20, 4);
|
||||
central.writeUInt16LE(20, 6);
|
||||
central.writeUInt16LE(0x0800, 8);
|
||||
central.writeUInt16LE(8, 10);
|
||||
central.writeUInt32LE(crc, 16);
|
||||
central.writeUInt32LE(body.length, 20);
|
||||
central.writeUInt32LE(raw.length, 24);
|
||||
central.writeUInt16LE(name.length, 28);
|
||||
central.writeUInt32LE(0, 42);
|
||||
name.copy(central, 46);
|
||||
const eocd = Buffer.alloc(22);
|
||||
eocd.writeUInt32LE(0x06054b50, 0);
|
||||
eocd.writeUInt16LE(1, 8);
|
||||
eocd.writeUInt16LE(1, 10);
|
||||
eocd.writeUInt32LE(central.length, 12);
|
||||
eocd.writeUInt32LE(local.length + body.length, 16);
|
||||
return new Uint8Array(Buffer.concat([local, body, central, eocd]));
|
||||
}
|
||||
|
||||
describe("resolveChunkedImportZip", () => {
|
||||
|
||||
it("returns null for a zip at or under the threshold", async () => {
|
||||
const dir = await makeTempDir();
|
||||
const zipPath = path.join(dir, "small.zip");
|
||||
await writeFile(zipPath, Buffer.alloc(1024));
|
||||
|
||||
expect(await resolveChunkedImportZip(zipPath)).toBeNull();
|
||||
});
|
||||
|
||||
it("takes the chunked path for a small zip whose entries inflate past the threshold", async () => {
|
||||
const dir = await makeTempDir();
|
||||
const zipPath = path.join(dir, "dense-package.zip");
|
||||
// ~64 MB of repetitive text DEFLATEs to a tiny file: far under the raw
|
||||
// 48 MB threshold, but the inline body would carry the inflated entries,
|
||||
// so the estimated request size sends the zip down the chunked path.
|
||||
const zipBytes = buildDeflateZip("dense-package/NOTES.md", "paperclip agent docs\n".repeat(3_200_000));
|
||||
await writeFile(zipPath, zipBytes);
|
||||
|
||||
const resolved = await resolveChunkedImportZip(zipPath);
|
||||
expect(resolved).not.toBeNull();
|
||||
expect(resolved!.rootPath).toBe("dense-package");
|
||||
expect(sha256Hex(resolved!.zipBytes)).toBe(sha256Hex(zipBytes));
|
||||
});
|
||||
|
||||
it("uses the lower existing-company threshold for the chunk decision", async () => {
|
||||
const dir = await makeTempDir();
|
||||
const packageDir = path.join(dir, "midsize-package");
|
||||
await mkdir(path.join(packageDir, "blobs"), { recursive: true });
|
||||
await writeFile(path.join(packageDir, "COMPANY.md"), "# Company\n");
|
||||
// ~9 MB of blob bytes: estimated inline ~12 MB — fine for the generic
|
||||
// import path (64 MB parser) but over the existing-company path's
|
||||
// default 10 MB parser, so only the existing-target threshold chunks it.
|
||||
await writeFile(path.join(packageDir, "blobs", "1a2b3c4d"), Buffer.alloc(9 * 1024 * 1024, 5));
|
||||
|
||||
expect(await resolveChunkedImportZip(packageDir)).toBeNull();
|
||||
const chunked = await resolveChunkedImportZip(
|
||||
packageDir,
|
||||
EXISTING_COMPANY_CHUNKED_IMPORT_THRESHOLD_BYTES,
|
||||
);
|
||||
expect(chunked).not.toBeNull();
|
||||
expect(chunked!.rootPath).toBe("midsize-package");
|
||||
});
|
||||
|
||||
it("keeps a small zip inline when its entries stay under the estimated threshold", async () => {
|
||||
const dir = await makeTempDir();
|
||||
const zipPath = path.join(dir, "modest-package.zip");
|
||||
await writeFile(zipPath, buildDeflateZip("modest-package/COMPANY.md", "# Company\n"));
|
||||
|
||||
expect(await resolveChunkedImportZip(zipPath)).toBeNull();
|
||||
});
|
||||
|
||||
it("reads an oversized zip file as-is so its declared hashes match the file on disk", async () => {
|
||||
const dir = await makeTempDir();
|
||||
const zipPath = path.join(dir, "big-package.zip");
|
||||
const zipBytes = Buffer.alloc(CHUNKED_IMPORT_THRESHOLD_BYTES + 1024, 7);
|
||||
await writeFile(zipPath, zipBytes);
|
||||
|
||||
const resolved = await resolveChunkedImportZip(zipPath);
|
||||
expect(resolved).not.toBeNull();
|
||||
expect(resolved!.rootPath).toBe("big-package");
|
||||
expect(resolved!.zipBytes.length).toBe(zipBytes.length);
|
||||
expect(sha256Hex(resolved!.zipBytes)).toBe(sha256Hex(zipBytes));
|
||||
});
|
||||
|
||||
it("returns null for a folder whose portable content is under the threshold", async () => {
|
||||
const dir = await makeTempDir();
|
||||
const packageDir = path.join(dir, "small-package");
|
||||
await mkdir(packageDir, { recursive: true });
|
||||
await writeFile(path.join(packageDir, "COMPANY.md"), "# Company\n");
|
||||
|
||||
expect(await resolveChunkedImportZip(packageDir)).toBeNull();
|
||||
});
|
||||
|
||||
it("takes the chunked path for a binary-heavy folder whose inline body outgrows the threshold", async () => {
|
||||
const dir = await makeTempDir();
|
||||
const packageDir = path.join(dir, "binary-package");
|
||||
await mkdir(path.join(packageDir, "blobs"), { recursive: true });
|
||||
await writeFile(path.join(packageDir, "COMPANY.md"), "# Company\n");
|
||||
// 40 MB of raw blob bytes: under the 48 MB raw threshold, but the inline
|
||||
// JSON body would carry them base64-inflated (~53 MB) — past the
|
||||
// threshold on the estimated request size, so the zip travels chunked.
|
||||
await writeFile(
|
||||
path.join(packageDir, "blobs", "9a1b2c3d"),
|
||||
Buffer.alloc(40 * 1024 * 1024, 3),
|
||||
);
|
||||
|
||||
const resolved = await resolveChunkedImportZip(packageDir);
|
||||
expect(resolved).not.toBeNull();
|
||||
expect(resolved!.rootPath).toBe("binary-package");
|
||||
const archive = await readZipArchive(resolved!.zipBytes);
|
||||
expect(Object.keys(archive.files).sort()).toEqual(["COMPANY.md", "blobs/9a1b2c3d"]);
|
||||
});
|
||||
|
||||
it("keeps a text folder under both the raw and estimated measures inline", async () => {
|
||||
const dir = await makeTempDir();
|
||||
const packageDir = path.join(dir, "text-package");
|
||||
await mkdir(packageDir, { recursive: true });
|
||||
await writeFile(path.join(packageDir, "COMPANY.md"), "# Company\n");
|
||||
// Sizable but nowhere near the threshold on either measure: text entries
|
||||
// travel JSON-escaped, close to their raw size, so no base64 inflation
|
||||
// pushes this folder onto the chunked path.
|
||||
await writeFile(path.join(packageDir, "NOTES.md"), "agent docs line\n".repeat(200_000));
|
||||
|
||||
expect(await resolveChunkedImportZip(packageDir)).toBeNull();
|
||||
});
|
||||
|
||||
it("zips an oversized folder in memory with the same walk filters as the inline path", async () => {
|
||||
const dir = await makeTempDir();
|
||||
const packageDir = path.join(dir, "big-package");
|
||||
await mkdir(path.join(packageDir, "blobs"), { recursive: true });
|
||||
await mkdir(path.join(packageDir, ".git"), { recursive: true });
|
||||
await writeFile(path.join(packageDir, "COMPANY.md"), "# Company\n");
|
||||
await writeFile(path.join(packageDir, "notes.txt"), "not portable\n");
|
||||
await writeFile(path.join(packageDir, ".git", "HEAD"), "ref: refs/heads/main\n");
|
||||
await writeFile(
|
||||
path.join(packageDir, "blobs", "4f2d1c9a"),
|
||||
Buffer.alloc(CHUNKED_IMPORT_THRESHOLD_BYTES + 1024, 9),
|
||||
);
|
||||
|
||||
const resolved = await resolveChunkedImportZip(packageDir);
|
||||
expect(resolved).not.toBeNull();
|
||||
expect(resolved!.rootPath).toBe("big-package");
|
||||
|
||||
// The archive unzips back into the same bundle the inline source carries.
|
||||
const archive = await readZipArchive(resolved!.zipBytes);
|
||||
expect(archive.rootPath).toBe("big-package");
|
||||
expect(Object.keys(archive.files).sort()).toEqual(["COMPANY.md", "blobs/4f2d1c9a"]);
|
||||
expect(archive.files["COMPANY.md"]).toBe("# Company\n");
|
||||
});
|
||||
});
|
||||
|
||||
describe("uploadCompanyImportTransfer", () => {
|
||||
const zipBytes = buildTwoPartZipBytes();
|
||||
type TransferApi = Parameters<typeof uploadCompanyImportTransfer>[0];
|
||||
|
||||
function fakeApi(overrides: { post?: ReturnType<typeof vi.fn>; putRaw?: ReturnType<typeof vi.fn> } = {}) {
|
||||
const post = overrides.post
|
||||
?? vi.fn().mockResolvedValue({
|
||||
transferId: "transfer-1",
|
||||
status: "running",
|
||||
alreadyCompleted: false,
|
||||
totalParts: 2,
|
||||
missingParts: [0, 1],
|
||||
});
|
||||
const putRaw = overrides.putRaw ?? vi.fn().mockResolvedValue({ ok: true });
|
||||
return { api: { post, putRaw } as unknown as TransferApi, post, putRaw };
|
||||
}
|
||||
|
||||
it("uploads only the parts the server reports missing", async () => {
|
||||
const { api, post, putRaw } = fakeApi({
|
||||
post: vi.fn().mockResolvedValue({
|
||||
transferId: "transfer-1",
|
||||
status: "running",
|
||||
alreadyCompleted: false,
|
||||
totalParts: 2,
|
||||
missingParts: [1],
|
||||
}),
|
||||
});
|
||||
const progress: number[] = [];
|
||||
|
||||
const transferId = await uploadCompanyImportTransfer(api, zipBytes, {
|
||||
onProgress: (update) => progress.push(update.uploadedParts),
|
||||
});
|
||||
|
||||
expect(transferId).toBe("transfer-1");
|
||||
expect(post).toHaveBeenCalledWith(
|
||||
"/api/companies/import/transfers",
|
||||
expect.objectContaining({ totalBytes: zipBytes.length }),
|
||||
);
|
||||
expect(putRaw).toHaveBeenCalledTimes(1);
|
||||
expect(putRaw.mock.calls[0]![0]).toBe("/api/companies/import/transfers/transfer-1/parts/1");
|
||||
expect(putRaw.mock.calls[0]![1]).toHaveLength(3);
|
||||
expect(progress).toEqual([2]);
|
||||
});
|
||||
|
||||
it("retries a failed part before succeeding", async () => {
|
||||
const putRaw = vi.fn()
|
||||
.mockRejectedValueOnce(new Error("socket hang up"))
|
||||
.mockRejectedValueOnce(new Error("socket hang up"))
|
||||
.mockResolvedValue({ ok: true });
|
||||
const { api } = fakeApi({ putRaw });
|
||||
|
||||
await expect(uploadCompanyImportTransfer(api, zipBytes)).resolves.toBe("transfer-1");
|
||||
// Part 0 took three attempts; part 1 succeeded first try.
|
||||
expect(putRaw).toHaveBeenCalledTimes(4);
|
||||
});
|
||||
|
||||
it("surfaces the upload error after exhausting the per-part attempts", async () => {
|
||||
const putRaw = vi.fn().mockRejectedValue(new Error("socket hang up"));
|
||||
const { api } = fakeApi({ putRaw });
|
||||
|
||||
await expect(uploadCompanyImportTransfer(api, zipBytes)).rejects.toThrow("socket hang up");
|
||||
expect(putRaw).toHaveBeenCalledTimes(3);
|
||||
});
|
||||
|
||||
it("refuses a transfer whose content was already applied", async () => {
|
||||
const { api, putRaw } = fakeApi({
|
||||
post: vi.fn().mockResolvedValue({
|
||||
transferId: "transfer-1",
|
||||
status: "completed",
|
||||
alreadyCompleted: true,
|
||||
totalParts: 2,
|
||||
missingParts: [],
|
||||
}),
|
||||
});
|
||||
|
||||
await expect(uploadCompanyImportTransfer(api, zipBytes)).rejects.toThrow(/already imported/);
|
||||
expect(putRaw).not.toHaveBeenCalled();
|
||||
});
|
||||
});
|
||||
|
||||
describe("company import command over the chunked transfer path", () => {
|
||||
let fetchMock: ReturnType<typeof vi.fn>;
|
||||
let logSpy: ReturnType<typeof vi.spyOn>;
|
||||
|
||||
beforeEach(() => {
|
||||
process.env = { ...ORIGINAL_ENV };
|
||||
delete process.env.PAPERCLIP_API_URL;
|
||||
delete process.env.PAPERCLIP_API_KEY;
|
||||
delete process.env.PAPERCLIP_COMPANY_ID;
|
||||
fetchMock = vi.fn();
|
||||
vi.stubGlobal("fetch", fetchMock);
|
||||
logSpy = vi.spyOn(console, "log").mockImplementation(() => undefined);
|
||||
vi.spyOn(console, "error").mockImplementation(() => undefined);
|
||||
});
|
||||
|
||||
afterEach(() => {
|
||||
process.env = { ...ORIGINAL_ENV };
|
||||
vi.unstubAllGlobals();
|
||||
vi.restoreAllMocks();
|
||||
});
|
||||
|
||||
async function runCommand(args: string[]): Promise<void> {
|
||||
const program = new Command();
|
||||
program.exitOverride();
|
||||
program.configureOutput({
|
||||
writeOut: () => undefined,
|
||||
writeErr: () => undefined,
|
||||
});
|
||||
registerCompanyCommands(program);
|
||||
await program.parseAsync(args, { from: "user" });
|
||||
}
|
||||
|
||||
function jsonResponse(body: unknown, status = 200): Response {
|
||||
return new Response(JSON.stringify(body), {
|
||||
status,
|
||||
headers: { "content-type": "application/json" },
|
||||
});
|
||||
}
|
||||
|
||||
function minimalPreview() {
|
||||
return {
|
||||
include: { company: true, agents: true, projects: true, issues: true },
|
||||
targetCompanyId: null,
|
||||
targetCompanyName: null,
|
||||
collisionStrategy: "rename",
|
||||
selectedAgentSlugs: [],
|
||||
plan: { companyAction: "create", agentPlans: [], projectPlans: [], issuePlans: [] },
|
||||
manifest: { agents: [], projects: [], issues: [], skills: [], company: null },
|
||||
files: {},
|
||||
envInputs: [],
|
||||
warnings: [],
|
||||
errors: [],
|
||||
};
|
||||
}
|
||||
|
||||
it("slices an oversized local zip into a transfer and applies it against the spool", async () => {
|
||||
const dir = await makeTempDir();
|
||||
const zipPath = path.join(dir, "big-package.zip");
|
||||
await writeFile(zipPath, Buffer.alloc(CHUNKED_IMPORT_THRESHOLD_BYTES + 1024, 5));
|
||||
|
||||
fetchMock
|
||||
.mockResolvedValueOnce(jsonResponse({
|
||||
transferId: "transfer-1",
|
||||
status: "running",
|
||||
alreadyCompleted: false,
|
||||
totalParts: 2,
|
||||
missingParts: [0, 1],
|
||||
}))
|
||||
.mockResolvedValueOnce(jsonResponse({ ok: true, index: 0, alreadyCompleted: false }))
|
||||
.mockResolvedValueOnce(jsonResponse({ ok: true, index: 1, alreadyCompleted: false }))
|
||||
.mockResolvedValueOnce(jsonResponse(minimalPreview()))
|
||||
.mockResolvedValueOnce(jsonResponse({
|
||||
company: { id: "company-9", name: "Imported", action: "created" },
|
||||
agents: [],
|
||||
skills: [],
|
||||
projects: [],
|
||||
routines: [],
|
||||
envInputs: [],
|
||||
warnings: [],
|
||||
}));
|
||||
|
||||
await runCommand([
|
||||
"company",
|
||||
"import",
|
||||
zipPath,
|
||||
"--target",
|
||||
"new",
|
||||
"--yes",
|
||||
"--json",
|
||||
"--api-base",
|
||||
"http://paperclip.test",
|
||||
"--api-key",
|
||||
"board-token",
|
||||
]);
|
||||
|
||||
expect(fetchMock).toHaveBeenNthCalledWith(
|
||||
1,
|
||||
"http://paperclip.test/api/companies/import/transfers",
|
||||
expect.objectContaining({ method: "POST" }),
|
||||
);
|
||||
const declared = JSON.parse(String(fetchMock.mock.calls[0]![1].body));
|
||||
expect(declared.totalBytes).toBe(CHUNKED_IMPORT_THRESHOLD_BYTES + 1024);
|
||||
expect(declared.parts).toHaveLength(2);
|
||||
|
||||
expect(fetchMock).toHaveBeenNthCalledWith(
|
||||
2,
|
||||
"http://paperclip.test/api/companies/import/transfers/transfer-1/parts/0",
|
||||
expect.objectContaining({
|
||||
method: "PUT",
|
||||
headers: expect.objectContaining({ "content-type": "application/octet-stream" }),
|
||||
}),
|
||||
);
|
||||
expect(fetchMock).toHaveBeenNthCalledWith(
|
||||
3,
|
||||
"http://paperclip.test/api/companies/import/transfers/transfer-1/parts/1",
|
||||
expect.objectContaining({ method: "PUT" }),
|
||||
);
|
||||
expect(fetchMock).toHaveBeenNthCalledWith(
|
||||
4,
|
||||
"http://paperclip.test/api/companies/import/transfers/transfer-1/preview",
|
||||
expect.objectContaining({ method: "POST" }),
|
||||
);
|
||||
// Preview and apply carry the meta fields, never an inline source.
|
||||
const previewBody = JSON.parse(String(fetchMock.mock.calls[3]![1].body));
|
||||
expect(previewBody.target).toEqual({ mode: "new_company", newCompanyName: null });
|
||||
expect(previewBody).not.toHaveProperty("source");
|
||||
expect(fetchMock).toHaveBeenNthCalledWith(
|
||||
5,
|
||||
"http://paperclip.test/api/companies/import/transfers/transfer-1/apply",
|
||||
expect.objectContaining({ method: "POST" }),
|
||||
);
|
||||
const applyBody = JSON.parse(String(fetchMock.mock.calls[4]![1].body));
|
||||
expect(applyBody).not.toHaveProperty("source");
|
||||
|
||||
expect(JSON.parse(String(logSpy.mock.calls.at(-1)?.[0]))).toMatchObject({
|
||||
company: { id: "company-9" },
|
||||
});
|
||||
});
|
||||
|
||||
it("keeps small local zips on the inline single-shot path", async () => {
|
||||
const dir = await makeTempDir();
|
||||
const zipPath = path.join(dir, "small-package.zip");
|
||||
await writeFile(zipPath, createStoredZipArchive({ "COMPANY.md": "# Company\n" }, "small-package"));
|
||||
|
||||
fetchMock.mockResolvedValueOnce(jsonResponse(minimalPreview()));
|
||||
|
||||
await runCommand([
|
||||
"company",
|
||||
"import",
|
||||
zipPath,
|
||||
"--target",
|
||||
"new",
|
||||
"--dry-run",
|
||||
"--json",
|
||||
"--api-base",
|
||||
"http://paperclip.test",
|
||||
"--api-key",
|
||||
"board-token",
|
||||
]);
|
||||
|
||||
expect(fetchMock).toHaveBeenCalledTimes(1);
|
||||
expect(fetchMock).toHaveBeenCalledWith(
|
||||
"http://paperclip.test/api/companies/import/preview",
|
||||
expect.objectContaining({ method: "POST" }),
|
||||
);
|
||||
const body = JSON.parse(String(fetchMock.mock.calls[0]![1].body));
|
||||
expect(body.source.type).toBe("inline");
|
||||
expect(body.source.files["COMPANY.md"]).toBe("# Company\n");
|
||||
});
|
||||
});
|
||||
|
|
@ -88,6 +88,15 @@ export class PaperclipApiClient {
|
|||
}, opts);
|
||||
}
|
||||
|
||||
/** Raw binary upload (e.g. one chunked import-transfer part); the body travels as-is. */
|
||||
putRaw<T>(path: string, body: Uint8Array, opts?: RequestOptions): Promise<T | null> {
|
||||
return this.request<T>(path, {
|
||||
method: "PUT",
|
||||
body: body as unknown as BodyInit,
|
||||
headers: { "content-type": "application/octet-stream" },
|
||||
}, opts);
|
||||
}
|
||||
|
||||
delete<T>(path: string, opts?: RequestOptions): Promise<T | null> {
|
||||
return this.request<T>(path, { method: "DELETE" }, opts);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -1,4 +1,5 @@
|
|||
import { Command } from "commander";
|
||||
import { createHash } from "node:crypto";
|
||||
import { mkdir, readdir, readFile, stat, writeFile } from "node:fs/promises";
|
||||
import path from "node:path";
|
||||
import * as p from "@clack/prompts";
|
||||
|
|
@ -12,10 +13,24 @@ import type {
|
|||
CompanyPortabilityPreviewResult,
|
||||
CompanyPortabilityImportResult,
|
||||
} from "@paperclipai/shared";
|
||||
import {
|
||||
companyImportTransferApplyPath,
|
||||
companyImportTransferPartPath,
|
||||
companyImportTransferPreviewPath,
|
||||
COMPANY_IMPORT_TRANSFERS_ROUTE_PATH,
|
||||
type CompanyImportTransferCreated,
|
||||
type CompanyImportTransferDeclaration,
|
||||
} from "@paperclipai/shared/company-import-transfer";
|
||||
import { getTelemetryClient, trackCompanyImported } from "../../telemetry.js";
|
||||
import { ApiRequestError } from "../../client/http.js";
|
||||
import { ApiRequestError, type PaperclipApiClient } from "../../client/http.js";
|
||||
import { openUrl } from "../../client/board-auth.js";
|
||||
import { binaryContentTypeByExtension, bytesToPortableFileEntry, isBlobStorePath, readZipArchive } from "./zip.js";
|
||||
import {
|
||||
binaryContentTypeByExtension,
|
||||
bytesToPortableFileEntry,
|
||||
createStoredZipArchive,
|
||||
isBlobStorePath,
|
||||
readZipArchive,
|
||||
} from "./zip.js";
|
||||
import {
|
||||
addCommonClientOptions,
|
||||
apiPath,
|
||||
|
|
@ -927,23 +942,23 @@ async function pathExists(inputPath: string): Promise<boolean> {
|
|||
}
|
||||
}
|
||||
|
||||
async function collectPackageFiles(
|
||||
async function collectPackageFileBytes(
|
||||
root: string,
|
||||
current: string,
|
||||
files: Record<string, CompanyPortabilityFileEntry>,
|
||||
files: Record<string, Uint8Array>,
|
||||
): Promise<void> {
|
||||
const entries = await readdir(current, { withFileTypes: true });
|
||||
for (const entry of entries) {
|
||||
if (entry.name.startsWith(".git")) continue;
|
||||
const absolutePath = path.join(current, entry.name);
|
||||
if (entry.isDirectory()) {
|
||||
await collectPackageFiles(root, absolutePath, files);
|
||||
await collectPackageFileBytes(root, absolutePath, files);
|
||||
continue;
|
||||
}
|
||||
if (!entry.isFile()) continue;
|
||||
const relativePath = path.relative(root, absolutePath).replace(/\\/g, "/");
|
||||
if (!shouldIncludePortableFile(relativePath)) continue;
|
||||
files[relativePath] = bytesToPortableFileEntry(relativePath, await readFile(absolutePath));
|
||||
files[relativePath] = await readFile(absolutePath);
|
||||
}
|
||||
}
|
||||
|
||||
|
|
@ -965,14 +980,232 @@ export async function resolveInlineSourceFromPath(inputPath: string): Promise<{
|
|||
}
|
||||
|
||||
const rootDir = resolvedStat.isDirectory() ? resolved : path.dirname(resolved);
|
||||
const files: Record<string, CompanyPortabilityFileEntry> = {};
|
||||
await collectPackageFiles(rootDir, rootDir, files);
|
||||
const fileBytes: Record<string, Uint8Array> = {};
|
||||
await collectPackageFileBytes(rootDir, rootDir, fileBytes);
|
||||
return {
|
||||
rootPath: path.basename(rootDir),
|
||||
files,
|
||||
files: Object.fromEntries(
|
||||
Object.entries(fileBytes).map(([relativePath, bytes]) => [
|
||||
relativePath,
|
||||
bytesToPortableFileEntry(relativePath, bytes),
|
||||
]),
|
||||
),
|
||||
};
|
||||
}
|
||||
|
||||
// ── Chunked transfer flow for large local packages ───────────────────
|
||||
//
|
||||
// A local package over the threshold is not posted as one inline JSON body:
|
||||
// its zip is declared as a chunked transfer (whole-file and per-part sha256),
|
||||
// the parts are uploaded individually with per-part retries, and preview and
|
||||
// apply run server-side against the assembled spool. Re-declaring the same
|
||||
// content — after a failure or an interrupted run — resumes the prior
|
||||
// transfer, so only the parts the server is missing are ever re-uploaded.
|
||||
|
||||
export const CHUNKED_IMPORT_THRESHOLD_BYTES = 48 * 1024 * 1024;
|
||||
// Imports into an EXISTING company post to /api/companies/:id/imports/*,
|
||||
// which sits behind the server's default 10 MB JSON parser — only the
|
||||
// generic /api/companies/import path carries the 64 MB portable limit. The
|
||||
// chunk decision for existing targets therefore uses this lower threshold
|
||||
// (margin under 10 MB for the envelope), or the inline body would 413.
|
||||
export const EXISTING_COMPANY_CHUNKED_IMPORT_THRESHOLD_BYTES = 8 * 1024 * 1024;
|
||||
export const IMPORT_TRANSFER_PART_SIZE_BYTES = 32 * 1024 * 1024;
|
||||
const IMPORT_TRANSFER_PART_ATTEMPTS = 3;
|
||||
|
||||
// ── Inline request size estimation ───────────────────────────────────
|
||||
//
|
||||
// Mirrors `estimateInlineImportBytes` in ui/src/lib/import-preflight.ts (the
|
||||
// CLI cannot import from ui/) — keep the math on both sides in sync. The
|
||||
// server enforces its body limit on raw request bytes, so each entry is
|
||||
// measured the way it actually travels: JSON-escaped UTF-8 for text
|
||||
// (multi-byte characters and escape sequences both inflate past
|
||||
// `String.length`), and the base64 payload plus its object structure for
|
||||
// binary entries (base64 and MIME types are ASCII, one byte per character).
|
||||
|
||||
const inlineEstimateUtf8 = new TextEncoder();
|
||||
|
||||
// Fixed serialization overhead of a base64 entry object around its data and
|
||||
// contentType values: {"encoding":"base64","data":"…","contentType":"…"}.
|
||||
const BASE64_ENTRY_STRUCTURE_BYTES = '{"encoding":"base64","data":"","contentType":""}'.length;
|
||||
|
||||
// Allowance for everything in the request body besides the files map itself
|
||||
// (rootPath, include flags, target, collision strategy, adapter overrides,
|
||||
// braces and commas). Deliberately generous so the estimate never undercounts.
|
||||
const REQUEST_ENVELOPE_ALLOWANCE_BYTES = 256 * 1024;
|
||||
|
||||
function fileEntryInlineBytes(entry: CompanyPortabilityFileEntry): number {
|
||||
if (typeof entry === "string") return inlineEstimateUtf8.encode(JSON.stringify(entry)).length;
|
||||
return BASE64_ENTRY_STRUCTURE_BYTES + entry.data.length + (entry.contentType?.length ?? 0);
|
||||
}
|
||||
|
||||
/**
|
||||
* Approximate JSON request size of an inline import: JSON-escaped UTF-8 text
|
||||
* bytes, base64 payloads with their entry structure, the serialized file-path
|
||||
* keys (thousands of paths are real bytes), and an envelope allowance for the
|
||||
* rest of the request body.
|
||||
*/
|
||||
function estimateInlineImportBytes(files: Record<string, CompanyPortabilityFileEntry>): number {
|
||||
let total = REQUEST_ENVELOPE_ALLOWANCE_BYTES;
|
||||
for (const [filePath, entry] of Object.entries(files)) {
|
||||
// "path": entry, → key bytes + colon + comma.
|
||||
total += inlineEstimateUtf8.encode(JSON.stringify(filePath)).length + 2 + fileEntryInlineBytes(entry);
|
||||
}
|
||||
return total;
|
||||
}
|
||||
|
||||
export interface ImportTransferUploadProgress {
|
||||
uploadedParts: number;
|
||||
totalParts: number;
|
||||
uploadedBytes: number;
|
||||
totalBytes: number;
|
||||
}
|
||||
|
||||
export function buildImportTransferManifest(zipBytes: Uint8Array): CompanyImportTransferDeclaration {
|
||||
const sha256 = (bytes: Uint8Array) => createHash("sha256").update(bytes).digest("hex");
|
||||
const parts: CompanyImportTransferDeclaration["parts"] = [];
|
||||
for (let offset = 0; offset < zipBytes.length; offset += IMPORT_TRANSFER_PART_SIZE_BYTES) {
|
||||
const byteSize = Math.min(IMPORT_TRANSFER_PART_SIZE_BYTES, zipBytes.length - offset);
|
||||
parts.push({
|
||||
index: parts.length,
|
||||
byteSize,
|
||||
sha256: sha256(zipBytes.subarray(offset, offset + byteSize)),
|
||||
});
|
||||
}
|
||||
return {
|
||||
totalBytes: zipBytes.length,
|
||||
zipSha256: sha256(zipBytes),
|
||||
partSizeBytes: IMPORT_TRANSFER_PART_SIZE_BYTES,
|
||||
parts,
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Resolve a local import source into raw zip bytes when its package is too
|
||||
* large to travel as one inline JSON body: a .zip file is read as-is (so its
|
||||
* declared hashes match the file on disk), a folder is packaged as a stored
|
||||
* zip in memory with the same walk filters the inline path uses. Both source
|
||||
* kinds are measured twice — raw bytes as a fast path, then the estimated
|
||||
* inline request size, because base64 inflates binary entries ~4/3 and a
|
||||
* compressed zip can expand far past its file size. Returns null for sources
|
||||
* under the threshold on both measures — those keep the inline JSON path.
|
||||
*/
|
||||
export async function resolveChunkedImportZip(
|
||||
inputPath: string,
|
||||
thresholdBytes: number = CHUNKED_IMPORT_THRESHOLD_BYTES,
|
||||
): Promise<{
|
||||
zipBytes: Uint8Array;
|
||||
rootPath: string;
|
||||
} | null> {
|
||||
const resolved = path.resolve(inputPath);
|
||||
const resolvedStat = await stat(resolved);
|
||||
if (resolvedStat.isFile() && path.extname(resolved).toLowerCase() === ".zip") {
|
||||
const zipBytes = new Uint8Array(await readFile(resolved));
|
||||
const rootPath = path.basename(resolved, ".zip");
|
||||
if (resolvedStat.size > thresholdBytes) return { zipBytes, rootPath };
|
||||
// A small compressed zip can still expand past server caps as inline
|
||||
// JSON (text compresses well and binary re-inflates ~4/3 as base64), so
|
||||
// the stay-inline decision uses the estimated request size of the same
|
||||
// entries the inline path would send. An unreadable zip stays inline so
|
||||
// that path surfaces its canonical parse error.
|
||||
let archive: Awaited<ReturnType<typeof readZipArchive>>;
|
||||
try {
|
||||
archive = await readZipArchive(zipBytes);
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
if (estimateInlineImportBytes(archive.files) <= thresholdBytes) return null;
|
||||
return { zipBytes, rootPath };
|
||||
}
|
||||
if (!resolvedStat.isDirectory()) return null;
|
||||
const fileBytes: Record<string, Uint8Array> = {};
|
||||
await collectPackageFileBytes(resolved, resolved, fileBytes);
|
||||
const rootPath = path.basename(resolved);
|
||||
// Content bytes alone already past the threshold means the stored zip
|
||||
// (content plus headers) is too.
|
||||
const contentBytes = Object.values(fileBytes).reduce((sum, bytes) => sum + bytes.length, 0);
|
||||
if (contentBytes <= thresholdBytes) {
|
||||
// Raw bytes under the threshold can still blow past server caps once the
|
||||
// inline body is built (binary entries travel base64-inflated), so the
|
||||
// stay-inline decision is made on the estimated request size — the same
|
||||
// entries the inline path would send.
|
||||
const inlineEntries = Object.fromEntries(
|
||||
Object.entries(fileBytes).map(([relativePath, bytes]) => [
|
||||
relativePath,
|
||||
bytesToPortableFileEntry(relativePath, bytes),
|
||||
]),
|
||||
);
|
||||
if (estimateInlineImportBytes(inlineEntries) <= thresholdBytes) return null;
|
||||
}
|
||||
return { zipBytes: createStoredZipArchive(fileBytes, rootPath), rootPath };
|
||||
}
|
||||
|
||||
/**
|
||||
* Declare (or resume) the transfer for these zip bytes and upload every part
|
||||
* the server reports missing, sequentially with per-part retries. Resolves
|
||||
* with the transfer id once the server holds every part.
|
||||
*/
|
||||
export async function uploadCompanyImportTransfer(
|
||||
api: Pick<PaperclipApiClient, "post" | "putRaw">,
|
||||
zipBytes: Uint8Array,
|
||||
opts: { onProgress?: (progress: ImportTransferUploadProgress) => void } = {},
|
||||
): Promise<string> {
|
||||
const manifest = buildImportTransferManifest(zipBytes);
|
||||
const created = await api.post<CompanyImportTransferCreated>(
|
||||
`/api/companies${COMPANY_IMPORT_TRANSFERS_ROUTE_PATH}`,
|
||||
manifest,
|
||||
);
|
||||
if (!created) {
|
||||
throw new Error("Import transfer declaration returned no data.");
|
||||
}
|
||||
if (created.alreadyCompleted) {
|
||||
// The server keys transfers by content, and this exact zip already
|
||||
// finished an apply — its spooled parts are gone, so it cannot re-run.
|
||||
throw new Error(
|
||||
"This exact package was already imported by a completed transfer. Re-export the package to import it again.",
|
||||
);
|
||||
}
|
||||
const missing = new Set(created.missingParts);
|
||||
let uploadedParts = manifest.parts.length - missing.size;
|
||||
let uploadedBytes = manifest.parts.reduce(
|
||||
(sum, part) => (missing.has(part.index) ? sum : sum + part.byteSize),
|
||||
0,
|
||||
);
|
||||
for (const part of manifest.parts) {
|
||||
if (!missing.has(part.index)) continue;
|
||||
const offset = part.index * manifest.partSizeBytes;
|
||||
const bytes = zipBytes.subarray(offset, offset + part.byteSize);
|
||||
let lastError: unknown = null;
|
||||
let uploaded = false;
|
||||
for (let attempt = 0; attempt < IMPORT_TRANSFER_PART_ATTEMPTS && !uploaded; attempt += 1) {
|
||||
try {
|
||||
await api.putRaw(
|
||||
`/api/companies${companyImportTransferPartPath(created.transferId, part.index)}`,
|
||||
bytes,
|
||||
);
|
||||
uploaded = true;
|
||||
} catch (err) {
|
||||
lastError = err;
|
||||
}
|
||||
}
|
||||
if (!uploaded) {
|
||||
// Parts already uploaded stay spooled server-side; re-running the
|
||||
// import resumes from them instead of starting over.
|
||||
throw lastError instanceof Error
|
||||
? lastError
|
||||
: new Error(`Import transfer part ${part.index} failed to upload.`);
|
||||
}
|
||||
uploadedParts += 1;
|
||||
uploadedBytes += part.byteSize;
|
||||
opts.onProgress?.({
|
||||
uploadedParts,
|
||||
totalParts: manifest.parts.length,
|
||||
uploadedBytes,
|
||||
totalBytes: manifest.totalBytes,
|
||||
});
|
||||
}
|
||||
return created.transferId;
|
||||
}
|
||||
|
||||
export async function writeExportToFolder(outDir: string, exported: CompanyPortabilityExportResult): Promise<void> {
|
||||
const root = path.resolve(outDir);
|
||||
await mkdir(root, { recursive: true });
|
||||
|
|
@ -1464,6 +1697,7 @@ export function registerCompanyCommands(program: Command): void {
|
|||
let sourcePayload:
|
||||
| { type: "inline"; rootPath?: string | null; files: Record<string, CompanyPortabilityFileEntry> }
|
||||
| { type: "github"; url: string };
|
||||
let chunkedZip: { zipBytes: Uint8Array; rootPath: string } | null = null;
|
||||
|
||||
const treatAsLocalPath = !isHttpUrl(from) && await pathExists(from);
|
||||
const isGithubSource = looksLikeRepoUrl(from) || (isGithubShorthand(from) && !treatAsLocalPath);
|
||||
|
|
@ -1480,12 +1714,24 @@ export function registerCompanyCommands(program: Command): void {
|
|||
if (opts.ref?.trim()) {
|
||||
throw new Error("--ref is only supported for GitHub import sources.");
|
||||
}
|
||||
const inline = await resolveInlineSourceFromPath(from);
|
||||
sourcePayload = {
|
||||
type: "inline",
|
||||
rootPath: inline.rootPath,
|
||||
files: inline.files,
|
||||
};
|
||||
chunkedZip = await resolveChunkedImportZip(
|
||||
from,
|
||||
target === "existing"
|
||||
? EXISTING_COMPANY_CHUNKED_IMPORT_THRESHOLD_BYTES
|
||||
: CHUNKED_IMPORT_THRESHOLD_BYTES,
|
||||
);
|
||||
if (chunkedZip) {
|
||||
// Too large for one request: the zip travels as a chunked
|
||||
// transfer, so the inline files map is never built or sent.
|
||||
sourcePayload = { type: "inline", rootPath: chunkedZip.rootPath, files: {} };
|
||||
} else {
|
||||
const inline = await resolveInlineSourceFromPath(from);
|
||||
sourcePayload = {
|
||||
type: "inline",
|
||||
rootPath: inline.rootPath,
|
||||
files: inline.files,
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
const sourceLabel = formatSourceLabel(sourcePayload);
|
||||
|
|
@ -1496,15 +1742,40 @@ export function registerCompanyCommands(program: Command): void {
|
|||
companyId: targetPayload.mode === "existing_company" ? targetPayload.companyId : null,
|
||||
});
|
||||
|
||||
// The transfer meta mirrors the inline preview payload minus its
|
||||
// `source` — the source is the assembled zip, spooled server-side.
|
||||
const transferMeta = {
|
||||
include,
|
||||
target: targetPayload,
|
||||
agents,
|
||||
collisionStrategy: collision,
|
||||
};
|
||||
let transferId: string | null = null;
|
||||
if (chunkedZip) {
|
||||
transferId = await uploadCompanyImportTransfer(ctx.api, chunkedZip.zipBytes, {
|
||||
onProgress: ctx.json
|
||||
? undefined
|
||||
: ({ uploadedParts, totalParts, uploadedBytes, totalBytes }) => {
|
||||
console.log(
|
||||
pc.dim(
|
||||
`Uploaded part ${uploadedParts}/${totalParts} (${Math.round(uploadedBytes / (1024 * 1024))} of ${Math.round(totalBytes / (1024 * 1024))} MB)`,
|
||||
),
|
||||
);
|
||||
},
|
||||
});
|
||||
}
|
||||
const transferPreviewPath = transferId
|
||||
? `/api/companies${companyImportTransferPreviewPath(transferId)}`
|
||||
: null;
|
||||
|
||||
let selectedFiles: string[] | undefined;
|
||||
if (interactiveView && !opts.yes && !opts.include?.trim()) {
|
||||
const initialPreview = await ctx.api.post<CompanyPortabilityPreviewResult>(previewApiPath, {
|
||||
source: sourcePayload,
|
||||
include,
|
||||
target: targetPayload,
|
||||
agents,
|
||||
collisionStrategy: collision,
|
||||
});
|
||||
const initialPreview = transferPreviewPath
|
||||
? await ctx.api.post<CompanyPortabilityPreviewResult>(transferPreviewPath, transferMeta)
|
||||
: await ctx.api.post<CompanyPortabilityPreviewResult>(previewApiPath, {
|
||||
source: sourcePayload,
|
||||
...transferMeta,
|
||||
});
|
||||
if (!initialPreview) {
|
||||
throw new Error("Import preview returned no data.");
|
||||
}
|
||||
|
|
@ -1513,13 +1784,15 @@ export function registerCompanyCommands(program: Command): void {
|
|||
|
||||
const previewPayload = {
|
||||
source: sourcePayload,
|
||||
include,
|
||||
target: targetPayload,
|
||||
agents,
|
||||
collisionStrategy: collision,
|
||||
...transferMeta,
|
||||
selectedFiles,
|
||||
};
|
||||
const preview = await ctx.api.post<CompanyPortabilityPreviewResult>(previewApiPath, previewPayload);
|
||||
const preview = transferPreviewPath
|
||||
? await ctx.api.post<CompanyPortabilityPreviewResult>(transferPreviewPath, {
|
||||
...transferMeta,
|
||||
selectedFiles,
|
||||
})
|
||||
: await ctx.api.post<CompanyPortabilityPreviewResult>(previewApiPath, previewPayload);
|
||||
if (!preview) {
|
||||
throw new Error("Import preview returned no data.");
|
||||
}
|
||||
|
|
@ -1576,10 +1849,15 @@ export function registerCompanyCommands(program: Command): void {
|
|||
targetMode: targetPayload.mode,
|
||||
companyId: targetPayload.mode === "existing_company" ? targetPayload.companyId : null,
|
||||
});
|
||||
const imported = await ctx.api.post<CompanyPortabilityImportResult>(importApiPath, {
|
||||
...previewPayload,
|
||||
adapterOverrides,
|
||||
});
|
||||
const imported = transferId
|
||||
? await ctx.api.post<CompanyPortabilityImportResult>(
|
||||
`/api/companies${companyImportTransferApplyPath(transferId)}`,
|
||||
{ ...transferMeta, selectedFiles, adapterOverrides },
|
||||
)
|
||||
: await ctx.api.post<CompanyPortabilityImportResult>(importApiPath, {
|
||||
...previewPayload,
|
||||
adapterOverrides,
|
||||
});
|
||||
if (!imported) {
|
||||
throw new Error("Import request returned no data.");
|
||||
}
|
||||
|
|
|
|||
|
|
@ -8,3 +8,110 @@ export {
|
|||
isBlobStorePath,
|
||||
readZipArchive,
|
||||
} from "@paperclipai/shared/portability-zip";
|
||||
|
||||
// STORE-only zip writer used to package a local folder in memory for the
|
||||
// chunked import transfer path. STORE keeps the writer dependency-free; the
|
||||
// transfer slices the raw archive bytes, so compression only trades CPU for
|
||||
// part count. Classic zip only: entry counts and offsets past the 16/32-bit
|
||||
// header fields would need zip64, which the reader side does not require and
|
||||
// this writer refuses to emit.
|
||||
|
||||
const ZIP_MAX_ENTRIES = 0xffff;
|
||||
const ZIP_MAX_OFFSET_BYTES = 0xffffffff;
|
||||
|
||||
function writeUint16(target: Uint8Array, offset: number, value: number) {
|
||||
target[offset] = value & 0xff;
|
||||
target[offset + 1] = (value >>> 8) & 0xff;
|
||||
}
|
||||
|
||||
function writeUint32(target: Uint8Array, offset: number, value: number) {
|
||||
target[offset] = value & 0xff;
|
||||
target[offset + 1] = (value >>> 8) & 0xff;
|
||||
target[offset + 2] = (value >>> 16) & 0xff;
|
||||
target[offset + 3] = (value >>> 24) & 0xff;
|
||||
}
|
||||
|
||||
function crc32(bytes: Uint8Array) {
|
||||
let crc = 0xffffffff;
|
||||
for (const byte of bytes) {
|
||||
crc ^= byte;
|
||||
for (let bit = 0; bit < 8; bit += 1) {
|
||||
crc = (crc & 1) === 1 ? (crc >>> 1) ^ 0xedb88320 : crc >>> 1;
|
||||
}
|
||||
}
|
||||
return (crc ^ 0xffffffff) >>> 0;
|
||||
}
|
||||
|
||||
/**
|
||||
* Build a stored (uncompressed) zip of `files` under a single `rootPath/`
|
||||
* top-level directory — the layout `readZipArchive` folds back into the
|
||||
* inline `{ rootPath, files }` bundle. Entries are written in sorted path
|
||||
* order and carry no timestamps, so the same content always produces the
|
||||
* same bytes and a re-run resumes its content-addressed transfer.
|
||||
*/
|
||||
export function createStoredZipArchive(files: Record<string, Uint8Array>, rootPath: string): Uint8Array {
|
||||
const entries = Object.entries(files).sort(([left], [right]) => left.localeCompare(right));
|
||||
if (entries.length > ZIP_MAX_ENTRIES) {
|
||||
throw new Error(`Package has too many files to zip (${entries.length}; the zip format caps at ${ZIP_MAX_ENTRIES}).`);
|
||||
}
|
||||
const encoder = new TextEncoder();
|
||||
const localChunks: Uint8Array[] = [];
|
||||
const centralChunks: Uint8Array[] = [];
|
||||
let localOffset = 0;
|
||||
|
||||
for (const [relativePath, body] of entries) {
|
||||
const fileName = encoder.encode(`${rootPath}/${relativePath}`);
|
||||
const checksum = crc32(body);
|
||||
|
||||
const localHeader = new Uint8Array(30 + fileName.length);
|
||||
writeUint32(localHeader, 0, 0x04034b50);
|
||||
writeUint16(localHeader, 4, 20);
|
||||
writeUint16(localHeader, 6, 0x0800);
|
||||
writeUint16(localHeader, 8, 0);
|
||||
writeUint32(localHeader, 14, checksum);
|
||||
writeUint32(localHeader, 18, body.length);
|
||||
writeUint32(localHeader, 22, body.length);
|
||||
writeUint16(localHeader, 26, fileName.length);
|
||||
localHeader.set(fileName, 30);
|
||||
|
||||
const centralHeader = new Uint8Array(46 + fileName.length);
|
||||
writeUint32(centralHeader, 0, 0x02014b50);
|
||||
writeUint16(centralHeader, 4, 20);
|
||||
writeUint16(centralHeader, 6, 20);
|
||||
writeUint16(centralHeader, 8, 0x0800);
|
||||
writeUint16(centralHeader, 10, 0);
|
||||
writeUint32(centralHeader, 16, checksum);
|
||||
writeUint32(centralHeader, 20, body.length);
|
||||
writeUint32(centralHeader, 24, body.length);
|
||||
writeUint16(centralHeader, 28, fileName.length);
|
||||
writeUint32(centralHeader, 42, localOffset);
|
||||
centralHeader.set(fileName, 46);
|
||||
|
||||
localChunks.push(localHeader, body);
|
||||
centralChunks.push(centralHeader);
|
||||
localOffset += localHeader.length + body.length;
|
||||
if (body.length > ZIP_MAX_OFFSET_BYTES || localOffset > ZIP_MAX_OFFSET_BYTES) {
|
||||
throw new Error("Package is too large to zip in memory (zip64 archives are not supported).");
|
||||
}
|
||||
}
|
||||
|
||||
const centralDirectoryLength = centralChunks.reduce((sum, chunk) => sum + chunk.length, 0);
|
||||
const archive = new Uint8Array(localOffset + centralDirectoryLength + 22);
|
||||
let offset = 0;
|
||||
for (const chunk of localChunks) {
|
||||
archive.set(chunk, offset);
|
||||
offset += chunk.length;
|
||||
}
|
||||
const centralDirectoryOffset = offset;
|
||||
for (const chunk of centralChunks) {
|
||||
archive.set(chunk, offset);
|
||||
offset += chunk.length;
|
||||
}
|
||||
writeUint32(archive, offset, 0x06054b50);
|
||||
writeUint16(archive, offset + 8, entries.length);
|
||||
writeUint16(archive, offset + 10, entries.length);
|
||||
writeUint32(archive, offset + 12, centralDirectoryLength);
|
||||
writeUint32(archive, offset + 16, centralDirectoryOffset);
|
||||
|
||||
return archive;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -68,7 +68,10 @@ vi.mock("../services/index.js", () => ({
|
|||
// Passthrough wrapper around the real spool module whose removeImportTransferSpool
|
||||
// can be made to throw, so tests can exercise a post-success cleanup failure
|
||||
// (e.g. EACCES on the spool dir) without touching real filesystem permissions.
|
||||
// assembleImportTransferZip can likewise be made to throw, standing in for the
|
||||
// spool being deleted out from under an unclaimed preview mid-assembly.
|
||||
const spoolRemovalFailure = vi.hoisted(() => ({ error: null as Error | null }));
|
||||
const assemblyFailure = vi.hoisted(() => ({ error: null as Error | null }));
|
||||
vi.mock("../services/company-import-transfers.js", async (importOriginal) => {
|
||||
const actual = await importOriginal<typeof import("../services/company-import-transfers.js")>();
|
||||
return {
|
||||
|
|
@ -77,6 +80,10 @@ vi.mock("../services/company-import-transfers.js", async (importOriginal) => {
|
|||
if (spoolRemovalFailure.error) throw spoolRemovalFailure.error;
|
||||
return actual.removeImportTransferSpool(spoolRoot, runId);
|
||||
},
|
||||
assembleImportTransferZip: async (spoolRoot: string, runId: string, partCount: number) => {
|
||||
if (assemblyFailure.error) throw assemblyFailure.error;
|
||||
return actual.assembleImportTransferZip(spoolRoot, runId, partCount);
|
||||
},
|
||||
};
|
||||
});
|
||||
|
||||
|
|
@ -260,6 +267,7 @@ describeEmbeddedPostgres("company import transfer routes", () => {
|
|||
beforeEach(() => {
|
||||
vi.clearAllMocks();
|
||||
spoolRemovalFailure.error = null;
|
||||
assemblyFailure.error = null;
|
||||
mockCompanyPortabilityService.importBundle.mockResolvedValue({
|
||||
company: { id: companyId, action: "updated" },
|
||||
agents: [{ id: "agent-1" }],
|
||||
|
|
@ -331,6 +339,173 @@ describeEmbeddedPostgres("company import transfer routes", () => {
|
|||
expect(finished.body.missingParts).toEqual([]);
|
||||
});
|
||||
|
||||
it("previews the assembled spool without consuming it, then applies the same transfer", async () => {
|
||||
mockCompanyPortabilityService.previewImport.mockResolvedValue({
|
||||
plan: { companyAction: "update" },
|
||||
warnings: [],
|
||||
errors: [],
|
||||
});
|
||||
const zip = buildFixtureZip();
|
||||
const { body, slices } = declareTransfer(zip, Math.ceil(zip.length / 3));
|
||||
const created = await request(app).post("/api/companies/import/transfers").send(body);
|
||||
const transferId = created.body.transferId as string;
|
||||
for (const [index, slice] of slices.entries()) {
|
||||
expect((await putPart(transferId, index, slice)).status).toBe(200);
|
||||
}
|
||||
|
||||
const previewed = await request(app)
|
||||
.post(`/api/companies/import/transfers/${transferId}/preview`)
|
||||
.send(importMeta);
|
||||
expect(previewed.status).toBe(200);
|
||||
expect(previewed.body.plan).toEqual({ companyAction: "update" });
|
||||
|
||||
// The assembled zip fed the existing preview pipeline unchanged.
|
||||
expect(mockCompanyPortabilityService.previewImport).toHaveBeenCalledTimes(1);
|
||||
const previewBody = mockCompanyPortabilityService.previewImport.mock.calls[0]![0];
|
||||
expect(previewBody.source.type).toBe("inline");
|
||||
expect(previewBody.source.files["COMPANY.md"]).toContain("Chunked Import");
|
||||
expect(previewBody.target).toEqual(importMeta.target);
|
||||
expect(mockCompanyPortabilityService.importBundle).not.toHaveBeenCalled();
|
||||
|
||||
// Preview consumed nothing: the run stays open, the spool stays on disk,
|
||||
// and no part needs re-uploading before the apply.
|
||||
const run = (await companyTransferRunService.getRun(db, transferId))!;
|
||||
expect(run.status).not.toBe("completed");
|
||||
await expect(fs.stat(path.join(spoolRoot, transferId))).resolves.toBeDefined();
|
||||
const status = await request(app).get(`/api/companies/import/transfers/${transferId}`);
|
||||
expect(status.body.missingParts).toEqual([]);
|
||||
|
||||
const applied = await request(app)
|
||||
.post(`/api/companies/import/transfers/${transferId}/apply`)
|
||||
.send(importMeta);
|
||||
expect(applied.status).toBe(200);
|
||||
expect(mockCompanyPortabilityService.importBundle).toHaveBeenCalledTimes(1);
|
||||
expect((await companyTransferRunService.getRun(db, transferId))!.status).toBe("completed");
|
||||
});
|
||||
|
||||
it("refuses to preview while parts are missing", async () => {
|
||||
const zip = buildFixtureZip();
|
||||
const { body, slices } = declareTransfer(zip, Math.ceil(zip.length / 3));
|
||||
const created = await request(app).post("/api/companies/import/transfers").send(body);
|
||||
const transferId = created.body.transferId as string;
|
||||
expect((await putPart(transferId, 1, slices[1]!)).status).toBe(200);
|
||||
|
||||
const previewed = await request(app)
|
||||
.post(`/api/companies/import/transfers/${transferId}/preview`)
|
||||
.send(importMeta);
|
||||
expect(previewed.status).toBe(409);
|
||||
expect(previewed.body.missingParts).toEqual([0, 2]);
|
||||
expect(mockCompanyPortabilityService.previewImport).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("refuses to preview while an apply holds the claim and leaves the spool untouched", async () => {
|
||||
const zip = buildFixtureZip();
|
||||
const { body, slices } = declareTransfer(zip, Math.ceil(zip.length / 2));
|
||||
const created = await request(app).post("/api/companies/import/transfers").send(body);
|
||||
const transferId = created.body.transferId as string;
|
||||
for (const [index, slice] of slices.entries()) {
|
||||
expect((await putPart(transferId, index, slice)).status).toBe(200);
|
||||
}
|
||||
|
||||
// Claim the run as an in-flight apply would: the preview must refuse
|
||||
// before reading the spool the apply is assembling (and deleting on
|
||||
// success) rather than race it.
|
||||
expect(await companyTransferRunService.claimApply(db, transferId)).toBe(true);
|
||||
try {
|
||||
const previewed = await request(app)
|
||||
.post(`/api/companies/import/transfers/${transferId}/preview`)
|
||||
.send(importMeta);
|
||||
expect(previewed.status).toBe(409);
|
||||
expect(previewed.body.error).toContain("apply is in progress");
|
||||
} finally {
|
||||
await companyTransferRunService.releaseApplyClaim(db, transferId, "test release");
|
||||
}
|
||||
expect(mockCompanyPortabilityService.previewImport).not.toHaveBeenCalled();
|
||||
// Every part file survived for the apply to read.
|
||||
for (const [index] of slices.entries()) {
|
||||
await expect(fs.stat(path.join(spoolRoot, transferId, `part-${index}`))).resolves.toBeDefined();
|
||||
}
|
||||
});
|
||||
|
||||
it("refuses to preview an already-applied transfer", async () => {
|
||||
const zip = buildFixtureZip();
|
||||
const { body, slices } = declareTransfer(zip, zip.length);
|
||||
const created = await request(app).post("/api/companies/import/transfers").send(body);
|
||||
const transferId = created.body.transferId as string;
|
||||
expect((await putPart(transferId, 0, slices[0]!)).status).toBe(200);
|
||||
expect(
|
||||
(await request(app).post(`/api/companies/import/transfers/${transferId}/apply`).send(importMeta)).status,
|
||||
).toBe(200);
|
||||
|
||||
const previewed = await request(app)
|
||||
.post(`/api/companies/import/transfers/${transferId}/preview`)
|
||||
.send(importMeta);
|
||||
expect(previewed.status).toBe(409);
|
||||
expect(previewed.body.error).toContain("already been applied");
|
||||
expect(mockCompanyPortabilityService.previewImport).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("returns 410 for a preview against a swept transfer", async () => {
|
||||
const zip = buildFixtureZip();
|
||||
const { body, slices } = declareTransfer(zip, Math.ceil(zip.length / 2));
|
||||
const created = await request(app).post("/api/companies/import/transfers").send(body);
|
||||
const transferId = created.body.transferId as string;
|
||||
expect((await putPart(transferId, 0, slices[0]!)).status).toBe(200);
|
||||
|
||||
const later = new Date(Date.now() + 25 * 60 * 60 * 1000);
|
||||
const swept = await sweepAbandonedImportTransferSpools(db, spoolRoot, { now: later });
|
||||
expect(swept.swept).toBeGreaterThanOrEqual(1);
|
||||
|
||||
const previewed = await request(app)
|
||||
.post(`/api/companies/import/transfers/${transferId}/preview`)
|
||||
.send(importMeta);
|
||||
expect(previewed.status).toBe(410);
|
||||
expect(previewed.body.error).toContain("expired");
|
||||
expect(mockCompanyPortabilityService.previewImport).not.toHaveBeenCalled();
|
||||
expect((await companyTransferRunService.getRun(db, transferId))!.status).toBe("cancelled");
|
||||
});
|
||||
|
||||
it("returns a clean conflict when an apply completes and sweeps the spool mid-preview-assembly", async () => {
|
||||
const zip = buildFixtureZip();
|
||||
const { body, slices } = declareTransfer(zip, Math.ceil(zip.length / 2));
|
||||
const created = await request(app).post("/api/companies/import/transfers").send(body);
|
||||
const transferId = created.body.transferId as string;
|
||||
for (const [index, slice] of slices.entries()) {
|
||||
expect((await putPart(transferId, index, slice)).status).toBe(200);
|
||||
}
|
||||
|
||||
// The race: the preview's initial read sees an open run, then a
|
||||
// concurrent apply settles the run completed and deletes the spool while
|
||||
// the preview is assembling it. The stale initial read comes from a
|
||||
// one-shot getRunForActor stub; the ledger itself already says completed
|
||||
// and the assembly blows up the way missing part files would.
|
||||
expect(await companyTransferRunService.claimApply(db, transferId)).toBe(true);
|
||||
await companyTransferRunService.complete(db, transferId);
|
||||
const completedRun = (await companyTransferRunService.getRun(db, transferId))!;
|
||||
expect(completedRun.status).toBe("completed");
|
||||
const staleRead = vi
|
||||
.spyOn(companyTransferRunService, "getRunForActor")
|
||||
.mockResolvedValueOnce({ ...completedRun, status: "running" });
|
||||
assemblyFailure.error = Object.assign(new Error("ENOENT: no such file or directory, open 'part-0'"), {
|
||||
code: "ENOENT",
|
||||
});
|
||||
try {
|
||||
const previewed = await request(app)
|
||||
.post(`/api/companies/import/transfers/${transferId}/preview`)
|
||||
.send(importMeta);
|
||||
// A clean already-applied conflict, not a 500 from the torn spool.
|
||||
expect(previewed.status).toBe(409);
|
||||
expect(previewed.body.error).toContain("already been applied");
|
||||
} finally {
|
||||
staleRead.mockRestore();
|
||||
assemblyFailure.error = null;
|
||||
}
|
||||
expect(mockCompanyPortabilityService.previewImport).not.toHaveBeenCalled();
|
||||
// Preview stayed read-only: the run keeps the terminal state the apply
|
||||
// gave it.
|
||||
expect((await companyTransferRunService.getRun(db, transferId))!.status).toBe("completed");
|
||||
});
|
||||
|
||||
it("applies through the async import job machinery when requested", async () => {
|
||||
const zip = buildFixtureZip();
|
||||
const { body, slices } = declareTransfer(zip, Math.ceil(zip.length / 2));
|
||||
|
|
|
|||
|
|
@ -836,16 +836,48 @@ export function companyRoutes(db: Db, storage?: StorageService, options?: Compan
|
|||
} satisfies CompanyImportTransferStatus);
|
||||
});
|
||||
|
||||
// Assemble the spooled parts back into the original zip, verify it against
|
||||
// the declared whole-file hash, and run it through the exact import path a
|
||||
// single-shot zip upload takes. The body carries the same meta fields the
|
||||
// multipart route's `meta` field does (include/target/collisionStrategy/...).
|
||||
router.post(`${COMPANY_IMPORT_TRANSFERS_ROUTE_PATH}/:transferId/apply`, async (req, res) => {
|
||||
assertBoard(req);
|
||||
/**
|
||||
* Resolve a completed transfer into the same raw preview/import body a
|
||||
* single-shot zip upload produces: load the run, require every part, then
|
||||
* assemble the spool and verify it against the declared whole-file hash.
|
||||
* Shared by the transfer preview and apply routes; returns null after
|
||||
* responding 409 when parts are still missing. A whole-file mismatch fails
|
||||
* closed for both callers: every part verified individually but the whole
|
||||
* does not match the declaration, so the spool is deleted and a resume
|
||||
* re-uploads every part instead of re-assembling the same corrupt bytes.
|
||||
*
|
||||
* With `claimApply` (the apply route), the run is atomically claimed before
|
||||
* the spool is touched: the import pipeline is not idempotent, so of any
|
||||
* overlapping applies of this transfer exactly one may run it — a single
|
||||
* guarded status flip to "applying" that losers see as 409. The claim is
|
||||
* settled by complete() on success or the guarded releaseApplyClaim() on
|
||||
* error (a released run stays retryable; a settled one is never reopened).
|
||||
* The preview route takes no claim — it never consumes the transfer — so
|
||||
* both its status gate and its assembly errors are re-checked against the
|
||||
* ledger (see the catch below) to keep a concurrent apply's spool cleanup
|
||||
* from surfacing as an internal error.
|
||||
*/
|
||||
async function resolveImportTransferBody(
|
||||
req: Request,
|
||||
res: Response,
|
||||
options: { claimApply?: boolean } = {},
|
||||
) {
|
||||
const run = await requireImportTransferRun(req);
|
||||
if (run.status === "completed") {
|
||||
throw conflict("Import transfer has already been applied");
|
||||
}
|
||||
if (run.status === "applying") {
|
||||
// An apply holds the claim right now: its assembly is reading the spool
|
||||
// and its success deletes it, so neither a preview nor a second apply
|
||||
// may touch the parts until the claim settles.
|
||||
throw conflict("Import transfer apply is in progress; retry after it settles");
|
||||
}
|
||||
if (run.status === "cancelled") {
|
||||
// The abandoned-spool sweep cancelled the run and deleted its parts —
|
||||
// same contract as the part-upload route: the caller re-creates it.
|
||||
res.status(410).json({ error: "Import transfer expired — re-create it" });
|
||||
return null;
|
||||
}
|
||||
const manifest = storedImportTransferManifest(run);
|
||||
const missingParts = await importTransferMissingParts(run, manifest);
|
||||
if (missingParts.length > 0) {
|
||||
|
|
@ -853,33 +885,82 @@ export function companyRoutes(db: Db, storage?: StorageService, options?: Compan
|
|||
error: "Import transfer is missing parts",
|
||||
missingParts,
|
||||
});
|
||||
return;
|
||||
return null;
|
||||
}
|
||||
// Atomic claim: the import pipeline is not idempotent, so of any
|
||||
// overlapping applies of this transfer exactly one may run it. The claim
|
||||
// is a single guarded status flip to "applying"; complete() or the
|
||||
// guarded releaseApplyClaim() below settle it (a released run stays
|
||||
// retryable). Losers get 409.
|
||||
if (!(await companyTransferRunService.claimApply(db, run.id))) {
|
||||
if (options.claimApply && !(await companyTransferRunService.claimApply(db, run.id))) {
|
||||
throw conflict("Import transfer apply is already in progress");
|
||||
}
|
||||
try {
|
||||
const zipBytes = await assembleImportTransferZip(importTransferSpoolRoot, run.id, manifest.parts.length);
|
||||
const zipSha256 = createHash("sha256").update(zipBytes).digest("hex");
|
||||
if (zipBytes.length !== manifest.totalBytes || zipSha256 !== manifest.zipSha256) {
|
||||
// Fail closed: every part verified individually but the whole does not
|
||||
// match the declaration. The spool is deleted so a resume re-uploads
|
||||
// every part instead of re-assembling the same corrupt bytes.
|
||||
await companyTransferRunService.fail(db, run.id, "Assembled import package failed whole-file verification");
|
||||
await removeImportTransferSpool(importTransferSpoolRoot, run.id);
|
||||
throw unprocessable("Assembled import package failed verification; upload the transfer again");
|
||||
}
|
||||
const archive = await readImportZipArchive(zipBytes);
|
||||
const meta = importTransferApplyMeta(req.body);
|
||||
const rawImportBody = {
|
||||
...meta,
|
||||
source: { type: "inline", rootPath: archive.rootPath, files: archive.files },
|
||||
return {
|
||||
run,
|
||||
rawBody: {
|
||||
...importTransferApplyMeta(req.body),
|
||||
source: { type: "inline", rootPath: archive.rootPath, files: archive.files },
|
||||
},
|
||||
};
|
||||
} catch (error) {
|
||||
// Whatever escapes here (a malformed zip, the verification failure
|
||||
// above) must not leave a claimed run parked in "applying": release
|
||||
// the claim, which fails the run and keeps it retryable. The release
|
||||
// is guarded on "applying", so the verification path that already
|
||||
// failed the run keeps its more specific terminal state.
|
||||
if (options.claimApply) {
|
||||
await companyTransferRunService.releaseApplyClaim(db, run.id, errorMessage(error));
|
||||
throw error;
|
||||
}
|
||||
// Unclaimed (preview) path: assembly read the spool without freezing
|
||||
// it, so a concurrent apply may have completed — or the sweep cancelled
|
||||
// the run — and deleted the part files mid-assembly. Re-read the ledger
|
||||
// and surface those races as the same clean signals the status gate
|
||||
// above sends; only a run that is genuinely still open rethrows as a
|
||||
// real assembly failure.
|
||||
const current = await companyTransferRunService.getRun(db, run.id);
|
||||
if (current?.status === "completed") {
|
||||
throw conflict("Import transfer has already been applied");
|
||||
}
|
||||
if (current?.status === "applying") {
|
||||
throw conflict("Import transfer apply is in progress; retry after it settles");
|
||||
}
|
||||
if (current?.status === "cancelled") {
|
||||
res.status(410).json({ error: "Import transfer expired — re-create it" });
|
||||
return null;
|
||||
}
|
||||
throw error;
|
||||
}
|
||||
}
|
||||
|
||||
// Run the import preview against the assembled spool without consuming the
|
||||
// transfer: the ledger run stays open and the parts stay spooled, so the
|
||||
// subsequent apply reuses them instead of re-uploading. The body carries the
|
||||
// same meta fields the multipart preview route's `meta` field does.
|
||||
router.post(`${COMPANY_IMPORT_TRANSFERS_ROUTE_PATH}/:transferId/preview`, async (req, res) => {
|
||||
assertBoard(req);
|
||||
const resolved = await resolveImportTransferBody(req, res);
|
||||
if (!resolved) return;
|
||||
const body = companyPortabilityPreviewSchema.parse(resolved.rawBody);
|
||||
assertImportTargetAccess(req, body.target);
|
||||
const preview = await portability.previewImport(body);
|
||||
res.json(preview);
|
||||
});
|
||||
|
||||
// Assemble the spooled parts back into the original zip, verify it against
|
||||
// the declared whole-file hash, and run it through the exact import path a
|
||||
// single-shot zip upload takes. The body carries the same meta fields the
|
||||
// multipart route's `meta` field does (include/target/collisionStrategy/...).
|
||||
router.post(`${COMPANY_IMPORT_TRANSFERS_ROUTE_PATH}/:transferId/apply`, async (req, res) => {
|
||||
assertBoard(req);
|
||||
const resolved = await resolveImportTransferBody(req, res, { claimApply: true });
|
||||
if (!resolved) return;
|
||||
const { run, rawBody: rawImportBody } = resolved;
|
||||
try {
|
||||
await executeImportRequest(req, res, rawImportBody, {
|
||||
onSuccess: async (result) => {
|
||||
// Settle the run the moment the import is committed: every later
|
||||
|
|
@ -944,12 +1025,11 @@ export function companyRoutes(db: Db, storage?: StorageService, options?: Compan
|
|||
},
|
||||
});
|
||||
} catch (error) {
|
||||
// Whatever escapes here (bad meta, malformed zip, an import error
|
||||
// rethrown after onFailure) must not leave the run parked in
|
||||
// "applying": release the claim, which keeps the run retryable. The
|
||||
// release is guarded on "applying", so a run that already settled —
|
||||
// failed by onFailure, or completed before a late error — keeps its
|
||||
// terminal state.
|
||||
// Whatever escapes here (bad meta, an import error rethrown after
|
||||
// onFailure) must not leave the run parked in "applying": release the
|
||||
// claim, which keeps the run retryable. The release is guarded on
|
||||
// "applying", so a run that already settled — failed by onFailure, or
|
||||
// completed before a late error — keeps its terminal state.
|
||||
await companyTransferRunService.releaseApplyClaim(db, run.id, errorMessage(error));
|
||||
throw error;
|
||||
}
|
||||
|
|
|
|||
|
|
@ -6007,6 +6007,34 @@ registry.registerPath({
|
|||
responses: { 200: r.ok(), 401: r.unauthorized, 404: r.notFound },
|
||||
});
|
||||
|
||||
registry.registerPath({
|
||||
method: "post",
|
||||
path: `${COMPANY_IMPORT_TRANSFERS_API_PATH}/{transferId}/preview`,
|
||||
tags: ["companies"],
|
||||
summary: "Preview a completed company import transfer",
|
||||
description:
|
||||
"Runs the import preview against the assembled spool without consuming the transfer: the " +
|
||||
"ledger run stays open and the parts stay spooled, so the subsequent apply reuses them " +
|
||||
"instead of re-uploading. The JSON body carries the same fields as the multipart preview " +
|
||||
"route's `meta` field (include, target, collisionStrategy, ...).",
|
||||
request: {
|
||||
params: z.object({ transferId: z.string() }),
|
||||
body: jsonBody(companyPortabilityPreviewSchema.omit({ source: true })),
|
||||
},
|
||||
responses: {
|
||||
200: r.ok(),
|
||||
400: r.badRequest,
|
||||
401: r.unauthorized,
|
||||
404: r.notFound,
|
||||
409: {
|
||||
description:
|
||||
"Parts are still missing, an apply is in progress, or the transfer was already applied",
|
||||
},
|
||||
410: { description: "The transfer expired and its spooled parts were deleted" },
|
||||
422: r.unprocessable,
|
||||
},
|
||||
});
|
||||
|
||||
registry.registerPath({
|
||||
method: "post",
|
||||
path: `${COMPANY_IMPORT_TRANSFERS_API_PATH}/{transferId}/apply`,
|
||||
|
|
@ -6034,6 +6062,7 @@ registry.registerPath({
|
|||
description:
|
||||
"Parts are still missing, an apply is already in progress, or the transfer was already applied",
|
||||
},
|
||||
410: { description: "The transfer expired and its spooled parts were deleted" },
|
||||
422: r.unprocessable,
|
||||
},
|
||||
});
|
||||
|
|
|
|||
|
|
@ -165,6 +165,14 @@ export const api = {
|
|||
}),
|
||||
put: <T>(path: string, body: unknown, options?: RequestOptions) =>
|
||||
request<T>(path, { method: "PUT", body: JSON.stringify(body), signal: options?.signal }),
|
||||
/** Raw binary upload (e.g. one chunked import-transfer part); the body travels as-is. */
|
||||
putRaw: <T>(path: string, body: Blob, options?: RequestOptions) =>
|
||||
request<T>(path, {
|
||||
method: "PUT",
|
||||
body,
|
||||
signal: options?.signal,
|
||||
headers: { "Content-Type": "application/octet-stream", ...(options?.headers ?? {}) },
|
||||
}),
|
||||
patch: <T>(path: string, body: unknown, options?: RequestOptions) =>
|
||||
request<T>(path, { method: "PATCH", body: JSON.stringify(body), signal: options?.signal }),
|
||||
delete: <T>(path: string, bodyOrOptions?: unknown, options?: RequestOptions) => {
|
||||
|
|
|
|||
|
|
@ -10,6 +10,17 @@ import type {
|
|||
UpdateCompanyBranding,
|
||||
} from "@paperclipai/shared";
|
||||
import type { ExportFidelityReport } from "@paperclipai/shared/portability-fidelity";
|
||||
import {
|
||||
companyImportTransferApplyPath,
|
||||
companyImportTransferPartPath,
|
||||
companyImportTransferPath,
|
||||
companyImportTransferPreviewPath,
|
||||
COMPANY_IMPORT_TRANSFERS_ROUTE_PATH,
|
||||
type CompanyImportTransferCreated,
|
||||
type CompanyImportTransferDeclaration,
|
||||
type CompanyImportTransferPartUploadResult,
|
||||
type CompanyImportTransferStatus,
|
||||
} from "@paperclipai/shared/company-import-transfer";
|
||||
import { api } from "./client";
|
||||
|
||||
export type CompanyStats = Record<string, { agentCount: number; issueCount: number }>;
|
||||
|
|
@ -126,4 +137,29 @@ export const companiesApi = {
|
|||
api.postForm<CompanyImportJobAccepted>("/companies/import?async=1", importPackageForm(file, meta)),
|
||||
getImportJob: (jobId: string) =>
|
||||
api.get<CompanyImportJobStatus>(`/companies/import/jobs/${encodeURIComponent(jobId)}`),
|
||||
// Chunked resumable transfer for large local .zip packages: declare the
|
||||
// sliced zip (content-addressed, so re-declaring the same file resumes the
|
||||
// prior transfer with its uploaded parts intact), upload the missing parts,
|
||||
// then preview/apply against the server-side assembled spool. Shapes and
|
||||
// paths come from the shared transfer contract.
|
||||
importTransferCreate: (manifest: CompanyImportTransferDeclaration) =>
|
||||
api.post<CompanyImportTransferCreated>(`/companies${COMPANY_IMPORT_TRANSFERS_ROUTE_PATH}`, manifest),
|
||||
importTransferUploadPart: (transferId: string, index: number, bytes: Blob) =>
|
||||
api.putRaw<CompanyImportTransferPartUploadResult>(
|
||||
`/companies${companyImportTransferPartPath(transferId, index)}`,
|
||||
bytes,
|
||||
),
|
||||
importTransferStatus: (transferId: string) =>
|
||||
api.get<CompanyImportTransferStatus>(`/companies${companyImportTransferPath(transferId)}`),
|
||||
importTransferPreview: (transferId: string, meta: CompanyPortabilityPreviewMeta) =>
|
||||
api.post<CompanyPortabilityPreviewResult>(
|
||||
`/companies${companyImportTransferPreviewPath(transferId)}`,
|
||||
meta,
|
||||
),
|
||||
/** Apply a fully uploaded transfer as an async job (same 202/409 contract as importBundlePackageAsync). */
|
||||
importTransferApply: (transferId: string, meta: CompanyPortabilityImportMeta) =>
|
||||
api.post<CompanyImportJobAccepted>(
|
||||
`/companies${companyImportTransferApplyPath(transferId)}?async=1`,
|
||||
meta,
|
||||
),
|
||||
};
|
||||
|
|
|
|||
|
|
@ -36,6 +36,9 @@ function fileEntryInlineBytes(entry: CompanyPortabilityFileEntry): number {
|
|||
* bytes, base64 payloads with their entry structure, the serialized file-path
|
||||
* keys (thousands of paths are real bytes), and an envelope allowance for the
|
||||
* rest of the request body.
|
||||
*
|
||||
* Mirrored by `estimateInlineImportBytes` in cli/src/commands/client/company.ts
|
||||
* (the CLI cannot import from ui/) — keep the math on both sides in sync.
|
||||
*/
|
||||
export function estimateInlineImportBytes(files: Record<string, CompanyPortabilityFileEntry>): number {
|
||||
let total = REQUEST_ENVELOPE_ALLOWANCE_BYTES;
|
||||
|
|
|
|||
|
|
@ -0,0 +1,46 @@
|
|||
// Chunked resumable upload plan for large local .zip imports. Zips over the
|
||||
// threshold are sliced into fixed byte-range parts; each part and the whole
|
||||
// file are content-addressed with sha256 so the server can verify every part
|
||||
// on arrival and a resumed transfer re-uploads only the parts it lost. The
|
||||
// declaration's wire shape is the shared transfer contract.
|
||||
|
||||
import type {
|
||||
CompanyImportTransferDeclaration,
|
||||
CompanyImportTransferDeclaredPart,
|
||||
} from "@paperclipai/shared/company-import-transfer";
|
||||
|
||||
/** Local zips larger than this take the chunked transfer path. */
|
||||
export const CHUNKED_IMPORT_THRESHOLD_BYTES = 48 * 1024 * 1024;
|
||||
/** Declared byte-range size of every part except the last. */
|
||||
export const IMPORT_TRANSFER_PART_SIZE_BYTES = 32 * 1024 * 1024;
|
||||
/** Upload attempts per part before the transfer surfaces a failure. */
|
||||
export const IMPORT_TRANSFER_PART_ATTEMPTS = 3;
|
||||
|
||||
async function sha256Hex(bytes: BufferSource): Promise<string> {
|
||||
const digest = await crypto.subtle.digest("SHA-256", bytes);
|
||||
return Array.from(new Uint8Array(digest), (byte) => byte.toString(16).padStart(2, "0")).join("");
|
||||
}
|
||||
|
||||
/**
|
||||
* Hash the zip and its byte-range parts. Works on the single ArrayBuffer the
|
||||
* caller reads from the File: the whole-file digest hashes the buffer itself
|
||||
* and each part digest hashes a view into it, so no second copy is ever held.
|
||||
*/
|
||||
export async function buildImportTransferManifest(buffer: ArrayBuffer): Promise<CompanyImportTransferDeclaration> {
|
||||
const zipSha256 = await sha256Hex(buffer);
|
||||
const parts: CompanyImportTransferDeclaredPart[] = [];
|
||||
for (let offset = 0; offset < buffer.byteLength; offset += IMPORT_TRANSFER_PART_SIZE_BYTES) {
|
||||
const byteSize = Math.min(IMPORT_TRANSFER_PART_SIZE_BYTES, buffer.byteLength - offset);
|
||||
parts.push({
|
||||
index: parts.length,
|
||||
byteSize,
|
||||
sha256: await sha256Hex(new Uint8Array(buffer, offset, byteSize)),
|
||||
});
|
||||
}
|
||||
return {
|
||||
totalBytes: buffer.byteLength,
|
||||
zipSha256,
|
||||
partSizeBytes: IMPORT_TRANSFER_PART_SIZE_BYTES,
|
||||
parts,
|
||||
};
|
||||
}
|
||||
|
|
@ -1,5 +1,6 @@
|
|||
// @vitest-environment jsdom
|
||||
|
||||
import { webcrypto } from "node:crypto";
|
||||
import type { ReactNode } from "react";
|
||||
import { createRoot, type Root } from "react-dom/client";
|
||||
import { QueryClient, QueryClientProvider } from "@tanstack/react-query";
|
||||
|
|
@ -9,12 +10,23 @@ import { ApiError } from "../api/client";
|
|||
import type { CompanyImportJobAccepted } from "../api/companies";
|
||||
import { CompanyImport } from "./CompanyImport";
|
||||
|
||||
// jsdom's crypto has no SubtleCrypto; the chunked transfer path hashes parts
|
||||
// with WebCrypto, so back the global with Node's implementation.
|
||||
if (!globalThis.crypto?.subtle) {
|
||||
vi.stubGlobal("crypto", webcrypto);
|
||||
}
|
||||
|
||||
const mockCompaniesApi = vi.hoisted(() => ({
|
||||
importPreview: vi.fn(),
|
||||
importPreviewPackage: vi.fn(),
|
||||
importBundle: vi.fn(),
|
||||
importBundleAsync: vi.fn(),
|
||||
importBundlePackageAsync: vi.fn(),
|
||||
importTransferCreate: vi.fn(),
|
||||
importTransferUploadPart: vi.fn(),
|
||||
importTransferStatus: vi.fn(),
|
||||
importTransferPreview: vi.fn(),
|
||||
importTransferApply: vi.fn(),
|
||||
getImportJob: vi.fn(),
|
||||
get: vi.fn(),
|
||||
}));
|
||||
|
|
@ -217,6 +229,31 @@ function buildAccepted(id = "job-1"): CompanyImportJobAccepted {
|
|||
return { job: { id, status: "running" }, statusUrl: `/companies/import/jobs/${id}` };
|
||||
}
|
||||
|
||||
function buildTransferCreated(missingParts: number[], alreadyCompleted = false) {
|
||||
return {
|
||||
transferId: "transfer-1",
|
||||
status: "running",
|
||||
alreadyCompleted,
|
||||
totalParts: 2,
|
||||
missingParts,
|
||||
};
|
||||
}
|
||||
|
||||
const TRANSFER_PART_SIZE = 32 * 1024 * 1024;
|
||||
|
||||
/**
|
||||
* A zip whose declared size crosses the 48 MB chunked-transfer threshold. Its
|
||||
* real content is one full part plus a 1 KB tail so the manifest slices into
|
||||
* two parts without allocating 48 MB in the test.
|
||||
*/
|
||||
function buildLargeZipFile(): File {
|
||||
const bytes = new Uint8Array(TRANSFER_PART_SIZE + 1024);
|
||||
const file = new File([bytes], "big-package.zip", { type: "application/zip" });
|
||||
Object.defineProperty(file, "size", { value: 49 * 1024 * 1024 });
|
||||
Object.defineProperty(file, "arrayBuffer", { value: async () => bytes.buffer });
|
||||
return file;
|
||||
}
|
||||
|
||||
function buildSucceededJob(id = "job-1") {
|
||||
return { job: { id, status: "succeeded" as const, importResult: buildImportResult() } };
|
||||
}
|
||||
|
|
@ -243,6 +280,10 @@ describe("CompanyImport", () => {
|
|||
// the job already finished with the full result. Individual tests override.
|
||||
mockCompaniesApi.importBundleAsync.mockResolvedValue(buildAccepted());
|
||||
mockCompaniesApi.importBundlePackageAsync.mockResolvedValue(buildAccepted());
|
||||
mockCompaniesApi.importTransferCreate.mockResolvedValue(buildTransferCreated([0, 1]));
|
||||
mockCompaniesApi.importTransferUploadPart.mockResolvedValue({ ok: true, index: 0, alreadyCompleted: false });
|
||||
mockCompaniesApi.importTransferPreview.mockResolvedValue(buildPreviewResult());
|
||||
mockCompaniesApi.importTransferApply.mockResolvedValue(buildAccepted());
|
||||
mockCompaniesApi.getImportJob.mockResolvedValue(buildSucceededJob());
|
||||
mockCompaniesApi.get.mockResolvedValue({ id: "company-2", name: "Imported Test", issuePrefix: "IMP" });
|
||||
mockSidebarPreferencesApi.updateProjectOrder.mockResolvedValue(undefined);
|
||||
|
|
@ -305,6 +346,17 @@ describe("CompanyImport", () => {
|
|||
await flushReact();
|
||||
}
|
||||
|
||||
async function chooseLocalZip(file: File) {
|
||||
await clickButton((text) => text.includes("Local zip"));
|
||||
const fileInput = container.querySelector<HTMLInputElement>('input[type="file"]');
|
||||
expect(fileInput).toBeTruthy();
|
||||
Object.defineProperty(fileInput!, "files", { value: [file], configurable: true });
|
||||
await act(async () => {
|
||||
fileInput!.dispatchEvent(new Event("change", { bubbles: true }));
|
||||
});
|
||||
await flushReact();
|
||||
}
|
||||
|
||||
async function renderPageAndImport() {
|
||||
await renderPage();
|
||||
await enterGithubUrl();
|
||||
|
|
@ -415,6 +467,151 @@ describe("CompanyImport", () => {
|
|||
expect(meta.pauseAutomations).toBe(true);
|
||||
// The bundle itself is never expanded into the request; only the raw zip travels.
|
||||
expect(meta).not.toHaveProperty("source");
|
||||
// A zip under the chunked threshold never touches the transfer machinery.
|
||||
expect(mockCompaniesApi.importTransferCreate).not.toHaveBeenCalled();
|
||||
expect(mockCompaniesApi.importTransferUploadPart).not.toHaveBeenCalled();
|
||||
});
|
||||
|
||||
it("uploads a large local .zip through the chunked transfer path for preview and import", async () => {
|
||||
mockReadZipArchive.mockResolvedValue({ rootPath: "big-package", files: previewFiles });
|
||||
mockCompaniesApi.importTransferCreate
|
||||
.mockResolvedValueOnce(buildTransferCreated([0, 1]))
|
||||
// The import's re-declaration resumes the transfer the preview uploaded.
|
||||
.mockResolvedValueOnce(buildTransferCreated([]));
|
||||
|
||||
// Hold the last part so the visible upload progress can be observed.
|
||||
let resolveLastPart!: (value: { ok: true; index: number; alreadyCompleted: boolean }) => void;
|
||||
mockCompaniesApi.importTransferUploadPart.mockImplementation((_id: string, index: number) => {
|
||||
if (index === 0) return Promise.resolve({ ok: true, index, alreadyCompleted: false });
|
||||
return new Promise((resolve) => {
|
||||
resolveLastPart = resolve;
|
||||
});
|
||||
});
|
||||
|
||||
await renderPage();
|
||||
await chooseLocalZip(buildLargeZipFile());
|
||||
await clickButton((text) => text === "Preview import");
|
||||
await vi.waitFor(() => {
|
||||
expect(mockCompaniesApi.importTransferUploadPart).toHaveBeenCalledTimes(2);
|
||||
});
|
||||
await flushReact();
|
||||
|
||||
expect(container.textContent).toContain("Uploading part 2 of 2");
|
||||
expect(container.textContent).toContain("32 MB");
|
||||
|
||||
await act(async () => {
|
||||
resolveLastPart({ ok: true, index: 1, alreadyCompleted: false });
|
||||
});
|
||||
await vi.waitFor(() => {
|
||||
expect(mockCompaniesApi.importTransferPreview).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
await settle();
|
||||
|
||||
// The declaration describes the sliced zip: whole-file hash plus one
|
||||
// content hash per 32 MB byte range.
|
||||
const manifest = mockCompaniesApi.importTransferCreate.mock.calls[0]![0] as {
|
||||
totalBytes: number;
|
||||
zipSha256: string;
|
||||
partSizeBytes: number;
|
||||
parts: Array<{ index: number; byteSize: number; sha256: string }>;
|
||||
};
|
||||
expect(manifest.totalBytes).toBe(TRANSFER_PART_SIZE + 1024);
|
||||
expect(manifest.partSizeBytes).toBe(TRANSFER_PART_SIZE);
|
||||
expect(manifest.zipSha256).toMatch(/^[0-9a-f]{64}$/);
|
||||
expect(manifest.parts.map((part) => ({ index: part.index, byteSize: part.byteSize }))).toEqual([
|
||||
{ index: 0, byteSize: TRANSFER_PART_SIZE },
|
||||
{ index: 1, byteSize: 1024 },
|
||||
]);
|
||||
expect(manifest.parts.every((part) => /^[0-9a-f]{64}$/.test(part.sha256))).toBe(true);
|
||||
|
||||
// Both parts traveled, then the preview ran against the assembled spool —
|
||||
// never the single-shot multipart endpoints.
|
||||
expect(mockCompaniesApi.importTransferUploadPart).toHaveBeenNthCalledWith(1, "transfer-1", 0, expect.anything());
|
||||
expect(mockCompaniesApi.importTransferUploadPart).toHaveBeenNthCalledWith(2, "transfer-1", 1, expect.anything());
|
||||
expect(mockCompaniesApi.importTransferPreview).toHaveBeenCalledWith(
|
||||
"transfer-1",
|
||||
expect.objectContaining({ collisionStrategy: "rename", target: { mode: "new_company", newCompanyName: null } }),
|
||||
);
|
||||
expect(mockCompaniesApi.importPreviewPackage).not.toHaveBeenCalled();
|
||||
expect(container.textContent).toContain("Import preview");
|
||||
|
||||
await clickButton((text) => text.startsWith("Import 3 file"));
|
||||
await vi.waitFor(() => {
|
||||
expect(mockCompaniesApi.importTransferApply).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
await settle();
|
||||
|
||||
// The resumed transfer had nothing left to upload, so the apply reused
|
||||
// the spooled parts and ran as the usual async job.
|
||||
expect(mockCompaniesApi.importTransferCreate).toHaveBeenCalledTimes(2);
|
||||
expect(mockCompaniesApi.importTransferUploadPart).toHaveBeenCalledTimes(2);
|
||||
expect(mockCompaniesApi.importTransferApply).toHaveBeenCalledWith(
|
||||
"transfer-1",
|
||||
expect.objectContaining({ pauseAutomations: true }),
|
||||
);
|
||||
expect(mockCompaniesApi.importBundlePackageAsync).not.toHaveBeenCalled();
|
||||
expect(mockCompaniesApi.getImportJob).toHaveBeenCalledWith("job-1");
|
||||
expect(container.textContent).toContain("Import complete");
|
||||
});
|
||||
|
||||
it("retries a failed part before surfacing the error panel, keeping the transfer resumable", async () => {
|
||||
mockReadZipArchive.mockResolvedValue({ rootPath: "big-package", files: previewFiles });
|
||||
mockCompaniesApi.importTransferCreate.mockResolvedValue(buildTransferCreated([0, 1]));
|
||||
mockCompaniesApi.importTransferUploadPart.mockImplementation((_id: string, index: number) => {
|
||||
if (index === 0) return Promise.resolve({ ok: true, index, alreadyCompleted: false });
|
||||
return Promise.reject(new Error("socket hang up"));
|
||||
});
|
||||
|
||||
await renderPage();
|
||||
await chooseLocalZip(buildLargeZipFile());
|
||||
await clickButton((text) => text === "Preview import");
|
||||
await vi.waitFor(() => {
|
||||
expect(container.textContent).toContain("Preview failed: socket hang up");
|
||||
});
|
||||
|
||||
// Part 1 was attempted three times before the failure surfaced; the
|
||||
// preview never ran and no state was consumed, so retrying can resume.
|
||||
const partOneAttempts = mockCompaniesApi.importTransferUploadPart.mock.calls.filter(
|
||||
(call) => call[1] === 1,
|
||||
);
|
||||
expect(partOneAttempts).toHaveLength(3);
|
||||
expect(mockCompaniesApi.importTransferPreview).not.toHaveBeenCalled();
|
||||
expect(findButton((text) => text === "Preview import")?.disabled).toBe(false);
|
||||
|
||||
// Retry: the re-declared transfer reports only the lost part missing, so
|
||||
// just that part travels again.
|
||||
mockCompaniesApi.importTransferCreate.mockResolvedValue(buildTransferCreated([1]));
|
||||
mockCompaniesApi.importTransferUploadPart.mockResolvedValue({ ok: true, index: 1, alreadyCompleted: false });
|
||||
mockCompaniesApi.importTransferUploadPart.mockClear();
|
||||
|
||||
await clickButton((text) => text === "Preview import");
|
||||
await vi.waitFor(() => {
|
||||
expect(mockCompaniesApi.importTransferPreview).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
await settle();
|
||||
|
||||
expect(mockCompaniesApi.importTransferUploadPart).toHaveBeenCalledTimes(1);
|
||||
expect(mockCompaniesApi.importTransferUploadPart).toHaveBeenCalledWith("transfer-1", 1, expect.anything());
|
||||
expect(container.textContent).toContain("Import preview");
|
||||
});
|
||||
|
||||
it("re-uploads only the missing parts when resuming an interrupted transfer", async () => {
|
||||
// A previous page load already uploaded part 0; re-declaring the same
|
||||
// file resumes that transfer, so only part 1 travels.
|
||||
mockReadZipArchive.mockResolvedValue({ rootPath: "big-package", files: previewFiles });
|
||||
mockCompaniesApi.importTransferCreate.mockResolvedValue(buildTransferCreated([1]));
|
||||
|
||||
await renderPage();
|
||||
await chooseLocalZip(buildLargeZipFile());
|
||||
await clickButton((text) => text === "Preview import");
|
||||
await vi.waitFor(() => {
|
||||
expect(mockCompaniesApi.importTransferPreview).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
await settle();
|
||||
|
||||
expect(mockCompaniesApi.importTransferUploadPart).toHaveBeenCalledTimes(1);
|
||||
expect(mockCompaniesApi.importTransferUploadPart).toHaveBeenCalledWith("transfer-1", 1, expect.anything());
|
||||
expect(container.textContent).toContain("Import preview");
|
||||
});
|
||||
|
||||
it("explains the disabled preview button until a package is chosen", async () => {
|
||||
|
|
|
|||
|
|
@ -52,6 +52,12 @@ import {
|
|||
} from "../components/FileTree";
|
||||
import { readZipArchive } from "../lib/zip";
|
||||
import { formatMegabytes } from "../lib/import-preflight";
|
||||
import type { CompanyImportTransferDeclaration } from "@paperclipai/shared/company-import-transfer";
|
||||
import {
|
||||
CHUNKED_IMPORT_THRESHOLD_BYTES,
|
||||
IMPORT_TRANSFER_PART_ATTEMPTS,
|
||||
buildImportTransferManifest,
|
||||
} from "../lib/import-transfer";
|
||||
import { getPortableFileDataUrl, getPortableFileText, isPortableImageFile } from "../lib/portable-files";
|
||||
import {
|
||||
clearStoredImportJob,
|
||||
|
|
@ -706,6 +712,33 @@ async function readLocalPackageZip(file: File): Promise<{
|
|||
};
|
||||
}
|
||||
|
||||
// ── Chunked transfer flow for large local zips ───────────────────────
|
||||
//
|
||||
// A local .zip over the threshold is not uploaded in one request: one dropped
|
||||
// connection would restart the whole multi-minute upload. Instead the file is
|
||||
// declared as a chunked transfer (whole-file and per-part sha256), the parts
|
||||
// are uploaded individually with per-part retries, and preview/apply run
|
||||
// server-side against the assembled spool. Re-declaring the same file — after
|
||||
// a failure, a refresh, or between preview and import — resumes the prior
|
||||
// transfer, so only the parts the server is missing are ever re-uploaded.
|
||||
|
||||
function usesChunkedTransfer(file: File): boolean {
|
||||
return file.size > CHUNKED_IMPORT_THRESHOLD_BYTES;
|
||||
}
|
||||
|
||||
/** Parts done / bytes uploaded, rendered inside the pending panels while parts upload. */
|
||||
interface ImportTransferProgress {
|
||||
uploadedParts: number;
|
||||
totalParts: number;
|
||||
uploadedBytes: number;
|
||||
totalBytes: number;
|
||||
}
|
||||
|
||||
function formatTransferProgress(progress: ImportTransferProgress): string {
|
||||
const currentPart = Math.min(progress.uploadedParts + 1, progress.totalParts);
|
||||
return `Uploading part ${currentPart} of ${progress.totalParts} — ${formatMegabytes(progress.uploadedBytes)} of ${formatMegabytes(progress.totalBytes)} uploaded.`;
|
||||
}
|
||||
|
||||
// ── Async import job flow ─────────────────────────────────────────────
|
||||
//
|
||||
// Imports run as server-side jobs: the submit returns 202 with a job id and
|
||||
|
|
@ -879,6 +912,87 @@ export function CompanyImport() {
|
|||
const [resumedWatchJobId, setResumedWatchJobId] = useState<string | null>(null);
|
||||
const resumeAttemptedRef = useRef(false);
|
||||
|
||||
// Chunked transfer state. The manifest cache is keyed by File identity so
|
||||
// preview and import hash the (large) package once; progress is set only
|
||||
// while parts are actually uploading, so the pending panels can report it.
|
||||
const transferManifestRef = useRef<{ file: File; manifest: CompanyImportTransferDeclaration } | null>(null);
|
||||
const [transferProgress, setTransferProgress] = useState<ImportTransferProgress | null>(null);
|
||||
|
||||
async function ensureTransferManifest(file: File): Promise<CompanyImportTransferDeclaration> {
|
||||
if (transferManifestRef.current?.file === file) {
|
||||
return transferManifestRef.current.manifest;
|
||||
}
|
||||
// The whole file is read once here for hashing; parts are later uploaded
|
||||
// as Blob slices of the File, so this buffer is not retained past hashing.
|
||||
const manifest = await buildImportTransferManifest(await file.arrayBuffer());
|
||||
transferManifestRef.current = { file, manifest };
|
||||
return manifest;
|
||||
}
|
||||
|
||||
/**
|
||||
* Declare (or resume) the transfer for this file and upload every part the
|
||||
* server reports missing, sequentially with per-part retries. Resolves with
|
||||
* the transfer id once the server holds every part.
|
||||
*/
|
||||
async function uploadImportTransfer(file: File): Promise<string> {
|
||||
const manifest = await ensureTransferManifest(file);
|
||||
const created = await companiesApi.importTransferCreate(manifest);
|
||||
if (created.alreadyCompleted) {
|
||||
// The server keys transfers by content, and this exact zip already
|
||||
// finished an apply — its parts are gone, so it cannot be re-run.
|
||||
throw new Error(
|
||||
"This exact package was already imported by a completed transfer. Re-export the package to import it again.",
|
||||
);
|
||||
}
|
||||
const missing = new Set(created.missingParts);
|
||||
let uploadedParts = manifest.parts.length - missing.size;
|
||||
let uploadedBytes = manifest.parts.reduce(
|
||||
(sum, part) => (missing.has(part.index) ? sum : sum + part.byteSize),
|
||||
0,
|
||||
);
|
||||
try {
|
||||
setTransferProgress({
|
||||
uploadedParts,
|
||||
totalParts: manifest.parts.length,
|
||||
uploadedBytes,
|
||||
totalBytes: manifest.totalBytes,
|
||||
});
|
||||
for (const part of manifest.parts) {
|
||||
if (!missing.has(part.index)) continue;
|
||||
const offset = part.index * manifest.partSizeBytes;
|
||||
const bytes = file.slice(offset, offset + part.byteSize);
|
||||
let lastError: unknown = null;
|
||||
let uploaded = false;
|
||||
for (let attempt = 0; attempt < IMPORT_TRANSFER_PART_ATTEMPTS && !uploaded; attempt += 1) {
|
||||
try {
|
||||
await companiesApi.importTransferUploadPart(created.transferId, part.index, bytes);
|
||||
uploaded = true;
|
||||
} catch (err) {
|
||||
lastError = err;
|
||||
}
|
||||
}
|
||||
if (!uploaded) {
|
||||
// The parts already uploaded stay spooled server-side; retrying the
|
||||
// preview/import resumes from them instead of starting over.
|
||||
throw lastError instanceof Error
|
||||
? lastError
|
||||
: new Error(`Part ${part.index + 1} of ${manifest.parts.length} failed to upload.`);
|
||||
}
|
||||
uploadedParts += 1;
|
||||
uploadedBytes += part.byteSize;
|
||||
setTransferProgress({
|
||||
uploadedParts,
|
||||
totalParts: manifest.parts.length,
|
||||
uploadedBytes,
|
||||
totalBytes: manifest.totalBytes,
|
||||
});
|
||||
}
|
||||
} finally {
|
||||
setTransferProgress(null);
|
||||
}
|
||||
return created.transferId;
|
||||
}
|
||||
|
||||
// Fetch current company agents to find CEO adapter type
|
||||
const { data: companyAgents } = useQuery({
|
||||
queryKey: selectedCompanyId ? queryKeys.agents.list(selectedCompanyId) : ["agents", "none"],
|
||||
|
|
@ -950,10 +1064,16 @@ export function CompanyImport() {
|
|||
|
||||
// Preview mutation
|
||||
const previewMutation = useMutation({
|
||||
mutationFn: (_generation: number) => {
|
||||
mutationFn: async (_generation: number) => {
|
||||
const meta = buildImportMetaCommon();
|
||||
if (sourceMode === "local") {
|
||||
if (!localPackage) throw new Error("No source configured.");
|
||||
if (usesChunkedTransfer(localPackage.file)) {
|
||||
// Too large for one request: upload (or resume) the chunked
|
||||
// transfer, then preview against the server-side assembled spool.
|
||||
const transferId = await uploadImportTransfer(localPackage.file);
|
||||
return companiesApi.importTransferPreview(transferId, meta);
|
||||
}
|
||||
// Upload the raw compressed zip; the server unzips it into the same
|
||||
// inline bundle the importer consumes.
|
||||
return companiesApi.importPreviewPackage(localPackage.file, meta);
|
||||
|
|
@ -1091,9 +1211,16 @@ export function CompanyImport() {
|
|||
const storageKey = currentImportJobStorageKey();
|
||||
let accepted: CompanyImportJobAccepted;
|
||||
try {
|
||||
accepted = localFile
|
||||
? await companiesApi.importBundlePackageAsync(localFile, meta)
|
||||
: await companiesApi.importBundleAsync({ source: githubSource!, ...meta });
|
||||
if (localFile && usesChunkedTransfer(localFile)) {
|
||||
// Same transfer the preview uploaded: re-declaring resumes it, so
|
||||
// normally no parts travel again and this goes straight to apply.
|
||||
const transferId = await uploadImportTransfer(localFile);
|
||||
accepted = await companiesApi.importTransferApply(transferId, meta);
|
||||
} else {
|
||||
accepted = localFile
|
||||
? await companiesApi.importBundlePackageAsync(localFile, meta)
|
||||
: await companiesApi.importBundleAsync({ source: githubSource!, ...meta });
|
||||
}
|
||||
} catch (err) {
|
||||
// 409: this user's previous import is still running. Adopt that job
|
||||
// and watch it — never fire a second import.
|
||||
|
|
@ -1817,9 +1944,11 @@ export function CompanyImport() {
|
|||
<div className="mt-3 flex items-start gap-2 rounded-md border border-border bg-muted/30 px-3 py-2.5">
|
||||
<Loader2 className="mt-0.5 h-3.5 w-3.5 shrink-0 animate-spin text-muted-foreground" />
|
||||
<p className="text-xs text-muted-foreground">
|
||||
Uploading and analyzing your package
|
||||
{localCompressedBytes !== null ? ` (${formatMegabytes(localCompressedBytes)} zip)` : ""} — large
|
||||
packages can take a few minutes. Keep this page open.
|
||||
{transferProgress
|
||||
? `${formatTransferProgress(transferProgress)} An interrupted upload resumes from the finished parts.`
|
||||
: `Uploading and analyzing your package${
|
||||
localCompressedBytes !== null ? ` (${formatMegabytes(localCompressedBytes)} zip)` : ""
|
||||
} — large packages can take a few minutes. Keep this page open.`}
|
||||
</p>
|
||||
</div>
|
||||
)}
|
||||
|
|
@ -1914,8 +2043,9 @@ export function CompanyImport() {
|
|||
<div className="mx-5 mt-3 flex items-start gap-2 rounded-md border border-border bg-muted/30 px-3 py-2.5">
|
||||
<Loader2 className="mt-0.5 h-3.5 w-3.5 shrink-0 animate-spin text-muted-foreground" />
|
||||
<p className="text-xs text-muted-foreground">
|
||||
Import running on the server — safe to keep waiting; reconnecting won't lose it.
|
||||
Large packages can take several minutes.
|
||||
{transferProgress
|
||||
? `${formatTransferProgress(transferProgress)} An interrupted upload resumes from the finished parts.`
|
||||
: "Import running on the server — safe to keep waiting; reconnecting won't lose it. Large packages can take several minutes."}
|
||||
</p>
|
||||
</div>
|
||||
)}
|
||||
|
|
|
|||
Loading…
Reference in New Issue