diff --git a/server/src/__tests__/onboarding-seed-route.test.ts b/server/src/__tests__/onboarding-seed-route.test.ts index 7fe3077c882..649617b0211 100644 --- a/server/src/__tests__/onboarding-seed-route.test.ts +++ b/server/src/__tests__/onboarding-seed-route.test.ts @@ -286,4 +286,61 @@ describeEmbeddedPostgres("POST /api/companies/:companyId/onboarding-seed", () => const companyIssues = await ctx.db.select().from(issues).where(eq(issues.companyId, companyId)); expect(companyIssues[0]?.projectId).toBe(companyProjects[0]?.id); }); + + it("does not duplicate entities when two identical pushes race", async () => { + const { companyId, app } = await seedCompany(); + + // Cloud's reconcile runs off portfolio fetches that can overlap, so the + // same revision can be pushed twice at once. The per-company advisory lock + // must serialize them: without it both pass the revision check before + // either writes the seed record and each creates a goal, an agent, a + // project and a task. + const [first, second] = await Promise.all([ + post(app, companyId, SEED), + post(app, companyId, SEED), + ]); + + expect(first.status).toBe(200); + expect(second.status).toBe(200); + + expect(await ctx.db.select().from(goals).where(eq(goals.companyId, companyId))).toHaveLength(1); + expect(await ctx.db.select().from(agents).where(eq(agents.companyId, companyId))).toHaveLength(1); + expect(await ctx.db.select().from(projects).where(eq(projects.companyId, companyId))).toHaveLength(1); + expect(await ctx.db.select().from(issues).where(eq(issues.companyId, companyId))).toHaveLength(1); + expect(await ctx.db.select().from(companyOnboardingSeeds)).toHaveLength(1); + }); + + it("refreshes a first task's assignee and goal when a later revision adds them", async () => { + const { companyId, app } = await seedCompany(); + + // First push seeds a task but no agent and no mission, so the issue is + // created unassigned and goal-less. + const taskOnly = await post(app, companyId, { + revision: "1".repeat(32), + firstTask: { title: "Draft the strategy" }, + }); + expect(taskOnly.status).toBe(200); + + const beforeIssue = (await ctx.db.select().from(issues).where(eq(issues.companyId, companyId)))[0]; + expect(beforeIssue?.assigneeAgentId).toBeNull(); + + // A later revision supplies the mission and the agent. The existing task is + // updated in place, and must pick up the newly-created assignee and goal + // rather than reporting them on the seed record while the issue row stays + // stale. + const withAgent = await post(app, companyId, { + revision: "2".repeat(32), + mission: "Make robotics boring enough to trust", + agent: { name: "Ada", role: "Chief of Staff" }, + firstTask: { title: "Draft the strategy" }, + }); + expect(withAgent.status).toBe(200); + + const companyIssues = await ctx.db.select().from(issues).where(eq(issues.companyId, companyId)); + expect(companyIssues).toHaveLength(1); + const companyAgents = await ctx.db.select().from(agents).where(eq(agents.companyId, companyId)); + const companyGoals = await ctx.db.select().from(goals).where(eq(goals.companyId, companyId)); + expect(companyIssues[0]?.assigneeAgentId).toBe(companyAgents[0]?.id); + expect(companyIssues[0]?.goalId).toBe(companyGoals[0]?.id); + }); }); diff --git a/server/src/routes/onboarding-seed.ts b/server/src/routes/onboarding-seed.ts index 74afcbce464..3f6e5f24d2c 100644 --- a/server/src/routes/onboarding-seed.ts +++ b/server/src/routes/onboarding-seed.ts @@ -2,6 +2,7 @@ import { Router } from "express"; import type { Db } from "@paperclipai/db"; import { applyOnboardingSeedSchema } from "@paperclipai/shared"; import { validate } from "../middleware/index.js"; +import { logger } from "../middleware/logger.js"; import { logActivity } from "../services/index.js"; import { onboardingSeedService } from "../services/onboarding-seed.js"; import { assertCompanyAccess, getActorInfo } from "./authz.js"; @@ -40,23 +41,37 @@ export function onboardingSeedRoutes(db: Db) { if (result.changed) { const actor = getActorInfo(req); - await logActivity(db, { - companyId, - actorType: actor.actorType, - actorId: actor.actorId, - agentId: actor.agentId, - runId: actor.runId, - agentApiKeyId: actor.agentApiKeyId, - action: "company.onboarding_seed_applied", - entityType: "company", - entityId: companyId, - details: { - revision: result.revision, - goalId: result.goalId, - agentId: result.agentId, - issueId: result.issueId, - }, - }); + // Best-effort. The seed and its revision are already committed by the + // time we get here, and Cloud reads any 2xx as "applied" and stops + // retrying. If this audit write threw and we let it 500, the retry + // would come back `changed: false` and skip the log for good — the + // entry would be permanently absent *and* the caller would have seen a + // spurious failure for an application that succeeded. So a logging + // failure is recorded server-side and the request still answers 200. + try { + await logActivity(db, { + companyId, + actorType: actor.actorType, + actorId: actor.actorId, + agentId: actor.agentId, + runId: actor.runId, + agentApiKeyId: actor.agentApiKeyId, + action: "company.onboarding_seed_applied", + entityType: "company", + entityId: companyId, + details: { + revision: result.revision, + goalId: result.goalId, + agentId: result.agentId, + issueId: result.issueId, + }, + }); + } catch (err) { + logger.error( + { err, companyId, revision: result.revision }, + "onboarding seed applied but its activity-log entry could not be written", + ); + } } res.status(200).json({ diff --git a/server/src/services/onboarding-seed.ts b/server/src/services/onboarding-seed.ts index c13d827f2f8..169bd45135c 100644 --- a/server/src/services/onboarding-seed.ts +++ b/server/src/services/onboarding-seed.ts @@ -1,4 +1,4 @@ -import { and, eq, ne } from "drizzle-orm"; +import { and, eq, ne, sql } from "drizzle-orm"; import type { Db } from "@paperclipai/db"; import { agents, companyOnboardingSeeds, goals, issues, projects } from "@paperclipai/db"; import type { ApplyOnboardingSeed } from "@paperclipai/shared"; @@ -62,22 +62,17 @@ export type OnboardingSeedApplication = { }; export function onboardingSeedService(db: Db) { - const agentSvc = agentService(db); - const goalSvc = goalService(db); - const projectSvc = projectService(db); - const issueSvc = issueService(db); - - async function readRecord(companyId: string) { - return db + async function readRecord(dbx: Db, companyId: string) { + return dbx .select() .from(companyOnboardingSeeds) .where(eq(companyOnboardingSeeds.companyId, companyId)) .then((rows) => rows[0] ?? null); } - async function goalStillExists(companyId: string, goalId: string | null) { + async function goalStillExists(dbx: Db, companyId: string, goalId: string | null) { if (!goalId) return false; - return db + return dbx .select({ id: goals.id }) .from(goals) .where(and(eq(goals.id, goalId), eq(goals.companyId, companyId))) @@ -90,9 +85,9 @@ export function onboardingSeedService(db: Db) { * already has. Built-in agents are excluded — they are provisioned by the * platform and are not the customer's first hire. */ - async function resolveTargetAgentId(companyId: string, recordedAgentId: string | null) { + async function resolveTargetAgentId(dbx: Db, companyId: string, recordedAgentId: string | null) { if (recordedAgentId) { - const recorded = await db + const recorded = await dbx .select({ id: agents.id }) .from(agents) .where(and(eq(agents.id, recordedAgentId), eq(agents.companyId, companyId))) @@ -100,7 +95,7 @@ export function onboardingSeedService(db: Db) { if (recorded) return recorded.id; } - const candidates = await db + const candidates = await dbx .select({ id: agents.id, metadata: agents.metadata }) .from(agents) .where(and( @@ -111,17 +106,22 @@ export function onboardingSeedService(db: Db) { return candidates.find((row) => !readBuiltInAgentMarker(row.metadata))?.id ?? null; } - async function issueStillExists(companyId: string, issueId: string | null) { + async function issueStillExists(dbx: Db, companyId: string, issueId: string | null) { if (!issueId) return false; - return db + return dbx .select({ id: issues.id }) .from(issues) .where(and(eq(issues.id, issueId), eq(issues.companyId, companyId))) .then((rows) => rows.length > 0); } - async function resolveOnboardingProjectId(companyId: string, goalId: string | null) { - const existing = await db + async function resolveOnboardingProjectId( + dbx: Db, + projectSvc: ReturnType, + companyId: string, + goalId: string | null, + ) { + const existing = await dbx .select({ id: projects.id, name: projects.name, status: projects.status }) .from(projects) .where(eq(projects.companyId, companyId)); @@ -141,24 +141,22 @@ export function onboardingSeedService(db: Db) { } /** - * Apply an onboarding seed to a company. - * - * Idempotent per `revision`: a replay of the revision already stored is a - * no-op that still reports success, because Cloud reads any 2xx as "the - * tenant holds this content" and retries otherwise. A *different* revision - * (the customer edited their answers in Cloud) updates the goal, agent and - * task this seed previously created rather than creating a second set. - * - * Every write happens before the caller responds — Cloud records the applied - * revision only on a 2xx, and the redirect into the tenant dashboard is - * gated on it, so a partially-applied seed must surface as a failure rather - * than as an acknowledged one. + * The seed application proper, run inside the per-company transaction the + * public `apply` opens. Every read and write goes through `dbx` — the locked + * transaction — so it is serialized against a concurrent push for the same + * company. Services are reconstructed on `dbx` for the same reason. */ - async function apply( + async function applyWithin( + dbx: Db, companyId: string, seed: ApplyOnboardingSeed, ): Promise { - const existing = await readRecord(companyId); + const agentSvc = agentService(dbx); + const goalSvc = goalService(dbx); + const projectSvc = projectService(dbx); + const issueSvc = issueService(dbx); + + const existing = await readRecord(dbx, companyId); if (existing && existing.revision === seed.revision) { return { revision: existing.revision, @@ -179,7 +177,7 @@ export function onboardingSeedService(db: Db) { let goalId = existing?.goalId ?? null; if (mission) { const parsed = parseSeedMission(mission); - const target = (await goalStillExists(companyId, goalId)) + const target = (await goalStillExists(dbx, companyId, goalId)) ? goalId : (await goalSvc.getDefaultCompanyGoal(companyId))?.id ?? null; if (target) { @@ -201,7 +199,7 @@ export function onboardingSeedService(db: Db) { // 2. Agent → the customer's first hire, the lead the first task is // assigned to. - let agentId = await resolveTargetAgentId(companyId, existing?.agentId ?? null); + let agentId = await resolveTargetAgentId(dbx, companyId, existing?.agentId ?? null); if (agentName) { if (agentId) { await agentSvc.update(agentId, { name: agentName, title: agentRole }); @@ -242,13 +240,24 @@ export function onboardingSeedService(db: Db) { // mission-only seed is what keeps it inert on the cloud path. let issueId = existing?.issueId ?? null; if (firstTaskTitle) { - if (await issueStillExists(companyId, issueId)) { - await issueSvc.update(issueId as string, { - title: firstTaskTitle, - description: firstTaskDetails, - }); + if (await issueStillExists(dbx, companyId, issueId)) { + await issueSvc.update( + issueId as string, + { + title: firstTaskTitle, + description: firstTaskDetails, + // Keep the task's relationships in step with a later revision that + // supplied the agent or goal after the task already existed — + // otherwise the record would report an assignee/goal the issue row + // does not actually carry. Only set them when resolved, so an + // absent value never clears an assignment the tenant made. + ...(agentId ? { assigneeAgentId: agentId } : {}), + ...(goalId ? { goalId } : {}), + }, + dbx, + ); } else { - const projectId = await resolveOnboardingProjectId(companyId, goalId); + const projectId = await resolveOnboardingProjectId(dbx, projectSvc, companyId, goalId); // The idempotency key is what protects two pushes that arrive at once // — Cloud's reconcile runs off portfolio fetches, which can overlap. // It is deliberately not revision-scoped: if the recorded issue is @@ -284,7 +293,7 @@ export function onboardingSeedService(db: Db) { appliedAt: now, updatedAt: now, }; - await db + await dbx .insert(companyOnboardingSeeds) .values(values) .onConflictDoUpdate({ @@ -307,5 +316,40 @@ export function onboardingSeedService(db: Db) { return { revision: seed.revision, changed: true, goalId, agentId, issueId }; } - return { apply, get: readRecord }; + /** + * Apply an onboarding seed to a company. + * + * Idempotent per `revision`: a replay of the revision already stored is a + * no-op that still reports success, because Cloud reads any 2xx as "the + * tenant holds this content" and retries otherwise. A *different* revision + * (the customer edited their answers in Cloud) updates the goal, agent and + * task this seed previously created rather than creating a second set. + * + * Every write happens before the caller responds — Cloud records the applied + * revision only on a 2xx, and the redirect into the tenant dashboard is + * gated on it, so a partially-applied seed must surface as a failure rather + * than as an acknowledged one. + * + * Concurrency: Cloud's reconcile runs off portfolio fetches, which can + * overlap, so two pushes for the same company can arrive at once. Both would + * otherwise pass the revision check before either wrote the seed record and + * each create a company goal, a lead agent and an Onboarding project. A + * per-company advisory lock held for the transaction serializes them — the + * same idiom `folders` and `decision-queues` use — so the second push sees + * the first push's writes (the record, the reused goal/agent/project) and + * updates in place instead of duplicating. + */ + async function apply( + companyId: string, + seed: ApplyOnboardingSeed, + ): Promise { + return db.transaction(async (tx) => { + await tx.execute( + sql`select pg_advisory_xact_lock(hashtextextended(${`paperclip:onboarding-seed:${companyId}`}, 0))`, + ); + return applyWithin(tx as unknown as Db, companyId, seed); + }); + } + + return { apply, get: (companyId: string) => readRecord(db, companyId) }; }