From 1c5cfa5ba80887086f194be76c954121f6257bea Mon Sep 17 00:00:00 2001 From: Dotta Date: Wed, 20 May 2026 08:29:01 -0500 Subject: [PATCH] Add async tenant import jobs Co-Authored-By: Paperclip --- .../company-portability-routes.test.ts | 131 ++++++++++++ server/src/routes/companies.ts | 202 ++++++++++++++++-- 2 files changed, 316 insertions(+), 17 deletions(-) diff --git a/server/src/__tests__/company-portability-routes.test.ts b/server/src/__tests__/company-portability-routes.test.ts index 26f6df16807..f8b763f2f2a 100644 --- a/server/src/__tests__/company-portability-routes.test.ts +++ b/server/src/__tests__/company-portability-routes.test.ts @@ -139,6 +139,58 @@ function createExportResult() { }; } +const importRequest = { + source: { type: "inline", files: { "COMPANY.md": "---\nname: Test\n---\n" } }, + include: { company: true, agents: true, projects: false, issues: false }, + target: { mode: "existing_company", companyId }, + collisionStrategy: "rename", +}; + +const cloudHeaders = { + "x-paperclip-cloud-stack-id": "stack-alpha", + "x-paperclip-cloud-paperclip-company-id": companyId, +}; + +function cloudTenantActor() { + return { + type: "board", + userId: "cloud-user-1", + userName: "Cloud User", + userEmail: "cloud-user@example.com", + companyIds: [companyId], + memberships: [{ companyId, membershipRole: "owner", status: "active" }], + isInstanceAdmin: true, + source: "cloud_tenant", + }; +} + +function createImportResult(action = "updated") { + return { + company: { id: companyId, action }, + agents: [{ id: "agent-1" }], + warnings: [], + }; +} + +async function waitForImportJobStatus(app: express.Express, statusUrl: string, status: string) { + for (let attempt = 0; attempt < 20; attempt += 1) { + const res = await request(app).get(statusUrl).set(cloudHeaders); + 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}`); +} + +async function waitForCondition(condition: () => boolean, label: string) { + for (let attempt = 0; attempt < 20; attempt += 1) { + if (condition()) return; + await new Promise((resolve) => setTimeout(resolve, 10)); + } + throw new Error(`Timed out waiting for ${label}`); +} + describe.sequential("company portability routes", () => { beforeEach(() => { vi.clearAllMocks(); @@ -426,4 +478,83 @@ describe.sequential("company portability routes", () => { expect(res.body.error).toContain("Instance admin"); expect(mockCompanyPortabilityService.importBundle).not.toHaveBeenCalled(); }); + + it.sequential("accepts trusted Cloud async import jobs and reports success by job id", async () => { + let resolveImport: (value: ReturnType) => void = () => undefined; + const pendingImport = new Promise>((resolve) => { + resolveImport = resolve; + }); + mockCompanyPortabilityService.importBundle.mockReturnValueOnce(pendingImport); + const app = await createApp(cloudTenantActor()); + + const accepted = await request(app) + .post("/api/companies/import") + .set("x-paperclip-cloud-async-import", "1") + .set(cloudHeaders) + .send(importRequest); + + expect(accepted.status).toBe(202); + expect(accepted.body.job.status).toBe("running"); + expect(accepted.body.statusUrl).toMatch(/^\/api\/companies\/import\/jobs\/tenant-import-/); + expect(accepted.body.retryAfterMs).toBe(1000); + await waitForCondition(() => mockCompanyPortabilityService.importBundle.mock.calls.length === 1, "import job start"); + expect(mockCompanyPortabilityService.importBundle).toHaveBeenCalledWith(importRequest, "cloud-user-1"); + expect(mockLogActivity).not.toHaveBeenCalled(); + + resolveImport(createImportResult("updated")); + const succeeded = await waitForImportJobStatus(app, accepted.body.statusUrl, "succeeded"); + + expect(succeeded.status).toBe(200); + expect(succeeded.body.job.status).toBe("succeeded"); + expect(succeeded.body.job.result.companyId).toBe(companyId); + expect(mockLogActivity).toHaveBeenCalledWith(expect.anything(), expect.objectContaining({ + action: "company.imported", + companyId, + details: expect.objectContaining({ + agentCount: 1, + warningCount: 0, + companyAction: "updated", + }), + })); + }); + + it.sequential("reports trusted Cloud async import job failures with the tenant error message", async () => { + mockCompanyPortabilityService.importBundle.mockRejectedValueOnce(new Error("tenant import exploded")); + const app = await createApp(cloudTenantActor()); + + const accepted = await request(app) + .post("/api/companies/import") + .set("x-paperclip-cloud-async-import", "1") + .set(cloudHeaders) + .send(importRequest); + + expect(accepted.status).toBe(202); + const failed = await waitForImportJobStatus(app, accepted.body.statusUrl, "failed"); + + expect(failed.status).toBe(200); + expect(failed.body.job.status).toBe("failed"); + expect(failed.body.job.error.message).toBe("tenant import exploded"); + expect(failed.body.message).toBe("tenant import exploded"); + expect(mockLogActivity).not.toHaveBeenCalled(); + }); + + it.sequential("keeps global import apply synchronous when Cloud async opt-in is absent", async () => { + mockCompanyPortabilityService.importBundle.mockResolvedValueOnce(createImportResult("created")); + const app = await createApp(cloudTenantActor()); + + const res = await request(app) + .post("/api/companies/import") + .set(cloudHeaders) + .send(importRequest); + + expect(res.status).toBe(200); + expect(res.body.company.id).toBe(companyId); + expect(res.body.company.action).toBe("created"); + expect(res.body.job).toBeUndefined(); + expect(mockCompanyPortabilityService.importBundle).toHaveBeenCalledWith(importRequest, "cloud-user-1"); + expect(mockLogActivity).toHaveBeenCalledWith(expect.anything(), expect.objectContaining({ + action: "company.imported", + companyId, + })); + }); }); diff --git a/server/src/routes/companies.ts b/server/src/routes/companies.ts index 6516e5efb51..e057ca87742 100644 --- a/server/src/routes/companies.ts +++ b/server/src/routes/companies.ts @@ -1,3 +1,4 @@ +import { randomUUID } from "node:crypto"; import { Router, type Request } from "express"; import type { Db } from "@paperclipai/db"; import { @@ -34,6 +35,7 @@ export function companyRoutes(db: Db, storage?: StorageService) { const access = accessService(db); const budgets = budgetService(db); const feedback = feedbackService(db); + const importJobs = new Map(); function parseBooleanQuery(value: unknown) { return value === true || value === "true" || value === "1"; @@ -176,27 +178,41 @@ export function companyRoutes(db: Db, storage?: StorageService) { res.json(preview); }); + router.get("/import/jobs/:jobId", async (req, res) => { + assertCloudTenantCaller(req); + const job = importJobs.get(req.params.jobId as string); + if (!job || job.cloudTenantKey !== cloudTenantRequestKey(req)) { + res.status(404).json({ error: "Import job not found" }); + return; + } + res.json(importJobResponse(job)); + }); + router.post("/import", validate(companyPortabilityImportSchema), async (req, res) => { assertBoard(req); assertImportTargetAccess(req, req.body.target); const actor = getActorInfo(req); - const result = await portability.importBundle(req.body, req.actor.type === "board" ? req.actor.userId : null); - await logActivity(db, { - companyId: result.company.id, - actorType: actor.actorType, - actorId: actor.actorId, - action: "company.imported", - entityType: "company", - entityId: result.company.id, - agentId: actor.agentId, - runId: actor.runId, - details: { - include: req.body.include ?? null, - agentCount: result.agents.length, - warningCount: result.warnings.length, - companyAction: result.company.action, - }, - }); + const importBody = req.body; + const boardUserId = req.actor.type === "board" ? req.actor.userId : null; + const activity = importedCompanyActivityContext(actor, importBody.include ?? null); + if (req.header("x-paperclip-cloud-async-import") === "1") { + assertCloudTenantCaller(req); + const job = createImportJob(cloudTenantRequestKey(req)); + importJobs.set(job.id, job); + const operation = async () => { + const result = await portability.importBundle(importBody, boardUserId); + await logImportedCompanyActivity(db, activity, result); + return result; + }; + res.status(202).json(importJobAcceptedResponse(job)); + setImmediate(() => { + void runImportJob(job, operation); + }); + return; + } + + const result = await portability.importBundle(importBody, boardUserId); + await logImportedCompanyActivity(db, activity, result); res.json(result); }); @@ -411,3 +427,155 @@ export function companyRoutes(db: Db, storage?: StorageService) { return router; } + +type CompanyImportResult = { + company: { id: string; action: unknown }; + agents: unknown[]; + warnings: unknown[]; +}; + +interface ImportJobRecord { + id: string; + cloudTenantKey: string; + status: "running" | "succeeded" | "failed"; + createdAt: string; + updatedAt: string; + completedAt?: string; + error?: { message: string }; + result?: { + companyId: string; + agentCount: number; + warningCount: number; + companyAction: unknown; + }; +} + +interface ImportedCompanyActivityContext { + actorType: "user" | "agent"; + actorId: string; + agentId: string | null; + runId: string | null; + 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 ?? "", + req.header("x-paperclip-cloud-stack-id")?.trim() ?? "", + req.header("x-paperclip-cloud-paperclip-company-id")?.trim() ?? "", + ].join(":"); +} + +function createImportJob(cloudTenantKey: string): ImportJobRecord { + const now = new Date().toISOString(); + return { + id: `tenant-import-${randomUUID()}`, + cloudTenantKey, + status: "running", + createdAt: now, + updatedAt: now, + }; +} + +async function runImportJob( + job: ImportJobRecord, + operation: () => Promise, +) { + try { + const result = await operation(); + const now = new Date().toISOString(); + job.status = "succeeded"; + job.updatedAt = now; + job.completedAt = now; + job.result = { + companyId: result.company.id, + agentCount: result.agents.length, + warningCount: result.warnings.length, + companyAction: result.company.action, + }; + } catch (error) { + const now = new Date().toISOString(); + job.status = "failed"; + job.updatedAt = now; + job.completedAt = now; + job.error = { message: errorMessage(error) }; + } +} + +function importedCompanyActivityContext( + actor: ReturnType, + include: unknown, +): ImportedCompanyActivityContext { + return { + actorType: actor.actorType, + actorId: actor.actorId, + agentId: actor.agentId, + runId: actor.runId, + include, + }; +} + +async function logImportedCompanyActivity( + db: Db, + activity: ImportedCompanyActivityContext, + result: CompanyImportResult, +) { + await logActivity(db, { + companyId: result.company.id, + actorType: activity.actorType, + actorId: activity.actorId, + action: "company.imported", + entityType: "company", + entityId: result.company.id, + agentId: activity.agentId, + runId: activity.runId, + details: { + include: activity.include, + agentCount: result.agents.length, + warningCount: result.warnings.length, + companyAction: result.company.action, + }, + }); +} + +function importJobAcceptedResponse(job: ImportJobRecord) { + return { + job: { + id: job.id, + status: job.status, + }, + statusUrl: `/api/companies/import/jobs/${encodeURIComponent(job.id)}`, + retryAfterMs: 1000, + }; +} + +function importJobResponse(job: ImportJobRecord) { + const response: Record = { + job: { + id: job.id, + status: job.status, + createdAt: job.createdAt, + updatedAt: job.updatedAt, + ...(job.completedAt ? { completedAt: job.completedAt } : {}), + ...(job.error ? { error: job.error } : {}), + ...(job.result ? { result: job.result } : {}), + }, + retryAfterMs: 1000, + }; + if (job.error?.message) { + response.error = job.error.message; + response.message = job.error.message; + response.reason = job.error.message; + } + return response; +} + +function errorMessage(error: unknown) { + return error instanceof Error && error.message.trim() ? error.message : String(error); +}