diff --git a/packages/shared/src/types/company-portability.ts b/packages/shared/src/types/company-portability.ts index 21134f3815..b43afdda2e 100644 --- a/packages/shared/src/types/company-portability.ts +++ b/packages/shared/src/types/company-portability.ts @@ -299,6 +299,13 @@ export type CompanyPortabilitySource = type: "inline"; rootPath?: string | null; files: Record; + /** + * Client-declared file count (`Object.keys(files).length`). The import + * path fails closed when the received file set is smaller, so a + * truncated inline body cannot import a fragment. Optional for callers + * that predate the check. + */ + expectedFileCount?: number; } | { type: "github"; diff --git a/packages/shared/src/validators/company-portability.ts b/packages/shared/src/validators/company-portability.ts index 8a86eb593f..eb766e8ecf 100644 --- a/packages/shared/src/validators/company-portability.ts +++ b/packages/shared/src/validators/company-portability.ts @@ -257,6 +257,12 @@ export const portabilitySourceSchema = z.discriminatedUnion("type", [ type: z.literal("inline"), rootPath: z.string().min(1).optional().nullable(), files: z.record(z.string(), portabilityFileEntrySchema), + // Self-describing completeness count. The client sets this to + // `Object.keys(files).length`; the import path rejects the payload when the + // received file set is smaller, so a truncated or re-framed body fails + // closed instead of importing a fragment. Optional for backwards + // compatibility with callers that predate the check. + expectedFileCount: z.number().int().nonnegative().optional(), }), z.object({ type: z.literal("github"), diff --git a/server/src/__tests__/company-portability-import-batching.test.ts b/server/src/__tests__/company-portability-import-batching.test.ts new file mode 100644 index 0000000000..7cc08ccbd1 --- /dev/null +++ b/server/src/__tests__/company-portability-import-batching.test.ts @@ -0,0 +1,384 @@ +import { afterAll, beforeAll, beforeEach, describe, expect, it } from "vitest"; +import { + assets, + createDb, + documentRevisions, + documents, + issueAttachments, + issueComments, + issueDocuments, + issueLabels, + issueRelations, + issueWorkProducts, + issues, +} from "@paperclipai/db"; +import { eq, sql } from "drizzle-orm"; +import { + getEmbeddedPostgresTestSupport, + startEmbeddedPostgresTestDatabase, +} from "./helpers/embedded-postgres.js"; +import { companyPortabilityService } from "../services/company-portability.js"; +import { workProductService } from "../services/work-products.js"; +import type { ImportIssueWorkProductRow } from "../services/import-write-types.js"; +import type { CompanyPortabilityFileEntry } from "@paperclipai/shared"; +import { randomUUID } from "node:crypto"; + +// This suite proves the company-import writers batch their inserts: it runs a +// real import against embedded Postgres and counts the SQL insert statements +// issued per table. The non-benchmark test guards against a regression to the +// old one-insert-per-row pattern in normal CI; the benchmark (opt-in via +// PORTABILITY_IMPORT_BENCH=1) reports wall-clock and statement counts at scale. + +const embeddedPostgresSupport = await getEmbeddedPostgresTestSupport(); +const describeEmbeddedPostgres = embeddedPostgresSupport.supported ? describe : describe.skip; + +// Tables whose insert statements we count, keyed by a readable label. +const COUNTED_TABLES: Array<[string, unknown]> = [ + ["issues", issues], + ["issue_labels", issueLabels], + ["issue_comments", issueComments], + ["documents", documents], + ["document_revisions", documentRevisions], + ["issue_documents", issueDocuments], + ["issue_work_products", issueWorkProducts], + ["assets", assets], + ["issue_attachments", issueAttachments], + ["issue_relations", issueRelations], +]; + +interface InsertCounts { + total: number; + byTable: Map; + reset: () => void; +} + +/** + * Patch a drizzle db so every `insert()` call (on the db and on any + * transaction handed to `db.transaction`) is counted. Each `insert(table)` + * call corresponds to exactly one SQL statement, so this counts statements. + */ +function instrumentInserts(db: Record): InsertCounts { + const counts: InsertCounts = { + total: 0, + byTable: new Map(), + reset() { + this.total = 0; + this.byTable.clear(); + }, + }; + const labelFor = (table: unknown): string | null => { + for (const [label, ref] of COUNTED_TABLES) { + if (table === ref) return label; + } + return null; + }; + const patchInsert = (target: Record) => { + const original = (target.insert as (table: unknown) => unknown).bind(target); + target.insert = (table: unknown) => { + counts.total += 1; + const label = labelFor(table); + if (label) counts.byTable.set(label, (counts.byTable.get(label) ?? 0) + 1); + return original(table); + }; + }; + patchInsert(db); + const originalTransaction = (db.transaction as (fn: unknown, config?: unknown) => unknown).bind(db); + db.transaction = (fn: (tx: Record) => unknown, config?: unknown) => + originalTransaction((tx: Record) => { + patchInsert(tx); + return fn(tx); + }, config); + return counts; +} + +// The `.paperclip.yaml` extension is consumed by the importer's own hand-rolled +// block-YAML parser (not a general YAML/JSON reader), so render it with the same +// block algorithm the exporter uses to guarantee a round-trip-compatible bundle. +function renderYamlScalar(value: unknown): string { + if (value === null) return "null"; + if (typeof value === "boolean" || typeof value === "number") return String(value); + return JSON.stringify(value); +} + +function isRenderableScalar(value: unknown): boolean { + return ( + value === null + || typeof value === "string" + || typeof value === "boolean" + || typeof value === "number" + || (Array.isArray(value) && value.length === 0) + || (value !== null && typeof value === "object" && !Array.isArray(value) && Object.keys(value).length === 0) + ); +} + +function renderYamlBlock(value: unknown, indentLevel: number): string[] { + const indent = " ".repeat(indentLevel); + if (Array.isArray(value)) { + if (value.length === 0) return [`${indent}[]`]; + const lines: string[] = []; + for (const entry of value) { + if (isRenderableScalar(entry)) { + lines.push(`${indent}- ${renderYamlScalar(entry)}`); + continue; + } + lines.push(`${indent}-`); + lines.push(...renderYamlBlock(entry, indentLevel + 1)); + } + return lines; + } + if (value !== null && typeof value === "object") { + const entries = Object.entries(value as Record); + if (entries.length === 0) return [`${indent}{}`]; + const lines: string[] = []; + for (const [key, entry] of entries) { + if (isRenderableScalar(entry)) { + lines.push(`${indent}${key}: ${renderYamlScalar(entry)}`); + continue; + } + lines.push(`${indent}${key}:`); + lines.push(...renderYamlBlock(entry, indentLevel + 1)); + } + return lines; + } + return [`${indent}${renderYamlScalar(value)}`]; +} + +function renderPaperclipYaml(value: Record): string { + return `${renderYamlBlock(value, 0).join("\n")}\n`; +} + +interface SyntheticBundleOptions { + issueCount: number; + commentsPerIssue: number; + documentsPerIssue: number; +} + +/** Build an inline import bundle with the requested shape, valid at schemaVersion 6. */ +function buildSyntheticBundle(options: SyntheticBundleOptions): { + rootPath: string; + files: Record; +} { + const files: Record = {}; + files["COMPANY.md"] = ['---', 'schema: "agentcompanies/v1"', 'name: "Batching Bench Co"', '---', '', 'Synthetic import bundle.', ''].join("\n"); + + const tasks: Record = {}; + for (let i = 1; i <= options.issueCount; i += 1) { + const slug = `task-${String(i).padStart(4, "0")}`; + files[`tasks/${slug}/TASK.md`] = ['---', `name: "Task ${i}"`, "kind: task", '---', '', `Description body for task ${i}. `.repeat(6), ''].join("\n"); + + const comments = []; + for (let c = 0; c < options.commentsPerIssue; c += 1) { + comments.push({ + body: `Comment ${c + 1} on task ${i}. `.repeat(4), + authorType: "system", + }); + } + const taskDocuments = []; + for (let d = 0; d < options.documentsPerIssue; d += 1) { + const key = d === 0 ? "spec" : `doc-${d + 1}`; + const docPath = `tasks/${slug}/documents/${key}.md`; + files[docPath] = `# ${key}\n\n${`Document body for task ${i} ${key}. `.repeat(8)}`; + taskDocuments.push({ key, title: key, format: "markdown", path: docPath }); + } + + tasks[slug] = { + status: "todo", + priority: "medium", + comments, + documents: taskDocuments, + }; + } + + files[".paperclip.yaml"] = renderPaperclipYaml({ schemaVersion: 6, tasks }); + + return { rootPath: "batching-bench", files }; +} + +describeEmbeddedPostgres("company import batches inserts", () => { + let tempDb: Awaited> | null = null; + let db!: ReturnType; + let counts!: InsertCounts; + + beforeAll(async () => { + tempDb = await startEmbeddedPostgresTestDatabase("paperclip-import-batching-"); + db = createDb(tempDb.connectionString); + counts = instrumentInserts(db as unknown as Record); + }, 30_000); + + afterAll(async () => { + await tempDb?.cleanup(); + }); + + beforeEach(() => { + counts.reset(); + }); + + it("imports many issues with a bounded number of issue-insert statements", async () => { + const issueCount = 50; + const commentsPerIssue = 2; + const documentsPerIssue = 1; + const bundle = buildSyntheticBundle({ issueCount, commentsPerIssue, documentsPerIssue }); + const portability = companyPortabilityService(db); + + const result = await portability.importBundle( + { + source: { type: "inline", rootPath: bundle.rootPath, files: bundle.files }, + include: { company: true, agents: false, projects: false, issues: true }, + target: { mode: "new_company", newCompanyName: "Batching Regression Co" }, + collisionStrategy: "rename", + }, + "user-batching-regression", + ); + + const companyId = result.company.id; + const issueInserts = counts.byTable.get("issues") ?? 0; + const commentInserts = counts.byTable.get("issue_comments") ?? 0; + + // The whole point: a 50-issue import must not issue one insert per issue. + expect(issueInserts).toBeLessThan(issueCount); + // 50 issues fit in a single chunk; the exact number is small and stable. + expect(issueInserts).toBeLessThanOrEqual(2); + expect(commentInserts).toBeLessThanOrEqual(2); + + // And the data must actually land, unchanged in shape. + const [{ issueRows }] = await db + .select({ issueRows: sql`count(*)::int` }) + .from(issues) + .where(eq(issues.companyId, companyId)); + const [{ commentRows }] = await db + .select({ commentRows: sql`count(*)::int` }) + .from(issueComments) + .where(eq(issueComments.companyId, companyId)); + // Count the issue-document links so company-bootstrap routine documents + // (seeded on company creation) don't skew the imported-document count. + const [{ documentLinkRows }] = await db + .select({ documentLinkRows: sql`count(*)::int` }) + .from(issueDocuments) + .where(eq(issueDocuments.companyId, companyId)); + + expect(issueRows).toBe(issueCount); + expect(commentRows).toBe(issueCount * commentsPerIssue); + expect(documentLinkRows).toBe(issueCount * documentsPerIssue); + expect(result.warnings).toEqual([]); + + // Identifiers are a contiguous range allocated from the company counter. + const identifiers = await db + .select({ identifier: issues.identifier }) + .from(issues) + .where(eq(issues.companyId, companyId)); + const uniqueIdentifiers = new Set(identifiers.map((row) => row.identifier)); + expect(uniqueIdentifiers.size).toBe(issueCount); + }); + + it("rolls back the whole work-product batch when a later chunk fails", async () => { + // Seed a real company + issue to hang work products off of. + const bundle = buildSyntheticBundle({ issueCount: 1, commentsPerIssue: 0, documentsPerIssue: 0 }); + const portability = companyPortabilityService(db); + const seeded = await portability.importBundle( + { + source: { type: "inline", rootPath: bundle.rootPath, files: bundle.files }, + include: { company: true, agents: false, projects: false, issues: true }, + target: { mode: "new_company", newCompanyName: "Work Product Atomicity Co" }, + collisionStrategy: "rename", + }, + "user-work-product-atomicity", + ); + const companyId = seeded.company.id; + const [issueRow] = await db + .select({ id: issues.id }) + .from(issues) + .where(eq(issues.companyId, companyId)) + .limit(1); + expect(issueRow?.id).toBeTruthy(); + + const buildRow = (issueId: string, index: number): ImportIssueWorkProductRow => ({ + companyId, + issueId, + projectId: null, + type: "pull_request", + provider: "github", + externalId: null, + title: `Work product ${index}`, + url: null, + status: "open", + reviewState: "none", + isPrimary: false, + healthStatus: "unknown", + summary: null, + metadata: null, + executionWorkspaceId: null, + runtimeServiceId: null, + createdByRunId: null, + sourceTrust: null, + }); + + // 501 rows span two chunks (chunk size is 500). The first 500 are valid; + // the 501st points at a non-existent issue so the *second* chunk's insert + // fails after the first chunk already ran. Without a wrapping transaction + // the first 500 would have committed; with it, nothing may persist. + const rows: ImportIssueWorkProductRow[] = []; + for (let i = 0; i < 500; i += 1) rows.push(buildRow(issueRow!.id, i)); + rows.push(buildRow(randomUUID(), 500)); + + await expect(workProductService(db).createManyForImport(rows)).rejects.toThrow(); + + const [{ workProductRows }] = await db + .select({ workProductRows: sql`count(*)::int` }) + .from(issueWorkProducts) + .where(eq(issueWorkProducts.companyId, companyId)); + expect(workProductRows).toBe(0); + }); + + const benchmark = process.env.PORTABILITY_IMPORT_BENCH === "1" ? it : it.skip; + benchmark( + "benchmark: large import stays in the tens of statements, not thousands", + async () => { + const issueCount = 500; + const commentsPerIssue = 5; + const documentsPerIssue = 1; + const bundle = buildSyntheticBundle({ issueCount, commentsPerIssue, documentsPerIssue }); + const totalRows = + issueCount + issueCount * commentsPerIssue + issueCount * documentsPerIssue * 3; + const portability = companyPortabilityService(db); + + const startedAt = performance.now(); + const result = await portability.importBundle( + { + source: { type: "inline", rootPath: bundle.rootPath, files: bundle.files }, + include: { company: true, agents: false, projects: false, issues: true }, + target: { mode: "new_company", newCompanyName: "Batching Benchmark Co" }, + collisionStrategy: "rename", + }, + "user-batching-benchmark", + ); + const elapsedMs = performance.now() - startedAt; + + const perTable = Object.fromEntries(counts.byTable); + // Analytical "before": the old path inserted one row per statement. + const beforeStatements = totalRows; + const afterStatements = counts.total; + + // eslint-disable-next-line no-console + console.log( + `[import-batching-bench] ${issueCount} issues x ${commentsPerIssue} comments x ${documentsPerIssue} doc\n` + + ` wall-clock: ${elapsedMs.toFixed(0)}ms\n` + + ` insert statements (after batching): ${afterStatements}\n` + + ` insert statements (before, one per row): ~${beforeStatements}\n` + + ` reduction: ${(beforeStatements / Math.max(1, afterStatements)).toFixed(0)}x\n` + + ` per-table insert statements: ${JSON.stringify(perTable)}`, + ); + + const companyId = result.company.id; + const [{ issueRows }] = await db + .select({ issueRows: sql`count(*)::int` }) + .from(issues) + .where(eq(issues.companyId, companyId)); + expect(issueRows).toBe(issueCount); + + // Two orders of magnitude fewer statements than the per-row baseline. + expect(afterStatements * 50).toBeLessThan(beforeStatements); + expect(counts.byTable.get("issues") ?? 0).toBeLessThanOrEqual(2); + }, + 120_000, + ); +}); diff --git a/server/src/__tests__/company-portability-routes.test.ts b/server/src/__tests__/company-portability-routes.test.ts index 088d38b1d4..8899620086 100644 --- a/server/src/__tests__/company-portability-routes.test.ts +++ b/server/src/__tests__/company-portability-routes.test.ts @@ -204,6 +204,62 @@ async function waitForCondition(condition: () => boolean, label: string) { throw new Error(`Timed out waiting for ${label}`); } +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 single companyRoutes instance (one shared in-memory job map) whose board +// actor is chosen per request from the `x-test-user-id` header. This lets one +// app exercise cross-user isolation and the per-user duplicate guard. +async function createBoardApp() { + registerCompanyRouteMocks(); + appImportCounter += 1; + const routeModulePath = `../routes/companies.js?company-portability-routes-${appImportCounter}`; + const middlewareModulePath = `../middleware/index.js?company-portability-routes-${appImportCounter}`; + const [{ companyRoutes }, { errorHandler }] = await Promise.all([ + import(routeModulePath) as Promise, + import(middlewareModulePath) as Promise, + ]); + const app = express(); + app.use(express.json()); + app.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(); + }); + app.use("/api/companies", companyRoutes({} as any)); + app.use(errorHandler); + return app; +} + +async function waitForImportJobStatusAs( + app: express.Express, + statusUrl: string, + status: string, + headers: Record, +) { + for (let attempt = 0; attempt < 20; attempt += 1) { + const res = await request(app).get(statusUrl).set(headers); + if (res.body.job?.status === status) { + return res; + } + await new Promise((resolve) => setTimeout(resolve, 10)); + } + throw new Error(`Timed out waiting for import job to reach ${status}`); +} + describe.sequential("company portability routes", () => { beforeEach(() => { vi.clearAllMocks(); @@ -751,6 +807,185 @@ describe.sequential("company portability routes", () => { ); }); + it.sequential("runs board-session async imports as jobs and reports the full result by job id", async () => { + let resolveImport: (value: ReturnType) => void = () => undefined; + const pendingImport = new Promise>((resolve) => { + resolveImport = resolve; + }); + mockCompanyPortabilityService.importBundle.mockReturnValueOnce(pendingImport); + const app = await createBoardApp(); + + const accepted = await request(app) + .post("/api/companies/import") + .set("x-paperclip-cloud-async-import", "1") + .set(TEST_USER_HEADER, "board-user-a") + .send(importRequest); + + expect(accepted.status).toBe(202); + expect(accepted.body.job.status).toBe("running"); + // Board jobs get their own id prefix, distinct from the tenant-import prefix. + expect(accepted.body.statusUrl).toMatch(/^\/api\/companies\/import\/jobs\/import-/); + expect(accepted.body.statusUrl).not.toMatch(/\/jobs\/tenant-import-/); + await waitForCondition(() => mockCompanyPortabilityService.importBundle.mock.calls.length === 1, "board import start"); + expect(mockCompanyPortabilityService.importBundle).toHaveBeenCalledWith(importRequest, "board-user-a", { pauseAutomations: false }); + + const fullResult = createImportResult("created"); + resolveImport(fullResult); + const succeeded = await waitForImportJobStatusAs(app, accepted.body.statusUrl, "succeeded", { + [TEST_USER_HEADER]: "board-user-a", + }); + + expect(succeeded.status).toBe(200); + expect(succeeded.body.job.status).toBe("succeeded"); + expect(succeeded.body.job.result.companyId).toBe(companyId); + // Parity with the synchronous response: the board job carries the full + // import result so the import page can run the same activation path. + expect(succeeded.body.job.importResult).toEqual(fullResult); + expect(mockLogActivity).toHaveBeenCalledWith(expect.anything(), expect.objectContaining({ + action: "company.imported", + companyId, + })); + }); + + it.sequential("hides a board import job from other board users", async () => { + mockCompanyPortabilityService.importBundle.mockReturnValueOnce(new Promise(() => undefined)); + const app = await createBoardApp(); + + const accepted = await request(app) + .post("/api/companies/import") + .set("x-paperclip-cloud-async-import", "1") + .set(TEST_USER_HEADER, "board-user-a") + .send(importRequest); + + expect(accepted.status).toBe(202); + + const crossUser = await request(app).get(accepted.body.statusUrl).set(TEST_USER_HEADER, "board-user-b"); + expect(crossUser.status).toBe(404); + expect(crossUser.body.error).toBe("Import job not found"); + + const owner = await request(app).get(accepted.body.statusUrl).set(TEST_USER_HEADER, "board-user-a"); + expect(owner.status).toBe(200); + expect(owner.body.job.status).toBe("running"); + }); + + it.sequential("returns 409 with the running job when a board user resubmits an import", async () => { + mockCompanyPortabilityService.importBundle.mockReturnValueOnce(new Promise(() => undefined)); + const app = await createBoardApp(); + + const first = await request(app) + .post("/api/companies/import") + .set("x-paperclip-cloud-async-import", "1") + .set(TEST_USER_HEADER, "board-user-a") + .send(importRequest); + + expect(first.status).toBe(202); + + const duplicate = await request(app) + .post("/api/companies/import") + .set("x-paperclip-cloud-async-import", "1") + .set(TEST_USER_HEADER, "board-user-a") + .send(importRequest); + + expect(duplicate.status).toBe(409); + expect(duplicate.body.job.id).toBe(first.body.job.id); + expect(duplicate.body.statusUrl).toBe(first.body.statusUrl); + // The duplicate submit must not start a second import of the same bundle. + expect(mockCompanyPortabilityService.importBundle).toHaveBeenCalledTimes(1); + + // A different board user is not blocked by the first user's running job. + mockCompanyPortabilityService.importBundle.mockReturnValueOnce(new Promise(() => undefined)); + const otherUser = await request(app) + .post("/api/companies/import") + .set("x-paperclip-cloud-async-import", "1") + .set(TEST_USER_HEADER, "board-user-b") + .send(importRequest); + + expect(otherUser.status).toBe(202); + expect(otherUser.body.job.id).not.toBe(first.body.job.id); + }); + + it.sequential( + "rejects a concurrent different import without adopting the running job", + async () => { + mockCompanyPortabilityService.importBundle.mockReturnValueOnce(new Promise(() => undefined)); + const app = await createBoardApp(); + + const first = await request(app) + .post("/api/companies/import") + .set("x-paperclip-cloud-async-import", "1") + .set(TEST_USER_HEADER, "board-user-a") + .send(importRequest); + + expect(first.status).toBe(202); + + // Same user, a *different* import (another destination) while the first + // still runs. It must NOT adopt the first job — adopting would show the + // wrong result and switch to the wrong company — so the 409 carries no + // job to watch, and no second import starts. + const different = await request(app) + .post("/api/companies/import") + .set("x-paperclip-cloud-async-import", "1") + .set(TEST_USER_HEADER, "board-user-a") + .send({ ...importRequest, target: { mode: "new_company", newCompanyName: "A Different Destination" } }); + + expect(different.status).toBe(409); + expect(different.body.job).toBeUndefined(); + expect(different.body.statusUrl).toBeUndefined(); + expect(different.body.error).toMatch(/different import is already running/i); + expect(mockCompanyPortabilityService.importBundle).toHaveBeenCalledTimes(1); + }, + ); + + it.sequential("fails a board async import job when the bundle import throws", async () => { + mockCompanyPortabilityService.importBundle.mockRejectedValueOnce(new Error("import payload is incomplete")); + const app = await createBoardApp(); + + const accepted = await request(app) + .post("/api/companies/import") + .set("x-paperclip-cloud-async-import", "1") + .set(TEST_USER_HEADER, "board-user-a") + .send(importRequest); + + expect(accepted.status).toBe(202); + + const failed = await waitForImportJobStatusAs(app, accepted.body.statusUrl, "failed", { + [TEST_USER_HEADER]: "board-user-a", + }); + + expect(failed.status).toBe(200); + expect(failed.body.job.status).toBe("failed"); + expect(failed.body.job.error.message).toBe("import payload is incomplete"); + expect(failed.body.job.importResult).toBeUndefined(); + // A resubmit is allowed once the job is terminal. + expect(mockLogActivity).not.toHaveBeenCalled(); + }); + + it.sequential("keeps the import job status route board-only", async () => { + const app = await createApp({ + type: "agent", + agentId: engineerAgentId, + companyId, + source: "agent_key", + runId: "run-1", + }); + + const res = await request(app).get("/api/companies/import/jobs/import-does-not-exist"); + + expect(res.status).toBe(403); + expect(res.body.error).toContain("Board access required"); + }); + + it.sequential("returns 404 for an unknown import job id", async () => { + const app = await createBoardApp(); + + const res = await request(app) + .get("/api/companies/import/jobs/import-missing") + .set(TEST_USER_HEADER, "board-user-a"); + + expect(res.status).toBe(404); + expect(res.body.error).toBe("Import job not found"); + }); + it.sequential("forwards pauseAutomations from CEO-safe import apply bodies to the portability service", async () => { mockCompanyPortabilityService.importBundle.mockResolvedValueOnce(createImportResult("created")); const app = await createApp({ diff --git a/server/src/__tests__/company-portability.test.ts b/server/src/__tests__/company-portability.test.ts index 3dd5d0110d..6f93494ec8 100644 --- a/server/src/__tests__/company-portability.test.ts +++ b/server/src/__tests__/company-portability.test.ts @@ -47,16 +47,21 @@ const issueSvc = { getRelationSummaries: vi.fn(), listAttachments: vi.fn(), createAttachment: vi.fn(), + importIssues: vi.fn(), + addImportedComments: vi.fn(), + addImportedAttachments: vi.fn(), }; const documentSvc = { listIssueDocuments: vi.fn(), upsertIssueDocument: vi.fn(), + createIssueDocumentsForImport: vi.fn(), }; const workProductSvc = { listForIssue: vi.fn(), createForIssue: vi.fn(), + createManyForImport: vi.fn(), }; const routineSvc = { @@ -1240,11 +1245,13 @@ describe("company portability", () => { expect(projectSvc.create).toHaveBeenCalledWith("company-imported", expect.objectContaining({ icon: "rocket", })); - expect(issueSvc.create).toHaveBeenCalledWith("company-imported", expect.objectContaining({ - projectId: "project-imported", - projectWorkspaceId: "workspace-imported", - title: "Write launch task", - })); + expect(issueSvc.importIssues).toHaveBeenCalledWith("company-imported", expect.arrayContaining([ + expect.objectContaining({ + projectId: "project-imported", + projectWorkspaceId: "workspace-imported", + title: "Write launch task", + }), + ])); }); it("normalizes invalid imported project icon names to null", async () => { @@ -1797,6 +1804,111 @@ describe("company portability", () => { expect(secretSvc.remove).toHaveBeenCalledWith("secret-created-for-failed-import"); }); + it("fails closed on an inline import that arrived with fewer files than declared", async () => { + const portability = companyPortabilityService({} as any); + agentSvc.list.mockResolvedValue([]); + + // The client declared four files, but the body was truncated in transit and + // only three arrived. The import must reject the fragment before writing any + // rows — not create a company and import a partial bundle. + await expect(portability.importBundle({ + source: { + type: "inline", + expectedFileCount: 4, + files: { + "COMPANY.md": [ + "---", + "name: Import", + "includes:", + " - agents/coder/AGENTS.md", + "---", + "", + ].join("\n"), + "agents/coder/AGENTS.md": [ + "---", + "name: Coder", + "slug: coder", + "kind: agent", + "---", + "", + "# Coder", + "", + ].join("\n"), + ".paperclip.yaml": [ + "schema: paperclip/v1", + "agents:", + " coder:", + " adapter:", + " type: codex_local", + " config: {}", + "", + ].join("\n"), + }, + }, + include: { company: true, agents: true, projects: false, issues: false }, + target: { mode: "new_company", newCompanyName: "Imported" }, + collisionStrategy: "rename", + }, "user-1")).rejects.toMatchObject({ + status: 422, + details: { code: "import_payload_incomplete", expectedFileCount: 4, receivedFileCount: 3 }, + }); + + expect(companySvc.create).not.toHaveBeenCalled(); + expect(agentSvc.create).not.toHaveBeenCalled(); + }); + + it("imports an inline bundle whose file count matches the declared count", async () => { + const portability = companyPortabilityService({} as any); + agentSvc.list.mockResolvedValue([]); + agentSvc.create.mockImplementation(async (_companyId: string, input: Record) => ({ + id: "agent-imported", + name: input.name, + adapterType: input.adapterType, + adapterConfig: input.adapterConfig, + status: input.status, + })); + + const files = { + "COMPANY.md": [ + "---", + "name: Import", + "includes:", + " - agents/coder/AGENTS.md", + "---", + "", + ].join("\n"), + "agents/coder/AGENTS.md": [ + "---", + "name: Coder", + "slug: coder", + "kind: agent", + "---", + "", + "# Coder", + "", + ].join("\n"), + ".paperclip.yaml": [ + "schema: paperclip/v1", + "agents:", + " coder:", + " adapter:", + " type: codex_local", + " config: {}", + "", + ].join("\n"), + }; + + const result = await portability.importBundle({ + source: { type: "inline", expectedFileCount: Object.keys(files).length, files }, + include: { company: false, agents: true, projects: false, issues: false }, + target: { mode: "existing_company", companyId: "company-1" }, + collisionStrategy: "rename", + }, "user-1"); + + expect(result.company.id).toBe("company-1"); + expect(agentSvc.create).toHaveBeenCalledTimes(1); + }); + it("reparents imported roots to pre-existing target managers before resolving imported hierarchy", async () => { const portability = companyPortabilityService({} as any); agentSvc.list.mockResolvedValue([ @@ -2372,7 +2484,7 @@ describe("company portability", () => { signingMode: "hmac_sha256", replayWindowSec: 120, }), expect.any(Object)); - expect(issueSvc.create).not.toHaveBeenCalled(); + expect(issueSvc.importIssues).not.toHaveBeenCalled(); expect(result.routines).toEqual([ { slug: "monday-review", id: "routine-created", action: "created", title: "Monday Review", status: "paused" }, ]); @@ -2588,7 +2700,7 @@ describe("company portability", () => { cronExpression: "0 9 * * 1", timezone: "America/Chicago", }), expect.any(Object)); - expect(issueSvc.create).not.toHaveBeenCalled(); + expect(issueSvc.importIssues).not.toHaveBeenCalled(); }); it("imports recurring tasks without a project or assignee as paused routines", async () => { @@ -3450,11 +3562,13 @@ describe("company portability", () => { expect(issueSvc.createLabel).toHaveBeenCalledWith("company-imported", { name: "bug", color: "#ff0000" }); expect(issueSvc.createLabel).toHaveBeenCalledWith("company-imported", { name: "urgent", color: "#00ff00" }); expect(issueSvc.createLabel).toHaveBeenCalledTimes(2); - expect(issueSvc.create).toHaveBeenCalledWith( + expect(issueSvc.importIssues).toHaveBeenCalledWith( "company-imported", - expect.objectContaining({ - labelIds: ["label-created-bug", "label-created-urgent"], - }), + expect.arrayContaining([ + expect.objectContaining({ + labelIds: ["label-created-bug", "label-created-urgent"], + }), + ]), ); }); @@ -3506,11 +3620,13 @@ describe("company portability", () => { expect(issueSvc.createLabel).toHaveBeenCalledTimes(1); expect(issueSvc.createLabel).toHaveBeenCalledWith("company-1", { name: "urgent", color: "#00ff00" }); - expect(issueSvc.create).toHaveBeenCalledWith( + expect(issueSvc.importIssues).toHaveBeenCalledWith( "company-1", - expect.objectContaining({ - labelIds: ["target-bug", "label-created-urgent"], - }), + expect.arrayContaining([ + expect.objectContaining({ + labelIds: ["target-bug", "label-created-urgent"], + }), + ]), ); expect(result.warnings).toContain( "Existing label color was kept for bug; the imported bundle used different colors.", @@ -3564,9 +3680,11 @@ describe("company portability", () => { }, "user-1"); expect(issueSvc.createLabel).not.toHaveBeenCalled(); - expect(issueSvc.create).toHaveBeenCalledWith( + expect(issueSvc.importIssues).toHaveBeenCalledWith( "company-imported", - expect.objectContaining({ labelIds: [] }), + expect.arrayContaining([ + expect.objectContaining({ labelIds: [] }), + ]), ); expect(result.warnings).toContain( "Task kickoff dropped 2 label references because the bundle carries raw label ids that do not exist in the target company.", @@ -3754,11 +3872,6 @@ describe("company portability", () => { companySvc.create.mockResolvedValue({ id: "company-imported", name: "Imported" }); accessSvc.ensureMembership.mockResolvedValue(undefined); agentSvc.list.mockResolvedValue([]); - issueSvc.create.mockImplementation(async (_companyId: string, input: Record) => ({ - id: input.title === "Alpha task" ? "issue-imported-1" : "issue-imported-2", - title: input.title, - projectId: null, - })); const result = await portability.importBundle({ source: { type: "inline", rootPath: exported.rootPath, files: exported.files }, @@ -3768,45 +3881,67 @@ describe("company portability", () => { collisionStrategy: "rename", }, "user-1"); - expect(documentSvc.upsertIssueDocument).toHaveBeenCalledWith({ - issueId: "issue-imported-1", - key: "spec", - title: "Spec", - format: "markdown", - body: "# Spec\n\nDetails.", - createdByUserId: "user-1", - }); - expect(workProductSvc.createForIssue).toHaveBeenCalledWith( - "issue-imported-1", - "company-imported", - expect.objectContaining({ - type: "pull_request", - provider: "github", - externalId: "42", - title: "Fix bug", - status: "merged", - reviewState: "approved", - isPrimary: true, - healthStatus: "healthy", - executionWorkspaceId: null, - runtimeServiceId: null, - createdByRunId: null, - sourceTrust: null, - }), + // Ids are pre-generated by the batched importer; correlate them by title. + const importedIssues = issueSvc.importIssues.mock.calls[0]![1] as Array<{ + id: string; + title: string; + monitorNotes: string | null; + monitorScheduledBy: string | null; + }>; + const alphaId = importedIssues.find((row) => row.title === "Alpha task")!.id; + const betaId = importedIssues.find((row) => row.title === "Beta task")!.id; + + expect(documentSvc.createIssueDocumentsForImport).toHaveBeenCalledWith( + expect.arrayContaining([ + expect.objectContaining({ + issueId: alphaId, + key: "spec", + title: "Spec", + format: "markdown", + body: "# Spec\n\nDetails.", + createdByUserId: "user-1", + }), + ]), + ); + expect(workProductSvc.createManyForImport).toHaveBeenCalledWith( + expect.arrayContaining([ + expect.objectContaining({ + issueId: alphaId, + companyId: "company-imported", + type: "pull_request", + provider: "github", + externalId: "42", + title: "Fix bug", + status: "merged", + reviewState: "approved", + isPrimary: true, + healthStatus: "healthy", + executionWorkspaceId: null, + runtimeServiceId: null, + createdByRunId: null, + sourceTrust: null, + }), + ]), ); expect(insertedRelationValues).toEqual([ { companyId: "company-imported", - issueId: "issue-imported-1", - relatedIssueId: "issue-imported-2", + issueId: alphaId, + relatedIssueId: betaId, type: "blocks", createdByAgentId: null, createdByUserId: "user-1", }, ]); - expect(monitorUpdates).toEqual([ - { monitorNotes: "Check deploy daily", monitorScheduledBy: "agent" }, - ]); + // Monitor notes/provenance ride on the issue row itself now, so there is no + // separate post-insert update. The monitor still lands un-armed. + expect(monitorUpdates).toEqual([]); + expect(importedIssues.find((row) => row.title === "Alpha task")).toEqual( + expect.objectContaining({ + monitorNotes: "Check deploy daily", + monitorScheduledBy: "agent", + }), + ); expect(result.warnings).toContain( "1 monitor was imported un-armed; re-arm it from the task page to resume checks.", ); @@ -3835,13 +3970,15 @@ describe("company portability", () => { selectedFiles: ["COMPANY.md", ".paperclip.yaml", "tasks/pap-2/TASK.md"], }, "user-1"); - expect(issueSvc.create).toHaveBeenCalledTimes(1); - expect(issueSvc.create).toHaveBeenCalledWith( + expect(issueSvc.importIssues.mock.calls[0]![1]).toHaveLength(1); + expect(issueSvc.importIssues).toHaveBeenCalledWith( "company-imported", - expect.objectContaining({ title: "Beta task" }), + expect.arrayContaining([ + expect.objectContaining({ title: "Beta task" }), + ]), ); - expect(documentSvc.upsertIssueDocument).not.toHaveBeenCalled(); - expect(workProductSvc.createForIssue).not.toHaveBeenCalled(); + expect(documentSvc.createIssueDocumentsForImport).not.toHaveBeenCalled(); + expect(workProductSvc.createManyForImport).not.toHaveBeenCalled(); expect(insertedRelationValues).toEqual([]); expect(result.warnings).toContain( "Task pap-2 blocker pap-1 was skipped because that task was not imported.", @@ -3999,9 +4136,6 @@ describe("company portability", () => { accessSvc.ensureMembership.mockResolvedValue(undefined); agentSvc.list.mockResolvedValue([]); issueSvc.create.mockResolvedValue({ id: "issue-imported", title: "Attachment task", projectId: null }); - issueSvc.addComment - .mockResolvedValueOnce({ id: "comment-imported-1" }) - .mockResolvedValueOnce({ id: "comment-imported-2" }); const result = await portability.importBundle({ source: { type: "inline", rootPath: exported.rootPath, files: exported.files }, @@ -4011,30 +4145,40 @@ describe("company portability", () => { collisionStrategy: "rename", }, "user-1"); + // Ids are pre-generated, so capture them to resolve the storage namespace + // and the comment-scoped attachment reference. + const importedIssueId = (issueSvc.importIssues.mock.calls[0]![1] as Array<{ id: string }>)[0]!.id; + const importedCommentIds = (issueSvc.addImportedComments.mock.calls[0]![0] as Array<{ id: string }>).map((row) => row.id); + expect(storage.putFile).toHaveBeenCalledTimes(2); expect(storage.putFile).toHaveBeenCalledWith(expect.objectContaining({ companyId: "company-imported", - namespace: "issues/issue-imported", + namespace: `issues/${importedIssueId}`, originalFilename: "screenshot.png", contentType: "image/png", body: Buffer.from("png-bytes"), })); - expect(issueSvc.createAttachment).toHaveBeenCalledTimes(2); - expect(issueSvc.createAttachment).toHaveBeenCalledWith(expect.objectContaining({ - issueId: "issue-imported", - issueCommentId: null, - originalFilename: "notes.bin", - contentType: "application/octet-stream", - sha256: sha, - byteSize: 9, - createdByUserId: "user-1", - })); - expect(issueSvc.createAttachment).toHaveBeenCalledWith(expect.objectContaining({ - issueId: "issue-imported", - issueCommentId: "comment-imported-2", - originalFilename: "screenshot.png", - contentType: "image/png", - })); + const attachmentRows = issueSvc.addImportedAttachments.mock.calls[0]![0] as Array>; + expect(attachmentRows).toHaveLength(2); + expect(issueSvc.addImportedAttachments).toHaveBeenCalledWith(expect.arrayContaining([ + expect.objectContaining({ + issueId: importedIssueId, + issueCommentId: null, + originalFilename: "notes.bin", + contentType: "application/octet-stream", + sha256: sha, + byteSize: 9, + createdByUserId: "user-1", + }), + ])); + expect(issueSvc.addImportedAttachments).toHaveBeenCalledWith(expect.arrayContaining([ + expect.objectContaining({ + issueId: importedIssueId, + issueCommentId: importedCommentIds[1], + originalFilename: "screenshot.png", + contentType: "image/png", + }), + ])); expect(result.warnings.filter((warning) => warning.includes("attachment"))).toEqual([]); }); @@ -4075,7 +4219,7 @@ describe("company portability", () => { collisionStrategy: "rename", }, "user-1"); - expect(issueSvc.createAttachment).not.toHaveBeenCalled(); + expect(issueSvc.addImportedAttachments).not.toHaveBeenCalled(); expect(result.warnings).toContain("Skipped 2 attachments because storage is unavailable."); }); @@ -4106,11 +4250,11 @@ describe("company portability", () => { agents: "all", collisionStrategy: "rename", }, "user-1")).rejects.toThrow(/does not match its declared sha256/); - expect(issueSvc.createAttachment).not.toHaveBeenCalled(); + expect(issueSvc.addImportedAttachments).not.toHaveBeenCalled(); // Blob verification runs before any write, so a tampered package cannot // leave a partially imported company behind. expect(companySvc.create).not.toHaveBeenCalled(); - expect(issueSvc.create).not.toHaveBeenCalled(); + expect(issueSvc.importIssues).not.toHaveBeenCalled(); }); it("skips oversized and missing-blob attachments with warnings instead of failing", async () => { @@ -4152,7 +4296,7 @@ describe("company portability", () => { collisionStrategy: "rename", }, "user-1"); - expect(issueSvc.createAttachment).not.toHaveBeenCalled(); + expect(issueSvc.addImportedAttachments).not.toHaveBeenCalled(); expect(result.warnings).toContain( `Task pap-1 attachment notes.bin was skipped because its blob is missing from the package: blobs/${sha}`, ); @@ -4290,17 +4434,19 @@ describe("company portability", () => { })); // Every reference now points at the minted asset id, not the source id. - const importedDescription = issueSvc.create.mock.calls[0]![1].description as string; + const importedDescription = (issueSvc.importIssues.mock.calls[0]![1] as Array<{ description: string }>)[0]!.description; expect(importedDescription).toContain(embeddedAssetUrl("asset-imported-1")); expect(importedDescription).not.toContain(EMBEDDED_ASSET_ID); - const importedCommentBody = issueSvc.addComment.mock.calls[0]![1] as string; + const importedCommentBody = (issueSvc.addImportedComments.mock.calls[0]![0] as Array<{ body: string }>)[0]!.body; expect(importedCommentBody).toContain(embeddedAssetUrl("asset-imported-1")); expect(importedCommentBody).not.toContain(EMBEDDED_ASSET_ID); - expect(documentSvc.upsertIssueDocument).toHaveBeenCalledWith(expect.objectContaining({ - key: "spec", - body: expect.stringContaining(embeddedAssetUrl("asset-imported-1")), - })); - const importedDocumentBody = documentSvc.upsertIssueDocument.mock.calls[0]![0].body as string; + expect(documentSvc.createIssueDocumentsForImport).toHaveBeenCalledWith(expect.arrayContaining([ + expect.objectContaining({ + key: "spec", + body: expect.stringContaining(embeddedAssetUrl("asset-imported-1")), + }), + ])); + const importedDocumentBody = (documentSvc.createIssueDocumentsForImport.mock.calls[0]![0] as Array<{ body: string }>)[0]!.body; expect(importedDocumentBody).not.toContain(EMBEDDED_ASSET_ID); expect(result.warnings.filter((warning) => warning.includes("embedded"))).toEqual([]); }); @@ -4373,9 +4519,9 @@ describe("company portability", () => { expect(result.warnings).toContain( `Embedded image asset embed.png was skipped because its blob is missing from the package: blobs/${sha}; its references were left unchanged.`, ); - const importedDescription = issueSvc.create.mock.calls[0]![1].description as string; + const importedDescription = (issueSvc.importIssues.mock.calls[0]![1] as Array<{ description: string }>)[0]!.description; expect(importedDescription).toContain(embeddedAssetUrl(EMBEDDED_ASSET_ID)); - const importedCommentBody = issueSvc.addComment.mock.calls[0]![1] as string; + const importedCommentBody = (issueSvc.addImportedComments.mock.calls[0]![0] as Array<{ body: string }>)[0]!.body; expect(importedCommentBody).toContain(embeddedAssetUrl(EMBEDDED_ASSET_ID)); }); @@ -4454,9 +4600,11 @@ describe("company portability", () => { expect(preview.warnings).toContain(v5Warning); const result = await portability.importBundle(request, "user-1"); - expect(issueSvc.create).toHaveBeenCalledWith( + expect(issueSvc.importIssues).toHaveBeenCalledWith( "company-imported", - expect.objectContaining({ title: "Kickoff" }), + expect.arrayContaining([ + expect.objectContaining({ title: "Kickoff" }), + ]), ); expect(result.warnings).toContain(v5Warning); }); @@ -4487,7 +4635,7 @@ describe("company portability", () => { agents: "all", collisionStrategy: "rename", }, "user-1")).rejects.toThrow(/newer Paperclip/); - expect(issueSvc.create).not.toHaveBeenCalled(); + expect(issueSvc.importIssues).not.toHaveBeenCalled(); }); it("preserves issue comment presentation fields through export and import", async () => { @@ -4556,17 +4704,17 @@ describe("company portability", () => { collisionStrategy: "rename", }, "user-1"); - expect(issueSvc.addComment).toHaveBeenCalledWith( - "issue-imported", - "Paperclip needs a disposition before this issue can continue.", - { agentId: undefined, userId: undefined }, - { + expect(issueSvc.addImportedComments).toHaveBeenCalledWith(expect.arrayContaining([ + expect.objectContaining({ + body: "Paperclip needs a disposition before this issue can continue.", authorType: "system", + authorAgentId: null, + authorUserId: null, presentation, metadata, createdAt: "2026-05-04T12:00:00.000Z", - }, - ); + }), + ])); }); it("does not export raw comment author user ids", async () => { @@ -4672,17 +4820,17 @@ describe("company portability", () => { collisionStrategy: "rename", }, null); - expect(issueSvc.addComment).toHaveBeenCalledWith( - "issue-imported", - "Need private follow-up.", - { agentId: undefined, userId: undefined }, - { + expect(issueSvc.addImportedComments).toHaveBeenCalledWith(expect.arrayContaining([ + expect.objectContaining({ + body: "Need private follow-up.", authorType: "system", + authorAgentId: null, + authorUserId: null, presentation: null, metadata: null, createdAt: "2026-05-04T12:00:00.000Z", - }, - ); + }), + ])); expect(result.warnings).toContain( "Comment on task pap-1 was imported as a system comment because no importing user was available.", ); @@ -4938,7 +5086,7 @@ describe("company portability", () => { sourceCompanyId: "company-1", })).rejects.toThrow("Safe import does not allow task review executionWorkspaceSettings."); - expect(issueSvc.create).not.toHaveBeenCalled(); + expect(issueSvc.importIssues).not.toHaveBeenCalled(); expect(routineSvc.createTrigger).not.toHaveBeenCalled(); }); @@ -5179,12 +5327,14 @@ describe("company portability", () => { expect(agentResult!.action).toBe("skipped"); // Issue should still be created and reference the existing agent - expect(issueSvc.create).toHaveBeenCalled(); - const issueCreateCall = issueSvc.create.mock.calls[0]; + expect(issueSvc.importIssues).toHaveBeenCalled(); + const issueImportCall = issueSvc.importIssues.mock.calls[0]; // The assigneeAgentId should resolve to the existing agent via existingSlugToAgentId - expect(issueCreateCall[1]).toEqual(expect.objectContaining({ - assigneeAgentId: "agent-1", - })); + expect(issueImportCall[1]).toEqual(expect.arrayContaining([ + expect.objectContaining({ + assigneeAgentId: "agent-1", + }), + ])); }); it("handles a package with only skills (no agents or projects)", async () => { diff --git a/server/src/routes/companies.ts b/server/src/routes/companies.ts index 4ae00a5b4c..3ac4003bb9 100644 --- a/server/src/routes/companies.ts +++ b/server/src/routes/companies.ts @@ -1,9 +1,10 @@ -import { randomUUID } from "node:crypto"; +import { createHash, randomUUID } from "node:crypto"; import { Router, type Request } from "express"; import { and, count as countFn, eq } from "drizzle-orm"; import { z } from "zod"; import type { Db } from "@paperclipai/db"; import { agents as agentsTable } from "@paperclipai/db"; +import type { CompanyPortabilityImportResult } from "@paperclipai/shared"; import { DEFAULT_FEEDBACK_DATA_SHARING_TERMS_VERSION, companyArtifactsQuerySchema, @@ -270,10 +271,15 @@ export function companyRoutes(db: Db, storage?: StorageService) { }); router.get("/import/jobs/:jobId", async (req, res) => { - assertCloudTenantCaller(req); + // Board sessions and trusted Cloud tenants both poll here. A job is + // readable only by the actor key that created it; every other caller — + // like every unknown or expired id — gets the same 404, so job ids + // cannot be probed across users or tenants. Jobs live in memory only + // (see the async block below), so a restart also surfaces as this 404. + assertBoard(req); cleanupTerminalImportJobs(importJobs, importJobTerminalRetentionMs); const job = importJobs.get(req.params.jobId as string); - if (!job || job.cloudTenantKey !== cloudTenantRequestKey(req)) { + if (!job || job.actorKey !== importJobActorKey(req)) { res.status(404).json({ error: "Import job not found" }); return; } @@ -286,9 +292,41 @@ export function companyRoutes(db: Db, storage?: StorageService) { const actor = getActorInfo(req); const boardUserId = req.actor.type === "board" ? req.actor.userId : null; if (req.header("x-paperclip-cloud-async-import") === "1") { - assertCloudTenantCaller(req); + // Async job path. Two kinds of callers opt in: + // - trusted Cloud tenants (original behavior, kept byte-identical), + // keyed by their tenant identity headers; + // - any other board session, keyed by its user id, so long imports + // survive proxies cutting the connection while the server finishes. + // Jobs are held in memory only and are lost on restart — existing + // semantics; the status route above 404s unknown ids, so a client + // that can no longer see its job treats it as gone and resubmits. cleanupTerminalImportJobs(importJobs, importJobTerminalRetentionMs); - const job = createImportJob(cloudTenantRequestKey(req)); + const isCloudTenant = req.actor.source === "cloud_tenant"; + const actorKey = importJobActorKey(req); + const signature = importRequestSignature(rawImportBody); + if (!isCloudTenant) { + // One live import per board actor: while a job for this key is still + // running, a resubmit of the *same* import gets 409 carrying the + // running job's id and status URL so the client adopts it instead of + // importing the same bundle twice. A *different* import (another tab, + // another target) gets a 409 without a job to adopt, so the client + // surfaces it as an error rather than switching to the wrong result. + // Terminal jobs never block a resubmit. + const running = findRunningImportJob(importJobs, actorKey); + if (running) { + if (running.signature === signature) { + res.status(409).json(importJobConflictResponse(running)); + } else { + res.status(409).json(importAlreadyRunningResponse()); + } + return; + } + } + const job = createImportJob( + actorKey, + isCloudTenant ? "cloud_tenant" : "board", + signature, + ); importJobs.set(job.id, job); const operation = async () => { const importBody = companyPortabilityImportSchema.parse(rawImportBody); @@ -563,9 +601,25 @@ type CompanyImportResult = { warnings: unknown[]; }; +type ImportJobActorKind = "cloud_tenant" | "board"; + +/** + * In-memory only: import jobs do not survive a server restart, and terminal + * jobs are dropped after `importJobTerminalRetentionMs`. Both cases surface + * to pollers as the status route's 404 for an unknown id. + */ interface ImportJobRecord { id: string; - cloudTenantKey: string; + /** Identity that created the job; the only key allowed to read it. */ + actorKey: string; + actorKind: ImportJobActorKind; + /** + * Fingerprint of the import request body. A resubmit is adopted (409 → + * watch the running job) only when it carries the same signature, so a + * *different* import from another tab is rejected instead of silently + * adopting an unrelated job (and its target). Board jobs only. + */ + signature?: string; status: "running" | "succeeded" | "failed"; createdAt: string; updatedAt: string; @@ -577,6 +631,12 @@ interface ImportJobRecord { warningCount: number; companyAction: unknown; }; + /** + * Full import result, retained for board-created jobs only so the import + * page can run the same success path as the synchronous response. Cloud + * tenant job responses keep their original summary-only shape. + */ + fullResult?: CompanyPortabilityImportResult; } interface ImportedCompanyActivityContext { @@ -587,12 +647,6 @@ interface ImportedCompanyActivityContext { include: unknown; } -function assertCloudTenantCaller(req: Request) { - if (req.actor.source !== "cloud_tenant") { - throw forbidden("Trusted Cloud tenant access required"); - } -} - function cloudTenantRequestKey(req: Request) { return [ req.actor.userId ?? "", @@ -601,20 +655,59 @@ function cloudTenantRequestKey(req: Request) { ].join(":"); } -function createImportJob(cloudTenantKey: string): ImportJobRecord { +/** + * Identity a job is created under and authorized against. Cloud tenant + * callers keep their header-derived tenant key; every other board session is + * keyed by its user id. The distinct prefixes keep the two namespaces + * disjoint, so crafted tenant headers can never collide with a board key. + */ +function importJobActorKey(req: Request) { + if (req.actor.source === "cloud_tenant") { + return `cloud-tenant:${cloudTenantRequestKey(req)}`; + } + return `board:${req.actor.userId ?? "local-board"}`; +} + +function createImportJob( + actorKey: string, + actorKind: ImportJobActorKind, + signature?: string, +): ImportJobRecord { const now = new Date().toISOString(); return { - id: `tenant-import-${randomUUID()}`, - cloudTenantKey, + // Cloud tenant job ids keep their original prefix; board jobs get their own. + id: `${actorKind === "cloud_tenant" ? "tenant-import" : "import"}-${randomUUID()}`, + actorKey, + actorKind, + signature, status: "running", createdAt: now, updatedAt: now, }; } +/** + * Stable fingerprint of an import request body. The same client resubmitting + * the same import produces a byte-identical body (no nonce/timestamp), so its + * signature matches; a different import differs. Used to tell a duplicate + * submit apart from a concurrent, unrelated import. + */ +function importRequestSignature(body: unknown): string { + return createHash("sha256").update(JSON.stringify(body) ?? "").digest("hex"); +} + +function findRunningImportJob(importJobs: Map, actorKey: string) { + for (const job of importJobs.values()) { + if (job.status === "running" && job.actorKey === actorKey) { + return job; + } + } + return undefined; +} + async function runImportJob( job: ImportJobRecord, - operation: () => Promise, + operation: () => Promise, ) { try { const result = await operation(); @@ -628,6 +721,9 @@ async function runImportJob( warningCount: result.warnings.length, companyAction: result.company.action, }; + if (job.actorKind === "board") { + job.fullResult = result; + } } catch (error) { const now = new Date().toISOString(); job.status = "failed"; @@ -673,13 +769,41 @@ async function logImportedCompanyActivity( }); } +function importJobStatusUrl(job: ImportJobRecord) { + return `/api/companies/import/jobs/${encodeURIComponent(job.id)}`; +} + function importJobAcceptedResponse(job: ImportJobRecord) { return { job: { id: job.id, status: job.status, }, - statusUrl: `/api/companies/import/jobs/${encodeURIComponent(job.id)}`, + statusUrl: importJobStatusUrl(job), + retryAfterMs: 1000, + }; +} + +/** + * 409 body for a concurrent, *different* import while one is already running. + * Carries no job to adopt, so the client reports it as an error rather than + * watching (and switching to) an unrelated import's result. + */ +function importAlreadyRunningResponse() { + return { + error: "A different import is already running for this account. Wait for it to finish before starting another.", + }; +} + +/** 409 body for a duplicate submit: points at the job already running. */ +function importJobConflictResponse(job: ImportJobRecord) { + return { + error: "An import is already running for this account", + job: { + id: job.id, + status: job.status, + }, + statusUrl: importJobStatusUrl(job), retryAfterMs: 1000, }; } @@ -695,6 +819,9 @@ function importJobResponse(job: ImportJobRecord) { ...(job.completedAt ? { completedAt: job.completedAt } : {}), ...(job.error ? { error: job.error } : {}), ...(job.result ? { result: job.result } : {}), + // Board jobs additionally carry the full import result (parity with + // the synchronous response); cloud tenant jobs never set it. + ...(job.fullResult ? { importResult: job.fullResult } : {}), }, ...(isTerminal ? {} : { retryAfterMs: 1000 }), }; diff --git a/server/src/routes/openapi.ts b/server/src/routes/openapi.ts index 87255b8bd7..f06316a3ff 100644 --- a/server/src/routes/openapi.ts +++ b/server/src/routes/openapi.ts @@ -715,6 +715,7 @@ const BOARD_ONLY_OPERATIONS = new Set([ "POST /api/companies", "GET /api/companies/stats", "GET /api/companies/issues", + "GET /api/companies/import/jobs/{jobId}", "POST /api/board-claim/{token}/claim", "GET /api/cli-auth/me", "POST /api/companies/{companyId}/invites", @@ -5247,8 +5248,19 @@ registry.registerPath({ path: "/api/companies/import", tags: ["companies"], summary: "Apply a company import (legacy route)", + description: + "Board sessions and trusted Cloud tenants can opt into asynchronous processing with the " + + "`x-paperclip-cloud-async-import: 1` header: the server responds 202 with a job id and status " + + "URL instead of holding the connection open for the whole import. While a board actor already " + + "has an async job running, a resubmit returns 409 carrying the running job's id and status URL. " + + "Jobs are held in memory and are lost on restart.", request: { body: jsonBody(companyPortabilityImportSchema) }, - responses: { 200: r.ok(), 400: r.badRequest, 401: r.unauthorized }, + responses: { + 200: r.ok(), + 400: r.badRequest, + 401: r.unauthorized, + 409: { description: "An async import job is already running for this actor" }, + }, }); // ─── Board claim & CLI auth ─────────────────────────────────────────────────── @@ -5495,8 +5507,21 @@ registerCurrentRoute({ summary: "Revoke a board API key", }); +registry.registerPath({ + method: "get", + path: "/api/companies/import/jobs/{jobId}", + tags: ["companies"], + summary: "Get company import job status", + description: + "A job is readable only by the actor that created it — the board user or the trusted Cloud " + + "tenant identity from the async import submission. Any other caller gets the same 404 as an " + + "unknown id. Jobs are held in memory: they are dropped a few minutes after finishing and do " + + "not survive a server restart.", + request: { params: z.object({ jobId: z.string() }) }, + responses: { 200: r.ok(), 400: r.badRequest, 401: r.unauthorized, 404: r.notFound }, +}); + for (const route of [ - ["get", "/api/companies/import/jobs/{jobId}", "Get company import job status"], ["get", "/api/companies/{companyId}/search", "Search company data"], ["get", "/api/companies/{companyId}/search/extract", "Extract company search matches"], ["get", "/api/companies/{companyId}/issues/count", "Count issues in a company"], diff --git a/server/src/services/batch-insert.ts b/server/src/services/batch-insert.ts new file mode 100644 index 0000000000..07700e1718 --- /dev/null +++ b/server/src/services/batch-insert.ts @@ -0,0 +1,69 @@ +// Chunked multi-row insert helper. +// +// PostgreSQL caps a single statement at 65535 bind parameters. A multi-row +// insert binds `columnsPerRow * rowCount` parameters, so large imports must +// split their rows into chunks that stay under that ceiling. This helper keeps +// the arithmetic (and the "insert nothing when there is nothing" guard) in one +// place so every import writer batches identically. + +// One below the hard 65535 ceiling so the arithmetic never lands exactly on it. +export const POSTGRES_MAX_BIND_PARAMS = 65534; + +// A conservative row cap so a single statement stays small even for narrow +// tables. Large enough that the common import tables issue a handful of +// statements, small enough to avoid pathological statement sizes. +export const DEFAULT_INSERT_CHUNK_ROWS = 500; + +// drizzle types `insert` as a per-table generic (``), which a +// table-agnostic helper cannot satisfy; `any` on the boundary is the standard +// escape hatch. Callers pass concretely-typed `db`/`tx` handles and tables. +type InsertExecutor = { + // eslint-disable-next-line @typescript-eslint/no-explicit-any + insert: (table: any) => { values: (rows: any) => unknown }; +}; + +/** + * Insert `rows` into `table` using chunked multi-row statements. + * + * Rows are normalized to a shared column set (the union of keys across the + * batch, with missing/`undefined` values written as `null`) so a single + * multi-row `values()` call stays well-formed even if a caller omitted an + * optional column on some rows. Callers must therefore set every NOT NULL / + * default-backed column they rely on explicitly; only genuinely nullable + * columns should be left absent. + */ +export async function insertRowsInChunks( + executor: InsertExecutor, + table: unknown, + rows: Array>, + options?: { maxRows?: number }, +): Promise { + if (rows.length === 0) return; + + const keys = new Set(); + for (const row of rows) { + for (const key of Object.keys(row)) keys.add(key); + } + const columns = [...keys]; + const normalized = rows.map((row) => { + const out: Record = {}; + for (const key of columns) { + const value = row[key]; + out[key] = value === undefined ? null : value; + } + return out; + }); + + const columnsPerRow = Math.max(1, columns.length); + const chunkSize = Math.max( + 1, + Math.min( + options?.maxRows ?? DEFAULT_INSERT_CHUNK_ROWS, + Math.floor(POSTGRES_MAX_BIND_PARAMS / columnsPerRow), + ), + ); + + for (let start = 0; start < normalized.length; start += chunkSize) { + await executor.insert(table).values(normalized.slice(start, start + chunkSize)); + } +} diff --git a/server/src/services/company-portability.ts b/server/src/services/company-portability.ts index 0e5b5baf85..c49c8d2bf4 100644 --- a/server/src/services/company-portability.ts +++ b/server/src/services/company-portability.ts @@ -7,7 +7,6 @@ import { and, eq, inArray } from "drizzle-orm"; import { builtInManagedResources, issueRelations, - issues as issuesTable, principalPermissionGrants, type Db, } from "@paperclipai/db"; @@ -98,6 +97,13 @@ import { } from "./catalog-provenance.js"; import { readBuiltInAgentMarker } from "./built-in-agent-metadata.js"; import { normalizePortablePath } from "./portable-path.js"; +import type { + ImportIssueRow, + ImportIssueCommentRow, + ImportIssueDocumentRow, + ImportIssueWorkProductRow, + ImportIssueAttachmentRow, +} from "./import-write-types.js"; /** Build OrgNode tree from manifest agent list (slug + reportsToSlug). */ function buildOrgTreeFromManifest(agents: CompanyPortabilityManifest["agents"]): OrgNode[] { @@ -171,6 +177,31 @@ function resolveImportMode(options?: ImportBehaviorOptions): ImportMode { return options?.mode ?? "board_full"; } +/** + * Reject an inline import whose received file set is smaller than the count the + * client declared. The bundle manifest and per-entry blob hashes seal each + * file's contents, but nothing else proves the *set* of files is whole: a + * truncated or proxy-re-framed body can parse into valid JSON with entries + * silently dropped. `expectedFileCount` is the client's assertion of how many + * files it sent, so a shortfall means the payload is incomplete and must fail + * closed instead of importing a fragment. A larger-than-declared set is not a + * truncation symptom and is left alone; the count is optional, so older callers + * that omit it are unaffected. + */ +function assertInlineSourceComplete(source: CompanyPortabilityImport["source"]) { + if (source.type !== "inline") return; + const expected = source.expectedFileCount; + if (expected == null) return; + const received = Object.keys(source.files).length; + if (received < expected) { + throw unprocessable( + `Import payload is incomplete: the request declared ${expected} file(s) but only ${received} arrived. ` + + "The upload was likely truncated; retry the import.", + { code: "import_payload_incomplete", expectedFileCount: expected, receivedFileCount: received }, + ); + } +} + function resolveSkillConflictStrategy(mode: ImportMode, collisionStrategy: CompanyPortabilityCollisionStrategy) { if (mode === "board_full") return "replace" as const; return collisionStrategy === "skip" ? "skip" as const : "rename" as const; @@ -4993,6 +5024,12 @@ export function companyPortabilityService(db: Db, storage?: StorageService) { actorUserId: string | null | undefined, options?: ImportBehaviorOptions, ): Promise { + // Fail closed before any preview or write work when an inline body arrived + // incomplete. A truncated or re-framed request can hand the parser a + // structurally valid JSON object with fewer files than the client sent; + // the declared count is the only signal that distinguishes it from a + // deliberately small bundle. Reject the fragment rather than importing it. + assertInlineSourceComplete(input.source); const mode = resolveImportMode(options); const plan = await buildPreview(input, options); if (plan.preview.errors.length > 0) { @@ -5669,6 +5706,17 @@ export function companyPortabilityService(db: Db, storage?: StorageService) { let attachmentsSkippedNoStorage = 0; const attachmentMaxBytes = normalizeIssueAttachmentMaxBytes(targetCompany.attachmentMaxBytes ?? null); + // Import writes every issue and its children as a single batch instead + // of one network round-trip per row. The loop below resolves each + // manifest issue (ids pre-generated so children reference parents + // without waiting) and buffers the resulting rows; the buffers are + // flushed through the batched writers once resolution is complete. + const issueRows: ImportIssueRow[] = []; + const commentRows: ImportIssueCommentRow[] = []; + const documentRows: ImportIssueDocumentRow[] = []; + const workProductRows: ImportIssueWorkProductRow[] = []; + const attachmentRows: ImportIssueAttachmentRow[] = []; + for (const manifestIssue of sourceManifest.issues) { const markdownRaw = readPortableTextFile(plan.source.files, manifestIssue.path); const parsed = markdownRaw ? parseFrontmatterMarkdown(markdownRaw) : null; @@ -5814,21 +5862,10 @@ export function companyPortabilityService(db: Db, storage?: StorageService) { resolvedLabelIds.push(labelId); } } - const createdIssue = await issues.create(targetCompany.id, { - projectId, - projectWorkspaceId, - title: manifestIssue.title, - description, - assigneeAgentId, - status: issueStatus, - priority: manifestIssue.priority && ISSUE_PRIORITIES.includes(manifestIssue.priority as any) - ? manifestIssue.priority as typeof ISSUE_PRIORITIES[number] - : "medium", - billingCode: manifestIssue.billingCode, - assigneeAdapterOverrides: manifestIssue.assigneeAdapterOverrides, - executionWorkspaceSettings: manifestIssue.executionWorkspaceSettings, - labelIds: resolvedLabelIds, - }); + // Pre-generate the issue id so this issue's comments, documents, and + // attachments can reference it without waiting on a per-issue insert + // round-trip. The row itself is buffered and flushed after the loop. + const issueId = randomUUID(); // Created comment ids are captured positionally so attachment // entries can resolve their commentIndex against them. const createdCommentIds: Array = []; @@ -5849,18 +5886,22 @@ export function companyPortabilityService(db: Db, storage?: StorageService) { : comment.authorType === "user" && actorUserId ? "user" : "system"; - const createdComment = await issues.addComment(createdIssue.id, rewriteEmbeddedAssetUrls(comment.body, embeddedAssetIdMap), { - agentId: authorAgentId ?? undefined, - userId: authorType === "user" ? actorUserId ?? undefined : undefined, - }, { + const commentId = randomUUID(); + commentRows.push({ + id: commentId, + companyId: targetCompany.id, + issueId, + body: rewriteEmbeddedAssetUrls(comment.body, embeddedAssetIdMap), authorType, - presentation: comment.presentation, - metadata: comment.metadata, - createdAt: comment.createdAt, + authorAgentId: authorAgentId ?? null, + authorUserId: authorType === "user" ? actorUserId ?? null : null, + presentation: comment.presentation ?? null, + metadata: comment.metadata ?? null, + createdAt: comment.createdAt ?? null, }); - createdCommentIds.push(createdComment?.id ?? null); + createdCommentIds.push(commentId); } - importedIssueIdBySlug.set(manifestIssue.slug, createdIssue.id); + importedIssueIdBySlug.set(manifestIssue.slug, issueId); if ((manifestIssue.blockedBy ?? []).length > 0) { blockedByBySlug.set(manifestIssue.slug, manifestIssue.blockedBy ?? []); } @@ -5870,33 +5911,35 @@ export function companyPortabilityService(db: Db, storage?: StorageService) { warnings.push(`Task ${manifestIssue.slug} document ${documentEntry.key} was skipped because its file is missing from the package: ${documentEntry.path}`); continue; } - try { - await documentsSvc.upsertIssueDocument({ - issueId: createdIssue.id, - key: documentEntry.key, - title: documentEntry.title, - format: documentEntry.format, - body: rewriteEmbeddedAssetUrls(documentBody, embeddedAssetIdMap), - createdByUserId: actorUserId ?? null, - }); - } catch (error) { - warnings.push(`Task ${manifestIssue.slug} document ${documentEntry.key} could not be imported: ${error instanceof Error ? error.message : String(error)}`); - } + documentRows.push({ + companyId: targetCompany.id, + issueId, + key: documentEntry.key, + title: documentEntry.title, + format: documentEntry.format, + body: rewriteEmbeddedAssetUrls(documentBody, embeddedAssetIdMap), + createdByAgentId: null, + createdByUserId: actorUserId ?? null, + createdByRunId: null, + sourceTrust: null, + }); } for (const workProductEntry of manifestIssue.workProducts ?? []) { - await workProductsSvc.createForIssue(createdIssue.id, targetCompany.id, { - projectId: createdIssue.projectId ?? projectId ?? null, + workProductRows.push({ + companyId: targetCompany.id, + issueId, + projectId: projectId ?? null, type: workProductEntry.type, provider: workProductEntry.provider, - externalId: workProductEntry.externalId, + externalId: workProductEntry.externalId ?? null, title: workProductEntry.title, - url: workProductEntry.url, + url: workProductEntry.url ?? null, status: workProductEntry.status, - reviewState: workProductEntry.reviewState, - isPrimary: workProductEntry.isPrimary, - healthStatus: workProductEntry.healthStatus, - summary: workProductEntry.summary, - metadata: workProductEntry.metadata, + reviewState: workProductEntry.reviewState ?? "none", + isPrimary: workProductEntry.isPrimary ?? false, + healthStatus: workProductEntry.healthStatus ?? "unknown", + summary: workProductEntry.summary ?? null, + metadata: workProductEntry.metadata ?? null, // Workspace/run references never travel across boards. executionWorkspaceId: null, runtimeServiceId: null, @@ -5904,17 +5947,15 @@ export function companyPortabilityService(db: Db, storage?: StorageService) { sourceTrust: null, }); } + // Monitors land un-armed: notes and provenance are restored on the + // issue row itself but monitorNextCheckAt stays NULL until an operator + // re-arms them. + let monitorNotes: string | null = null; + let monitorScheduledBy: string | null = null; if (manifestIssue.monitor) { - // Monitors land un-armed: notes and provenance are restored but - // monitorNextCheckAt stays NULL until an operator re-arms them. if (manifestIssue.monitor.notes !== null || manifestIssue.monitor.scheduledBy !== null) { - await db - .update(issuesTable) - .set({ - monitorNotes: manifestIssue.monitor.notes, - monitorScheduledBy: manifestIssue.monitor.scheduledBy, - }) - .where(eq(issuesTable.id, createdIssue.id)); + monitorNotes = manifestIssue.monitor.notes; + monitorScheduledBy = manifestIssue.monitor.scheduledBy; } unarmedMonitorCount += 1; } @@ -5950,13 +5991,14 @@ export function companyPortabilityService(db: Db, storage?: StorageService) { try { const stored = await storage.putFile({ companyId: targetCompany.id, - namespace: `issues/${createdIssue.id}`, + namespace: `issues/${issueId}`, originalFilename: attachmentEntry.originalFilename, contentType: attachmentEntry.contentType, body, }); - await issues.createAttachment({ - issueId: createdIssue.id, + attachmentRows.push({ + companyId: targetCompany.id, + issueId, issueCommentId, provider: stored.provider, objectKey: stored.objectKey, @@ -5971,8 +6013,39 @@ export function companyPortabilityService(db: Db, storage?: StorageService) { warnings.push(`Task ${manifestIssue.slug} attachment ${attachmentLabel} could not be imported: ${error instanceof Error ? error.message : String(error)}`); } } + issueRows.push({ + id: issueId, + ref: manifestIssue.slug, + projectId: projectId ?? null, + projectWorkspaceId: projectWorkspaceId ?? null, + title: manifestIssue.title, + description, + assigneeAgentId, + status: issueStatus, + priority: manifestIssue.priority && ISSUE_PRIORITIES.includes(manifestIssue.priority as any) + ? manifestIssue.priority as typeof ISSUE_PRIORITIES[number] + : "medium", + billingCode: manifestIssue.billingCode ?? null, + assigneeAdapterOverrides: manifestIssue.assigneeAdapterOverrides ?? null, + executionWorkspaceSettings: manifestIssue.executionWorkspaceSettings ?? null, + labelIds: resolvedLabelIds, + monitorNotes, + monitorScheduledBy, + }); } + // Flush the buffered rows in dependency order: issues first (parents of + // every other row), then comments (attachments may reference them), then + // the remaining children. Each writer inserts in chunked multi-row + // statements, turning what used to be one round-trip per row into a + // handful per table. Empty buffers are skipped so an issues-free import + // (e.g. routines only) issues no writes at all. + if (issueRows.length > 0) await issues.importIssues(targetCompany.id, issueRows); + if (commentRows.length > 0) await issues.addImportedComments(commentRows); + if (documentRows.length > 0) await documentsSvc.createIssueDocumentsForImport(documentRows); + if (workProductRows.length > 0) await workProductsSvc.createManyForImport(workProductRows); + if (attachmentRows.length > 0) await issues.addImportedAttachments(attachmentRows); + if (blockedByBySlug.size > 0) { const acceptedAdjacency = new Map(); const wouldCreateBlockingCycle = (blockedIssueId: string, blockerIssueId: string) => { diff --git a/server/src/services/documents.ts b/server/src/services/documents.ts index ff35742650..b30f79b22b 100644 --- a/server/src/services/documents.ts +++ b/server/src/services/documents.ts @@ -1,8 +1,11 @@ +import { randomUUID } from "node:crypto"; import { and, asc, desc, eq } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; import { documentRevisions, documents, issueDocuments, issues } from "@paperclipai/db"; import { isSystemIssueDocumentKey, issueDocumentKeySchema } from "@paperclipai/shared"; import { conflict, notFound, unprocessable } from "../errors.js"; +import { insertRowsInChunks } from "./batch-insert.js"; +import type { ImportIssueDocumentRow } from "./import-write-types.js"; function normalizeDocumentKey(key: string) { const normalized = key.trim().toLowerCase(); @@ -512,6 +515,74 @@ export function documentService(db: Db) { throw conflict("Unable to choose a new document key for locked document", { key }); }, + /** + * Batched issue-document insert for company import. + * + * Every imported document is a fresh create (the issue is brand new), so we + * skip {@link upsertIssueDocument}'s per-row existence/lock/base-revision + * dance and the follow-up latest-revision update: ids are pre-generated so + * `latest_revision_id` can be written inline. Documents, their initial + * revisions, and the issue links are each inserted in chunked statements. + */ + createIssueDocumentsForImport: async (rows: ImportIssueDocumentRow[]): Promise => { + if (rows.length === 0) return; + const now = new Date(); + const documentRows: Array> = []; + const revisionRows: Array> = []; + const linkRows: Array> = []; + for (const row of rows) { + const key = normalizeDocumentKey(row.key); + const documentId = randomUUID(); + const revisionId = randomUUID(); + documentRows.push({ + id: documentId, + companyId: row.companyId, + title: row.title ?? null, + format: row.format, + latestBody: row.body, + latestRevisionId: revisionId, + latestRevisionNumber: 1, + createdByAgentId: row.createdByAgentId ?? null, + createdByUserId: row.createdByUserId ?? null, + updatedByAgentId: row.createdByAgentId ?? null, + updatedByUserId: row.createdByUserId ?? null, + lockedAt: null, + lockedByAgentId: null, + lockedByUserId: null, + sourceTrust: row.sourceTrust ?? null, + createdAt: now, + updatedAt: now, + }); + revisionRows.push({ + id: revisionId, + companyId: row.companyId, + documentId, + revisionNumber: 1, + title: row.title ?? null, + format: row.format, + body: row.body, + changeSummary: null, + createdByAgentId: row.createdByAgentId ?? null, + createdByUserId: row.createdByUserId ?? null, + createdByRunId: row.createdByRunId ?? null, + createdAt: now, + }); + linkRows.push({ + companyId: row.companyId, + issueId: row.issueId, + documentId, + key, + createdAt: now, + updatedAt: now, + }); + } + await db.transaction(async (tx) => { + await insertRowsInChunks(tx, documents, documentRows); + await insertRowsInChunks(tx, documentRevisions, revisionRows); + await insertRowsInChunks(tx, issueDocuments, linkRows); + }); + }, + restoreIssueDocumentRevision: async (input: { issueId: string; key: string; diff --git a/server/src/services/import-write-types.ts b/server/src/services/import-write-types.ts new file mode 100644 index 0000000000..264f80f97a --- /dev/null +++ b/server/src/services/import-write-types.ts @@ -0,0 +1,103 @@ +// Row shapes handed to the batched company-import writers. +// +// Company import resolves every entity up front (slugs → ids, label remaps, +// status downgrades, blob verification, embedded-asset rewrites) and then hands +// fully-resolved rows to these writers, which insert them in chunked multi-row +// statements. Ids are pre-generated by the caller so children can reference +// their parents without a per-row `.returning()` round-trip. + +import type { + IssueCommentAuthorType, + IssueCommentMetadata, + IssueCommentPresentation, + IssuePriority, + IssueStatus, + SourceTrustMetadata, +} from "@paperclipai/shared"; + +/** A resolved issue row with a pre-generated id, ready for batch insert. */ +export interface ImportIssueRow { + /** Pre-generated issue id (app-supplied uuid). */ + id: string; + /** Source slug, so the caller can correlate the row after the batch. */ + ref: string; + projectId: string | null; + projectWorkspaceId: string | null; + title: string; + description: string | null; + assigneeAgentId: string | null; + status: IssueStatus; + priority: IssuePriority; + billingCode: string | null; + assigneeAdapterOverrides: Record | null; + executionWorkspaceSettings: Record | null; + labelIds: string[]; + /** Imported monitors land un-armed; only notes/provenance are restored. */ + monitorNotes: string | null; + monitorScheduledBy: string | null; +} + +/** A resolved comment row with a pre-generated id, ready for batch insert. */ +export interface ImportIssueCommentRow { + id: string; + companyId: string; + issueId: string; + body: string; + authorType: IssueCommentAuthorType; + authorAgentId: string | null; + authorUserId: string | null; + presentation: IssueCommentPresentation | null; + metadata: IssueCommentMetadata | null; + createdAt: Date | string | null; +} + +/** A resolved attachment (asset + link) row, ready for batch insert. */ +export interface ImportIssueAttachmentRow { + companyId: string; + issueId: string; + issueCommentId: string | null; + provider: string; + objectKey: string; + contentType: string; + byteSize: number; + sha256: string; + originalFilename: string | null; + createdByAgentId: string | null; + createdByUserId: string | null; +} + +/** A resolved issue-document row, ready for batch insert. */ +export interface ImportIssueDocumentRow { + companyId: string; + issueId: string; + key: string; + title: string | null; + format: string; + body: string; + createdByAgentId: string | null; + createdByUserId: string | null; + createdByRunId: string | null; + sourceTrust: SourceTrustMetadata | null; +} + +/** A resolved work-product row, ready for batch insert. */ +export interface ImportIssueWorkProductRow { + companyId: string; + issueId: string; + projectId: string | null; + type: string; + provider: string; + externalId: string | null; + title: string; + url: string | null; + status: string; + reviewState: string; + isPrimary: boolean; + healthStatus: string; + summary: string | null; + metadata: Record | null; + executionWorkspaceId: string | null; + runtimeServiceId: string | null; + createdByRunId: string | null; + sourceTrust: SourceTrustMetadata | null; +} diff --git a/server/src/services/issues.ts b/server/src/services/issues.ts index b5e6811cf3..f30bc272f0 100644 --- a/server/src/services/issues.ts +++ b/server/src/services/issues.ts @@ -1,5 +1,5 @@ import { Buffer } from "node:buffer"; -import { createHash } from "node:crypto"; +import { createHash, randomUUID } from "node:crypto"; import { and, asc, desc, eq, gt, gte, inArray, isNull, like, lt, ne, notInArray, or, sql, type SQL } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; import { @@ -92,6 +92,12 @@ import { resolveIssueGoalId, resolveNextIssueGoalId } from "./issue-goal-fallbac import { getRunLogStore } from "./run-log-store.js"; import { getDefaultCompanyGoal } from "./goals.js"; import { assertAssignableAgent } from "./agent-assignability.js"; +import { insertRowsInChunks } from "./batch-insert.js"; +import type { + ImportIssueRow, + ImportIssueCommentRow, + ImportIssueAttachmentRow, +} from "./import-write-types.js"; import { summarizeIssueWatchdog, upsertIssueWatchdogForIssue, @@ -6562,6 +6568,245 @@ export function issueService(db: Db) { }); }, + /** + * Batched issue insert for company import. + * + * Company import used to call {@link create} once per issue — each call a + * separate network round-trip that inserted the issue, then serialized the + * issue's comments/documents behind the returned id. This inserts a whole + * bundle of pre-resolved issues (ids already generated by the caller) in + * chunked multi-row statements, so a thousand-issue import issues a handful + * of statements instead of thousands. + * + * The per-issue derivations {@link create} performs are reproduced for the + * import subset: a single contiguous identifier range is allocated from the + * company counter, per-project goal/workspace/policy defaults are computed + * once and cached, assignable-agent and workspace validation run per + * distinct id, and monitor notes land un-armed on the row. Dedup, + * idempotency, watchdogs, workspace inheritance and blocked-by wiring — none + * of which import uses — are intentionally omitted. + */ + importIssues: async (companyId: string, rows: ImportIssueRow[]): Promise => { + if (rows.length === 0) return; + const isolatedWorkspacesEnabled = (await instanceSettings.getExperimental()).enableIsolatedWorkspaces; + await db.transaction(async (tx) => { + // Self-correcting counter: seed from max(issue_number) so a drifted + // company counter cannot mint colliding identifiers, then reserve the + // whole range in one bump instead of one-per-issue. + const [maxRow] = await tx + .select({ maxNum: sql`coalesce(max(${issues.issueNumber}), 0)` }) + .from(issues) + .where(eq(issues.companyId, companyId)); + const currentMax = maxRow?.maxNum ?? 0; + const [company] = await tx + .select({ issueCounter: companies.issueCounter, issuePrefix: companies.issuePrefix }) + .from(companies) + .where(eq(companies.id, companyId)); + if (!company) throw notFound("Target company not found"); + const base = Math.max(company.issueCounter ?? 0, currentMax); + await tx + .update(companies) + .set({ issueCounter: base + rows.length }) + .where(eq(companies.id, companyId)); + + const defaultCompanyGoal = await getDefaultCompanyGoal(tx, companyId); + const defaultGoalId = defaultCompanyGoal?.id ?? null; + + // Project-scoped derivations depend only on the project, so resolve each + // distinct project once and reuse it across that project's issues. + const projectDerivedCache = new Map< + string, + { + goalId: string | null; + defaultProjectWorkspaceId: string | null; + defaultExecutionWorkspaceSettings: Record | null; + } + >(); + const loadProjectDerived = async (projectId: string) => { + const cached = projectDerivedCache.get(projectId); + if (cached) return cached; + const projectRow = await tx + .select({ + goalId: projects.goalId, + executionWorkspacePolicy: projects.executionWorkspacePolicy, + }) + .from(projects) + .where(and(eq(projects.id, projectId), eq(projects.companyId, companyId))) + .then((r) => r[0] ?? null); + const policy = parseProjectExecutionWorkspacePolicy(projectRow?.executionWorkspacePolicy); + let defaultProjectWorkspaceId = policy?.defaultProjectWorkspaceId ?? null; + if (!defaultProjectWorkspaceId) { + defaultProjectWorkspaceId = await tx + .select({ id: projectWorkspaces.id }) + .from(projectWorkspaces) + .where(and(eq(projectWorkspaces.projectId, projectId), eq(projectWorkspaces.companyId, companyId))) + .orderBy(desc(projectWorkspaces.isPrimary), asc(projectWorkspaces.createdAt), asc(projectWorkspaces.id)) + .then((r) => r[0]?.id ?? null); + } + const defaultExecutionWorkspaceSettings = defaultIssueExecutionWorkspaceSettingsForProject( + gateProjectExecutionWorkspacePolicy(policy, isolatedWorkspacesEnabled), + ) as Record | null; + const derived = { goalId: projectRow?.goalId ?? null, defaultProjectWorkspaceId, defaultExecutionWorkspaceSettings }; + projectDerivedCache.set(projectId, derived); + return derived; + }; + + const validatedAgentIds = new Set(); + const validatedWorkspaceKeys = new Set(); + const issueRows: Array> = []; + const labelRows: Array<{ issueId: string; labelId: string; companyId: string }> = []; + + let counter = base; + for (const row of rows) { + counter += 1; + const issueNumber = counter; + const identifier = `${company.issuePrefix}-${issueNumber}`; + + if (row.assigneeAgentId) { + if (!validatedAgentIds.has(row.assigneeAgentId)) { + await assertAssignableAgent(tx as unknown as Db, companyId, row.assigneeAgentId, { kind: "work" }); + validatedAgentIds.add(row.assigneeAgentId); + } + } + if (row.status === "in_progress" && !row.assigneeAgentId) { + throw unprocessable("in_progress issues require an assignee"); + } + + const projectId = row.projectId ?? null; + let projectWorkspaceId = row.projectWorkspaceId ?? null; + // Isolated-workspace fields are gated the same way create() gates them: + // when the experiment is off the imported settings are dropped. + let executionWorkspaceSettings = isolatedWorkspacesEnabled + ? (row.executionWorkspaceSettings ?? null) + : null; + let projectGoalId: string | null = null; + if (projectId) { + const derived = await loadProjectDerived(projectId); + projectGoalId = derived.goalId; + if (!projectWorkspaceId) projectWorkspaceId = derived.defaultProjectWorkspaceId; + if (executionWorkspaceSettings == null) { + executionWorkspaceSettings = derived.defaultExecutionWorkspaceSettings; + } + } + if (projectWorkspaceId) { + const workspaceKey = `${projectId ?? ""}:${projectWorkspaceId}`; + if (!validatedWorkspaceKeys.has(workspaceKey)) { + await assertValidProjectWorkspace(companyId, projectId, projectWorkspaceId, tx); + validatedWorkspaceKeys.add(workspaceKey); + } + } + + const goalId = resolveIssueGoalId({ + projectId, + goalId: null, + projectGoalId, + defaultGoalId, + }); + + issueRows.push({ + id: row.id, + companyId, + issueNumber, + identifier, + title: row.title, + description: row.description ?? null, + assigneeAgentId: row.assigneeAgentId ?? null, + status: row.status, + priority: row.priority, + billingCode: row.billingCode ?? null, + assigneeAdapterOverrides: row.assigneeAdapterOverrides ?? null, + projectId, + projectWorkspaceId, + executionWorkspaceSettings, + goalId, + responsibleUserId: null, + requestDepth: clampIssueRequestDepth(undefined), + originKind: "manual", + startedAt: row.status === "in_progress" ? new Date() : null, + completedAt: row.status === "done" ? new Date() : null, + cancelledAt: row.status === "cancelled" ? new Date() : null, + monitorNotes: row.monitorNotes ?? null, + monitorScheduledBy: row.monitorScheduledBy ?? null, + }); + for (const labelId of new Set(row.labelIds ?? [])) { + labelRows.push({ issueId: row.id, labelId, companyId }); + } + } + + await insertRowsInChunks(tx, issues, issueRows); + await insertRowsInChunks(tx, issueLabels, labelRows); + }); + }, + + /** + * Batched comment insert for company import. Comment ids are pre-generated + * by the caller so attachments can reference them without a round-trip. + */ + addImportedComments: async (rows: ImportIssueCommentRow[]): Promise => { + if (rows.length === 0) return; + const censorUsernameInLogs = (await instanceSettings.getGeneral()).censorUsernameInLogs; + await db.transaction(async (tx) => { + const commentRows = rows.map((row) => { + const createdAt = row.createdAt ? new Date(row.createdAt) : null; + return { + id: row.id, + companyId: row.companyId, + issueId: row.issueId, + authorAgentId: row.authorAgentId ?? null, + authorUserId: row.authorUserId ?? null, + authorType: row.authorType, + createdByRunId: null, + body: redactCurrentUserText(row.body, { enabled: censorUsernameInLogs }), + presentation: row.presentation ?? null, + metadata: row.metadata ?? null, + sourceTrust: null, + createdAt: createdAt && !Number.isNaN(createdAt.getTime()) ? createdAt : new Date(), + }; + }); + await insertRowsInChunks(tx, issueComments, commentRows); + // Mirror addComment's recency bump, once per affected issue. + const issueIds = [...new Set(rows.map((row) => row.issueId))]; + if (issueIds.length > 0) { + await tx.update(issues).set({ updatedAt: new Date() }).where(inArray(issues.id, issueIds)); + } + }); + }, + + /** + * Batched attachment insert for company import: each row mints an asset and + * links it to its issue (and optionally comment) in two chunked statements. + */ + addImportedAttachments: async (rows: ImportIssueAttachmentRow[]): Promise => { + if (rows.length === 0) return; + await db.transaction(async (tx) => { + const assetRows: Array> = []; + const attachmentRows: Array> = []; + for (const row of rows) { + const assetId = randomUUID(); + assetRows.push({ + id: assetId, + companyId: row.companyId, + provider: row.provider, + objectKey: row.objectKey, + contentType: row.contentType, + byteSize: row.byteSize, + sha256: row.sha256, + originalFilename: row.originalFilename ?? null, + createdByAgentId: row.createdByAgentId ?? null, + createdByUserId: row.createdByUserId ?? null, + }); + attachmentRows.push({ + companyId: row.companyId, + issueId: row.issueId, + assetId, + issueCommentId: row.issueCommentId ?? null, + }); + } + await insertRowsInChunks(tx, assets, assetRows); + await insertRowsInChunks(tx, issueAttachments, attachmentRows); + }); + }, + update: async ( id: string, data: Partial & { diff --git a/server/src/services/work-products.ts b/server/src/services/work-products.ts index 61c88c771a..55d2a8c714 100644 --- a/server/src/services/work-products.ts +++ b/server/src/services/work-products.ts @@ -2,6 +2,8 @@ import { and, desc, eq } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; import { issueWorkProducts } from "@paperclipai/db"; import type { IssueWorkProduct } from "@paperclipai/shared"; +import { insertRowsInChunks } from "./batch-insert.js"; +import type { ImportIssueWorkProductRow } from "./import-write-types.js"; type IssueWorkProductRow = typeof issueWorkProducts.$inferSelect; @@ -110,6 +112,52 @@ export function workProductService(db: Db) { return row ? toIssueWorkProduct(row) : null; }, + /** + * Batched work-product insert for company import. + * + * {@link createForIssue} clears the prior primary of the same type on every + * call; imported issues are brand new, so the only primaries in play are the + * imported rows themselves. We reproduce "last primary wins" within each + * (issue, type) group and insert the whole batch in chunked statements. + */ + createManyForImport: async (rows: ImportIssueWorkProductRow[]): Promise => { + if (rows.length === 0) return; + const lastPrimaryIndexByGroup = new Map(); + rows.forEach((row, index) => { + if (row.isPrimary) lastPrimaryIndexByGroup.set(`${row.issueId}:${row.type}`, index); + }); + const values = rows.map((row, index) => ({ + companyId: row.companyId, + issueId: row.issueId, + projectId: row.projectId ?? null, + type: row.type, + provider: row.provider, + externalId: row.externalId ?? null, + title: row.title, + url: row.url ?? null, + status: row.status, + reviewState: row.reviewState, + isPrimary: row.isPrimary + ? lastPrimaryIndexByGroup.get(`${row.issueId}:${row.type}`) === index + : false, + healthStatus: row.healthStatus, + summary: row.summary ?? null, + metadata: row.metadata ?? null, + executionWorkspaceId: row.executionWorkspaceId ?? null, + runtimeServiceId: row.runtimeServiceId ?? null, + createdByRunId: row.createdByRunId ?? null, + sourceTrust: row.sourceTrust ?? null, + })); + // Chunked writes are wrapped in a single transaction so a large import + // that spans multiple insert statements is atomic: if a later chunk + // fails, the earlier chunks roll back rather than leaving a partial + // prefix behind (which a retry would then duplicate). Mirrors the + // per-writer transaction the batched issue/document writers use. + await db.transaction(async (tx) => { + await insertRowsInChunks(tx, issueWorkProducts, values); + }); + }, + remove: async (id: string) => { const row = await db .delete(issueWorkProducts) diff --git a/ui/src/api/client.ts b/ui/src/api/client.ts index a90efb8fc1..7505278c20 100644 --- a/ui/src/api/client.ts +++ b/ui/src/api/client.ts @@ -17,6 +17,8 @@ export class ApiError extends Error { export interface RequestOptions { /** Abort signal wired through to `fetch` and coalescing (per-caller). */ signal?: AbortSignal; + /** Extra request headers (e.g. the async-import opt-in). Mutations only. */ + headers?: Record; } function abortError(): DOMException { @@ -146,7 +148,12 @@ function isRequestOptions(value: unknown): value is RequestOptions { export const api = { get: (path: string, options?: RequestOptions) => coalescedGet(path, options), post: (path: string, body: unknown, options?: RequestOptions) => - request(path, { method: "POST", body: JSON.stringify(body), signal: options?.signal }), + request(path, { + method: "POST", + body: JSON.stringify(body), + signal: options?.signal, + ...(options?.headers ? { headers: options.headers } : {}), + }), postForm: (path: string, body: FormData, options?: RequestOptions) => request(path, { method: "POST", body, signal: options?.signal }), put: (path: string, body: unknown, options?: RequestOptions) => diff --git a/ui/src/api/companies.ts b/ui/src/api/companies.ts index 222262d352..40b334aeb2 100644 --- a/ui/src/api/companies.ts +++ b/ui/src/api/companies.ts @@ -14,6 +14,29 @@ import { api } from "./client"; export type CompanyStats = Record; +export type CompanyImportJobState = "running" | "succeeded" | "failed"; + +/** 202 body from the async opt-in on POST /companies/import (and the 409 body when a job is already running). */ +export interface CompanyImportJobAccepted { + job: { id: string; status: CompanyImportJobState }; + statusUrl: string; + retryAfterMs?: number; +} + +export interface CompanyImportJobStatus { + job: { + id: string; + status: CompanyImportJobState; + createdAt?: string; + updatedAt?: string; + completedAt?: string; + error?: { message: string }; + /** Board-created jobs carry the full result for parity with the sync response. */ + importResult?: CompanyPortabilityImportResult; + }; + retryAfterMs?: number; +} + export const companiesApi = { list: () => api.get("/companies"), get: (companyId: string) => api.get(`/companies/${companyId}`), @@ -61,4 +84,11 @@ export const companiesApi = { api.post("/companies/import/preview", data), importBundle: (data: CompanyPortabilityImportRequest) => api.post("/companies/import", data), + /** Submit an import as a server-side job: 202 with a job id to poll, or 409 with the already-running job. */ + importBundleAsync: (data: CompanyPortabilityImportRequest) => + api.post("/companies/import", data, { + headers: { "x-paperclip-cloud-async-import": "1" }, + }), + getImportJob: (jobId: string) => + api.get(`/companies/import/jobs/${encodeURIComponent(jobId)}`), }; diff --git a/ui/src/lib/import-job-watch.ts b/ui/src/lib/import-job-watch.ts new file mode 100644 index 0000000000..21ae039bf3 --- /dev/null +++ b/ui/src/lib/import-job-watch.ts @@ -0,0 +1,98 @@ +/** + * Client-side bookkeeping for async company import jobs. + * + * The import page submits imports as server-side jobs and polls their status, + * so a dropped connection or a page reload must not lose track of a running + * import. The job id is mirrored into `sessionStorage` (keyed per company and + * package) when a job is accepted and removed once it reaches a terminal + * state, letting the page resume watching after a reload. Jobs live in server + * memory only: a stored id the server no longer knows (restart, retention + * expiry) surfaces as a 404 and the stored entry is discarded. + */ + +export const IMPORT_JOB_POLL_INTERVAL_MS = 3000; + +const STORAGE_PREFIX = "paperclip:company-import-job"; + +export interface StoredImportJob { + jobId: string; + pauseAutomations: boolean; +} + +export function importJobStorageKey(companyId: string, packageName: string): string { + return `${STORAGE_PREFIX}:${companyId}:${packageName}`; +} + +export function writeStoredImportJob(storageKey: string, value: StoredImportJob): void { + try { + sessionStorage.setItem(storageKey, JSON.stringify(value)); + } catch { + // Storage may be unavailable (quota, privacy mode); polling still works, + // only reload-resume is lost. + } +} + +export function clearStoredImportJob(storageKey: string): void { + try { + sessionStorage.removeItem(storageKey); + } catch { + // Ignore storage failures; a stale entry is discarded on the next read. + } +} + +/** + * Find a stored job for this company. Keys are scoped per package name, so + * scan the company's prefix; the server allows one live job per user, so at + * most one non-terminal entry is expected. + */ +export function readStoredImportJob( + companyId: string, +): (StoredImportJob & { storageKey: string }) | null { + try { + const prefix = `${STORAGE_PREFIX}:${companyId}:`; + for (let index = 0; index < sessionStorage.length; index += 1) { + const key = sessionStorage.key(index); + if (!key || !key.startsWith(prefix)) continue; + const raw = sessionStorage.getItem(key); + if (raw) { + try { + const parsed: unknown = JSON.parse(raw); + if ( + typeof parsed === "object" && parsed !== null && + typeof (parsed as { jobId?: unknown }).jobId === "string" + ) { + return { + jobId: (parsed as { jobId: string }).jobId, + pauseAutomations: (parsed as { pauseAutomations?: unknown }).pauseAutomations === true, + storageKey: key, + }; + } + } catch { + // Fall through to discard the malformed entry. + } + } + sessionStorage.removeItem(key); + index -= 1; + } + } catch { + // Storage unavailable — nothing to resume. + } + return null; +} + +function sleep(ms: number): Promise { + return new Promise((resolve) => setTimeout(resolve, ms)); +} + +/** + * Wait one poll interval before the next status request. While the tab is + * hidden, keep waiting instead of polling: the job runs server-side, so a + * backgrounded page skips status requests until it is visible again. + */ +export async function waitForNextImportJobPoll( + intervalMs: number = IMPORT_JOB_POLL_INTERVAL_MS, +): Promise { + do { + await sleep(intervalMs); + } while (typeof document !== "undefined" && document.hidden); +} diff --git a/ui/src/pages/CompanyImport.test.tsx b/ui/src/pages/CompanyImport.test.tsx index bd644ac769..c02ec36bdf 100644 --- a/ui/src/pages/CompanyImport.test.tsx +++ b/ui/src/pages/CompanyImport.test.tsx @@ -5,11 +5,15 @@ import { createRoot, type Root } from "react-dom/client"; import { QueryClient, QueryClientProvider } from "@tanstack/react-query"; import type { CompanyPortabilityImportResult, CompanyPortabilityPreviewResult } from "@paperclipai/shared"; import { afterEach, beforeEach, describe, expect, it, vi } from "vitest"; +import { ApiError } from "../api/client"; +import type { CompanyImportJobAccepted } from "../api/companies"; import { CompanyImport } from "./CompanyImport"; const mockCompaniesApi = vi.hoisted(() => ({ importPreview: vi.fn(), importBundle: vi.fn(), + importBundleAsync: vi.fn(), + getImportJob: vi.fn(), get: vi.fn(), })); const mockAgentsApi = vi.hoisted(() => ({ @@ -33,6 +37,13 @@ vi.mock("../api/companies", () => ({ companiesApi: mockCompaniesApi, })); +// Keep the real sessionStorage bookkeeping so reload-resume is exercised, but +// make the poll delay instant so tests never wait the real 3s interval. +vi.mock("../lib/import-job-watch", async (importOriginal) => { + const actual = await importOriginal(); + return { ...actual, waitForNextImportJobPoll: () => Promise.resolve() }; +}); + vi.mock("../lib/zip", () => ({ readZipArchive: mockReadZipArchive, })); @@ -102,6 +113,14 @@ async function flushReact() { }); } +// The async import runs submit → poll → onSuccess, several awaits deep. Drain a +// handful of macrotask turns so the outcome/activation panels have committed. +async function settle(times = 6) { + for (let index = 0; index < times; index += 1) { + await flushReact(); + } +} + const previewFiles = { ".paperclip.yaml": 'schema: "paperclip/v1"\n', "agents/coder/AGENTS.md": "---\nname: Coder\n---\n\nYou write code.\n", @@ -146,11 +165,20 @@ function buildImportResult(): CompanyPortabilityImportResult { }; } +function buildAccepted(id = "job-1"): CompanyImportJobAccepted { + return { job: { id, status: "running" }, statusUrl: `/companies/import/jobs/${id}` }; +} + +function buildSucceededJob(id = "job-1") { + return { job: { id, status: "succeeded" as const, importResult: buildImportResult() } }; +} + describe("CompanyImport", () => { let container: HTMLDivElement; let root: Root | null = null; beforeEach(() => { + sessionStorage.clear(); container = document.createElement("div"); document.body.appendChild(container); mockAuthApi.getSession.mockResolvedValue({ user: { id: "user-1" } }); @@ -158,7 +186,10 @@ describe("CompanyImport", () => { mockAgentsApi.resume.mockResolvedValue({ id: "agent-1", status: "idle" }); mockRoutinesApi.update.mockResolvedValue({ id: "routine-1", status: "active" }); mockCompaniesApi.importPreview.mockResolvedValue(buildPreviewResult()); - mockCompaniesApi.importBundle.mockResolvedValue(buildImportResult()); + // Default async flow: the submit is accepted (202) and the first poll finds + // the job already finished with the full result. Individual tests override. + mockCompaniesApi.importBundleAsync.mockResolvedValue(buildAccepted()); + mockCompaniesApi.getImportJob.mockResolvedValue(buildSucceededJob()); mockCompaniesApi.get.mockResolvedValue({ id: "company-2", name: "Imported Test", issuePrefix: "IMP" }); mockSidebarPreferencesApi.updateProjectOrder.mockResolvedValue(undefined); }); @@ -173,6 +204,7 @@ describe("CompanyImport", () => { } container.remove(); document.body.innerHTML = ""; + sessionStorage.clear(); vi.clearAllMocks(); }); @@ -191,7 +223,7 @@ describe("CompanyImport", () => { await flushReact(); } - async function renderPageAndImport() { + async function renderPage() { root = createRoot(container); const queryClient = new QueryClient({ defaultOptions: { queries: { retry: false } } }); const currentRoot = root; @@ -204,17 +236,24 @@ describe("CompanyImport", () => { ); }); await flushReact(); + } + async function enterGithubUrl(url = "https://github.com/acme/starter/tree/main/company") { const urlInput = container.querySelector( 'input[placeholder="https://github.com/owner/repo/tree/main/company"]', ); expect(urlInput).toBeTruthy(); await act(async () => { const setter = Object.getOwnPropertyDescriptor(HTMLInputElement.prototype, "value")!.set!; - setter.call(urlInput!, "https://github.com/acme/starter/tree/main/company"); + setter.call(urlInput!, url); urlInput!.dispatchEvent(new Event("input", { bubbles: true })); }); await flushReact(); + } + + async function renderPageAndImport() { + await renderPage(); + await enterGithubUrl(); await clickButton((text) => text === "Preview import"); @@ -225,14 +264,18 @@ describe("CompanyImport", () => { expect(pauseLabel?.querySelector('input[type="checkbox"]')?.checked).toBe(true); await clickButton((text) => text.startsWith("Import 3 file")); + await settle(); } - it("shows the activation panel after import and activates selected agents and routines", async () => { + it("submits the import as an async job, then activates selected agents and routines", async () => { await renderPageAndImport(); - expect(mockCompaniesApi.importBundle).toHaveBeenCalledWith( + expect(mockCompaniesApi.importBundleAsync).toHaveBeenCalledWith( expect.objectContaining({ pauseAutomations: true }), ); + // The page polls the job it was handed and runs the same activation path + // the synchronous response used to drive. + expect(mockCompaniesApi.getImportJob).toHaveBeenCalledWith("job-1"); expect(container.textContent).toContain("Import complete"); expect(container.textContent).toContain("Activate imported agents and routines"); expect(container.textContent).toContain("Coder"); @@ -260,7 +303,7 @@ describe("CompanyImport", () => { expect(mockPushToast).toHaveBeenCalledWith(expect.objectContaining({ tone: "error" })); }); - it("blocks oversized local packages until attachments are dropped", async () => { + it("blocks oversized local packages and sends the declared file count once attachments are dropped", async () => { // A synthetic parsed package: the base64 blob payload alone exceeds the // inline import limit, so no real 60MB zip needs to be built. mockReadZipArchive.mockResolvedValue({ @@ -276,17 +319,7 @@ describe("CompanyImport", () => { }, }); - root = createRoot(container); - const queryClient = new QueryClient({ defaultOptions: { queries: { retry: false } } }); - const currentRoot = root; - await act(async () => { - currentRoot.render( - - - , - ); - }); - await flushReact(); + await renderPage(); await clickButton((text) => text.includes("Local zip")); @@ -301,21 +334,357 @@ describe("CompanyImport", () => { await flushReact(); expect(container.textContent).toContain("CLI folder import"); + expect(container.textContent).toContain("Package too large for browser import"); expect(findButton((text) => text === "Preview import")?.disabled).toBe(true); await clickButton((text) => text === "Continue without attachments"); expect(container.textContent).not.toContain("CLI folder import"); + expect(container.textContent).not.toContain("Package too large for browser import"); expect(findButton((text) => text === "Preview import")?.disabled).toBe(false); await clickButton((text) => text === "Preview import"); + // The button label reflects the preview's file count (3); the sent source + // is the local package, which is down to two files after the blob is + // dropped. await clickButton((text) => text.startsWith("Import 3 file")); + await settle(); - expect(mockCompaniesApi.importBundle).toHaveBeenCalledTimes(1); - const request = mockCompaniesApi.importBundle.mock.calls[0]![0] as { - source: { type: string; files: Record }; + expect(mockCompaniesApi.importBundleAsync).toHaveBeenCalledTimes(1); + const request = mockCompaniesApi.importBundleAsync.mock.calls[0]![0] as { + source: { type: string; files: Record; expectedFileCount?: number }; }; expect(request.source.type).toBe("inline"); expect(Object.keys(request.source.files).sort()).toEqual([".paperclip.yaml", "COMPANY.md"]); + // The client declares the file count so the server can reject a truncated + // upload instead of importing a fragment. + expect(request.source.expectedFileCount).toBe(2); + }); + + it("explains the disabled preview button until a package is chosen", async () => { + await renderPage(); + + expect(findButton((text) => text === "Preview import")?.disabled).toBe(true); + expect(container.textContent).toContain("Choose a package above to enable the preview."); + + await enterGithubUrl(); + + expect(findButton((text) => text === "Preview import")?.disabled).toBe(false); + expect(container.textContent).not.toContain("Choose a package above to enable the preview."); + }); + + it("shows a progress panel while the preview runs and a durable error panel when it fails", async () => { + let rejectPreview!: (err: Error) => void; + mockCompaniesApi.importPreview.mockImplementation( + () => + new Promise((_resolve, reject) => { + rejectPreview = reject; + }), + ); + await renderPage(); + await enterGithubUrl(); + + await clickButton((text) => text === "Preview import"); + + expect(container.textContent).toContain("Uploading and analyzing your package"); + expect(container.textContent).toContain("Keep this page open."); + + // A mid-flight config edit keeps the progress panel visible (the request + // really is still running) but supersedes the request, so its failure + // settles silently instead of describing a package no longer selected. + await enterGithubUrl("https://github.com/acme/starter-b/tree/main/company"); + + expect(container.textContent).toContain("Uploading and analyzing your package"); + + await act(async () => { + rejectPreview(new Error("stream disconnected")); + }); + await flushReact(); + + expect(container.textContent).not.toContain("Uploading and analyzing your package"); + expect(container.textContent).not.toContain("Preview failed:"); + expect(mockPushToast).not.toHaveBeenCalled(); + + // A failure of the currently configured request renders a durable panel. + await clickButton((text) => text === "Preview import"); + await act(async () => { + rejectPreview(new Error("stream disconnected")); + }); + await flushReact(); + + expect(container.textContent).toContain("Preview failed: stream disconnected"); + expect(container.textContent).toContain("Retry, or use the CLI folder import"); + expect(mockPushToast).toHaveBeenCalledWith(expect.objectContaining({ tone: "error" })); + + // Changing the package supersedes the failed request: the error panel resets. + await enterGithubUrl("https://github.com/acme/other-starter/tree/main/company"); + + expect(container.textContent).not.toContain("Preview failed:"); + }); + + it("shows a progress panel while the import runs and a durable error panel when it fails", async () => { + // The submit stays pending across the whole job, so the progress panel and + // structural locks cover it. Rejecting the submit surfaces the error panel. + let rejectImport!: (err: Error) => void; + mockCompaniesApi.importBundleAsync.mockImplementation( + () => + new Promise((_resolve, reject) => { + rejectImport = reject; + }), + ); + await renderPage(); + await enterGithubUrl(); + await clickButton((text) => text === "Preview import"); + + await clickButton((text) => text.startsWith("Import 3 file")); + + expect(container.textContent).toContain("Import running on the server"); + expect(container.textContent).toContain("safe to keep waiting"); + + // While the import runs, previewing and the structural package/settings + // controls are locked (and explained), so nothing can replace or unmount + // the plan the import started from. + expect(findButton((text) => text === "Preview import")?.disabled).toBe(true); + expect(container.textContent).toContain( + "Import in progress — the package and settings unlock when it finishes.", + ); + const lockedUrlInput = container.querySelector( + 'input[placeholder="https://github.com/owner/repo/tree/main/company"]', + ); + expect(lockedUrlInput?.disabled).toBe(true); + const lockedSelects = Array.from(container.querySelectorAll("select")).filter( + (select) => select.value === "new" || select.value === "rename", + ); + expect(lockedSelects).toHaveLength(2); + expect(lockedSelects.every((select) => select.disabled)).toBe(true); + + // Config edits while the import is in flight must not detach it from + // the UI: the progress panel keeps reporting it until it settles. + const midFlightNameInput = container.querySelector( + 'input[placeholder="Imported Company"]', + ); + expect(midFlightNameInput).toBeTruthy(); + await act(async () => { + const setter = Object.getOwnPropertyDescriptor(HTMLInputElement.prototype, "value")!.set!; + setter.call(midFlightNameInput!, "Mid Flight"); + midFlightNameInput!.dispatchEvent(new Event("input", { bubbles: true })); + }); + await flushReact(); + + expect(container.textContent).toContain("Import running on the server"); + + await act(async () => { + rejectImport(new Error("connection reset")); + }); + await flushReact(); + + expect(container.textContent).not.toContain("Import running on the server"); + expect(container.textContent).toContain("Import failed: connection reset"); + expect(container.textContent).toContain("check the target company before retrying."); + expect(mockPushToast).toHaveBeenCalledWith(expect.objectContaining({ tone: "error" })); + + // The new-company name feeds the request payload too: editing it clears + // the stale error panel without discarding the rendered preview. + const nameInput = container.querySelector('input[placeholder="Imported Company"]'); + expect(nameInput).toBeTruthy(); + await act(async () => { + const setter = Object.getOwnPropertyDescriptor(HTMLInputElement.prototype, "value")!.set!; + setter.call(nameInput!, "Renamed Import"); + nameInput!.dispatchEvent(new Event("input", { bubbles: true })); + }); + await flushReact(); + + expect(container.textContent).not.toContain("Import failed:"); + expect(findButton((text) => text.startsWith("Import 3 file"))).toBeTruthy(); + expect(findButton((text) => text === "Preview import")?.disabled).toBe(false); + expect( + container.querySelector( + 'input[placeholder="https://github.com/owner/repo/tree/main/company"]', + )?.disabled, + ).toBe(false); + + // The pause toggle feeds the request payload too: after another failed + // attempt, toggling it also supersedes the request and clears the panel. + await clickButton((text) => text.startsWith("Import 3 file")); + await act(async () => { + rejectImport(new Error("second failure")); + }); + await flushReact(); + + expect(container.textContent).toContain("Import failed: second failure"); + + const pauseInput = Array.from(container.querySelectorAll("label")) + .find((label) => label.textContent?.includes("Start imported agents and routines paused")) + ?.querySelector('input[type="checkbox"]'); + expect(pauseInput).toBeTruthy(); + await act(async () => { + pauseInput!.click(); + }); + await flushReact(); + + expect(container.textContent).not.toContain("Import failed:"); + + // File-selection edits feed the payload too and supersede a failure. + await clickButton((text) => text.startsWith("Import 3 file")); + await act(async () => { + rejectImport(new Error("third failure")); + }); + await flushReact(); + + expect(container.textContent).toContain("Import failed: third failure"); + + const fileCheckbox = Array.from( + container.querySelectorAll('input[type="checkbox"]'), + ).find((input) => !input.closest("label")?.textContent?.includes("Start imported agents")); + expect(fileCheckbox).toBeTruthy(); + await act(async () => { + fileCheckbox!.click(); + }); + await flushReact(); + + expect(container.textContent).not.toContain("Import failed:"); + + // Changing the package supersedes the failed import: previewing a fresh + // package must not resurface the stale import error panel. + await enterGithubUrl("https://github.com/acme/other-starter/tree/main/company"); + await clickButton((text) => text === "Preview import"); + + expect(findButton((text) => text.startsWith("Import 3 file"))).toBeTruthy(); + expect(container.textContent).not.toContain("Import failed:"); + + // The outcome reports the submitted pause option, not the live checkbox: + // a mid-flight toggle must not change what the completed import did. The + // submit stays pending until we resolve it to the accepted job. + let resolveAccepted!: (value: CompanyImportJobAccepted) => void; + mockCompaniesApi.importBundleAsync.mockImplementation( + () => + new Promise((resolve) => { + resolveAccepted = resolve; + }), + ); + mockCompaniesApi.getImportJob.mockResolvedValue(buildSucceededJob("job-final")); + await clickButton((text) => text.startsWith("Import 3 file")); + + expect(mockCompaniesApi.importBundleAsync).toHaveBeenLastCalledWith( + expect.objectContaining({ pauseAutomations: false }), + ); + + const midFlightPauseInput = Array.from(container.querySelectorAll("label")) + .find((label) => label.textContent?.includes("Start imported agents and routines paused")) + ?.querySelector('input[type="checkbox"]'); + expect(midFlightPauseInput).toBeTruthy(); + await act(async () => { + midFlightPauseInput!.click(); + }); + await flushReact(); + + await act(async () => { + resolveAccepted(buildAccepted("job-final")); + }); + await settle(); + + expect(container.textContent).toContain("Import complete"); + expect(container.textContent).not.toContain("Activate imported agents and routines"); + }); + + it("adopts the already-running job when the server reports a 409", async () => { + mockCompaniesApi.importBundleAsync.mockRejectedValueOnce( + new ApiError("An import is already running for this account", 409, { + job: { id: "job-existing", status: "running" }, + statusUrl: "/companies/import/jobs/job-existing", + }), + ); + mockCompaniesApi.getImportJob.mockResolvedValue(buildSucceededJob("job-existing")); + + await renderPageAndImport(); + + // The duplicate submit adopts the running job from the 409 body and polls + // it instead of firing a second import. + expect(mockCompaniesApi.getImportJob).toHaveBeenCalledWith("job-existing"); + expect(container.textContent).toContain("Import complete"); + expect(container.textContent).toContain("Activate imported agents and routines"); + }); + + it("surfaces a failed import job in the durable error panel", async () => { + mockCompaniesApi.getImportJob.mockResolvedValue({ + job: { id: "job-1", status: "failed", error: { message: "blob mismatch" } }, + }); + + await renderPage(); + await enterGithubUrl(); + await clickButton((text) => text === "Preview import"); + await clickButton((text) => text.startsWith("Import 3 file")); + await settle(); + + expect(container.textContent).toContain("Import failed: blob mismatch"); + expect(mockPushToast).toHaveBeenCalledWith(expect.objectContaining({ tone: "error" })); + }); + + it("terminates the import when polling hits a permanent auth error", async () => { + // A 403 (expired or revoked board session) will never recover by polling + // again, so the watch must stop and surface an error instead of leaving the + // import locked in its running state forever. + mockCompaniesApi.getImportJob.mockRejectedValue( + new ApiError("Board access required", 403, { error: "Board access required" }), + ); + + await renderPage(); + await enterGithubUrl(); + await clickButton((text) => text === "Preview import"); + await clickButton((text) => text.startsWith("Import 3 file")); + await settle(); + + expect(container.textContent).not.toContain("Import running on the server"); + expect(container.textContent).toContain("your session may have expired"); + expect(mockPushToast).toHaveBeenCalledWith(expect.objectContaining({ tone: "error" })); + }); + + it("keeps watching through a transient poll failure", async () => { + // A one-off 5xx is transient: the job is still running server-side, so the + // watch must keep polling and settle on the eventual success rather than + // treating the blip as a terminal failure. + mockCompaniesApi.getImportJob + .mockRejectedValueOnce(new ApiError("upstream unavailable", 503, null)) + .mockResolvedValue(buildSucceededJob("job-1")); + + await renderPageAndImport(); + + expect(container.textContent).not.toContain("Import failed:"); + expect(container.textContent).toContain("Import complete"); + }); + + it("resumes watching a stored import job on mount", async () => { + // A previous page load persisted a running job; reloading must resume + // watching it rather than showing the stale form. + sessionStorage.setItem( + "paperclip:company-import-job:company-1:acme/starter", + JSON.stringify({ jobId: "job-resume", pauseAutomations: false }), + ); + + let resolveFirstPoll!: (value: ReturnType) => void; + mockCompaniesApi.getImportJob.mockImplementationOnce( + () => + new Promise((resolve) => { + resolveFirstPoll = resolve; + }), + ); + + await renderPage(); + + // The resume panel is shown while the first poll is in flight, and no new + // submit was fired — the job is only being watched. + expect(container.textContent).toContain("Resume watching import"); + expect(mockCompaniesApi.importBundleAsync).not.toHaveBeenCalled(); + expect(mockCompaniesApi.getImportJob).toHaveBeenCalledWith("job-resume"); + + await act(async () => { + resolveFirstPoll(buildSucceededJob("job-resume")); + }); + await settle(); + + expect(container.textContent).not.toContain("Resume watching import"); + expect(container.textContent).toContain("Import complete"); + // The stored entry is cleared once the job settles. + expect(sessionStorage.getItem("paperclip:company-import-job:company-1:acme/starter")).toBeNull(); }); }); diff --git a/ui/src/pages/CompanyImport.tsx b/ui/src/pages/CompanyImport.tsx index 2e5d3ddd7b..4972598126 100644 --- a/ui/src/pages/CompanyImport.tsx +++ b/ui/src/pages/CompanyImport.tsx @@ -12,7 +12,8 @@ import { useCompany } from "../context/CompanyContext"; import { useBreadcrumbs } from "../context/BreadcrumbContext"; import { useToastActions } from "../context/ToastContext"; import { authApi } from "../api/auth"; -import { companiesApi } from "../api/companies"; +import { ApiError } from "../api/client"; +import { companiesApi, type CompanyImportJobAccepted } from "../api/companies"; import { agentsApi } from "../api/agents"; import { routinesApi } from "../api/routines"; import { sidebarPreferencesApi } from "../api/sidebarPreferences"; @@ -29,6 +30,7 @@ import { ChevronRight, Download, Github, + Loader2, Package, Upload, } from "lucide-react"; @@ -56,6 +58,13 @@ import { stripBlobFiles, } from "../lib/import-preflight"; import { getPortableFileDataUrl, getPortableFileText, isPortableImageFile } from "../lib/portable-files"; +import { + clearStoredImportJob, + importJobStorageKey, + readStoredImportJob, + waitForNextImportJobPoll, + writeStoredImportJob, +} from "../lib/import-job-watch"; import { Badge } from "@/components/ui/badge"; // ── Import-specific helpers ─────────────────────────────────────────── @@ -675,6 +684,78 @@ async function readLocalPackageZip(file: File): Promise<{ }; } +// ── Async import job flow ───────────────────────────────────────────── +// +// Imports run as server-side jobs: the submit returns 202 with a job id and +// the page polls the status route until the job settles. The connection is +// no longer load-bearing — a proxy timeout, network blip, or page reload +// cannot kill the import, and the stored job id lets the page resume +// watching. Jobs are held in server memory only, so an id the server no +// longer knows (restart, retention expiry) polls as 404. + +/** A 409 on submit means this user's previous import is still running; adopt it instead of importing twice. */ +function runningImportJobFromError(err: unknown): CompanyImportJobAccepted | null { + if (!(err instanceof ApiError) || err.status !== 409) return null; + const body = err.body as { job?: { id?: unknown }; statusUrl?: unknown } | null; + const jobId = body?.job?.id; + if (typeof jobId !== "string" || jobId.length === 0) return null; + return { + job: { id: jobId, status: "running" }, + statusUrl: typeof body?.statusUrl === "string" ? body.statusUrl : `/companies/import/jobs/${jobId}`, + }; +} + +/** Poll a job to a terminal state; resolves with the import result or throws the job's error. */ +async function watchImportJob( + jobId: string, + storageKey: string, +): Promise { + for (;;) { + let job: Awaited>["job"] | null = null; + try { + job = (await companiesApi.getImportJob(jobId)).job; + } catch (err) { + if (err instanceof ApiError && err.status === 404) { + clearStoredImportJob(storageKey); + throw new Error( + "The server no longer reports this import job — it may have restarted while the import ran.", + ); + } + if ( + err instanceof ApiError + && err.status >= 400 + && err.status < 500 + && err.status !== 429 + ) { + // A permanent client error (an expired board session, lost board + // access, or a bad request) will never recover by polling again, so + // stop instead of leaving the import locked in its running state. + // 429 (rate limited) and 5xx stay transient and fall through below. + clearStoredImportJob(storageKey); + throw new Error( + "The import status can no longer be read — your session may have expired. Reload and sign in to check on it.", + ); + } + // Any other poll failure is treated as transient (network blip, + // dropped connection, rate limit, or a 5xx): the job keeps running + // server-side, so keep watching rather than reporting a failure that + // may not exist. + } + if (job?.status === "succeeded") { + clearStoredImportJob(storageKey); + if (!job.importResult) { + throw new Error("The import finished, but its result is no longer available."); + } + return job.importResult; + } + if (job?.status === "failed") { + clearStoredImportJob(storageKey); + throw new Error(job.error?.message ?? "Import failed on the server."); + } + await waitForNextImportJobPoll(); + } +} + // ── Main page ───────────────────────────────────────────────────────── export function CompanyImport() { @@ -736,6 +817,11 @@ export function CompanyImport() { const [activationFailures, setActivationFailures] = useState>({}); const [isActivating, setIsActivating] = useState(false); + // A still-running job from a previous page load being re-attached to. While + // set, the page shows a "resume watching" panel instead of the stale form. + const [resumedWatchJobId, setResumedWatchJobId] = useState(null); + const resumeAttemptedRef = useRef(false); + // Fetch current company agents to find CEO adapter type const { data: companyAgents } = useQuery({ queryKey: selectedCompanyId ? queryKeys.agents.list(selectedCompanyId) : ["agents", "none"], @@ -762,16 +848,31 @@ export function CompanyImport() { function buildSource(): CompanyPortabilitySource | null { if (sourceMode === "local") { if (!localPackage) return null; - return { type: "inline", rootPath: localPackage.rootPath, files: localPackage.files }; + // Declare how many files we are sending so the server can reject a + // truncated upload instead of importing a fragment (see the server-side + // completeness check in importBundle). + return { + type: "inline", + rootPath: localPackage.rootPath, + files: localPackage.files, + expectedFileCount: Object.keys(localPackage.files).length, + }; } const url = importUrl.trim(); if (!url) return null; return { type: "github", url }; } + // Monotonic id for preview requests. Structural configuration changes bump + // it, so an in-flight preview they supersede settles silently instead of + // publishing a result or error for a package that is no longer selected. + // Imports are not gated this way: they mutate the server, so their outcome + // is always published. + const previewGenerationRef = useRef(0); + // Preview mutation const previewMutation = useMutation({ - mutationFn: () => { + mutationFn: (_generation: number) => { const source = buildSource(); if (!source) throw new Error("No source configured."); return companiesApi.importPreview({ @@ -784,7 +885,8 @@ export function CompanyImport() { collisionStrategy, }); }, - onSuccess: (result) => { + onSuccess: (result, generation) => { + if (generation !== previewGenerationRef.current) return; setImportPreview(result); // Build conflicts and set default name overrides with prefix @@ -847,7 +949,8 @@ export function CompanyImport() { const firstFile = Object.keys(result.files)[0]; if (firstFile) setSelectedFile(firstFile); }, - onError: (err) => { + onError: (err, generation) => { + if (generation !== previewGenerationRef.current) return; pushToast({ tone: "error", title: "Preview failed", @@ -873,26 +976,64 @@ export function CompanyImport() { return selected.length > 0 ? selected : undefined; } - // Apply mutation + /** Storage key for the pending submission, scoped per company + package. */ + function currentImportJobStorageKey(): string { + const packageName = + sourceMode === "local" + ? localPackage?.name ?? "package" + : importUrl.trim() || "package"; + return importJobStorageKey(selectedCompanyId ?? "unknown-company", packageName); + } + + // Apply mutation. The preview the import was started from and the + // submitted pause option ride along as the mutation variables so the + // request and its callbacks never read state that a later edit replaced. + // The import runs as a server-side job: the mutation stays pending across + // the 202 submit and every poll, so the existing progress panel, error + // panel, and structural locks cover the whole job, not just one request. const importMutation = useMutation({ - mutationFn: () => { + mutationFn: async (variables: { + previewForImport: CompanyPortabilityPreviewResult | null; + pauseAutomations: boolean; + /** Re-attach to a job stored by a previous page load instead of submitting. */ + resume?: { jobId: string; storageKey: string }; + }) => { + if (variables.resume) { + return watchImportJob(variables.resume.jobId, variables.resume.storageKey); + } const source = buildSource(); if (!source) throw new Error("No source configured."); - return companiesApi.importBundle({ - source, - include: { company: true, agents: true, projects: true, issues: true }, - target: - targetMode === "new" - ? { mode: "new_company", newCompanyName: newCompanyName || null } - : { mode: "existing_company", companyId: selectedCompanyId! }, - collisionStrategy, - nameOverrides: buildFinalNameOverrides(), - selectedFiles: buildSelectedFiles(), - adapterOverrides: buildFinalAdapterOverrides(), - pauseAutomations, + const storageKey = currentImportJobStorageKey(); + let accepted: CompanyImportJobAccepted; + try { + accepted = await companiesApi.importBundleAsync({ + source, + include: { company: true, agents: true, projects: true, issues: true }, + target: + targetMode === "new" + ? { mode: "new_company", newCompanyName: newCompanyName || null } + : { mode: "existing_company", companyId: selectedCompanyId! }, + collisionStrategy, + nameOverrides: buildFinalNameOverrides(), + selectedFiles: buildSelectedFiles(), + adapterOverrides: buildFinalAdapterOverrides(), + pauseAutomations: variables.pauseAutomations, + }); + } catch (err) { + // 409: this user's previous import is still running. Adopt that job + // and watch it — never fire a second import. + const running = runningImportJobFromError(err); + if (!running) throw err; + accepted = running; + } + writeStoredImportJob(storageKey, { + jobId: accepted.job.id, + pauseAutomations: variables.pauseAutomations, }); + return watchImportJob(accepted.job.id, storageKey); }, - onSuccess: async (result) => { + onSuccess: async (result, { previewForImport, pauseAutomations: submittedPauseAutomations }) => { + setResumedWatchJobId(null); await queryClient.invalidateQueries({ queryKey: queryKeys.companies.all }); const importedCompany = await companiesApi.get(result.company.id); const refreshedSession = currentUserId @@ -906,7 +1047,7 @@ export function CompanyImport() { ?? refreshedSession?.user?.id ?? refreshedSession?.session?.userId ?? null; - await applyImportedSidebarOrder(importPreview, result, sidebarOrderUserId); + await applyImportedSidebarOrder(previewForImport, result, sidebarOrderUserId); setSelectedCompanyId(importedCompany.id); setActivationChecked(new Set(buildActivationItems(result).map((item) => item.key))); setActivatedKeys(new Set()); @@ -914,10 +1055,11 @@ export function CompanyImport() { setImportOutcome({ result, dashboardPath: `/${importedCompany.issuePrefix}/dashboard`, - pausedAutomations: pauseAutomations, + pausedAutomations: submittedPauseAutomations, }); }, onError: (err) => { + setResumedWatchJobId(null); pushToast({ tone: "error", title: "Import failed", @@ -926,13 +1068,51 @@ export function CompanyImport() { }, }); + // On mount, re-attach to a job stored by a previous page load. This is what + // makes dropped connections and reloads harmless: the job kept running + // server-side, so resume watching it instead of showing the stale form. + const resumeImportWatch = importMutation.mutate; + useEffect(() => { + if (resumeAttemptedRef.current || !selectedCompanyId) return; + resumeAttemptedRef.current = true; + const stored = readStoredImportJob(selectedCompanyId); + if (!stored) return; + setResumedWatchJobId(stored.jobId); + resumeImportWatch({ + previewForImport: null, + pauseAutomations: stored.pauseAutomations, + resume: { jobId: stored.jobId, storageKey: stored.storageKey }, + }); + }, [selectedCompanyId, resumeImportWatch]); + + // Any change to the import configuration supersedes the request a settled + // progress/error panel describes, so it clears settled mutation state. A + // pending request is never detached: its panel keeps reporting it (and the + // action buttons stay disabled) until it settles. Payload edits made + // against a rendered preview (new-company name, pause toggle, file + // selection, conflict choices, adapter overrides, dropping attachments) + // reset only the panels; structural changes also discard the preview + // itself. Structural controls are locked while an import runs, so this can + // never unmount the preview section that hosts a running import's status + // panels. + function resetMutationState() { + if (!previewMutation.isPending) previewMutation.reset(); + if (!importMutation.isPending) importMutation.reset(); + } + + function resetImportFlowState() { + previewGenerationRef.current += 1; + setImportPreview(null); + resetMutationState(); + } + async function handleChooseLocalPackage(e: ChangeEvent) { const fileList = e.target.files; if (!fileList || fileList.length === 0) return; try { const pkg = await readLocalPackageZip(fileList[0]!); setLocalPackage(pkg); - setImportPreview(null); + resetImportFlowState(); } catch (err) { pushToast({ tone: "error", @@ -991,6 +1171,7 @@ export function CompanyImport() { function handleToggleCheck(path: string, kind: "file" | "dir") { if (!importPreview) return; + resetMutationState(); setCheckedFiles((prev) => { const next = new Set(prev); if (kind === "file") { @@ -1023,6 +1204,7 @@ export function CompanyImport() { } function handleConflictRename(slug: string, newName: string) { + resetMutationState(); setNameOverrides((prev) => ({ ...prev, [slug]: newName })); // Editing the name un-confirms setConfirmedSlugs((prev) => { @@ -1034,6 +1216,7 @@ export function CompanyImport() { } function handleConflictToggleConfirm(slug: string) { + resetMutationState(); setConfirmedSlugs((prev) => { const next = new Set(prev); if (next.has(slug)) next.delete(slug); @@ -1043,6 +1226,7 @@ export function CompanyImport() { } function handleConflictToggleSkip(slug: string, filePath: string | null) { + resetMutationState(); setSkippedSlugs((prev) => { const next = new Set(prev); const wasSkipped = next.has(slug); @@ -1070,6 +1254,7 @@ export function CompanyImport() { } function handleAdapterChange(slug: string, adapterType: string) { + resetMutationState(); setAdapterOverrides((prev) => ({ ...prev, [slug]: adapterType })); // Reset config values when adapter type changes setAdapterConfigValues((prev) => { @@ -1089,6 +1274,7 @@ export function CompanyImport() { } function handleAdapterConfigChange(slug: string, patch: Partial) { + resetMutationState(); setAdapterConfigValues((prev) => ({ ...prev, [slug]: { ...(prev[slug] ?? { ...defaultCreateValues, adapterType: adapterOverrides[slug] ?? "claude_local" }), ...patch }, @@ -1176,6 +1362,7 @@ export function CompanyImport() { function handleContinueWithoutAttachments() { if (!localPackage) return; + resetMutationState(); setLocalPackage({ ...localPackage, files: stripBlobFiles(localPackage.files) }); setCheckedFiles((prev) => new Set([...prev].filter((filePath) => !isBlobStoreFilePath(filePath)))); } @@ -1277,6 +1464,28 @@ export function CompanyImport() { ); } + // Resuming a job from a previous page load: show a watch panel instead of + // the stale form. Cleared when the job settles (success renders the + // outcome above; failure returns to the form with an error toast). + if (resumedWatchJobId) { + return ( +
+
+

Resume watching import

+

+ An import you started earlier is still running on the server. +

+
+
+ +

+ Import running on the server — safe to keep waiting; reconnecting won't lose it. +

+
+
+ ); + } + if (!selectedCompanyId) { return ; } @@ -1307,10 +1516,12 @@ export function CompanyImport() { sourceMode === key ? "border-foreground bg-accent" : "border-border hover:bg-accent/50", + importMutation.isPending && "cursor-not-allowed opacity-50", )} + disabled={importMutation.isPending} onClick={() => { setSourceMode(key); - setImportPreview(null); + resetImportFlowState(); }} >
@@ -1335,6 +1546,7 @@ export function CompanyImport() { size="sm" variant="outline" onClick={() => packageInputRef.current?.click()} + disabled={importMutation.isPending} > Choose zip @@ -1376,9 +1588,10 @@ export function CompanyImport() { type="text" value={importUrl} placeholder="https://github.com/owner/repo/tree/main/company" + disabled={importMutation.isPending} onChange={(e) => { setImportUrl(e.target.value); - setImportPreview(null); + resetImportFlowState(); }} /> @@ -1388,9 +1601,10 @@ export function CompanyImport() { { setCollisionStrategy(e.target.value as CompanyPortabilityCollisionStrategy); - setImportPreview(null); + resetImportFlowState(); }} > @@ -1437,12 +1655,52 @@ export function CompanyImport() { + {!hasSource && !previewMutation.isPending && ( + + Choose a package above to enable the preview. + + )} + {importMutation.isPending && ( + + Import in progress — the package and settings unlock when it finishes. + + )} + {inlineImportBlocked && ( + + Package too large for browser import — see the notice above. + + )}
+ {previewMutation.isPending && ( +
+ +

+ Uploading and analyzing your package + {inlinePreflight ? ` (about ${formatMegabytes(inlinePreflight.estimatedBytes)})` : ""} — large + packages can take a few minutes. Keep this page open. +

+
+ )} + {previewMutation.isError && + !previewMutation.isPending && + previewMutation.variables === previewGenerationRef.current && ( +
+

+ Preview failed:{" "} + {previewMutation.error instanceof Error + ? previewMutation.error.message + : "the request did not complete."}{" "} + Retry, or use the CLI folder import for very large packages. +

+
+ )} {/* Preview results */} @@ -1498,14 +1756,17 @@ export function CompanyImport() { setPauseAutomations(e.target.checked)} + onChange={(e) => { + setPauseAutomations(e.target.checked); + resetMutationState(); + }} className="accent-foreground" /> Start imported agents and routines paused + {importMutation.isPending && ( +
+ +

+ Import running on the server — safe to keep waiting; reconnecting won't lose it. + Large packages can take several minutes. +

+
+ )} + {importMutation.isError && !importMutation.isPending && ( +
+

+ Import failed:{" "} + {importMutation.error instanceof Error + ? importMutation.error.message + : "the request did not complete."}{" "} + Nothing may have been created, or the import stopped partway — check the target company + before retrying. +

+
+ )} {/* Warnings */} {importPreview.warnings.length > 0 && (