feat(server): chunked resumable company import transfers (#11223)
## Thinking Path > - Paperclip is the open source app people use to manage AI agents for work > - Company import moves large packages into an instance, and since the upload cap rose to 1 GB, the transport is the weak point: one HTTP request, buffered fully in memory, with no resume > - A dropped connection at 90% of an 800 MB upload starts the whole transfer over, and a server restart loses all progress > - This pull request adds the server side of chunked resumable import transfers: a durable run ledger and routes that accept the same import zip as verified ~32 MB parts spooled to disk > - An interrupted transfer resumes from the parts already uploaded — across dropped connections, page refreshes, and server restarts — and peak upload memory drops from the whole package to one part > - The benefit is that large imports become reliable on real-world connections instead of all-or-nothing ## Linked Issues or Issue Description **What happened?** Large company imports travel as a single HTTP upload. On a slow or flaky connection, any interruption discards all progress and the upload restarts from zero. The server buffers the entire compressed package in memory during upload. A server restart mid-upload loses the transfer entirely. With the upload cap now at 1 GB, these failure modes govern exactly the imports the cap was raised for. **Expected behavior** A large import upload survives interruptions: already-transferred data is kept and verified, only the missing remainder is re-sent, and the server's memory use during upload is bounded by a part, not the package. **Steps to reproduce** 1. Import a multi-hundred-MB company package over a connection that drops mid-upload. 2. The upload fails; retrying starts from byte zero. 3. Repeat on an unstable connection and the import may never complete. ## What Changed - New `company_transfer_runs` table (drizzle schema + migration) and `companyTransferRunService`: one row per transfer with a content-derived idempotency key, per-part completion recorded atomically and idempotently, resume scoped to actor and direction, completed runs short-circuiting retries of identical content. - New transfer routes beside the existing import routes, same authorization: declare a sliced zip (`POST /import/transfers` — validates cap, 64 MB part ceiling, contiguity, size sums, sha256 format), upload parts (`PUT .../parts/:n` — raw body, hash-and-size verified before an atomic write to a disk spool under the instance root; re-uploads are no-op successes), poll resume state (`GET .../:id` — missing parts recomputed from disk), and apply (`POST .../:id/apply` — requires all parts, re-verifies the assembled zip against the whole-file hash fail-closed, then feeds the existing import pipeline through factored helpers rather than duplicated logic). - Hourly sweep fails and cleans spools idle for 24 h; a swept transfer honestly reports all parts missing on resume. - Strict UUID gating on run ids before any filesystem path construction. - The existing single-shot upload path is untouched; clients arrive in the follow-up PR. ## Verification - Transfer route suite (embedded Postgres): create/upload/status/apply round-trip with a real imported company, out-of-order parts, wrong-hash part rejected and unrecorded, re-upload no-op, apply-with-missing-parts rejection, resume after failure with prior progress intact, assembled-hash mismatch failing closed with spool deletion, actor scoping 404s, async-job apply, sweep followed by honest resume. - Ledger suite (embedded Postgres): part idempotency, actor/direction scoping, completed-run short-circuit, cancelled runs staying cancelled. - Existing portability route suite unchanged and green; server + db typechecks clean. Exact counts in the PR checks. ## Risks - New routes are additive; the existing import path is untouched. The transfer routes carry the same board authorization as the import routes they sit beside. - Disk spool: bounded by the existing upload cap per transfer, cleaned on success, failure, hash mismatch, and by the 24 h sweep. Spool paths are strict-UUID-gated. - The apply step still materializes the assembled zip in memory once (same profile as today's single-shot import at apply time); upload-time memory drops to one part. - Known limitation, deliberate: transfers are keyed on content alone, so identical package content cannot be imported twice without re-exporting (surfaced explicitly to the caller). Acceptable for v1; noted for review. ## 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
2494a2a0fe
commit
23a1b025c2
|
|
@ -0,0 +1,24 @@
|
|||
CREATE TABLE "company_transfer_runs" (
|
||||
"id" uuid PRIMARY KEY DEFAULT gen_random_uuid() NOT NULL,
|
||||
"company_id" uuid,
|
||||
"direction" text NOT NULL,
|
||||
"status" text DEFAULT 'pending' NOT NULL,
|
||||
"actor_key" text NOT NULL,
|
||||
"container_ref" jsonb NOT NULL,
|
||||
"idempotency_key" text NOT NULL,
|
||||
"manifest_sha256" text,
|
||||
"manifest" jsonb,
|
||||
"chunk_count" integer DEFAULT 0 NOT NULL,
|
||||
"blob_count" integer DEFAULT 0 NOT NULL,
|
||||
"completed_parts" jsonb DEFAULT '[]'::jsonb NOT NULL,
|
||||
"error" text,
|
||||
"started_at" timestamp with time zone,
|
||||
"finished_at" timestamp with time zone,
|
||||
"created_at" timestamp with time zone DEFAULT now() NOT NULL,
|
||||
"updated_at" timestamp with time zone DEFAULT now() NOT NULL
|
||||
);
|
||||
--> statement-breakpoint
|
||||
ALTER TABLE "company_transfer_runs" ADD CONSTRAINT "company_transfer_runs_company_id_companies_id_fk" FOREIGN KEY ("company_id") REFERENCES "public"."companies"("id") ON DELETE set null ON UPDATE no action;--> statement-breakpoint
|
||||
CREATE INDEX "company_transfer_runs_company_idx" ON "company_transfer_runs" USING btree ("company_id");--> statement-breakpoint
|
||||
CREATE INDEX "company_transfer_runs_idempotency_direction_idx" ON "company_transfer_runs" USING btree ("idempotency_key","direction");--> statement-breakpoint
|
||||
CREATE INDEX "company_transfer_runs_actor_status_idx" ON "company_transfer_runs" USING btree ("actor_key","status");
|
||||
File diff suppressed because it is too large
Load Diff
|
|
@ -1478,6 +1478,13 @@
|
|||
"when": 1786388759523,
|
||||
"tag": "0212_onboarding_first_task_unique",
|
||||
"breakpoints": true
|
||||
},
|
||||
{
|
||||
"idx": 213,
|
||||
"version": "7",
|
||||
"when": 1786467951626,
|
||||
"tag": "0213_complete_mystique",
|
||||
"breakpoints": true
|
||||
}
|
||||
]
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,49 @@
|
|||
import { pgTable, uuid, text, integer, jsonb, timestamp, index } from "drizzle-orm/pg-core";
|
||||
import { companies } from "./companies.js";
|
||||
|
||||
// Durable ledger for chunked company transfers (export publishes and import
|
||||
// applies). One row per run; `completedParts` records every container part
|
||||
// (chunk or blob name) that has been fully processed and verified, so an
|
||||
// interrupted run resumes from the parts it already finished instead of
|
||||
// restarting. `idempotencyKey` is the transfer manifest's content-derived key:
|
||||
// a retried publish/apply of identical content finds its prior run.
|
||||
export const companyTransferRuns = pgTable(
|
||||
"company_transfer_runs",
|
||||
{
|
||||
id: uuid("id").primaryKey().defaultRandom(),
|
||||
// Null while an import targeting a new company has not created it yet;
|
||||
// set as soon as the destination company exists.
|
||||
companyId: uuid("company_id").references(() => companies.id, { onDelete: "set null" }),
|
||||
direction: text("direction").notNull(), // "export" | "import"
|
||||
status: text("status").notNull().default("pending"), // pending | running | applying | completed | failed | cancelled
|
||||
// The actor the run belongs to (board session / API key identity); run
|
||||
// visibility and resume are scoped to the same actor.
|
||||
actorKey: text("actor_key").notNull(),
|
||||
// Credential-free description of where the container lives (zip upload
|
||||
// reference or relay bucket/prefix). Secrets never land in this column.
|
||||
containerRef: jsonb("container_ref").notNull(),
|
||||
idempotencyKey: text("idempotency_key").notNull(),
|
||||
manifestSha256: text("manifest_sha256"),
|
||||
// The parsed transfer manifest (chunk + blob index). Persisted so resume
|
||||
// can re-verify parts without refetching manifest.json first.
|
||||
manifest: jsonb("manifest"),
|
||||
chunkCount: integer("chunk_count").notNull().default(0),
|
||||
blobCount: integer("blob_count").notNull().default(0),
|
||||
// Container part names ("chunks/0000.json.gz", "blobs/<sha256>") that
|
||||
// completed processing, in completion order.
|
||||
completedParts: jsonb("completed_parts").notNull().default([]),
|
||||
error: text("error"),
|
||||
startedAt: timestamp("started_at", { withTimezone: true }),
|
||||
finishedAt: timestamp("finished_at", { withTimezone: true }),
|
||||
createdAt: timestamp("created_at", { withTimezone: true }).notNull().defaultNow(),
|
||||
updatedAt: timestamp("updated_at", { withTimezone: true }).notNull().defaultNow(),
|
||||
},
|
||||
(table) => ({
|
||||
companyIdx: index("company_transfer_runs_company_idx").on(table.companyId),
|
||||
idempotencyDirectionIdx: index("company_transfer_runs_idempotency_direction_idx").on(
|
||||
table.idempotencyKey,
|
||||
table.direction,
|
||||
),
|
||||
actorStatusIdx: index("company_transfer_runs_actor_status_idx").on(table.actorKey, table.status),
|
||||
}),
|
||||
);
|
||||
|
|
@ -1,5 +1,6 @@
|
|||
export { companies } from "./companies.js";
|
||||
export { companyLogos } from "./company_logos.js";
|
||||
export { companyTransferRuns } from "./company_transfer_runs.js";
|
||||
export { authUsers, authSessions, authAccounts, authVerifications } from "./auth.js";
|
||||
export { instanceSettings } from "./instance_settings.js";
|
||||
export { instanceUserRoles } from "./instance_user_roles.js";
|
||||
|
|
|
|||
|
|
@ -0,0 +1,98 @@
|
|||
import { z } from "zod";
|
||||
|
||||
// Wire contract for chunked resumable company import transfers. The server
|
||||
// routes validate against these schemas and the browser and CLI clients type
|
||||
// their requests/responses against the inferred types, so all three sides
|
||||
// share one description of the payload shapes and route paths.
|
||||
|
||||
/** Transfer routes relative to the companies router (`/api/companies`). */
|
||||
export const COMPANY_IMPORT_TRANSFERS_ROUTE_PATH = "/import/transfers";
|
||||
export const COMPANY_IMPORT_TRANSFERS_API_PATH = `/api/companies${COMPANY_IMPORT_TRANSFERS_ROUTE_PATH}`;
|
||||
|
||||
/**
|
||||
* Paths for one transfer, relative to the companies API root — callers prefix
|
||||
* their own mount ("/companies" in the browser client, "/api/companies" in
|
||||
* the CLI; the server registers the same shapes as express params).
|
||||
*/
|
||||
export function companyImportTransferPath(transferId: string): string {
|
||||
return `${COMPANY_IMPORT_TRANSFERS_ROUTE_PATH}/${encodeURIComponent(transferId)}`;
|
||||
}
|
||||
|
||||
export function companyImportTransferPartPath(transferId: string, partIndex: number): string {
|
||||
return `${companyImportTransferPath(transferId)}/parts/${partIndex}`;
|
||||
}
|
||||
|
||||
export function companyImportTransferPreviewPath(transferId: string): string {
|
||||
return `${companyImportTransferPath(transferId)}/preview`;
|
||||
}
|
||||
|
||||
export function companyImportTransferApplyPath(transferId: string): string {
|
||||
return `${companyImportTransferPath(transferId)}/apply`;
|
||||
}
|
||||
|
||||
/** Hard cap on the number of declared byte-range parts per transfer. */
|
||||
export const COMPANY_IMPORT_TRANSFER_MAX_PARTS = 4096;
|
||||
|
||||
const SHA256_HEX_RE = /^[0-9a-f]{64}$/;
|
||||
|
||||
export const companyImportTransferDeclaredPartSchema = z.object({
|
||||
index: z.number().int().min(0),
|
||||
byteSize: z.number().int().min(1),
|
||||
sha256: z
|
||||
.string()
|
||||
.regex(SHA256_HEX_RE, "sha256 must be 64 lowercase hex characters")
|
||||
.describe("sha256 of this part, 64 lowercase hex characters"),
|
||||
});
|
||||
|
||||
/**
|
||||
* Declaration body for POST /api/companies/import/transfers: the caller's
|
||||
* existing .zip described as contiguous content-addressed byte-range parts.
|
||||
*/
|
||||
export const companyImportTransferDeclarationSchema = z.object({
|
||||
totalBytes: z.number().int().min(1),
|
||||
zipSha256: z
|
||||
.string()
|
||||
.regex(SHA256_HEX_RE, "sha256 must be 64 lowercase hex characters")
|
||||
.describe("sha256 of the whole .zip, 64 lowercase hex characters"),
|
||||
partSizeBytes: z.number().int().min(1),
|
||||
parts: z.array(companyImportTransferDeclaredPartSchema).min(1).max(COMPANY_IMPORT_TRANSFER_MAX_PARTS),
|
||||
});
|
||||
|
||||
export type CompanyImportTransferDeclaredPart = z.infer<typeof companyImportTransferDeclaredPartSchema>;
|
||||
export type CompanyImportTransferDeclaration = z.infer<typeof companyImportTransferDeclarationSchema>;
|
||||
|
||||
/** Ledger states a transfer run moves through (see company_transfer_runs). */
|
||||
export type CompanyImportTransferRunStatus =
|
||||
| "pending"
|
||||
| "running"
|
||||
| "applying"
|
||||
| "completed"
|
||||
| "failed"
|
||||
| "cancelled";
|
||||
|
||||
/** Response of declaring (or resuming) a chunked import transfer. */
|
||||
export interface CompanyImportTransferCreated {
|
||||
transferId: string;
|
||||
status: CompanyImportTransferRunStatus;
|
||||
/** True when this exact content already finished a prior apply. */
|
||||
alreadyCompleted: boolean;
|
||||
totalParts: number;
|
||||
missingParts: number[];
|
||||
}
|
||||
|
||||
/** Response of the resume-polling GET for one transfer. */
|
||||
export interface CompanyImportTransferStatus {
|
||||
transferId: string;
|
||||
status: CompanyImportTransferRunStatus;
|
||||
totalParts: number;
|
||||
completedParts: number;
|
||||
missingParts: number[];
|
||||
}
|
||||
|
||||
/** Response of uploading one declared part. */
|
||||
export interface CompanyImportTransferPartUploadResult {
|
||||
ok: true;
|
||||
index: number;
|
||||
/** True when the part was already spooled and verified by a prior upload. */
|
||||
alreadyCompleted: boolean;
|
||||
}
|
||||
|
|
@ -0,0 +1,803 @@
|
|||
import { createHash } from "node:crypto";
|
||||
import { promises as fs } from "node:fs";
|
||||
import os from "node:os";
|
||||
import path from "node:path";
|
||||
import express from "express";
|
||||
import request from "supertest";
|
||||
import { afterAll, beforeAll, beforeEach, describe, expect, it, vi } from "vitest";
|
||||
import { companies, createDb } from "@paperclipai/db";
|
||||
import {
|
||||
getEmbeddedPostgresTestSupport,
|
||||
startEmbeddedPostgresTestDatabase,
|
||||
} from "./helpers/embedded-postgres.js";
|
||||
import { companyTransferRunService } from "../services/company-transfer-runs.js";
|
||||
import { sweepAbandonedImportTransferSpools } from "../services/company-import-transfers.js";
|
||||
|
||||
const mockCompanyService = vi.hoisted(() => ({
|
||||
list: vi.fn(),
|
||||
stats: vi.fn(),
|
||||
getById: vi.fn(),
|
||||
create: vi.fn(),
|
||||
update: vi.fn(),
|
||||
archive: vi.fn(),
|
||||
remove: vi.fn(),
|
||||
}));
|
||||
|
||||
const mockAgentService = vi.hoisted(() => ({
|
||||
getById: vi.fn(),
|
||||
}));
|
||||
|
||||
const mockAccessService = vi.hoisted(() => ({
|
||||
ensureMembership: vi.fn(),
|
||||
}));
|
||||
|
||||
const mockBudgetService = vi.hoisted(() => ({
|
||||
upsertPolicy: vi.fn(),
|
||||
}));
|
||||
|
||||
const mockCompanyPortabilityService = vi.hoisted(() => ({
|
||||
exportBundle: vi.fn(),
|
||||
previewExport: vi.fn(),
|
||||
previewImport: vi.fn(),
|
||||
importBundle: vi.fn(),
|
||||
}));
|
||||
|
||||
const mockCompanyArtifactsService = vi.hoisted(() => ({
|
||||
list: vi.fn(),
|
||||
}));
|
||||
|
||||
const mockLogActivity = vi.hoisted(() => vi.fn());
|
||||
const mockFeedbackService = vi.hoisted(() => ({
|
||||
listIssueVotesForUser: vi.fn(),
|
||||
listFeedbackTraces: vi.fn(),
|
||||
getFeedbackTraceById: vi.fn(),
|
||||
saveIssueVote: vi.fn(),
|
||||
}));
|
||||
|
||||
vi.mock("../services/index.js", () => ({
|
||||
accessService: () => mockAccessService,
|
||||
agentService: () => mockAgentService,
|
||||
budgetService: () => mockBudgetService,
|
||||
companyArtifactsService: () => mockCompanyArtifactsService,
|
||||
companyPortabilityService: () => mockCompanyPortabilityService,
|
||||
companyService: () => mockCompanyService,
|
||||
feedbackService: () => mockFeedbackService,
|
||||
logActivity: mockLogActivity,
|
||||
}));
|
||||
|
||||
// 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.
|
||||
const spoolRemovalFailure = 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 {
|
||||
...actual,
|
||||
removeImportTransferSpool: async (spoolRoot: string, runId: string) => {
|
||||
if (spoolRemovalFailure.error) throw spoolRemovalFailure.error;
|
||||
return actual.removeImportTransferSpool(spoolRoot, runId);
|
||||
},
|
||||
};
|
||||
});
|
||||
|
||||
const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport();
|
||||
const describeEmbeddedPostgres = embeddedPostgresSupport.supported
|
||||
? describe.sequential
|
||||
: describe.skip;
|
||||
|
||||
const companyId = "11111111-1111-4111-8111-111111111111";
|
||||
const TEST_USER_HEADER = "x-test-user-id";
|
||||
|
||||
function boardActor(userId: string) {
|
||||
return {
|
||||
type: "board",
|
||||
userId,
|
||||
userName: "Board User",
|
||||
userEmail: `${userId}@example.com`,
|
||||
companyIds: [companyId],
|
||||
memberships: [{ companyId, membershipRole: "owner", status: "active" }],
|
||||
isInstanceAdmin: true,
|
||||
source: "session",
|
||||
};
|
||||
}
|
||||
|
||||
// A minimal STORE-only zip writer (same layout as the portability routes test)
|
||||
// so apply can be exercised end-to-end through the real readZipArchive path.
|
||||
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;
|
||||
}
|
||||
|
||||
function buildStoreZip(files: Record<string, string>, rootPath: string): Buffer {
|
||||
const encoder = new TextEncoder();
|
||||
const localChunks: Buffer[] = [];
|
||||
const centralChunks: Buffer[] = [];
|
||||
let localOffset = 0;
|
||||
const entries = Object.entries(files);
|
||||
|
||||
for (const [relativePath, content] of entries) {
|
||||
const fileName = encoder.encode(`${rootPath}/${relativePath}`);
|
||||
const body = Buffer.from(encoder.encode(content));
|
||||
const checksum = crc32(body);
|
||||
|
||||
const localHeader = Buffer.alloc(30 + fileName.length);
|
||||
localHeader.writeUInt32LE(0x04034b50, 0);
|
||||
localHeader.writeUInt16LE(20, 4);
|
||||
localHeader.writeUInt16LE(0x0800, 6);
|
||||
localHeader.writeUInt32LE(checksum, 14);
|
||||
localHeader.writeUInt32LE(body.length, 18);
|
||||
localHeader.writeUInt32LE(body.length, 22);
|
||||
localHeader.writeUInt16LE(fileName.length, 26);
|
||||
Buffer.from(fileName).copy(localHeader, 30);
|
||||
|
||||
const centralHeader = Buffer.alloc(46 + fileName.length);
|
||||
centralHeader.writeUInt32LE(0x02014b50, 0);
|
||||
centralHeader.writeUInt16LE(20, 4);
|
||||
centralHeader.writeUInt16LE(20, 6);
|
||||
centralHeader.writeUInt16LE(0x0800, 8);
|
||||
centralHeader.writeUInt32LE(checksum, 16);
|
||||
centralHeader.writeUInt32LE(body.length, 20);
|
||||
centralHeader.writeUInt32LE(body.length, 24);
|
||||
centralHeader.writeUInt16LE(fileName.length, 28);
|
||||
centralHeader.writeUInt32LE(localOffset, 42);
|
||||
Buffer.from(fileName).copy(centralHeader, 46);
|
||||
|
||||
localChunks.push(localHeader, body);
|
||||
centralChunks.push(centralHeader);
|
||||
localOffset += localHeader.length + body.length;
|
||||
}
|
||||
|
||||
const centralDirectory = Buffer.concat(centralChunks);
|
||||
const eocd = Buffer.alloc(22);
|
||||
eocd.writeUInt32LE(0x06054b50, 0);
|
||||
eocd.writeUInt16LE(entries.length, 8);
|
||||
eocd.writeUInt16LE(entries.length, 10);
|
||||
eocd.writeUInt32LE(centralDirectory.length, 12);
|
||||
eocd.writeUInt32LE(localOffset, 16);
|
||||
|
||||
return Buffer.concat([...localChunks, centralDirectory, eocd]);
|
||||
}
|
||||
|
||||
function sha256Hex(bytes: Buffer): string {
|
||||
return createHash("sha256").update(bytes).digest("hex");
|
||||
}
|
||||
|
||||
// Distinct zip content per call: the transfer declaration is content-addressed
|
||||
// (same content resumes the prior run), so tests that must not share runs
|
||||
// build distinct fixtures.
|
||||
let zipSeq = 0;
|
||||
function buildFixtureZip(): Buffer {
|
||||
zipSeq += 1;
|
||||
return buildStoreZip(
|
||||
{
|
||||
"COMPANY.md": `---\nname: Chunked Import ${zipSeq}\n---\n`,
|
||||
"agents/ceo/AGENTS.md": "---\nname: CEO\n---\n",
|
||||
},
|
||||
"paperclip",
|
||||
);
|
||||
}
|
||||
|
||||
function sliceIntoParts(zip: Buffer, partSizeBytes: number): Buffer[] {
|
||||
const parts: Buffer[] = [];
|
||||
for (let offset = 0; offset < zip.length; offset += partSizeBytes) {
|
||||
parts.push(zip.subarray(offset, Math.min(offset + partSizeBytes, zip.length)));
|
||||
}
|
||||
return parts;
|
||||
}
|
||||
|
||||
function declareTransfer(zip: Buffer, partSizeBytes: number) {
|
||||
const slices = sliceIntoParts(zip, partSizeBytes);
|
||||
return {
|
||||
slices,
|
||||
body: {
|
||||
totalBytes: zip.length,
|
||||
zipSha256: sha256Hex(zip),
|
||||
partSizeBytes,
|
||||
parts: slices.map((slice, index) => ({
|
||||
index,
|
||||
byteSize: slice.length,
|
||||
sha256: sha256Hex(slice),
|
||||
})),
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
const importMeta = {
|
||||
include: { company: true, agents: true, projects: false, issues: false },
|
||||
target: { mode: "existing_company", companyId },
|
||||
collisionStrategy: "rename",
|
||||
};
|
||||
|
||||
describeEmbeddedPostgres("company import transfer routes", () => {
|
||||
let tempDb: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | null = null;
|
||||
let db!: ReturnType<typeof createDb>;
|
||||
let spoolRoot!: string;
|
||||
let app!: express.Express;
|
||||
let appImportCounter = 0;
|
||||
|
||||
async function createApp() {
|
||||
appImportCounter += 1;
|
||||
const routeModulePath = `../routes/companies.js?company-import-transfer-routes-${appImportCounter}`;
|
||||
const middlewareModulePath = `../middleware/index.js?company-import-transfer-routes-${appImportCounter}`;
|
||||
const [{ companyRoutes }, { errorHandler }] = await Promise.all([
|
||||
import(routeModulePath) as Promise<typeof import("../routes/companies.js")>,
|
||||
import(middlewareModulePath) as Promise<typeof import("../middleware/index.js")>,
|
||||
]);
|
||||
const built = express();
|
||||
built.use(express.json());
|
||||
built.use((req, _res, next) => {
|
||||
const header = req.headers[TEST_USER_HEADER];
|
||||
const userId = typeof header === "string" && header.length > 0 ? header : "board-user-a";
|
||||
(req as any).actor = boardActor(userId);
|
||||
next();
|
||||
});
|
||||
built.use("/api/companies", companyRoutes(db, undefined, { importTransferSpoolRoot: spoolRoot }));
|
||||
built.use(errorHandler);
|
||||
return built;
|
||||
}
|
||||
|
||||
beforeAll(async () => {
|
||||
tempDb = await startEmbeddedPostgresTestDatabase("paperclip-import-transfer-routes-");
|
||||
db = createDb(tempDb.connectionString);
|
||||
// The mocked import result points at this company; the ledger's
|
||||
// company_id foreign key needs the row to exist.
|
||||
await db.insert(companies).values({ id: companyId, name: "Chunked Import Target" });
|
||||
spoolRoot = await fs.mkdtemp(path.join(os.tmpdir(), "paperclip-import-transfer-spool-"));
|
||||
app = await createApp();
|
||||
});
|
||||
|
||||
afterAll(async () => {
|
||||
await tempDb?.cleanup();
|
||||
await fs.rm(spoolRoot, { recursive: true, force: true });
|
||||
});
|
||||
|
||||
beforeEach(() => {
|
||||
vi.clearAllMocks();
|
||||
spoolRemovalFailure.error = null;
|
||||
mockCompanyPortabilityService.importBundle.mockResolvedValue({
|
||||
company: { id: companyId, action: "updated" },
|
||||
agents: [{ id: "agent-1" }],
|
||||
warnings: [],
|
||||
});
|
||||
});
|
||||
|
||||
function putPart(transferId: string, index: number, bytes: Buffer, userId?: string) {
|
||||
let req = request(app)
|
||||
.put(`/api/companies/import/transfers/${transferId}/parts/${index}`)
|
||||
.set("content-type", "application/octet-stream");
|
||||
if (userId) req = req.set(TEST_USER_HEADER, userId);
|
||||
return req.send(bytes);
|
||||
}
|
||||
|
||||
it("uploads parts out of order, tracks progress, and applies through the import pipeline", async () => {
|
||||
const zip = buildFixtureZip();
|
||||
const { body, slices } = declareTransfer(zip, Math.ceil(zip.length / 3));
|
||||
expect(slices.length).toBe(3);
|
||||
|
||||
const created = await request(app).post("/api/companies/import/transfers").send(body);
|
||||
expect(created.status).toBe(200);
|
||||
const transferId = created.body.transferId as string;
|
||||
expect(transferId).toMatch(/^[0-9a-f-]{36}$/);
|
||||
expect(created.body.totalParts).toBe(3);
|
||||
expect(created.body.missingParts).toEqual([0, 1, 2]);
|
||||
expect(created.body.alreadyCompleted).toBe(false);
|
||||
|
||||
// Out of order: last part first.
|
||||
expect((await putPart(transferId, 2, slices[2]!)).status).toBe(200);
|
||||
expect((await putPart(transferId, 0, slices[0]!)).status).toBe(200);
|
||||
|
||||
const midway = await request(app).get(`/api/companies/import/transfers/${transferId}`);
|
||||
expect(midway.status).toBe(200);
|
||||
expect(midway.body.totalParts).toBe(3);
|
||||
expect(midway.body.completedParts).toBe(2);
|
||||
expect(midway.body.missingParts).toEqual([1]);
|
||||
|
||||
expect((await putPart(transferId, 1, slices[1]!)).status).toBe(200);
|
||||
|
||||
const applied = await request(app)
|
||||
.post(`/api/companies/import/transfers/${transferId}/apply`)
|
||||
.send(importMeta);
|
||||
expect(applied.status).toBe(200);
|
||||
expect(applied.body.company).toEqual({ id: companyId, action: "updated" });
|
||||
|
||||
// The assembled zip fed the existing import pipeline unchanged.
|
||||
expect(mockCompanyPortabilityService.importBundle).toHaveBeenCalledTimes(1);
|
||||
const importBody = mockCompanyPortabilityService.importBundle.mock.calls[0]![0];
|
||||
expect(importBody.source.type).toBe("inline");
|
||||
expect(importBody.source.files["COMPANY.md"]).toContain("Chunked Import");
|
||||
expect(importBody.target).toEqual(importMeta.target);
|
||||
expect(importBody.collisionStrategy).toBe("rename");
|
||||
|
||||
const run = (await companyTransferRunService.getRun(db, transferId))!;
|
||||
expect(run.status).toBe("completed");
|
||||
expect(run.companyId).toBe(companyId);
|
||||
// The shared import pipeline writes the same activity entry as the
|
||||
// single-shot upload route.
|
||||
expect(mockLogActivity).toHaveBeenCalledWith(
|
||||
expect.anything(),
|
||||
expect.objectContaining({ action: "company.imported", entityId: companyId }),
|
||||
);
|
||||
// Spool is deleted on success.
|
||||
await expect(fs.stat(path.join(spoolRoot, transferId))).rejects.toThrow();
|
||||
|
||||
const finished = await request(app).get(`/api/companies/import/transfers/${transferId}`);
|
||||
expect(finished.body.status).toBe("completed");
|
||||
expect(finished.body.missingParts).toEqual([]);
|
||||
});
|
||||
|
||||
it("applies through the async import job machinery when requested", 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);
|
||||
}
|
||||
|
||||
const accepted = await request(app)
|
||||
.post(`/api/companies/import/transfers/${transferId}/apply?async=1`)
|
||||
.send(importMeta);
|
||||
expect(accepted.status).toBe(202);
|
||||
const statusUrl = accepted.body.statusUrl as string;
|
||||
let job: Record<string, any> | undefined;
|
||||
for (let attempt = 0; attempt < 30; attempt += 1) {
|
||||
const polled = await request(app).get(statusUrl);
|
||||
job = polled.body.job;
|
||||
if (job?.status === "succeeded" || job?.status === "failed") break;
|
||||
await new Promise((resolve) => setTimeout(resolve, 10));
|
||||
}
|
||||
expect(job?.status).toBe("succeeded");
|
||||
expect(job?.result?.companyId).toBe(companyId);
|
||||
|
||||
const run = (await companyTransferRunService.getRun(db, transferId))!;
|
||||
expect(run.status).toBe("completed");
|
||||
await expect(fs.stat(path.join(spoolRoot, transferId))).rejects.toThrow();
|
||||
});
|
||||
|
||||
it("releases the claim when an async apply is refused by a running import", async () => {
|
||||
// Two complete transfers for the same board actor. The first async apply
|
||||
// parks a live job; the second transfer's async apply is refused by the
|
||||
// one-live-import-per-actor rule — and must NOT stay stranded in
|
||||
// "applying", or it would be unusable until a restart.
|
||||
const firstZip = buildFixtureZip();
|
||||
const first = declareTransfer(firstZip, Math.ceil(firstZip.length / 2));
|
||||
const firstCreated = await request(app).post("/api/companies/import/transfers").send(first.body);
|
||||
const firstId = firstCreated.body.transferId as string;
|
||||
for (const [index, slice] of first.slices.entries()) {
|
||||
expect((await putPart(firstId, index, slice)).status).toBe(200);
|
||||
}
|
||||
const secondZip = buildFixtureZip();
|
||||
const second = declareTransfer(secondZip, Math.ceil(secondZip.length / 2));
|
||||
const secondCreated = await request(app).post("/api/companies/import/transfers").send(second.body);
|
||||
const secondId = secondCreated.body.transferId as string;
|
||||
for (const [index, slice] of second.slices.entries()) {
|
||||
expect((await putPart(secondId, index, slice)).status).toBe(200);
|
||||
}
|
||||
|
||||
let releaseImport: (() => void) | undefined;
|
||||
mockCompanyPortabilityService.importBundle.mockImplementationOnce(
|
||||
async () => {
|
||||
await new Promise<void>((resolve) => {
|
||||
releaseImport = resolve;
|
||||
});
|
||||
return {
|
||||
company: { id: companyId, action: "updated" },
|
||||
agents: [{ id: "agent-1" }],
|
||||
warnings: [],
|
||||
};
|
||||
},
|
||||
);
|
||||
const accepted = await request(app)
|
||||
.post(`/api/companies/import/transfers/${firstId}/apply?async=1`)
|
||||
.send(importMeta);
|
||||
expect(accepted.status).toBe(202);
|
||||
|
||||
const refused = await request(app)
|
||||
.post(`/api/companies/import/transfers/${secondId}/apply?async=1`)
|
||||
.send(importMeta);
|
||||
expect(refused.status).toBe(409);
|
||||
// The refused transfer is failed (retryable), never stranded applying.
|
||||
expect((await companyTransferRunService.getRun(db, secondId))!.status).toBe("failed");
|
||||
|
||||
releaseImport!();
|
||||
let job: Record<string, any> | undefined;
|
||||
for (let attempt = 0; attempt < 30; attempt += 1) {
|
||||
const polled = await request(app).get(accepted.body.statusUrl as string);
|
||||
job = polled.body.job;
|
||||
if (job?.status === "succeeded" || job?.status === "failed") break;
|
||||
await new Promise((resolve) => setTimeout(resolve, 10));
|
||||
}
|
||||
expect(job?.status).toBe("succeeded");
|
||||
expect((await companyTransferRunService.getRun(db, firstId))!.status).toBe("completed");
|
||||
|
||||
// After the first import settles, the refused transfer applies cleanly.
|
||||
const retried = await request(app)
|
||||
.post(`/api/companies/import/transfers/${secondId}/apply`)
|
||||
.send(importMeta);
|
||||
expect(retried.status).toBe(200);
|
||||
expect((await companyTransferRunService.getRun(db, secondId))!.status).toBe("completed");
|
||||
});
|
||||
|
||||
it("freezes part uploads while an apply holds the claim", 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, then retry a part upload:
|
||||
// the route must refuse before touching the spool, and the previously
|
||||
// uploaded part file must survive untouched for the apply to read.
|
||||
expect(await companyTransferRunService.claimApply(db, transferId)).toBe(true);
|
||||
const frozen = await putPart(transferId, 0, slices[0]!);
|
||||
expect(frozen.status).toBe(409);
|
||||
expect(frozen.body.error).toContain("apply is in progress");
|
||||
await expect(
|
||||
fs.stat(path.join(spoolRoot, transferId, "part-0")),
|
||||
).resolves.toBeDefined();
|
||||
await companyTransferRunService.releaseApplyClaim(db, transferId, "test release");
|
||||
});
|
||||
|
||||
it("rejects declarations that are inconsistent or over the limits", async () => {
|
||||
const zip = buildFixtureZip();
|
||||
const { body } = declareTransfer(zip, Math.ceil(zip.length / 2));
|
||||
|
||||
const oversizedParts = await request(app)
|
||||
.post("/api/companies/import/transfers")
|
||||
.send({ ...body, partSizeBytes: 65 * 1024 * 1024 });
|
||||
expect(oversizedParts.status).toBe(422);
|
||||
|
||||
const outOfOrder = await request(app)
|
||||
.post("/api/companies/import/transfers")
|
||||
.send({ ...body, parts: [...body.parts].reverse() });
|
||||
expect(outOfOrder.status).toBe(422);
|
||||
|
||||
const sumMismatch = await request(app)
|
||||
.post("/api/companies/import/transfers")
|
||||
.send({ ...body, totalBytes: body.totalBytes + 1 });
|
||||
expect(sumMismatch.status).toBe(422);
|
||||
|
||||
const badHash = await request(app)
|
||||
.post("/api/companies/import/transfers")
|
||||
.send({
|
||||
...body,
|
||||
parts: body.parts.map((part) => ({ ...part, sha256: "not-hex" })),
|
||||
});
|
||||
expect(badHash.status).toBe(400);
|
||||
});
|
||||
|
||||
it("rejects a part whose bytes do not match the declaration and records nothing", 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;
|
||||
|
||||
// Right length, wrong content -> sha mismatch.
|
||||
const corrupted = Buffer.from(slices[0]!);
|
||||
corrupted[0] = corrupted[0]! ^ 0xff;
|
||||
const wrongSha = await putPart(transferId, 0, corrupted);
|
||||
expect(wrongSha.status).toBe(422);
|
||||
|
||||
// Wrong length.
|
||||
const wrongLength = await putPart(transferId, 0, slices[0]!.subarray(1));
|
||||
expect(wrongLength.status).toBe(422);
|
||||
|
||||
const status = await request(app).get(`/api/companies/import/transfers/${transferId}`);
|
||||
expect(status.body.completedParts).toBe(0);
|
||||
expect(status.body.missingParts).toEqual([0, 1]);
|
||||
const run = (await companyTransferRunService.getRun(db, transferId))!;
|
||||
expect(run.completedParts).toEqual([]);
|
||||
});
|
||||
|
||||
it("treats a re-upload of a completed part as a no-op success", 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 again = await putPart(transferId, 0, slices[0]!);
|
||||
expect(again.status).toBe(200);
|
||||
expect(again.body.alreadyCompleted).toBe(true);
|
||||
|
||||
const run = (await companyTransferRunService.getRun(db, transferId))!;
|
||||
expect(run.completedParts).toEqual(["part-0"]);
|
||||
});
|
||||
|
||||
it("refuses to apply 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, 0, slices[0]!)).status).toBe(200);
|
||||
|
||||
const applied = await request(app)
|
||||
.post(`/api/companies/import/transfers/${transferId}/apply`)
|
||||
.send(importMeta);
|
||||
expect(applied.status).toBe(409);
|
||||
expect(applied.body.missingParts).toEqual([1, 2]);
|
||||
expect(mockCompanyPortabilityService.importBundle).not.toHaveBeenCalled();
|
||||
expect((await companyTransferRunService.getRun(db, transferId))!.status).not.toBe("completed");
|
||||
});
|
||||
|
||||
it("resumes a re-declared transfer with its prior progress", 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, 0, slices[0]!)).status).toBe(200);
|
||||
|
||||
const redeclared = await request(app).post("/api/companies/import/transfers").send(body);
|
||||
expect(redeclared.status).toBe(200);
|
||||
expect(redeclared.body.transferId).toBe(transferId);
|
||||
expect(redeclared.body.missingParts).toEqual([1, 2]);
|
||||
expect(redeclared.body.alreadyCompleted).toBe(false);
|
||||
});
|
||||
|
||||
it("reports an already-applied transfer on re-declaration", 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 redeclared = await request(app).post("/api/companies/import/transfers").send(body);
|
||||
expect(redeclared.status).toBe(200);
|
||||
expect(redeclared.body.transferId).toBe(transferId);
|
||||
expect(redeclared.body.alreadyCompleted).toBe(true);
|
||||
expect(redeclared.body.missingParts).toEqual([]);
|
||||
|
||||
const reApplied = await request(app)
|
||||
.post(`/api/companies/import/transfers/${transferId}/apply`)
|
||||
.send(importMeta);
|
||||
expect(reApplied.status).toBe(409);
|
||||
});
|
||||
|
||||
it("keeps a completed run completed when post-success spool cleanup fails", 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 import itself succeeds; only the post-success spool deletion blows
|
||||
// up. The cleanup error must not surface as an apply failure and — above
|
||||
// all — must not flip the completed run back to a claimable state.
|
||||
spoolRemovalFailure.error = new Error("EACCES: spool dir is busy");
|
||||
try {
|
||||
const applied = await request(app)
|
||||
.post(`/api/companies/import/transfers/${transferId}/apply`)
|
||||
.send(importMeta);
|
||||
expect(applied.status).toBe(200);
|
||||
expect(applied.body.company).toEqual({ id: companyId, action: "updated" });
|
||||
} finally {
|
||||
spoolRemovalFailure.error = null;
|
||||
}
|
||||
expect(mockCompanyPortabilityService.importBundle).toHaveBeenCalledTimes(1);
|
||||
expect((await companyTransferRunService.getRun(db, transferId))!.status).toBe("completed");
|
||||
|
||||
// A follow-up apply short-circuits on the completed run instead of
|
||||
// claiming it for a duplicate import.
|
||||
const reApplied = await request(app)
|
||||
.post(`/api/companies/import/transfers/${transferId}/apply`)
|
||||
.send(importMeta);
|
||||
expect(reApplied.status).toBe(409);
|
||||
expect(reApplied.body.error).toContain("already been applied");
|
||||
const redeclared = await request(app).post("/api/companies/import/transfers").send(body);
|
||||
expect(redeclared.body.transferId).toBe(transferId);
|
||||
expect(redeclared.body.alreadyCompleted).toBe(true);
|
||||
// The import pipeline never ran a second time.
|
||||
expect(mockCompanyPortabilityService.importBundle).toHaveBeenCalledTimes(1);
|
||||
expect((await companyTransferRunService.getRun(db, transferId))!.status).toBe("completed");
|
||||
|
||||
// The leftover spool from the failed cleanup is not stranded: the sweep
|
||||
// collects terminal-run spools immediately, no idle wait, and the run
|
||||
// keeps its terminal status.
|
||||
await expect(fs.stat(path.join(spoolRoot, transferId))).resolves.toBeDefined();
|
||||
const swept = await sweepAbandonedImportTransferSpools(db, spoolRoot);
|
||||
expect(swept.swept).toBeGreaterThanOrEqual(1);
|
||||
await expect(fs.stat(path.join(spoolRoot, transferId))).rejects.toThrow();
|
||||
expect((await companyTransferRunService.getRun(db, transferId))!.status).toBe("completed");
|
||||
});
|
||||
|
||||
it("settles the run completed even when attaching the imported company fails", 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 import commits, then a post-settlement step (company attachment)
|
||||
// blows up. Settlement runs first, so the error must neither surface as
|
||||
// an apply failure nor release the claim into a re-importable state.
|
||||
const attachSpy = vi
|
||||
.spyOn(companyTransferRunService, "attachCompany")
|
||||
.mockRejectedValueOnce(new Error("registry write timed out"));
|
||||
try {
|
||||
const applied = await request(app)
|
||||
.post(`/api/companies/import/transfers/${transferId}/apply`)
|
||||
.send(importMeta);
|
||||
expect(applied.status).toBe(200);
|
||||
} finally {
|
||||
attachSpy.mockRestore();
|
||||
}
|
||||
expect(mockCompanyPortabilityService.importBundle).toHaveBeenCalledTimes(1);
|
||||
expect((await companyTransferRunService.getRun(db, transferId))!.status).toBe("completed");
|
||||
|
||||
const reApplied = await request(app)
|
||||
.post(`/api/companies/import/transfers/${transferId}/apply`)
|
||||
.send(importMeta);
|
||||
expect(reApplied.status).toBe(409);
|
||||
expect(mockCompanyPortabilityService.importBundle).toHaveBeenCalledTimes(1);
|
||||
});
|
||||
|
||||
it("fails closed when the assembled zip does not match the declared whole-file hash", async () => {
|
||||
const zip = buildFixtureZip();
|
||||
const { body, slices } = declareTransfer(zip, Math.ceil(zip.length / 2));
|
||||
// Per-part hashes are honest, but the declared whole-file hash is not.
|
||||
const lying = { ...body, zipSha256: sha256Hex(Buffer.from("something else entirely")) };
|
||||
const created = await request(app).post("/api/companies/import/transfers").send(lying);
|
||||
const transferId = created.body.transferId as string;
|
||||
for (const [index, slice] of slices.entries()) {
|
||||
expect((await putPart(transferId, index, slice)).status).toBe(200);
|
||||
}
|
||||
|
||||
const applied = await request(app)
|
||||
.post(`/api/companies/import/transfers/${transferId}/apply`)
|
||||
.send(importMeta);
|
||||
expect(applied.status).toBe(422);
|
||||
expect(mockCompanyPortabilityService.importBundle).not.toHaveBeenCalled();
|
||||
|
||||
const run = (await companyTransferRunService.getRun(db, transferId))!;
|
||||
expect(run.status).toBe("failed");
|
||||
// The spool is deleted so a resume re-uploads everything.
|
||||
await expect(fs.stat(path.join(spoolRoot, transferId))).rejects.toThrow();
|
||||
});
|
||||
|
||||
it("hides transfers from other actors", 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;
|
||||
|
||||
const otherStatus = await request(app)
|
||||
.get(`/api/companies/import/transfers/${transferId}`)
|
||||
.set(TEST_USER_HEADER, "board-user-b");
|
||||
expect(otherStatus.status).toBe(404);
|
||||
|
||||
const otherUpload = await putPart(transferId, 0, slices[0]!, "board-user-b");
|
||||
expect(otherUpload.status).toBe(404);
|
||||
|
||||
const otherApply = await request(app)
|
||||
.post(`/api/companies/import/transfers/${transferId}/apply`)
|
||||
.set(TEST_USER_HEADER, "board-user-b")
|
||||
.send(importMeta);
|
||||
expect(otherApply.status).toBe(404);
|
||||
|
||||
// Malformed ids never reach the filesystem or the database.
|
||||
const malformed = await request(app).get("/api/companies/import/transfers/..%2Fescape");
|
||||
expect(malformed.status).toBe(404);
|
||||
});
|
||||
|
||||
it("sweeps abandoned spools and cancels their open runs", 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);
|
||||
|
||||
// Fresh activity: nothing to sweep at the real threshold.
|
||||
expect(await sweepAbandonedImportTransferSpools(db, spoolRoot)).toEqual({ swept: 0 });
|
||||
await expect(fs.stat(path.join(spoolRoot, transferId))).resolves.toBeDefined();
|
||||
|
||||
// 25 hours later the run saw no activity: spool deleted, run cancelled.
|
||||
// Cancelled — not failed — so no apply can claim the run in the window
|
||||
// between the status flip and the spool deletion (claimApply accepts
|
||||
// failed runs but never cancelled ones).
|
||||
const later = new Date(Date.now() + 25 * 60 * 60 * 1000);
|
||||
const swept = await sweepAbandonedImportTransferSpools(db, spoolRoot, { now: later });
|
||||
expect(swept.swept).toBeGreaterThanOrEqual(1);
|
||||
await expect(fs.stat(path.join(spoolRoot, transferId))).rejects.toThrow();
|
||||
expect((await companyTransferRunService.getRun(db, transferId))!.status).toBe("cancelled");
|
||||
|
||||
// A redeclaration after the sweep starts a FRESH run — the cancelled run
|
||||
// is never resumed — with every part missing again, matching the part
|
||||
// upload's 410 "re-create it" contract.
|
||||
const redeclared = await request(app).post("/api/companies/import/transfers").send(body);
|
||||
expect(redeclared.status).toBe(200);
|
||||
expect(redeclared.body.transferId).not.toBe(transferId);
|
||||
expect(redeclared.body.alreadyCompleted).toBe(false);
|
||||
expect(redeclared.body.missingParts).toEqual([0, 1]);
|
||||
// The swept run itself stays terminal.
|
||||
expect((await companyTransferRunService.getRun(db, transferId))!.status).toBe("cancelled");
|
||||
});
|
||||
|
||||
it("skips the sweep when the run saw activity after the staleness read", 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);
|
||||
|
||||
// TOCTOU simulation: the sweep's initial read sees a 25h-stale updatedAt
|
||||
// — as if a part upload bumped the row right after that read — while the
|
||||
// guarded UPDATE sees the fresh row in the database and must miss.
|
||||
const real = (await companyTransferRunService.getRun(db, transferId))!;
|
||||
const staleRead = vi.spyOn(companyTransferRunService, "getRun").mockResolvedValue({
|
||||
...real,
|
||||
updatedAt: new Date(Date.now() - 25 * 60 * 60 * 1000),
|
||||
});
|
||||
try {
|
||||
expect(await sweepAbandonedImportTransferSpools(db, spoolRoot)).toEqual({ swept: 0 });
|
||||
} finally {
|
||||
staleRead.mockRestore();
|
||||
}
|
||||
|
||||
// The spool survived and the run is still open.
|
||||
await expect(fs.stat(path.join(spoolRoot, transferId))).resolves.toBeDefined();
|
||||
expect((await companyTransferRunService.getRun(db, transferId))!.status).toBe("running");
|
||||
});
|
||||
|
||||
it("returns 410 for a part upload against a swept transfer and records nothing", 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);
|
||||
expect((await companyTransferRunService.getRun(db, transferId))!.status).toBe("cancelled");
|
||||
|
||||
const upload = await putPart(transferId, 1, slices[1]!);
|
||||
expect(upload.status).toBe(410);
|
||||
const run = (await companyTransferRunService.getRun(db, transferId))!;
|
||||
expect(run.status).toBe("cancelled");
|
||||
expect(run.completedParts).toEqual(["part-0"]);
|
||||
// The stray part file was discarded, not left half-spooled.
|
||||
await expect(fs.stat(path.join(spoolRoot, transferId, "part-1"))).rejects.toThrow();
|
||||
});
|
||||
|
||||
it("runs exactly one of two concurrent applies", 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);
|
||||
}
|
||||
|
||||
// Hold the import open long enough for the losing apply to reach the
|
||||
// claim while the winner is still importing.
|
||||
mockCompanyPortabilityService.importBundle.mockImplementation(async () => {
|
||||
await new Promise((resolve) => setTimeout(resolve, 100));
|
||||
return { company: { id: companyId, action: "updated" }, agents: [], warnings: [] };
|
||||
});
|
||||
|
||||
const [first, second] = await Promise.all([
|
||||
request(app).post(`/api/companies/import/transfers/${transferId}/apply`).send(importMeta),
|
||||
request(app).post(`/api/companies/import/transfers/${transferId}/apply`).send(importMeta),
|
||||
]);
|
||||
expect([first.status, second.status].sort((a, b) => a - b)).toEqual([200, 409]);
|
||||
// The non-idempotent import ran exactly once and the claim settled.
|
||||
expect(mockCompanyPortabilityService.importBundle).toHaveBeenCalledTimes(1);
|
||||
expect((await companyTransferRunService.getRun(db, transferId))!.status).toBe("completed");
|
||||
await expect(fs.stat(path.join(spoolRoot, transferId))).rejects.toThrow();
|
||||
});
|
||||
});
|
||||
|
|
@ -0,0 +1,336 @@
|
|||
import { afterAll, beforeAll, describe, expect, it } from "vitest";
|
||||
import { createDb } from "@paperclipai/db";
|
||||
import { sha256HexOfBytes } from "@paperclipai/shared/portability-hash";
|
||||
import {
|
||||
getEmbeddedPostgresTestSupport,
|
||||
startEmbeddedPostgresTestDatabase,
|
||||
} from "./helpers/embedded-postgres.js";
|
||||
import {
|
||||
companyTransferRunService,
|
||||
type CompanyTransferPartsIndex,
|
||||
} from "../services/company-transfer-runs.js";
|
||||
|
||||
const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport();
|
||||
const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip;
|
||||
|
||||
const pngBytes = new Uint8Array([0x89, 0x50, 0x4e, 0x47, 0x0d, 0x0a, 0x1a, 0x0a]);
|
||||
|
||||
// Manifest-shaped fixture: the ledger only cares about an idempotency key and
|
||||
// part names, so the fixture is built inline rather than through any container
|
||||
// packer. Distinct content per call: identical content intentionally derives
|
||||
// the same idempotency key, which would make separate tests resume each
|
||||
// other's runs.
|
||||
let manifestSeq = 0;
|
||||
function demoManifest(): CompanyTransferPartsIndex & { idempotencyKey: string } {
|
||||
manifestSeq += 1;
|
||||
return {
|
||||
idempotencyKey: sha256HexOfBytes(Buffer.from(`transfer-run-fixture-${manifestSeq}`)),
|
||||
chunks: [{ name: "chunks/0000.json.gz" }, { name: "chunks/0001.json.gz" }],
|
||||
blobs: [{ name: `blobs/${sha256HexOfBytes(pngBytes)}` }],
|
||||
};
|
||||
}
|
||||
|
||||
describeEmbeddedPostgres("companyTransferRunService", () => {
|
||||
let tempDb: Awaited<ReturnType<typeof startEmbeddedPostgresTestDatabase>> | null = null;
|
||||
let db!: ReturnType<typeof createDb>;
|
||||
|
||||
beforeAll(async () => {
|
||||
tempDb = await startEmbeddedPostgresTestDatabase("paperclip-transfer-runs-");
|
||||
db = createDb(tempDb.connectionString);
|
||||
});
|
||||
|
||||
afterAll(async () => {
|
||||
await tempDb?.cleanup();
|
||||
});
|
||||
|
||||
it("creates a run, records parts idempotently, and resumes from the remainder", async () => {
|
||||
const manifest = demoManifest();
|
||||
const created = await companyTransferRunService.resumeOrCreate(db, {
|
||||
direction: "import",
|
||||
actorKey: "user:alice",
|
||||
idempotencyKey: manifest.idempotencyKey,
|
||||
containerRef: { kind: "zip-upload" },
|
||||
});
|
||||
expect(created.resumed).toBe(false);
|
||||
expect(created.run.status).toBe("pending");
|
||||
|
||||
await companyTransferRunService.recordManifest(
|
||||
db,
|
||||
created.run.id,
|
||||
manifest,
|
||||
sha256HexOfBytes(Buffer.from(JSON.stringify(manifest))),
|
||||
);
|
||||
await companyTransferRunService.start(db, created.run.id);
|
||||
|
||||
const firstChunk = manifest.chunks[0]!.name;
|
||||
await companyTransferRunService.completePart(db, created.run.id, firstChunk);
|
||||
// Re-completing the same part must not double-count.
|
||||
await companyTransferRunService.completePart(db, created.run.id, firstChunk);
|
||||
|
||||
const midway = (await companyTransferRunService.getRun(db, created.run.id))!;
|
||||
expect(midway.status).toBe("running");
|
||||
expect(midway.completedParts).toEqual([firstChunk]);
|
||||
expect(midway.chunkCount).toBe(manifest.chunks.length);
|
||||
expect(midway.blobCount).toBe(manifest.blobs.length);
|
||||
|
||||
const remaining = companyTransferRunService.remainingParts(midway, manifest);
|
||||
expect(remaining).toEqual([
|
||||
manifest.chunks[1]!.name,
|
||||
...manifest.blobs.map((blob) => blob.name),
|
||||
]);
|
||||
|
||||
// Simulate an interruption: the run fails, then the same content retries.
|
||||
await companyTransferRunService.fail(db, created.run.id, "connection dropped");
|
||||
const resumed = await companyTransferRunService.resumeOrCreate(db, {
|
||||
direction: "import",
|
||||
actorKey: "user:alice",
|
||||
idempotencyKey: manifest.idempotencyKey,
|
||||
containerRef: { kind: "zip-upload" },
|
||||
});
|
||||
expect(resumed.resumed).toBe(true);
|
||||
expect(resumed.alreadyCompleted).toBe(false);
|
||||
expect(resumed.run.id).toBe(created.run.id);
|
||||
expect(resumed.run.completedParts).toEqual([firstChunk]);
|
||||
|
||||
await companyTransferRunService.start(db, resumed.run.id);
|
||||
for (const part of companyTransferRunService.remainingParts(resumed.run, manifest)) {
|
||||
await companyTransferRunService.completePart(db, resumed.run.id, part);
|
||||
}
|
||||
await companyTransferRunService.complete(db, resumed.run.id);
|
||||
|
||||
const done = (await companyTransferRunService.getRun(db, created.run.id))!;
|
||||
expect(done.status).toBe("completed");
|
||||
expect(done.error).toBeNull();
|
||||
expect(companyTransferRunService.remainingParts(done, manifest)).toEqual([]);
|
||||
|
||||
// A retry of identical content after completion short-circuits.
|
||||
const retried = await companyTransferRunService.resumeOrCreate(db, {
|
||||
direction: "import",
|
||||
actorKey: "user:alice",
|
||||
idempotencyKey: manifest.idempotencyKey,
|
||||
containerRef: { kind: "zip-upload" },
|
||||
});
|
||||
expect(retried.alreadyCompleted).toBe(true);
|
||||
expect(retried.run.id).toBe(created.run.id);
|
||||
});
|
||||
|
||||
it("scopes resume to the actor and direction", async () => {
|
||||
const manifest = demoManifest();
|
||||
const aliceRun = await companyTransferRunService.resumeOrCreate(db, {
|
||||
direction: "export",
|
||||
actorKey: "user:alice",
|
||||
idempotencyKey: manifest.idempotencyKey,
|
||||
containerRef: { kind: "relay", prefix: "staging/a" },
|
||||
});
|
||||
|
||||
const bobRun = await companyTransferRunService.resumeOrCreate(db, {
|
||||
direction: "export",
|
||||
actorKey: "user:bob",
|
||||
idempotencyKey: manifest.idempotencyKey,
|
||||
containerRef: { kind: "relay", prefix: "staging/b" },
|
||||
});
|
||||
expect(bobRun.resumed).toBe(false);
|
||||
expect(bobRun.run.id).not.toBe(aliceRun.run.id);
|
||||
|
||||
const aliceImport = await companyTransferRunService.resumeOrCreate(db, {
|
||||
direction: "import",
|
||||
actorKey: "user:alice",
|
||||
idempotencyKey: manifest.idempotencyKey,
|
||||
containerRef: { kind: "relay", prefix: "staging/a" },
|
||||
});
|
||||
expect(aliceImport.resumed).toBe(false);
|
||||
expect(aliceImport.run.id).not.toBe(aliceRun.run.id);
|
||||
|
||||
expect(await companyTransferRunService.getRunForActor(db, aliceRun.run.id, "user:bob")).toBeNull();
|
||||
expect(await companyTransferRunService.getRunForActor(db, aliceRun.run.id, "user:alice")).not.toBeNull();
|
||||
});
|
||||
|
||||
it("creates a single run for concurrent identical declarations", async () => {
|
||||
const manifest = demoManifest();
|
||||
const input = {
|
||||
direction: "import" as const,
|
||||
actorKey: "user:dave",
|
||||
idempotencyKey: manifest.idempotencyKey,
|
||||
containerRef: { kind: "zip-upload" },
|
||||
};
|
||||
// Without the advisory lock both callers can miss the lookup and both
|
||||
// insert; with it, exactly one inserts and the other resumes that row.
|
||||
const [first, second] = await Promise.all([
|
||||
companyTransferRunService.resumeOrCreate(db, input),
|
||||
companyTransferRunService.resumeOrCreate(db, input),
|
||||
]);
|
||||
expect(first.run.id).toBe(second.run.id);
|
||||
expect([first.resumed, second.resumed].filter(Boolean)).toHaveLength(1);
|
||||
expect(first.alreadyCompleted).toBe(false);
|
||||
expect(second.alreadyCompleted).toBe(false);
|
||||
});
|
||||
|
||||
it("grants the apply claim to exactly one concurrent caller", async () => {
|
||||
const manifest = demoManifest();
|
||||
const { run } = await companyTransferRunService.resumeOrCreate(db, {
|
||||
direction: "import",
|
||||
actorKey: "user:erin",
|
||||
idempotencyKey: manifest.idempotencyKey,
|
||||
containerRef: { kind: "zip-upload" },
|
||||
});
|
||||
await companyTransferRunService.start(db, run.id);
|
||||
|
||||
const claims = await Promise.all([
|
||||
companyTransferRunService.claimApply(db, run.id),
|
||||
companyTransferRunService.claimApply(db, run.id),
|
||||
]);
|
||||
expect(claims.filter(Boolean)).toHaveLength(1);
|
||||
expect((await companyTransferRunService.getRun(db, run.id))!.status).toBe("applying");
|
||||
|
||||
// A concurrent re-declaration resumes the mid-apply run instead of
|
||||
// spawning a duplicate, but start() must not restart it.
|
||||
const redeclared = await companyTransferRunService.resumeOrCreate(db, {
|
||||
direction: "import",
|
||||
actorKey: "user:erin",
|
||||
idempotencyKey: manifest.idempotencyKey,
|
||||
containerRef: { kind: "zip-upload" },
|
||||
});
|
||||
expect(redeclared.run.id).toBe(run.id);
|
||||
expect(redeclared.resumed).toBe(true);
|
||||
expect(redeclared.alreadyCompleted).toBe(false);
|
||||
await companyTransferRunService.start(db, run.id);
|
||||
expect((await companyTransferRunService.getRun(db, run.id))!.status).toBe("applying");
|
||||
|
||||
// A failed apply releases the claim: the run is retryable again.
|
||||
await companyTransferRunService.fail(db, run.id, "import blew up");
|
||||
expect(await companyTransferRunService.claimApply(db, run.id)).toBe(true);
|
||||
await companyTransferRunService.complete(db, run.id);
|
||||
expect((await companyTransferRunService.getRun(db, run.id))!.status).toBe("completed");
|
||||
expect(await companyTransferRunService.claimApply(db, run.id)).toBe(false);
|
||||
});
|
||||
|
||||
it("records parts only while the run is open", async () => {
|
||||
const manifest = demoManifest();
|
||||
const { run } = await companyTransferRunService.resumeOrCreate(db, {
|
||||
direction: "import",
|
||||
actorKey: "user:frank",
|
||||
idempotencyKey: manifest.idempotencyKey,
|
||||
containerRef: { kind: "zip-upload" },
|
||||
});
|
||||
await companyTransferRunService.start(db, run.id);
|
||||
expect(await companyTransferRunService.completePart(db, run.id, "part-0")).toBe(true);
|
||||
|
||||
await companyTransferRunService.fail(db, run.id, "swept");
|
||||
expect(await companyTransferRunService.completePart(db, run.id, "part-1")).toBe(false);
|
||||
expect((await companyTransferRunService.getRun(db, run.id))!.completedParts).toEqual(["part-0"]);
|
||||
});
|
||||
|
||||
it("claims an abandoned run only while it is open and still idle", async () => {
|
||||
const manifest = demoManifest();
|
||||
const { run } = await companyTransferRunService.resumeOrCreate(db, {
|
||||
direction: "import",
|
||||
actorKey: "user:grace",
|
||||
idempotencyKey: manifest.idempotencyKey,
|
||||
containerRef: { kind: "zip-upload" },
|
||||
});
|
||||
await companyTransferRunService.start(db, run.id);
|
||||
|
||||
// Fresh activity: a cutoff in the past misses the guard.
|
||||
const past = new Date(Date.now() - 60 * 60 * 1000);
|
||||
expect(await companyTransferRunService.cancelIfIdle(db, run.id, past, "stale")).toBe(false);
|
||||
expect((await companyTransferRunService.getRun(db, run.id))!.status).toBe("running");
|
||||
|
||||
// Idle past the cutoff: exactly this claim cancels the run.
|
||||
const future = new Date(Date.now() + 60 * 60 * 1000);
|
||||
expect(await companyTransferRunService.cancelIfIdle(db, run.id, future, "stale")).toBe(true);
|
||||
const swept = (await companyTransferRunService.getRun(db, run.id))!;
|
||||
expect(swept.status).toBe("cancelled");
|
||||
expect(swept.error).toBe("stale");
|
||||
// Terminal already — a second claim finds nothing to do.
|
||||
expect(await companyTransferRunService.cancelIfIdle(db, run.id, future, "stale")).toBe(false);
|
||||
// Cancelled, not failed, so the swept run can never be claimed for an
|
||||
// apply while its spool is being deleted.
|
||||
expect(await companyTransferRunService.claimApply(db, run.id)).toBe(false);
|
||||
});
|
||||
|
||||
it("releases an apply claim without ever reopening a settled run", async () => {
|
||||
const manifest = demoManifest();
|
||||
const { run } = await companyTransferRunService.resumeOrCreate(db, {
|
||||
direction: "import",
|
||||
actorKey: "user:heidi",
|
||||
idempotencyKey: manifest.idempotencyKey,
|
||||
containerRef: { kind: "zip-upload" },
|
||||
});
|
||||
await companyTransferRunService.start(db, run.id);
|
||||
|
||||
// Releasing a held claim fails the run and makes it claimable again.
|
||||
expect(await companyTransferRunService.claimApply(db, run.id)).toBe(true);
|
||||
expect(await companyTransferRunService.releaseApplyClaim(db, run.id, "import blew up")).toBe(true);
|
||||
const released = (await companyTransferRunService.getRun(db, run.id))!;
|
||||
expect(released.status).toBe("failed");
|
||||
expect(released.error).toBe("import blew up");
|
||||
expect(await companyTransferRunService.claimApply(db, run.id)).toBe(true);
|
||||
|
||||
// Guarded on "applying": after complete() settles the claim, a late
|
||||
// release (e.g. a cleanup error caught after success) is a no-op — the
|
||||
// run stays completed and can never be flipped back to claimable.
|
||||
await companyTransferRunService.complete(db, run.id);
|
||||
expect(await companyTransferRunService.releaseApplyClaim(db, run.id, "cleanup failed")).toBe(false);
|
||||
const settled = (await companyTransferRunService.getRun(db, run.id))!;
|
||||
expect(settled.status).toBe("completed");
|
||||
expect(settled.error).toBeNull();
|
||||
expect(await companyTransferRunService.claimApply(db, run.id)).toBe(false);
|
||||
});
|
||||
|
||||
it("recovers runs stranded in applying by a restart, leaving live runs alone", async () => {
|
||||
const strandedManifest = demoManifest();
|
||||
const { run: stranded } = await companyTransferRunService.resumeOrCreate(db, {
|
||||
direction: "import",
|
||||
actorKey: "user:ivan",
|
||||
idempotencyKey: strandedManifest.idempotencyKey,
|
||||
containerRef: { kind: "zip-upload" },
|
||||
});
|
||||
await companyTransferRunService.start(db, stranded.id);
|
||||
expect(await companyTransferRunService.claimApply(db, stranded.id)).toBe(true);
|
||||
|
||||
const openManifest = demoManifest();
|
||||
const { run: open } = await companyTransferRunService.resumeOrCreate(db, {
|
||||
direction: "import",
|
||||
actorKey: "user:ivan",
|
||||
idempotencyKey: openManifest.idempotencyKey,
|
||||
containerRef: { kind: "zip-upload" },
|
||||
});
|
||||
await companyTransferRunService.start(db, open.id);
|
||||
|
||||
// The process "restarted": the in-memory apply for the claimed run is
|
||||
// gone, so startup recovery fails it and makes it claimable again.
|
||||
const recovered = await companyTransferRunService.recoverStrandedApplyingRuns(db);
|
||||
expect(recovered).toContain(stranded.id);
|
||||
expect(recovered).not.toContain(open.id);
|
||||
const failed = (await companyTransferRunService.getRun(db, stranded.id))!;
|
||||
expect(failed.status).toBe("failed");
|
||||
expect(failed.error).toBe("apply interrupted by a restart — verify whether the import completed before retrying");
|
||||
expect(await companyTransferRunService.claimApply(db, stranded.id)).toBe(true);
|
||||
|
||||
// Runs that were merely uploading are untouched.
|
||||
expect((await companyTransferRunService.getRun(db, open.id))!.status).toBe("running");
|
||||
});
|
||||
|
||||
it("does not restart a cancelled run", async () => {
|
||||
const manifest = demoManifest();
|
||||
const { run } = await companyTransferRunService.resumeOrCreate(db, {
|
||||
direction: "import",
|
||||
actorKey: "user:carol",
|
||||
idempotencyKey: manifest.idempotencyKey,
|
||||
containerRef: { kind: "zip-upload" },
|
||||
});
|
||||
await companyTransferRunService.cancel(db, run.id);
|
||||
await companyTransferRunService.start(db, run.id);
|
||||
expect((await companyTransferRunService.getRun(db, run.id))!.status).toBe("cancelled");
|
||||
|
||||
// Cancelled runs are not resumed; the same content starts a fresh run.
|
||||
const fresh = await companyTransferRunService.resumeOrCreate(db, {
|
||||
direction: "import",
|
||||
actorKey: "user:carol",
|
||||
idempotencyKey: manifest.idempotencyKey,
|
||||
containerRef: { kind: "zip-upload" },
|
||||
});
|
||||
expect(fresh.resumed).toBe(false);
|
||||
expect(fresh.run.id).not.toBe(run.id);
|
||||
});
|
||||
});
|
||||
|
|
@ -4,6 +4,7 @@ import path from "node:path";
|
|||
import { fileURLToPath } from "node:url";
|
||||
import request from "supertest";
|
||||
import { describe, expect, it } from "vitest";
|
||||
import { COMPANY_IMPORT_TRANSFERS_ROUTE_PATH } from "@paperclipai/shared/company-import-transfer";
|
||||
import { errorHandler } from "../middleware/index.js";
|
||||
import { buildOpenApiSpec, openApiRoutes } from "../routes/openapi.js";
|
||||
|
||||
|
|
@ -79,8 +80,18 @@ function createApp() {
|
|||
return app;
|
||||
}
|
||||
|
||||
// Route files may compose paths from shared path constants inside template
|
||||
// literals; substitute the constants' values before normalizing.
|
||||
const routePathConstantSubstitutions: Record<string, string> = {
|
||||
"${COMPANY_IMPORT_TRANSFERS_ROUTE_PATH}": COMPANY_IMPORT_TRANSFERS_ROUTE_PATH,
|
||||
};
|
||||
|
||||
function normalizeExpressPath(routePath: string) {
|
||||
return routePath
|
||||
let substituted = routePath;
|
||||
for (const [placeholder, value] of Object.entries(routePathConstantSubstitutions)) {
|
||||
substituted = substituted.split(placeholder).join(value);
|
||||
}
|
||||
return substituted
|
||||
.replace(/\*([A-Za-z0-9_]+)/g, "{$1}")
|
||||
.replace(/:([A-Za-z0-9_]+)/g, "{$1}")
|
||||
.replace(/\/+/g, "/");
|
||||
|
|
@ -126,6 +137,9 @@ function loadActualRoutes() {
|
|||
if (file === "companies.ts" && source.includes("router.post(COMPANY_IMPORT_ROUTE_PATH")) {
|
||||
routes.add("POST /api/companies/import");
|
||||
}
|
||||
if (file === "companies.ts" && source.includes("router.post(COMPANY_IMPORT_TRANSFERS_ROUTE_PATH")) {
|
||||
routes.add(`POST /api/companies${COMPANY_IMPORT_TRANSFERS_ROUTE_PATH}`);
|
||||
}
|
||||
}
|
||||
|
||||
return { routes, unknownRouteFiles: unknownRouteFiles.sort() };
|
||||
|
|
|
|||
|
|
@ -11,6 +11,12 @@ import { actorMiddleware } from "./middleware/auth.js";
|
|||
import { boardMutationGuard } from "./middleware/board-mutation-guard.js";
|
||||
import { privateHostnameGuard, resolvePrivateHostnameAllowSet } from "./middleware/private-hostname-guard.js";
|
||||
import { applyTrustProxy, parseTrustProxyEnv } from "./middleware/trust-proxy.js";
|
||||
import {
|
||||
IMPORT_TRANSFER_SPOOL_SWEEP_INTERVAL_MS,
|
||||
resolveDefaultImportTransferSpoolRoot,
|
||||
sweepAbandonedImportTransferSpools,
|
||||
} from "./services/company-import-transfers.js";
|
||||
import { companyTransferRunService } from "./services/company-transfer-runs.js";
|
||||
import { healthRoutes } from "./routes/health.js";
|
||||
import { cloudRoutes } from "./routes/cloud.js";
|
||||
import { companyRoutes } from "./routes/companies.js";
|
||||
|
|
@ -672,6 +678,48 @@ export async function createApp(
|
|||
if (opts.feedbackExportService) {
|
||||
void flushPendingFeedbackExports();
|
||||
}
|
||||
// Abandoned chunked-import spool sweep: hourly (plus once at startup),
|
||||
// deleting spool dirs whose transfer saw no activity for 24h and cancelling
|
||||
// their still-open ledger runs. Same setInterval + unref + shutdown-clear
|
||||
// shape as the feedback export flush above.
|
||||
const importTransferSpoolRoot = resolveDefaultImportTransferSpoolRoot();
|
||||
const sweepImportTransferSpools = () => {
|
||||
sweepAbandonedImportTransferSpools(db, importTransferSpoolRoot)
|
||||
.then((result) => {
|
||||
if (result.swept > 0) {
|
||||
logger.info(result, "swept abandoned company import transfer spools");
|
||||
}
|
||||
})
|
||||
.catch((err) => {
|
||||
logger.error({ err }, "abandoned company import transfer spool sweep failed");
|
||||
});
|
||||
};
|
||||
let importTransferSweepTimer: ReturnType<typeof setInterval> | null = setInterval(
|
||||
sweepImportTransferSpools,
|
||||
IMPORT_TRANSFER_SPOOL_SWEEP_INTERVAL_MS,
|
||||
);
|
||||
importTransferSweepTimer.unref?.();
|
||||
// Startup only (never on the hourly interval — that would kill live
|
||||
// applies): apply jobs are in-memory in this single process, so any run
|
||||
// still "applying" now was interrupted by the previous shutdown and would
|
||||
// otherwise 409 every retry forever. Fail those stranded runs — their
|
||||
// spooled parts stay reusable — then run the normal sweep once.
|
||||
void companyTransferRunService
|
||||
.recoverStrandedApplyingRuns(db)
|
||||
.then((recovered) => {
|
||||
if (recovered.length > 0) {
|
||||
logger.warn(
|
||||
{ count: recovered.length, runIds: recovered },
|
||||
"failed company transfer runs stranded in applying by a restart",
|
||||
);
|
||||
}
|
||||
})
|
||||
.catch((err) => {
|
||||
logger.error({ err }, "stranded company transfer apply recovery failed");
|
||||
})
|
||||
.finally(() => {
|
||||
sweepImportTransferSpools();
|
||||
});
|
||||
void toolDispatcher.initialize().catch((err) => {
|
||||
logger.error({ err }, "Failed to initialize plugin tool dispatcher");
|
||||
});
|
||||
|
|
@ -736,6 +784,10 @@ export async function createApp(
|
|||
if (appServicesShutdown) return;
|
||||
appServicesShutdown = true;
|
||||
disableFeedbackExportFlushes();
|
||||
if (importTransferSweepTimer) {
|
||||
clearInterval(importTransferSweepTimer);
|
||||
importTransferSweepTimer = null;
|
||||
}
|
||||
devWatcher?.close();
|
||||
viteHtmlRenderer?.dispose();
|
||||
hostServiceCleanup.disposeAll();
|
||||
|
|
|
|||
|
|
@ -24,9 +24,29 @@ import {
|
|||
updateCompanyBrandingSchema,
|
||||
updateCompanySchema,
|
||||
} from "@paperclipai/shared";
|
||||
import { badRequest, forbidden, unprocessable } from "../errors.js";
|
||||
import {
|
||||
COMPANY_IMPORT_TRANSFERS_ROUTE_PATH,
|
||||
companyImportTransferDeclarationSchema,
|
||||
type CompanyImportTransferCreated,
|
||||
type CompanyImportTransferDeclaration,
|
||||
type CompanyImportTransferPartUploadResult,
|
||||
type CompanyImportTransferStatus,
|
||||
} from "@paperclipai/shared/company-import-transfer";
|
||||
import { badRequest, conflict, forbidden, notFound, unprocessable } from "../errors.js";
|
||||
import { PORTABLE_ZIP_UPLOAD_LIMIT_BYTES } from "../http/body-limits.js";
|
||||
import { logger } from "../middleware/logger.js";
|
||||
import { validate } from "../middleware/validate.js";
|
||||
import {
|
||||
assembleImportTransferZip,
|
||||
importTransferPartName,
|
||||
importTransferPartSizeOnDisk,
|
||||
isImportTransferRunId,
|
||||
removeImportTransferPart,
|
||||
removeImportTransferSpool,
|
||||
resolveDefaultImportTransferSpoolRoot,
|
||||
writeImportTransferPart,
|
||||
} from "../services/company-import-transfers.js";
|
||||
import { companyTransferRunService } from "../services/company-transfer-runs.js";
|
||||
import {
|
||||
accessService,
|
||||
agentService,
|
||||
|
|
@ -63,6 +83,28 @@ const zipPackageUpload = multer({
|
|||
limits: { fileSize: PORTABLE_ZIP_UPLOAD_LIMIT_BYTES, files: 1 },
|
||||
});
|
||||
|
||||
// Chunked resumable variant of the zip upload above: the client slices the
|
||||
// exact same .zip into byte-range parts and uploads them individually, so a
|
||||
// dropped connection re-uploads one part instead of the whole package. Parts
|
||||
// are spooled to disk and reassembled at apply, then fed through the same
|
||||
// preview/import logic — no new container format, no change to import
|
||||
// semantics. Only one part is held in memory at a time during upload; the
|
||||
// full-zip buffer exists only while an apply runs (matching the single-shot
|
||||
// path's memory profile).
|
||||
const IMPORT_TRANSFER_PART_SIZE_LIMIT_BYTES = 64 * 1024 * 1024;
|
||||
|
||||
// The declaration body and response shapes are the shared wire contract in
|
||||
// @paperclipai/shared/company-import-transfer — the browser and CLI clients
|
||||
// type against the same schemas and path builders.
|
||||
const importTransferManifestSchema = companyImportTransferDeclarationSchema;
|
||||
|
||||
type ImportTransferManifest = CompanyImportTransferDeclaration;
|
||||
|
||||
const importTransferPartBodyParser = express.raw({
|
||||
type: () => true,
|
||||
limit: IMPORT_TRANSFER_PART_SIZE_LIMIT_BYTES,
|
||||
});
|
||||
|
||||
const rawZipBodyParser = express.raw({
|
||||
type: [...PORTABLE_ZIP_CONTENT_TYPES],
|
||||
limit: PORTABLE_ZIP_UPLOAD_LIMIT_BYTES,
|
||||
|
|
@ -113,6 +155,20 @@ function parseImportMeta(metaRaw: string | undefined): Record<string, unknown> {
|
|||
return rest;
|
||||
}
|
||||
|
||||
/**
|
||||
* The chunked-transfer apply body is the already-parsed JSON equivalent of the
|
||||
* multipart `meta` field: the import fields minus `source` (the source is
|
||||
* always the assembled zip). Mirrors `parseImportMeta` for object bodies.
|
||||
*/
|
||||
function importTransferApplyMeta(body: unknown): Record<string, unknown> {
|
||||
if (body === undefined || body === null) return {};
|
||||
if (typeof body !== "object" || Array.isArray(body)) {
|
||||
throw badRequest("Import transfer apply body must be a JSON object.");
|
||||
}
|
||||
const { source: _ignoredSource, ...rest } = body as Record<string, unknown>;
|
||||
return rest;
|
||||
}
|
||||
|
||||
/**
|
||||
* Resolve the object to hand the preview/import zod schemas. For a JSON request
|
||||
* this is the inline body unchanged. For a multipart or application/zip request
|
||||
|
|
@ -161,6 +217,20 @@ async function resolveImportPayload(req: Request, res: Response): Promise<unknow
|
|||
if (!zipBytes || zipBytes.length === 0) {
|
||||
throw badRequest("Import package upload was empty.");
|
||||
}
|
||||
const archive = await readImportZipArchive(zipBytes);
|
||||
return {
|
||||
...parseImportMeta(metaRaw),
|
||||
source: { type: "inline", rootPath: archive.rootPath, files: archive.files },
|
||||
};
|
||||
}
|
||||
|
||||
/**
|
||||
* Read an uploaded import zip with the scaled bomb guards. Shared by the
|
||||
* single-shot upload path (`resolveImportPayload`) and the chunked transfer
|
||||
* apply path, so both feed the exact same `{ rootPath, files }` bundle into
|
||||
* the unchanged preview/import logic.
|
||||
*/
|
||||
async function readImportZipArchive(zipBytes: Buffer) {
|
||||
let archive: Awaited<ReturnType<typeof readZipArchive>>;
|
||||
try {
|
||||
// Scale the bomb guards from the configured upload cap so a legitimately
|
||||
|
|
@ -179,10 +249,7 @@ async function resolveImportPayload(req: Request, res: Response): Promise<unknow
|
|||
if (Object.keys(archive.files).length === 0) {
|
||||
throw badRequest("Import package contained no files.");
|
||||
}
|
||||
return {
|
||||
...parseImportMeta(metaRaw),
|
||||
source: { type: "inline", rootPath: archive.rootPath, files: archive.files },
|
||||
};
|
||||
return archive;
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -197,8 +264,15 @@ function wantsAsyncImport(req: Request) {
|
|||
return req.query.async === "1" || req.header("x-paperclip-cloud-async-import") === "1";
|
||||
}
|
||||
|
||||
export function companyRoutes(db: Db, storage?: StorageService) {
|
||||
export interface CompanyRoutesOptions {
|
||||
/** Overridable in tests; defaults to `<instance root>/import-transfers`. */
|
||||
importTransferSpoolRoot?: string;
|
||||
}
|
||||
|
||||
export function companyRoutes(db: Db, storage?: StorageService, options?: CompanyRoutesOptions) {
|
||||
const router = Router();
|
||||
const importTransferSpoolRoot =
|
||||
options?.importTransferSpoolRoot ?? resolveDefaultImportTransferSpoolRoot();
|
||||
const svc = companyService(db);
|
||||
const agents = agentService(db);
|
||||
const portability = companyPortabilityService(db, storage);
|
||||
|
|
@ -453,15 +527,53 @@ export function companyRoutes(db: Db, storage?: StorageService) {
|
|||
res.json(importJobResponse(job));
|
||||
});
|
||||
|
||||
router.post(COMPANY_IMPORT_ROUTE_PATH, async (req, res) => {
|
||||
assertBoard(req);
|
||||
// Resolve the request body up front: a JSON caller's inline body is used
|
||||
// unchanged; a multipart/application-zip caller's uploaded zip is read into
|
||||
// the same `{ source: { type: "inline", rootPath, files } }` bundle here,
|
||||
// fast and in-memory, so the async job machinery below is transport-agnostic.
|
||||
const rawImportBody: unknown = await resolveImportPayload(req, res);
|
||||
/**
|
||||
* Post-resolution import execution, shared by the single-shot upload route
|
||||
* and the chunked transfer apply route: validate the resolved body, run the
|
||||
* import synchronously or through the in-memory async job machinery, and
|
||||
* fire the optional completion hooks (the transfer path uses them to settle
|
||||
* its ledger run) on the import's real outcome in either mode.
|
||||
*/
|
||||
async function executeImportRequest(
|
||||
req: Request,
|
||||
res: Response,
|
||||
rawImportBody: unknown,
|
||||
hooks?: {
|
||||
onSuccess?: (result: CompanyPortabilityImportResult) => Promise<void>;
|
||||
/** Fires when the request is refused before any import work starts (e.g. another import is already running for the actor). */
|
||||
onConflict?: () => Promise<void>;
|
||||
onFailure?: (message: string) => Promise<void>;
|
||||
},
|
||||
) {
|
||||
const actor = getActorInfo(req);
|
||||
const boardUserId = req.actor.type === "board" ? req.actor.userId : null;
|
||||
const operation = async () => {
|
||||
try {
|
||||
const importBody = companyPortabilityImportSchema.parse(rawImportBody);
|
||||
assertImportTargetAccess(req, importBody.target);
|
||||
const activity = importedCompanyActivityContext(actor, importBody.include ?? null);
|
||||
const result = await portability.importBundle(importBody, boardUserId, {
|
||||
pauseAutomations: importBody.pauseAutomations === true,
|
||||
});
|
||||
// The import is committed. Settlement (hooks) runs before the
|
||||
// best-effort audit entry so a logging failure cannot make a
|
||||
// committed import read as failed — or, on the transfer path, release
|
||||
// the apply claim and invite a duplicate re-import.
|
||||
await hooks?.onSuccess?.(result);
|
||||
try {
|
||||
await logImportedCompanyActivity(db, activity, result);
|
||||
} catch (activityError) {
|
||||
logger.warn(
|
||||
{ err: activityError, companyId: result.company.id },
|
||||
"failed to write the company.imported activity entry for a committed import",
|
||||
);
|
||||
}
|
||||
return result;
|
||||
} catch (error) {
|
||||
await hooks?.onFailure?.(errorMessage(error));
|
||||
throw error;
|
||||
}
|
||||
};
|
||||
if (wantsAsyncImport(req)) {
|
||||
// Async job path. Two kinds of callers opt in:
|
||||
// - trusted Cloud tenants (original behavior, kept byte-identical),
|
||||
|
|
@ -485,6 +597,10 @@ export function companyRoutes(db: Db, storage?: StorageService) {
|
|||
// Terminal jobs never block a resubmit.
|
||||
const running = findRunningImportJob(importJobs, actorKey);
|
||||
if (running) {
|
||||
// No import work starts on this path, so a caller holding a claim
|
||||
// (the transfer apply) must be told to release it — otherwise the
|
||||
// refused transfer would sit in "applying" until a restart.
|
||||
await hooks?.onConflict?.();
|
||||
if (running.signature === signature) {
|
||||
res.status(409).json(importJobConflictResponse(running));
|
||||
} else {
|
||||
|
|
@ -499,16 +615,6 @@ export function companyRoutes(db: Db, storage?: StorageService) {
|
|||
signature,
|
||||
);
|
||||
importJobs.set(job.id, job);
|
||||
const operation = async () => {
|
||||
const importBody = companyPortabilityImportSchema.parse(rawImportBody);
|
||||
assertImportTargetAccess(req, importBody.target);
|
||||
const activity = importedCompanyActivityContext(actor, importBody.include ?? null);
|
||||
const result = await portability.importBundle(importBody, boardUserId, {
|
||||
pauseAutomations: importBody.pauseAutomations === true,
|
||||
});
|
||||
await logImportedCompanyActivity(db, activity, result);
|
||||
return result;
|
||||
};
|
||||
res.status(202).json(importJobAcceptedResponse(job));
|
||||
setImmediate(() => {
|
||||
void runImportJob(job, operation);
|
||||
|
|
@ -516,14 +622,337 @@ export function companyRoutes(db: Db, storage?: StorageService) {
|
|||
return;
|
||||
}
|
||||
|
||||
const importBody = companyPortabilityImportSchema.parse(rawImportBody);
|
||||
assertImportTargetAccess(req, importBody.target);
|
||||
const activity = importedCompanyActivityContext(actor, importBody.include ?? null);
|
||||
const result = await portability.importBundle(importBody, boardUserId, {
|
||||
pauseAutomations: importBody.pauseAutomations === true,
|
||||
});
|
||||
await logImportedCompanyActivity(db, activity, result);
|
||||
const result = await operation();
|
||||
res.json(result);
|
||||
}
|
||||
|
||||
router.post(COMPANY_IMPORT_ROUTE_PATH, async (req, res) => {
|
||||
assertBoard(req);
|
||||
// Resolve the request body up front: a JSON caller's inline body is used
|
||||
// unchanged; a multipart/application-zip caller's uploaded zip is read into
|
||||
// the same `{ source: { type: "inline", rootPath, files } }` bundle here,
|
||||
// fast and in-memory, so the async job machinery below is transport-agnostic.
|
||||
const rawImportBody: unknown = await resolveImportPayload(req, res);
|
||||
await executeImportRequest(req, res, rawImportBody);
|
||||
});
|
||||
|
||||
/**
|
||||
* Load the transfer run for the URL's id, scoped to the caller the same way
|
||||
* import jobs are (`importJobActorKey`): an unknown id, another actor's id,
|
||||
* and a non-import run are all the same 404, so transfer ids cannot be
|
||||
* probed across users or tenants. The id is regex-validated as a UUID before
|
||||
* it is used anywhere (it is later joined into spool paths).
|
||||
*/
|
||||
async function requireImportTransferRun(req: Request) {
|
||||
const transferId = req.params.transferId as string;
|
||||
if (!isImportTransferRunId(transferId)) {
|
||||
throw notFound("Import transfer not found");
|
||||
}
|
||||
const run = await companyTransferRunService.getRunForActor(db, transferId, importJobActorKey(req));
|
||||
if (!run || run.direction !== "import") {
|
||||
throw notFound("Import transfer not found");
|
||||
}
|
||||
return run;
|
||||
}
|
||||
|
||||
/** The declared part list persisted on the run at create time. */
|
||||
function storedImportTransferManifest(run: { manifest: unknown }): ImportTransferManifest {
|
||||
const parsed = importTransferManifestSchema.safeParse(run.manifest);
|
||||
if (!parsed.success) {
|
||||
throw conflict("Import transfer has no recorded part manifest");
|
||||
}
|
||||
return parsed.data;
|
||||
}
|
||||
|
||||
/**
|
||||
* Indexes the client still has to upload. A part counts as done only when
|
||||
* the ledger recorded it AND its spool file is still on disk at the declared
|
||||
* size — so a swept or damaged spool surfaces as missing parts to re-upload
|
||||
* instead of a failing apply.
|
||||
*/
|
||||
async function importTransferMissingParts(
|
||||
run: { id: string; completedParts: string[] },
|
||||
manifest: ImportTransferManifest,
|
||||
): Promise<number[]> {
|
||||
const completed = new Set(run.completedParts);
|
||||
const missing: number[] = [];
|
||||
for (const part of manifest.parts) {
|
||||
if (!completed.has(importTransferPartName(part.index))) {
|
||||
missing.push(part.index);
|
||||
continue;
|
||||
}
|
||||
const sizeOnDisk = await importTransferPartSizeOnDisk(importTransferSpoolRoot, run.id, part.index);
|
||||
if (sizeOnDisk !== part.byteSize) {
|
||||
missing.push(part.index);
|
||||
}
|
||||
}
|
||||
return missing;
|
||||
}
|
||||
|
||||
// Declare a chunked import transfer. The declaration is content-addressed:
|
||||
// the idempotency key derives from the whole-zip hash plus every part hash,
|
||||
// so re-declaring the same zip resumes the prior run (with its uploaded
|
||||
// parts intact) instead of starting over.
|
||||
router.post(COMPANY_IMPORT_TRANSFERS_ROUTE_PATH, async (req, res) => {
|
||||
assertBoard(req);
|
||||
const declared = importTransferManifestSchema.parse(req.body);
|
||||
if (declared.totalBytes > PORTABLE_ZIP_UPLOAD_LIMIT_BYTES) {
|
||||
throw unprocessable(
|
||||
`Import package exceeds the ${Math.floor(PORTABLE_ZIP_UPLOAD_LIMIT_BYTES / (1024 * 1024))} MB upload limit`,
|
||||
);
|
||||
}
|
||||
if (declared.partSizeBytes > IMPORT_TRANSFER_PART_SIZE_LIMIT_BYTES) {
|
||||
throw unprocessable(
|
||||
`Import transfer parts may be at most ${Math.floor(IMPORT_TRANSFER_PART_SIZE_LIMIT_BYTES / (1024 * 1024))} MB`,
|
||||
);
|
||||
}
|
||||
declared.parts.forEach((part, position) => {
|
||||
if (part.index !== position) {
|
||||
throw unprocessable("Import transfer parts must be contiguous, ordered 0..n-1");
|
||||
}
|
||||
if (part.byteSize > declared.partSizeBytes) {
|
||||
throw unprocessable(`Part ${part.index} is larger than the declared partSizeBytes`);
|
||||
}
|
||||
});
|
||||
const declaredTotal = declared.parts.reduce((sum, part) => sum + part.byteSize, 0);
|
||||
if (declaredTotal !== declared.totalBytes) {
|
||||
throw unprocessable("Sum of part byte sizes must equal totalBytes");
|
||||
}
|
||||
|
||||
const idempotencyKey = createHash("sha256")
|
||||
.update(JSON.stringify([declared.zipSha256, declared.parts.map((part) => part.sha256)]))
|
||||
.digest("hex");
|
||||
const { run, alreadyCompleted } = await companyTransferRunService.resumeOrCreate(db, {
|
||||
direction: "import",
|
||||
actorKey: importJobActorKey(req),
|
||||
idempotencyKey,
|
||||
containerRef: { kind: "chunked_zip_upload" },
|
||||
});
|
||||
if (alreadyCompleted) {
|
||||
res.json({
|
||||
transferId: run.id,
|
||||
status: "completed",
|
||||
alreadyCompleted: true,
|
||||
totalParts: declared.parts.length,
|
||||
missingParts: [],
|
||||
} satisfies CompanyImportTransferCreated);
|
||||
return;
|
||||
}
|
||||
if (run.manifest === null || run.manifest === undefined) {
|
||||
const manifestSha256 = createHash("sha256").update(JSON.stringify(declared)).digest("hex");
|
||||
await companyTransferRunService.recordManifest(db, run.id, declared, manifestSha256, {
|
||||
chunkCount: declared.parts.length,
|
||||
blobCount: 0,
|
||||
});
|
||||
}
|
||||
await companyTransferRunService.start(db, run.id);
|
||||
res.json({
|
||||
transferId: run.id,
|
||||
status: "running",
|
||||
alreadyCompleted: false,
|
||||
totalParts: declared.parts.length,
|
||||
missingParts: await importTransferMissingParts(run, declared),
|
||||
} satisfies CompanyImportTransferCreated);
|
||||
});
|
||||
|
||||
// Upload one declared part as a raw body. Verified against the declared
|
||||
// size and sha256 before it is spooled; re-uploading an already completed
|
||||
// part is a no-op success, so clients can blindly retry.
|
||||
router.put(
|
||||
`${COMPANY_IMPORT_TRANSFERS_ROUTE_PATH}/:transferId/parts/:partIndex`,
|
||||
importTransferPartBodyParser,
|
||||
async (req, res) => {
|
||||
assertBoard(req);
|
||||
const run = await requireImportTransferRun(req);
|
||||
if (run.status === "completed") {
|
||||
throw conflict("Import transfer has already been applied");
|
||||
}
|
||||
if (run.status === "applying") {
|
||||
// An apply is assembling the spool right now; accepting writes would
|
||||
// let a retried upload race the files being read. Parts are frozen
|
||||
// until the apply settles (completed short-circuits, failed resumes).
|
||||
throw conflict("Import transfer apply is in progress; retry after it settles");
|
||||
}
|
||||
const manifest = storedImportTransferManifest(run);
|
||||
const rawIndex = req.params.partIndex as string;
|
||||
// Strict integer-in-bounds validation before the index goes anywhere
|
||||
// near a filesystem path.
|
||||
if (!/^\d{1,5}$/.test(rawIndex) || Number(rawIndex) >= manifest.parts.length) {
|
||||
throw notFound("Import transfer part not found");
|
||||
}
|
||||
const partIndex = Number(rawIndex);
|
||||
const declared = manifest.parts[partIndex]!;
|
||||
const partName = importTransferPartName(partIndex);
|
||||
if (run.completedParts.includes(partName)) {
|
||||
const sizeOnDisk = await importTransferPartSizeOnDisk(importTransferSpoolRoot, run.id, partIndex);
|
||||
if (sizeOnDisk === declared.byteSize) {
|
||||
res.json({ ok: true, index: partIndex, alreadyCompleted: true } satisfies CompanyImportTransferPartUploadResult);
|
||||
return;
|
||||
}
|
||||
}
|
||||
const bytes = Buffer.isBuffer(req.body) ? req.body : Buffer.alloc(0);
|
||||
if (bytes.length !== declared.byteSize) {
|
||||
throw unprocessable(
|
||||
`Part ${partIndex} upload is ${bytes.length} bytes but was declared as ${declared.byteSize} bytes`,
|
||||
);
|
||||
}
|
||||
const digest = createHash("sha256").update(bytes).digest("hex");
|
||||
if (digest !== declared.sha256) {
|
||||
throw unprocessable(`Part ${partIndex} content does not match its declared sha256`);
|
||||
}
|
||||
await writeImportTransferPart(importTransferSpoolRoot, run.id, partIndex, bytes);
|
||||
const recorded = await companyTransferRunService.completePart(db, run.id, partName);
|
||||
if (!recorded) {
|
||||
// The run left pending/running between the load above and this
|
||||
// record. Which way it went decides the file's fate: under a claimed
|
||||
// or completed apply the just-written bytes are hash-verified
|
||||
// identical to what the apply assembles, so the file must NOT be
|
||||
// deleted out from under it; a swept or failed run is dead, so the
|
||||
// stray part is dropped and the client told to re-create.
|
||||
const current = await companyTransferRunService.getRun(db, run.id);
|
||||
if (current?.status === "applying" || current?.status === "completed") {
|
||||
throw conflict("Import transfer apply is in progress; retry after it settles");
|
||||
}
|
||||
await removeImportTransferPart(importTransferSpoolRoot, run.id, partIndex);
|
||||
res.status(410).json({ error: "Import transfer expired — re-create it" });
|
||||
return;
|
||||
}
|
||||
res.json({ ok: true, index: partIndex, alreadyCompleted: false } satisfies CompanyImportTransferPartUploadResult);
|
||||
},
|
||||
);
|
||||
|
||||
// Resume polling: which parts are done, which are still missing.
|
||||
router.get(`${COMPANY_IMPORT_TRANSFERS_ROUTE_PATH}/:transferId`, async (req, res) => {
|
||||
assertBoard(req);
|
||||
const run = await requireImportTransferRun(req);
|
||||
const manifest = storedImportTransferManifest(run);
|
||||
const missingParts = run.status === "completed" ? [] : await importTransferMissingParts(run, manifest);
|
||||
res.json({
|
||||
transferId: run.id,
|
||||
status: run.status,
|
||||
totalParts: manifest.parts.length,
|
||||
completedParts: manifest.parts.length - missingParts.length,
|
||||
missingParts,
|
||||
} 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);
|
||||
const run = await requireImportTransferRun(req);
|
||||
if (run.status === "completed") {
|
||||
throw conflict("Import transfer has already been applied");
|
||||
}
|
||||
const manifest = storedImportTransferManifest(run);
|
||||
const missingParts = await importTransferMissingParts(run, manifest);
|
||||
if (missingParts.length > 0) {
|
||||
res.status(409).json({
|
||||
error: "Import transfer is missing parts",
|
||||
missingParts,
|
||||
});
|
||||
return;
|
||||
}
|
||||
// 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))) {
|
||||
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 },
|
||||
};
|
||||
await executeImportRequest(req, res, rawImportBody, {
|
||||
onSuccess: async (result) => {
|
||||
// Settle the run the moment the import is committed: every later
|
||||
// step is best-effort, because any error escaping this hook reads
|
||||
// as a failed apply and — before complete() — would release the
|
||||
// claim on an already-committed import. complete() gets one retry;
|
||||
// if both attempts fail the run is left "applying" (unclaimable, so
|
||||
// no duplicate import) and logged for the operator; a restart's
|
||||
// stranded-run recovery then makes it inspectable as failed.
|
||||
try {
|
||||
await companyTransferRunService.complete(db, run.id);
|
||||
} catch {
|
||||
try {
|
||||
await companyTransferRunService.complete(db, run.id);
|
||||
} catch (settleError) {
|
||||
logger.error(
|
||||
{ err: settleError, transferId: run.id },
|
||||
"import committed but the transfer run could not be marked completed",
|
||||
);
|
||||
return;
|
||||
}
|
||||
}
|
||||
try {
|
||||
await companyTransferRunService.attachCompany(db, run.id, result.company.id);
|
||||
} catch (attachError) {
|
||||
logger.warn(
|
||||
{ err: attachError, transferId: run.id },
|
||||
"failed to attach the imported company to its transfer run",
|
||||
);
|
||||
}
|
||||
// Best-effort cleanup: the import is committed and the run is
|
||||
// completed, so a spool deletion error must not escape — it would
|
||||
// reach the failure paths below and read as a failed apply. The
|
||||
// leftover dir is harmless (completed runs never assemble again)
|
||||
// and the sweep collects terminal-run spools immediately.
|
||||
try {
|
||||
await removeImportTransferSpool(importTransferSpoolRoot, run.id);
|
||||
} catch (cleanupError) {
|
||||
logger.warn(
|
||||
{ err: cleanupError, transferId: run.id },
|
||||
"failed to delete import transfer spool after a successful apply",
|
||||
);
|
||||
}
|
||||
},
|
||||
onFailure: async (message) => {
|
||||
// Parts remain spooled: the failure is in the import itself, not
|
||||
// the upload, so a retry of apply can reuse the verified parts.
|
||||
// Guarded release, not a blind fail: this hook also fires when an
|
||||
// error escapes onSuccess above, i.e. possibly after complete(),
|
||||
// and a completed run must never be flipped back to claimable.
|
||||
await companyTransferRunService.releaseApplyClaim(db, run.id, message);
|
||||
},
|
||||
onConflict: async () => {
|
||||
// The request was refused before any import work (another import is
|
||||
// running for this actor): release the claim so the transfer stays
|
||||
// retryable instead of stranded in "applying" until a restart.
|
||||
await companyTransferRunService.releaseApplyClaim(
|
||||
db,
|
||||
run.id,
|
||||
"another import is already running for this actor",
|
||||
);
|
||||
},
|
||||
});
|
||||
} 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.
|
||||
await companyTransferRunService.releaseApplyClaim(db, run.id, errorMessage(error));
|
||||
throw error;
|
||||
}
|
||||
});
|
||||
|
||||
router.post("/:companyId/exports/preview", async (req, res) => {
|
||||
|
|
|
|||
|
|
@ -225,6 +225,10 @@ import {
|
|||
toolPolicyTestRequestSchema,
|
||||
createToolMcpGatewaySchema,
|
||||
} from "@paperclipai/shared";
|
||||
import {
|
||||
COMPANY_IMPORT_TRANSFERS_API_PATH,
|
||||
companyImportTransferDeclarationSchema,
|
||||
} from "@paperclipai/shared/company-import-transfer";
|
||||
|
||||
type JsonSchema = Record<string, unknown>;
|
||||
type OpenApiResponse = Record<string, unknown>;
|
||||
|
|
@ -5946,6 +5950,94 @@ registry.registerPath({
|
|||
},
|
||||
});
|
||||
|
||||
registry.registerPath({
|
||||
method: "post",
|
||||
path: COMPANY_IMPORT_TRANSFERS_API_PATH,
|
||||
tags: ["companies"],
|
||||
summary: "Declare a chunked company import transfer",
|
||||
description:
|
||||
"Declares the caller's existing company package .zip as content-addressed byte-range parts " +
|
||||
"(whole-file plus per-part sha256). Re-declaring the same zip resumes the prior transfer " +
|
||||
"with its uploaded parts intact; the response carries the transfer id and the part indexes " +
|
||||
"still missing.",
|
||||
request: { body: jsonBody(companyImportTransferDeclarationSchema) },
|
||||
responses: { 200: r.ok(), 400: r.badRequest, 401: r.unauthorized, 422: r.unprocessable },
|
||||
});
|
||||
|
||||
registry.registerPath({
|
||||
method: "put",
|
||||
path: `${COMPANY_IMPORT_TRANSFERS_API_PATH}/{transferId}/parts/{partIndex}`,
|
||||
tags: ["companies"],
|
||||
summary: "Upload one declared part of a company import transfer",
|
||||
description:
|
||||
"Raw part bytes as the request body. The upload is verified against the declared byte size " +
|
||||
"and sha256 before it is spooled; re-uploading an already completed part is a no-op success. " +
|
||||
"Uploading against a transfer the abandoned-spool sweep has expired returns 410 — the client " +
|
||||
"re-creates the transfer.",
|
||||
request: {
|
||||
params: z.object({ transferId: z.string(), partIndex: z.string() }),
|
||||
body: {
|
||||
content: {
|
||||
"application/octet-stream": {
|
||||
schema: { type: "string", format: "binary", description: "The raw part bytes." },
|
||||
},
|
||||
},
|
||||
required: true as const,
|
||||
},
|
||||
},
|
||||
responses: {
|
||||
200: r.ok(),
|
||||
401: r.unauthorized,
|
||||
404: r.notFound,
|
||||
409: { description: "The transfer has already been applied" },
|
||||
410: { description: "The transfer expired and its spooled parts were deleted" },
|
||||
422: r.unprocessable,
|
||||
},
|
||||
});
|
||||
|
||||
registry.registerPath({
|
||||
method: "get",
|
||||
path: `${COMPANY_IMPORT_TRANSFERS_API_PATH}/{transferId}`,
|
||||
tags: ["companies"],
|
||||
summary: "Get company import transfer progress",
|
||||
description:
|
||||
"Resume polling for a chunked import transfer: the transfer status plus which declared " +
|
||||
"parts are completed and which are still missing.",
|
||||
request: { params: z.object({ transferId: z.string() }) },
|
||||
responses: { 200: r.ok(), 401: r.unauthorized, 404: r.notFound },
|
||||
});
|
||||
|
||||
registry.registerPath({
|
||||
method: "post",
|
||||
path: `${COMPANY_IMPORT_TRANSFERS_API_PATH}/{transferId}/apply`,
|
||||
tags: ["companies"],
|
||||
summary: "Apply a completed company import transfer",
|
||||
description:
|
||||
"Assembles the spooled parts back into the original zip, verifies the whole file against " +
|
||||
"the declared hash fail-closed, and runs it through the same import pipeline as the " +
|
||||
"single-shot upload — including the async import job machinery via the proxy-safe " +
|
||||
"`?async=1` query parameter. The JSON body carries the same import fields as the multipart " +
|
||||
"route's `meta` field (include, target, collisionStrategy, ...). Overlapping applies of " +
|
||||
"the same transfer are serialized: exactly one proceeds, the rest get 409.",
|
||||
request: {
|
||||
params: z.object({ transferId: z.string() }),
|
||||
query: z.object({ async: z.enum(["1"]).optional() }),
|
||||
body: jsonBody(companyPortabilityImportSchema.omit({ source: true })),
|
||||
},
|
||||
responses: {
|
||||
200: r.ok(),
|
||||
202: { description: "Async import job accepted" },
|
||||
400: r.badRequest,
|
||||
401: r.unauthorized,
|
||||
404: r.notFound,
|
||||
409: {
|
||||
description:
|
||||
"Parts are still missing, an apply is already in progress, or the transfer was already applied",
|
||||
},
|
||||
422: r.unprocessable,
|
||||
},
|
||||
});
|
||||
|
||||
// ─── Board claim & CLI auth ───────────────────────────────────────────────────
|
||||
|
||||
registry.registerPath({
|
||||
|
|
|
|||
|
|
@ -0,0 +1,183 @@
|
|||
import { promises as fs } from "node:fs";
|
||||
import path from "node:path";
|
||||
import type { Db } from "@paperclipai/db";
|
||||
import { resolvePaperclipInstanceRoot } from "../home-paths.js";
|
||||
import { companyTransferRunService } from "./company-transfer-runs.js";
|
||||
|
||||
// Disk spool for chunked resumable company imports. The client slices its
|
||||
// existing import .zip into byte-range parts and uploads them one at a time;
|
||||
// each verified part is spooled at
|
||||
//
|
||||
// <instance root>/import-transfers/<runId>/part-<index>
|
||||
//
|
||||
// until apply assembles them back into the original zip. The layout follows
|
||||
// the other per-instance state dirs that live directly under the instance root
|
||||
// (telemetry/, runtime-services/, skills/, data/run-logs/); like those
|
||||
// siblings it needs no backup/export exclusion wiring — database backups dump
|
||||
// embedded Postgres, they do not walk the instance root tree.
|
||||
|
||||
/** Spool dirs with no upload/apply activity for this long are abandoned. */
|
||||
export const IMPORT_TRANSFER_SPOOL_MAX_AGE_MS = 24 * 60 * 60 * 1000;
|
||||
/** How often the abandoned-spool sweep runs. */
|
||||
export const IMPORT_TRANSFER_SPOOL_SWEEP_INTERVAL_MS = 60 * 60 * 1000;
|
||||
|
||||
const UUID_RE = /^[0-9a-f]{8}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{4}-[0-9a-f]{12}$/i;
|
||||
|
||||
export function resolveDefaultImportTransferSpoolRoot(): string {
|
||||
return path.resolve(resolvePaperclipInstanceRoot(), "import-transfers");
|
||||
}
|
||||
|
||||
/**
|
||||
* Transfer run ids come from request URLs and are joined into filesystem
|
||||
* paths, so they are accepted only as canonical UUIDs — anything else (path
|
||||
* separators, dots, empty segments) is rejected before any path is built.
|
||||
*/
|
||||
export function isImportTransferRunId(value: string): boolean {
|
||||
return UUID_RE.test(value);
|
||||
}
|
||||
|
||||
export function importTransferPartName(index: number): string {
|
||||
if (!Number.isInteger(index) || index < 0) {
|
||||
throw new Error(`Invalid import transfer part index: ${String(index)}`);
|
||||
}
|
||||
return `part-${index}`;
|
||||
}
|
||||
|
||||
function spoolDirFor(spoolRoot: string, runId: string): string {
|
||||
if (!isImportTransferRunId(runId)) {
|
||||
throw new Error("Invalid import transfer run id");
|
||||
}
|
||||
return path.join(spoolRoot, runId);
|
||||
}
|
||||
|
||||
function partPathFor(spoolRoot: string, runId: string, index: number): string {
|
||||
return path.join(spoolDirFor(spoolRoot, runId), importTransferPartName(index));
|
||||
}
|
||||
|
||||
/** Atomic part write: temp file in the same dir, then rename over the target. */
|
||||
export async function writeImportTransferPart(
|
||||
spoolRoot: string,
|
||||
runId: string,
|
||||
index: number,
|
||||
bytes: Buffer,
|
||||
): Promise<void> {
|
||||
const target = partPathFor(spoolRoot, runId, index);
|
||||
await fs.mkdir(path.dirname(target), { recursive: true });
|
||||
const tempPath = `${target}.tmp-${Date.now()}-${Math.random().toString(36).slice(2, 8)}`;
|
||||
await fs.writeFile(tempPath, bytes);
|
||||
await fs.rename(tempPath, target);
|
||||
}
|
||||
|
||||
/** Byte size of a spooled part, or null when it is missing (or not a file). */
|
||||
export async function importTransferPartSizeOnDisk(
|
||||
spoolRoot: string,
|
||||
runId: string,
|
||||
index: number,
|
||||
): Promise<number | null> {
|
||||
try {
|
||||
const stat = await fs.stat(partPathFor(spoolRoot, runId, index));
|
||||
return stat.isFile() ? stat.size : null;
|
||||
} catch {
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Concatenate the spooled parts, in index order, back into the original zip.
|
||||
* The full-zip buffer this produces exists only for the duration of an apply —
|
||||
* the same memory profile as the single-shot zip upload path.
|
||||
*/
|
||||
export async function assembleImportTransferZip(
|
||||
spoolRoot: string,
|
||||
runId: string,
|
||||
partCount: number,
|
||||
): Promise<Buffer> {
|
||||
const buffers: Buffer[] = [];
|
||||
for (let index = 0; index < partCount; index += 1) {
|
||||
buffers.push(await fs.readFile(partPathFor(spoolRoot, runId, index)));
|
||||
}
|
||||
return Buffer.concat(buffers);
|
||||
}
|
||||
|
||||
export async function removeImportTransferSpool(spoolRoot: string, runId: string): Promise<void> {
|
||||
await fs.rm(spoolDirFor(spoolRoot, runId), { recursive: true, force: true });
|
||||
}
|
||||
|
||||
/**
|
||||
* Remove one spooled part file. Used to discard a stray part written for a
|
||||
* run that turned out to be dead (e.g. expired by the sweep mid-upload).
|
||||
*/
|
||||
export async function removeImportTransferPart(
|
||||
spoolRoot: string,
|
||||
runId: string,
|
||||
index: number,
|
||||
): Promise<void> {
|
||||
await fs.rm(partPathFor(spoolRoot, runId, index), { force: true });
|
||||
}
|
||||
|
||||
/**
|
||||
* Delete spool dirs whose transfer saw no activity for `maxAgeMs` and cancel
|
||||
* their still-open ledger runs. A run-backed spool is only deleted after an
|
||||
* atomic claim: one guarded UPDATE cancels the run while it is still open
|
||||
* (pending/running) AND its `updatedAt` (every part upload and resume bumps
|
||||
* it) is still older than the cutoff — so a transfer that resumed after this
|
||||
* sweep read the run makes the guard miss and keeps its spool. The claim
|
||||
* cancels rather than fails on purpose: a failed run would still be claimable
|
||||
* by `claimApply`, letting an apply assemble the spool while this sweep
|
||||
* deletes it; a cancelled run is unclaimable the instant the UPDATE commits,
|
||||
* before any file is removed. Terminal runs keep their status and their spool
|
||||
* (a failed run's verified parts stay reusable for an apply retry; a
|
||||
* completed run's spool was already removed at apply). Orphan dirs with no
|
||||
* run row fall back to the dir's mtime — no run means no upload can race the
|
||||
* delete. A swept run is never resumed: `resumeOrCreate` ignores cancelled
|
||||
* runs, so redeclaring the same content starts a fresh run with every part
|
||||
* missing — matching the 410 "re-create it" a part upload against the swept
|
||||
* run gets.
|
||||
*/
|
||||
export async function sweepAbandonedImportTransferSpools(
|
||||
db: Db,
|
||||
spoolRoot: string,
|
||||
options: { maxAgeMs?: number; now?: Date } = {},
|
||||
): Promise<{ swept: number }> {
|
||||
const maxAgeMs = options.maxAgeMs ?? IMPORT_TRANSFER_SPOOL_MAX_AGE_MS;
|
||||
const nowMs = (options.now ?? new Date()).getTime();
|
||||
const cutoff = new Date(nowMs - maxAgeMs);
|
||||
let entries;
|
||||
try {
|
||||
entries = await fs.readdir(spoolRoot, { withFileTypes: true });
|
||||
} catch {
|
||||
return { swept: 0 };
|
||||
}
|
||||
let swept = 0;
|
||||
for (const entry of entries) {
|
||||
if (!entry.isDirectory() || !isImportTransferRunId(entry.name)) continue;
|
||||
const runId = entry.name;
|
||||
const run = await companyTransferRunService.getRun(db, runId).catch(() => null);
|
||||
if (run) {
|
||||
if (run.status === "completed" || run.status === "cancelled") {
|
||||
// Terminal runs never assemble again; a leftover spool (e.g. a
|
||||
// best-effort post-apply cleanup that failed) is pure garbage and is
|
||||
// collected immediately, regardless of idle time.
|
||||
} else {
|
||||
const claimed = await companyTransferRunService.cancelIfIdle(
|
||||
db,
|
||||
runId,
|
||||
cutoff,
|
||||
"Abandoned import transfer: spooled parts were deleted after prolonged inactivity.",
|
||||
);
|
||||
if (!claimed) continue;
|
||||
}
|
||||
} else {
|
||||
let mtimeMs: number;
|
||||
try {
|
||||
mtimeMs = (await fs.stat(path.join(spoolRoot, runId))).mtimeMs;
|
||||
} catch {
|
||||
continue;
|
||||
}
|
||||
if (nowMs - mtimeMs <= maxAgeMs) continue;
|
||||
}
|
||||
await fs.rm(path.join(spoolRoot, runId), { recursive: true, force: true });
|
||||
swept += 1;
|
||||
}
|
||||
return { swept };
|
||||
}
|
||||
|
|
@ -0,0 +1,375 @@
|
|||
import { and, desc, eq, inArray, lt, sql } from "drizzle-orm";
|
||||
import type { Db } from "@paperclipai/db";
|
||||
import { companyTransferRuns } from "@paperclipai/db";
|
||||
|
||||
// Durable run ledger for chunked company transfers. A run tracks one export
|
||||
// publish or import apply of a transfer container. Progress is recorded per
|
||||
// container part (chunk or blob name) as each part finishes processing and
|
||||
// verification, so an interrupted run — process restart, dropped connection,
|
||||
// failed part — resumes from the parts it already completed. The transfer
|
||||
// manifest's content-derived idempotency key ties retries of the same content
|
||||
// to the same run instead of starting over.
|
||||
//
|
||||
// Status machine — every transition site, in one place:
|
||||
//
|
||||
// (insert) -> pending resumeOrCreate
|
||||
// pending|running|failed --start---------> running
|
||||
// pending|running|failed --claimApply----> applying exactly one winner of concurrent applies
|
||||
// applying --complete------> completed terminal; no path ever reopens it
|
||||
// applying --releaseApplyClaim-> failed guarded on "applying", so a cleanup error
|
||||
// after complete() is a no-op release
|
||||
// applying --recoverStrandedApplyingRuns-> failed
|
||||
// startup only: in-memory applies died with the process
|
||||
// pending|running (idle) --cancelIfIdle--> cancelled sweep claim; also guarded on a stale updatedAt
|
||||
// pending|running|failed --cancel--------> cancelled
|
||||
// (any) --fail----------> failed unguarded settle; callers use it only on runs
|
||||
// they know are open (else releaseApplyClaim)
|
||||
//
|
||||
// failed: retryable — in the resume lookup and claimable by claimApply.
|
||||
// completed: terminal — resumeOrCreate short-circuits with alreadyCompleted.
|
||||
// cancelled: terminal and unclaimable — excluded from the resume lookup, so
|
||||
// redeclaring the same content after a sweep starts a fresh run.
|
||||
// completePart records parts only under pending|running — never onto an
|
||||
// applying or terminal run.
|
||||
|
||||
export type CompanyTransferDirection = "export" | "import";
|
||||
export type CompanyTransferRunStatus =
|
||||
| "pending"
|
||||
| "running"
|
||||
| "applying"
|
||||
| "completed"
|
||||
| "failed"
|
||||
| "cancelled";
|
||||
|
||||
/**
|
||||
* Minimal structural view of a transfer manifest's part index: the ledger only
|
||||
* ever needs part names and counts, so it stays decoupled from any particular
|
||||
* container format. Manifests without this shape (e.g. the chunked zip-upload
|
||||
* manifest, which carries its own part list) pass explicit counts to
|
||||
* `recordManifest` and track parts by their own names.
|
||||
*/
|
||||
export interface CompanyTransferPartsIndex {
|
||||
chunks: Array<{ name: string }>;
|
||||
blobs: Array<{ name: string }>;
|
||||
}
|
||||
|
||||
function asPartsIndex(manifest: unknown): CompanyTransferPartsIndex | null {
|
||||
if (typeof manifest !== "object" || manifest === null) return null;
|
||||
const { chunks, blobs } = manifest as { chunks?: unknown; blobs?: unknown };
|
||||
const named = (value: unknown): value is Array<{ name: string }> =>
|
||||
Array.isArray(value) &&
|
||||
value.every(
|
||||
(entry) =>
|
||||
typeof entry === "object" &&
|
||||
entry !== null &&
|
||||
typeof (entry as { name?: unknown }).name === "string",
|
||||
);
|
||||
if (!named(chunks) || !named(blobs)) return null;
|
||||
return { chunks, blobs };
|
||||
}
|
||||
|
||||
const RESUMABLE_STATUSES: CompanyTransferRunStatus[] = ["pending", "running", "failed"];
|
||||
/** Statuses under which container parts may still be recorded as completed. */
|
||||
const PART_UPLOAD_STATUSES: CompanyTransferRunStatus[] = ["pending", "running"];
|
||||
|
||||
export interface CompanyTransferRunRow {
|
||||
id: string;
|
||||
companyId: string | null;
|
||||
direction: CompanyTransferDirection;
|
||||
status: CompanyTransferRunStatus;
|
||||
actorKey: string;
|
||||
containerRef: unknown;
|
||||
idempotencyKey: string;
|
||||
manifestSha256: string | null;
|
||||
manifest: unknown;
|
||||
chunkCount: number;
|
||||
blobCount: number;
|
||||
completedParts: string[];
|
||||
error: string | null;
|
||||
startedAt: Date | null;
|
||||
finishedAt: Date | null;
|
||||
createdAt: Date;
|
||||
updatedAt: Date;
|
||||
}
|
||||
|
||||
export interface ResumeOrCreateTransferRunInput {
|
||||
direction: CompanyTransferDirection;
|
||||
actorKey: string;
|
||||
idempotencyKey: string;
|
||||
/** Credential-free container location descriptor. Never put secrets here. */
|
||||
containerRef: unknown;
|
||||
companyId?: string | null;
|
||||
}
|
||||
|
||||
export interface ResumeOrCreateTransferRunResult {
|
||||
run: CompanyTransferRunRow;
|
||||
/** True when an existing run for the same content was picked up. */
|
||||
resumed: boolean;
|
||||
/** True when that existing run already finished — the caller can short-circuit. */
|
||||
alreadyCompleted: boolean;
|
||||
}
|
||||
|
||||
function rowFromRecord(record: typeof companyTransferRuns.$inferSelect): CompanyTransferRunRow {
|
||||
return {
|
||||
...record,
|
||||
direction: record.direction as CompanyTransferDirection,
|
||||
status: record.status as CompanyTransferRunStatus,
|
||||
manifest: record.manifest ?? null,
|
||||
completedParts: Array.isArray(record.completedParts) ? (record.completedParts as string[]) : [],
|
||||
};
|
||||
}
|
||||
|
||||
async function getRun(db: Db, runId: string): Promise<CompanyTransferRunRow | null> {
|
||||
const [record] = await db
|
||||
.select()
|
||||
.from(companyTransferRuns)
|
||||
.where(eq(companyTransferRuns.id, runId))
|
||||
.limit(1);
|
||||
return record ? rowFromRecord(record) : null;
|
||||
}
|
||||
|
||||
export const companyTransferRunService = {
|
||||
getRun,
|
||||
|
||||
async getRunForActor(db: Db, runId: string, actorKey: string): Promise<CompanyTransferRunRow | null> {
|
||||
const run = await getRun(db, runId);
|
||||
return run && run.actorKey === actorKey ? run : null;
|
||||
},
|
||||
|
||||
/**
|
||||
* Find the actor's run for this exact content (same idempotency key and
|
||||
* direction) or create a fresh one. A completed prior run is returned with
|
||||
* `alreadyCompleted` so the caller can skip the apply outright; a pending,
|
||||
* running, failed, or mid-apply prior run is resumed with its part progress
|
||||
* intact (an "applying" run resumes so a concurrent re-declaration never
|
||||
* spawns a duplicate, but `start` leaves it alone until the apply settles).
|
||||
*/
|
||||
async resumeOrCreate(db: Db, input: ResumeOrCreateTransferRunInput): Promise<ResumeOrCreateTransferRunResult> {
|
||||
return db.transaction(async (tx) => {
|
||||
const txDb = tx as unknown as Db;
|
||||
// No unique index backs the (actor, direction, content key) lookup, so
|
||||
// two concurrent identical declarations could both miss the lookup and
|
||||
// both insert. Serialize them with a transaction-scoped advisory lock
|
||||
// on the key: the second caller waits here until the first commits and
|
||||
// then resumes the first caller's row.
|
||||
const lockKey = `${input.actorKey}|${input.direction}|${input.idempotencyKey}`;
|
||||
await txDb.execute(sql`select pg_advisory_xact_lock(hashtextextended(${lockKey}, 0))`);
|
||||
const matches = await txDb
|
||||
.select()
|
||||
.from(companyTransferRuns)
|
||||
.where(
|
||||
and(
|
||||
eq(companyTransferRuns.idempotencyKey, input.idempotencyKey),
|
||||
eq(companyTransferRuns.direction, input.direction),
|
||||
eq(companyTransferRuns.actorKey, input.actorKey),
|
||||
inArray(companyTransferRuns.status, [...RESUMABLE_STATUSES, "applying", "completed"]),
|
||||
),
|
||||
)
|
||||
.orderBy(desc(companyTransferRuns.createdAt))
|
||||
.limit(1);
|
||||
const existing = matches[0] ? rowFromRecord(matches[0]) : null;
|
||||
if (existing) {
|
||||
return { run: existing, resumed: true, alreadyCompleted: existing.status === "completed" };
|
||||
}
|
||||
const [created] = await txDb
|
||||
.insert(companyTransferRuns)
|
||||
.values({
|
||||
direction: input.direction,
|
||||
actorKey: input.actorKey,
|
||||
idempotencyKey: input.idempotencyKey,
|
||||
containerRef: input.containerRef,
|
||||
companyId: input.companyId ?? null,
|
||||
})
|
||||
.returning();
|
||||
return { run: rowFromRecord(created!), resumed: false, alreadyCompleted: false };
|
||||
});
|
||||
},
|
||||
|
||||
/**
|
||||
* Persist the parsed manifest so resume can re-verify parts without
|
||||
* refetching it. Counts default to the manifest's chunk/blob index when it
|
||||
* has one; manifests with a different part layout pass them explicitly.
|
||||
*/
|
||||
async recordManifest(
|
||||
db: Db,
|
||||
runId: string,
|
||||
manifest: unknown,
|
||||
manifestSha256: string,
|
||||
counts?: { chunkCount: number; blobCount: number },
|
||||
): Promise<void> {
|
||||
const index = asPartsIndex(manifest);
|
||||
const chunkCount = counts?.chunkCount ?? index?.chunks.length ?? 0;
|
||||
const blobCount = counts?.blobCount ?? index?.blobs.length ?? 0;
|
||||
await db
|
||||
.update(companyTransferRuns)
|
||||
.set({
|
||||
manifest,
|
||||
manifestSha256,
|
||||
chunkCount,
|
||||
blobCount,
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(eq(companyTransferRuns.id, runId));
|
||||
},
|
||||
|
||||
/**
|
||||
* Atomically claim a run for apply. The apply itself is not idempotent, so
|
||||
* of any overlapping apply attempts exactly one caller gets `true` here —
|
||||
* a single guarded UPDATE moves the run to "applying" only while it is
|
||||
* still open (pending/running/failed). The winner settles the claim with
|
||||
* `complete` on success or `releaseApplyClaim`/`fail` on error (which makes
|
||||
* the run retryable again); every concurrent loser gets `false` and must
|
||||
* not run the apply.
|
||||
*/
|
||||
async claimApply(db: Db, runId: string): Promise<boolean> {
|
||||
const claimed = await db
|
||||
.update(companyTransferRuns)
|
||||
.set({ status: "applying", error: null, updatedAt: new Date() })
|
||||
.where(
|
||||
and(eq(companyTransferRuns.id, runId), inArray(companyTransferRuns.status, RESUMABLE_STATUSES)),
|
||||
)
|
||||
.returning({ id: companyTransferRuns.id });
|
||||
return claimed.length > 0;
|
||||
},
|
||||
|
||||
async start(db: Db, runId: string): Promise<void> {
|
||||
await db
|
||||
.update(companyTransferRuns)
|
||||
.set({ status: "running", error: null, startedAt: new Date(), finishedAt: null, updatedAt: new Date() })
|
||||
.where(
|
||||
and(eq(companyTransferRuns.id, runId), inArray(companyTransferRuns.status, RESUMABLE_STATUSES)),
|
||||
);
|
||||
},
|
||||
|
||||
async attachCompany(db: Db, runId: string, companyId: string): Promise<void> {
|
||||
await db
|
||||
.update(companyTransferRuns)
|
||||
.set({ companyId, updatedAt: new Date() })
|
||||
.where(eq(companyTransferRuns.id, runId));
|
||||
},
|
||||
|
||||
/**
|
||||
* Record one finished container part. Appends atomically in SQL and is
|
||||
* idempotent: re-completing an already recorded part leaves the ledger
|
||||
* unchanged, so a retried part never double-counts. Only open runs
|
||||
* (pending/running) accept parts; when the run has meanwhile gone terminal
|
||||
* (e.g. the abandoned-spool sweep cancelled it) or is mid-apply, nothing is
|
||||
* recorded and `false` comes back so the caller can discard the part.
|
||||
*/
|
||||
async completePart(db: Db, runId: string, partName: string): Promise<boolean> {
|
||||
const partJson = JSON.stringify([partName]);
|
||||
const updated = await db
|
||||
.update(companyTransferRuns)
|
||||
.set({
|
||||
completedParts: sql`case when ${companyTransferRuns.completedParts} @> ${partJson}::jsonb then ${companyTransferRuns.completedParts} else ${companyTransferRuns.completedParts} || ${partJson}::jsonb end`,
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(
|
||||
and(eq(companyTransferRuns.id, runId), inArray(companyTransferRuns.status, PART_UPLOAD_STATUSES)),
|
||||
)
|
||||
.returning({ id: companyTransferRuns.id });
|
||||
return updated.length > 0;
|
||||
},
|
||||
|
||||
/** Container part names the run still has to process, in manifest order. */
|
||||
remainingParts(run: CompanyTransferRunRow, manifest: CompanyTransferPartsIndex): string[] {
|
||||
const completed = new Set(run.completedParts);
|
||||
const remaining: string[] = [];
|
||||
for (const chunk of manifest.chunks) {
|
||||
if (!completed.has(chunk.name)) remaining.push(chunk.name);
|
||||
}
|
||||
for (const blob of manifest.blobs) {
|
||||
if (!completed.has(blob.name)) remaining.push(blob.name);
|
||||
}
|
||||
return remaining;
|
||||
},
|
||||
|
||||
async complete(db: Db, runId: string): Promise<void> {
|
||||
await db
|
||||
.update(companyTransferRuns)
|
||||
.set({ status: "completed", error: null, finishedAt: new Date(), updatedAt: new Date() })
|
||||
.where(eq(companyTransferRuns.id, runId));
|
||||
},
|
||||
|
||||
/**
|
||||
* Atomically claim an abandoned open run: cancel it only while it is still
|
||||
* open (pending/running) AND saw no activity since `idleSince` — one
|
||||
* guarded UPDATE, so any concurrent part upload or resume that bumped
|
||||
* `updatedAt` makes the guard miss and `false` comes back. Callers must
|
||||
* only touch the run's spooled data after a `true` claim.
|
||||
*
|
||||
* The terminal state is "cancelled", not "failed", precisely because failed
|
||||
* runs stay claimable: a failed sweep target could be claimed by an apply
|
||||
* and assembled while its spool is being deleted underneath it. A cancelled
|
||||
* run is unclaimable the instant this UPDATE commits — before any
|
||||
* filesystem deletion — and `resumeOrCreate` ignores it, so redeclaring the
|
||||
* same content afterwards starts a fresh run.
|
||||
*/
|
||||
async cancelIfIdle(db: Db, runId: string, idleSince: Date, error: string): Promise<boolean> {
|
||||
const claimed = await db
|
||||
.update(companyTransferRuns)
|
||||
.set({ status: "cancelled", error, finishedAt: new Date(), updatedAt: new Date() })
|
||||
.where(
|
||||
and(
|
||||
eq(companyTransferRuns.id, runId),
|
||||
inArray(companyTransferRuns.status, PART_UPLOAD_STATUSES),
|
||||
lt(companyTransferRuns.updatedAt, idleSince),
|
||||
),
|
||||
)
|
||||
.returning({ id: companyTransferRuns.id });
|
||||
return claimed.length > 0;
|
||||
},
|
||||
|
||||
/**
|
||||
* Release a held apply claim into "failed". Guarded on "applying", unlike
|
||||
* `fail`: catch-all error paths around an apply cannot know whether the run
|
||||
* already settled (a cleanup error thrown after `complete` reaches the same
|
||||
* catch as a real apply failure), and an unguarded fail there would flip a
|
||||
* completed run back to claimable and invite a duplicate import. When the
|
||||
* run already settled this is a no-op and returns false.
|
||||
*/
|
||||
async releaseApplyClaim(db: Db, runId: string, error: string): Promise<boolean> {
|
||||
const released = await db
|
||||
.update(companyTransferRuns)
|
||||
.set({ status: "failed", error, finishedAt: new Date(), updatedAt: new Date() })
|
||||
.where(and(eq(companyTransferRuns.id, runId), eq(companyTransferRuns.status, "applying")))
|
||||
.returning({ id: companyTransferRuns.id });
|
||||
return released.length > 0;
|
||||
},
|
||||
|
||||
/**
|
||||
* Fail every run stranded in "applying" by a dead process. Apply jobs run
|
||||
* in-memory in this single server process, so at boot any run still
|
||||
* "applying" belongs to an apply that died without settling its claim and
|
||||
* would otherwise 409 every retry forever. Failing it makes it claimable
|
||||
* again with its spooled parts intact. Startup only — running this while
|
||||
* applies may be live (e.g. from a periodic sweep) would kill them.
|
||||
*/
|
||||
async recoverStrandedApplyingRuns(db: Db): Promise<string[]> {
|
||||
const recovered = await db
|
||||
.update(companyTransferRuns)
|
||||
.set({
|
||||
status: "failed",
|
||||
error: "apply interrupted by a restart — verify whether the import completed before retrying",
|
||||
finishedAt: new Date(),
|
||||
updatedAt: new Date(),
|
||||
})
|
||||
.where(eq(companyTransferRuns.status, "applying"))
|
||||
.returning({ id: companyTransferRuns.id });
|
||||
return recovered.map((row) => row.id);
|
||||
},
|
||||
|
||||
async fail(db: Db, runId: string, error: string): Promise<void> {
|
||||
await db
|
||||
.update(companyTransferRuns)
|
||||
.set({ status: "failed", error, finishedAt: new Date(), updatedAt: new Date() })
|
||||
.where(eq(companyTransferRuns.id, runId));
|
||||
},
|
||||
|
||||
async cancel(db: Db, runId: string): Promise<void> {
|
||||
await db
|
||||
.update(companyTransferRuns)
|
||||
.set({ status: "cancelled", finishedAt: new Date(), updatedAt: new Date() })
|
||||
.where(and(eq(companyTransferRuns.id, runId), inArray(companyTransferRuns.status, RESUMABLE_STATUSES)));
|
||||
},
|
||||
};
|
||||
Loading…
Reference in New Issue