From bec4f4550e3cf497c6a6d0e6dcc7e4c161c26629 Mon Sep 17 00:00:00 2001 From: Tonio Date: Wed, 12 Aug 2026 12:03:13 -0700 Subject: [PATCH] fix(server): serialize seed application and keep its writes consistent MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Greptile P1s on the seed receiver: 1. Concurrent application duplicated entities. Cloud's reconcile runs off portfolio fetches that can overlap, so the same company could be pushed twice at once; both passed the revision check before either wrote the seed record, each creating a goal, an agent and an Onboarding project. The issue idempotency key only serialized issue creation. Wrap the whole application in a transaction that first takes a per-company advisory lock (`pg_advisory_xact_lock`), the same idiom `folders` and `decision-queues` use, so the second push observes the first's writes and updates in place. All reads/writes now go through the locked transaction. 2. A revised first task kept stale relationships. When a later revision added an agent or goal for a task that already existed, the update branch set only title/description, leaving the issue's `assigneeAgentId`/`goalId` null or stale while the seed record reported the new relationships. Carry the resolved agent and goal onto the update too (only when present, so an absent value never clears an assignment the tenant made). 3. A failed audit write left the entry permanently absent. The seed and its revision commit before the route logs `company.onboarding_seed_applied`; if that log threw, the request 500'd and the next Cloud retry returned `changed: false` and skipped logging for good — for an application that had in fact succeeded. Make the audit write best-effort: a failure is logged server-side and the request still answers 200, matching the "any 2xx means applied" contract Cloud relies on. Tests: two racing identical pushes yield exactly one goal/agent/project/ task, and a later revision that adds an agent refreshes the existing task's assignee and goal. `vitest run onboarding-seed-route.test.ts` — 13 passing. Co-Authored-By: Paperclip --- .../__tests__/onboarding-seed-route.test.ts | 57 ++++++++ server/src/routes/onboarding-seed.ts | 49 ++++--- server/src/services/onboarding-seed.ts | 126 ++++++++++++------ 3 files changed, 174 insertions(+), 58 deletions(-) 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) }; }