diff --git a/packages/hub-client/src/default-routines.ts b/packages/hub-client/src/default-routines.ts index fb9a31649..368d12d33 100644 --- a/packages/hub-client/src/default-routines.ts +++ b/packages/hub-client/src/default-routines.ts @@ -14,10 +14,17 @@ // what CL-6201 asks of the previously-stranded last-30-days-research // definition. // -// Idempotent by name, the same convention `seed.ts`'s own -// `ensureCatalogOffering`/`ensureWorkflowAsset` use: a re-seed lists -// existing routines first and skips any preset already present, never -// creating a duplicate. +// Idempotent by a stable `presetKey` (each preset's own `assetName`), +// enforced server-side by `@corbits/routines`' `createRoutineIfAbsent` +// (a real `INSERT ... ON CONFLICT DO NOTHING`, unique per +// `(tenantId, presetKey)` — see packages/routines/src/store.ts and +// migrations.ts' 0005). The list-then-skip check below is only a fast +// path that avoids a redundant deploy lookup and API round trip on the +// common case; it is never the thing that prevents a duplicate. Two +// overlapping `ensureDefaultRoutines` calls (e.g. two "finish setup" +// requests racing, as `pending-seed.ts` explicitly allows) can both +// pass this check and both POST — the server-side conflict target is +// what guarantees exactly one row and one "Created routine" notice. import { paginatedSchema } from "@intx/types"; import { type } from "arktype"; import { CliError } from "./errors"; @@ -170,6 +177,8 @@ export async function ensureDefaultRoutines( trigger: preset.trigger, scope: "bench", input: preset.input, + // The create-if-absent identity — see this file's header comment. + presetKey: preset.assetName, }; if (sharedDeliveryWorkbenchId !== undefined) { body.deliveryWorkbenchId = sharedDeliveryWorkbenchId; @@ -196,7 +205,12 @@ export async function ensureDefaultRoutines( ); continue; } - if (created.status !== 201) { + // 201: this call genuinely minted the row. 200: `presetKey` already + // resolved to an existing row (this preset's own prior seed, or the + // winner of a race against another overlapping seed call) — the + // server already skipped the "Created routine" notice and any + // fire, so there is nothing left to do but note it and move on. + if (created.status !== 201 && created.status !== 200) { throw new CliError( `the hub rejected creation of the default routine "${preset.name}" with status ${created.status}: ${JSON.stringify(created.data)}`, "check the hub logs for the underlying failure, then re-run: workbench seed", @@ -211,6 +225,11 @@ export async function ensureDefaultRoutines( sharedDeliveryWorkbenchId = row.deliveryWorkbenchId; } + if (created.status === 200) { + log(`routine "${preset.name}" already exists (skipped)`); + continue; + } + const disabled = await api( "PATCH", `/api/tenants/${tenantId}/routines/${row.id}`, diff --git a/packages/hub-client/test/default-routines.test.ts b/packages/hub-client/test/default-routines.test.ts index 6a159f0c7..97a55683f 100644 --- a/packages/hub-client/test/default-routines.test.ts +++ b/packages/hub-client/test/default-routines.test.ts @@ -223,6 +223,104 @@ describe("ensureDefaultRoutines", () => { ); }); + test("sends each preset's assetName as presetKey — the create-if-absent identity", async () => { + const { log } = collector(); + const createCalls: { body: unknown }[] = []; + const handler: FakeHandler = (method, path, body) => { + if ( + method === "GET" && + path === `/api/tenants/${TENANT_ID}/workflows/definitions` + ) { + return { + status: 200, + data: { + data: [definitionRow("wfd_digest", "workbench-digest")], + nextCursor: null, + }, + }; + } + if (method === "GET" && path === `/api/tenants/${TENANT_ID}/routines`) { + return { status: 200, data: { items: [] } }; + } + if (method === "POST" && path === `/api/tenants/${TENANT_ID}/routines`) { + createCalls.push({ body }); + return { + status: 201, + data: routineRow({ id: "rtn_1", name: "Daily digest" }), + }; + } + if ( + method === "PATCH" && + path.startsWith(`/api/tenants/${TENANT_ID}/routines/`) + ) { + return { status: 200, data: {} }; + } + return undefined; + }; + + await ensureDefaultRoutines(fakeAPI(handler), [], TENANT_ID, log); + + expect(createCalls[0]?.body).toMatchObject({ + presetKey: "workbench-digest", + }); + }); + + // CL-6375: even when the app-level "already present" check races + // (two overlapping seed calls both list zero existing routines, so + // both POST), the server's own create-if-absent guarantee means only + // one of the two POSTs actually mints a row. `ensureDefaultRoutines` + // must read a 200 as "already exists" — no duplicate disable, no + // treating it as a fresh seed. + test("a 200 create response (lost the create-if-absent race) is treated as already-seeded, not re-disabled", async () => { + const { lines, log } = collector(); + const patchCalls: { id: string }[] = []; + const handler: FakeHandler = (method, path) => { + if ( + method === "GET" && + path === `/api/tenants/${TENANT_ID}/workflows/definitions` + ) { + return { + status: 200, + data: { + data: [definitionRow("wfd_digest", "workbench-digest")], + nextCursor: null, + }, + }; + } + if (method === "GET" && path === `/api/tenants/${TENANT_ID}/routines`) { + // The app-level pre-check itself raced and saw nothing yet — + // the server is the one that actually caught the duplicate. + return { status: 200, data: { items: [] } }; + } + if (method === "POST" && path === `/api/tenants/${TENANT_ID}/routines`) { + return { + status: 200, + data: routineRow({ + id: "rtn_winner", + name: "Daily digest", + deliveryWorkbenchId: "ch_winner", + enabled: false, + }), + }; + } + if ( + method === "PATCH" && + path.startsWith(`/api/tenants/${TENANT_ID}/routines/`) + ) { + patchCalls.push({ id: path.split("/").pop() ?? "" }); + return { status: 200, data: {} }; + } + return undefined; + }; + + await ensureDefaultRoutines(fakeAPI(handler), [], TENANT_ID, log); + + expect(patchCalls).toHaveLength(0); + expect(lines.join("\n")).toContain( + 'routine "Daily digest" already exists (skipped)', + ); + }); + test("a non-201 create response is a loud failure naming the preset", async () => { const { log } = collector(); const handler: FakeHandler = (method, path) => { diff --git a/packages/routines/src/migrations.ts b/packages/routines/src/migrations.ts index 1e184d9b0..62f45717a 100644 --- a/packages/routines/src/migrations.ts +++ b/packages/routines/src/migrations.ts @@ -91,6 +91,24 @@ export const routineMigrations: readonly RoutineMigration[] = [ ALTER TABLE "routines"."routine_draft" RENAME COLUMN "delivery_channel_id" TO "delivery_workbench_id"; `, }, + // CL-6375: a template-minted routine (e.g. a `DEFAULT_ROUTINE_PRESETS` + // entry) carries a stable `preset_key`, unique per tenant while the + // row is live. This is what makes re-seeding a genuine create-if-absent + // rather than the app-level "list, then create if missing" race that + // let two overlapping seed calls each insert their own "Daily digest" + // routine (and each provision its own delivery workbench) — the + // partial index below is what a concurrent `INSERT ... ON CONFLICT DO + // NOTHING` targets. + { + name: "0005_routine_preset_key", + sql: ` + ALTER TABLE "routines"."routine" + ADD COLUMN IF NOT EXISTS "preset_key" text; + CREATE UNIQUE INDEX IF NOT EXISTS "routine_tenant_preset_key_idx" + ON "routines"."routine" ("tenant_id", "preset_key") + WHERE "preset_key" IS NOT NULL AND "deleted_at" IS NULL; + `, + }, ]; // Named distinctly from the platform's setup ledger and from any diff --git a/packages/routines/src/routes.ts b/packages/routines/src/routes.ts index 6865af92f..0af720961 100644 --- a/packages/routines/src/routes.ts +++ b/packages/routines/src/routes.ts @@ -240,6 +240,12 @@ const CreateRoutineBody = type({ // must never be forced to collect a workbench it would just discard. "deliveryWorkbenchId?": "string", "runOnceNow?": "boolean", + // Present only for a template-minted routine (e.g. a + // `DEFAULT_ROUTINE_PRESETS` entry seeded by `ensureDefaultRoutines`) — + // see `RoutineStore.createRoutineIfAbsent`'s own doc comment for the + // create-if-absent guarantee this unlocks. Absent for every + // person-authored create, unchanged prior behavior. + "presetKey?": "string", }); const UpdateRoutineBody = type({ @@ -623,17 +629,34 @@ export function createRoutineRoutes( body.deliveryWorkbenchId ?? provisionedSpace?.workbenchId ?? null; let row: RoutineRow; + let created = true; try { - row = await deps.store.createRoutine({ - tenantId: tenant.id, - name: body.name, - definitionId: body.definitionId, - trigger: body.trigger, - scope: body.scope, - input: body.input ?? {}, - deliveryWorkbenchId, - createdBy: principal.id, - }); + if (body.presetKey !== undefined) { + const result = await deps.store.createRoutineIfAbsent({ + tenantId: tenant.id, + name: body.name, + definitionId: body.definitionId, + trigger: body.trigger, + scope: body.scope, + input: body.input ?? {}, + deliveryWorkbenchId, + createdBy: principal.id, + presetKey: body.presetKey, + }); + row = result.row; + created = result.created; + } else { + row = await deps.store.createRoutine({ + tenantId: tenant.id, + name: body.name, + definitionId: body.definitionId, + trigger: body.trigger, + scope: body.scope, + input: body.input ?? {}, + deliveryWorkbenchId, + createdBy: principal.id, + }); + } } catch (err) { if (provisionedSpace !== undefined) { log.error( @@ -655,6 +678,30 @@ export function createRoutineRoutes( throw err; } + // Lost the create-if-absent race (or this is a genuine re-seed): + // the preset row already exists. Any space this request just + // provisioned points at nothing real and is compensated (deleted) + // rather than left orphaned — the winning request's own row keeps + // whatever delivery workbench it resolved to. No fire, no notice: + // both already happened (or are about to happen) on the winning + // request. + if (!created) { + if (provisionedSpace !== undefined) { + try { + await provisionedSpace.compensate(); + } catch (compensationErr) { + log.error( + "Compensation failed for orphaned space {workbenchId} " + + "after losing a routine create-if-absent race; this " + + "space now has no routine pointing at it and requires " + + "manual cleanup", + { workbenchId: provisionedSpace.workbenchId, compensationErr }, + ); + } + } + return c.json(routineView(row), 200); + } + if (body.runOnceNow === true) { await launchAndCorrelate( { store: deps.store, launcher: deps.launcher }, diff --git a/packages/routines/src/schema.ts b/packages/routines/src/schema.ts index 51073e92b..3b8138fa0 100644 --- a/packages/routines/src/schema.ts +++ b/packages/routines/src/schema.ts @@ -38,6 +38,13 @@ export const routine = routinesSchema.table("routine", { enabled: boolean("enabled").notNull().default(true), deliveryWorkbenchId: text("delivery_workbench_id"), createdBy: text("created_by").notNull(), + // A stable identity for a routine minted from a fixed template (e.g. + // a `DEFAULT_ROUTINE_PRESETS` entry) — `null` for an ordinary, + // person-authored routine. Enforced unique per tenant (see + // migrations.ts' 0005), so re-running the same seed/mint call twice — + // including two overlapping calls racing each other — is a real + // create-if-absent, not a check-then-insert that can double-create. + presetKey: text("preset_key"), // The due-fire clock: the next minute this routine's trigger matches, // recomputed on create, on every trigger/enabled change, and on each // fire. A scheduler tests `nextFireAt <= now`, not "does this exact diff --git a/packages/routines/src/store.ts b/packages/routines/src/store.ts index bf29b8e4d..c3a88ac93 100644 --- a/packages/routines/src/store.ts +++ b/packages/routines/src/store.ts @@ -3,7 +3,7 @@ // persistence from `routes.ts`. `RoutineStore` is the seam the route // layer depends on; `createDrizzleRoutineStore` is its one production // implementation, over the tables in `./schema.ts`. -import { and, desc, eq, isNull, lte } from "drizzle-orm"; +import { and, desc, eq, isNull, lte, sql } from "drizzle-orm"; import type { PostgresJsDatabase } from "drizzle-orm/postgres-js"; import { generateId } from "@intx/hub-common"; @@ -53,6 +53,9 @@ export interface RoutineRow { readonly deadLetteredAt: Date | null; readonly createdAt: Date; readonly updatedAt: Date; + /** See `schema.ts`' `preset_key` column doc comment. `null` for an + * ordinary, person-authored routine. */ + readonly presetKey: string | null; } export interface CreateRoutineInput { @@ -64,6 +67,24 @@ export interface CreateRoutineInput { readonly input: Record; readonly deliveryWorkbenchId?: string | null; readonly createdBy: string; + readonly presetKey?: string | null; +} + +/** `createRoutineIfAbsent`'s own input: `presetKey` is the whole point + * of this call (mandatory, unlike `CreateRoutineInput`'s optional + * field), since it is the identity the create-if-absent conflict + * target matches on. */ +export type CreateRoutineIfAbsentInput = Omit< + CreateRoutineInput, + "presetKey" +> & { readonly presetKey: string }; + +export interface CreateRoutineIfAbsentResult { + readonly row: RoutineRow; + /** `false` when a row for this `(tenantId, presetKey)` already + * existed — including one created by a request that raced this one — + * and this call is the loser that must not re-announce or re-fire. */ + readonly created: boolean; } export interface UpdateRoutineInput { @@ -91,6 +112,19 @@ export interface MarkFailedFireResult { export interface RoutineStore { createRoutine(input: CreateRoutineInput): Promise; + /** + * Real create-if-absent, keyed on `(tenantId, presetKey)`: a single + * atomic `INSERT ... ON CONFLICT DO NOTHING` (backed by + * `routine_tenant_preset_key_idx`, migrations.ts' 0005), never a + * check-then-insert. Two overlapping calls with the same + * `(tenantId, presetKey)` — including genuinely concurrent ones — are + * guaranteed exactly one winner (`created: true`) and any number of + * losers (`created: false`, returning the winner's own row), never + * two rows. + */ + createRoutineIfAbsent( + input: CreateRoutineIfAbsentInput, + ): Promise; /** `undefined` for an unknown OR a soft-deleted routine. */ getRoutine( tenantId: string, @@ -196,6 +230,7 @@ function mapRoutineRow(row: typeof routine.$inferSelect): RoutineRow { deadLetteredAt: row.deadLetteredAt, createdAt: row.createdAt, updatedAt: row.updatedAt, + presetKey: row.presetKey ?? null, }; } @@ -236,6 +271,7 @@ export function createDrizzleRoutineStore< deadLetteredAt: null, createdAt: now, updatedAt: now, + presetKey: input.presetKey ?? null, }) .returning(); if (row === undefined) { @@ -244,6 +280,61 @@ export function createDrizzleRoutineStore< return mapRoutineRow(row); }, + async createRoutineIfAbsent(input) { + const now = new Date(); + const [inserted] = await db + .insert(routine) + .values({ + id: generateId("workflowRun"), + tenantId: input.tenantId, + name: input.name, + definitionId: input.definitionId, + trigger: input.trigger, + scope: input.scope, + input: input.input, + enabled: true, + deliveryWorkbenchId: input.deliveryWorkbenchId ?? null, + createdBy: input.createdBy, + nextFireAt: computeNextFireAt(input.trigger, now), + lastFireAt: null, + deletedAt: null, + consecutiveFailures: 0, + deadLetteredAt: null, + createdAt: now, + updatedAt: now, + presetKey: input.presetKey, + }) + .onConflictDoNothing({ + target: [routine.tenantId, routine.presetKey], + where: sql`${routine.presetKey} is not null and ${routine.deletedAt} is null`, + }) + .returning(); + if (inserted !== undefined) { + return { row: mapRoutineRow(inserted), created: true }; + } + + // Lost the race (or a genuine re-seed): the winner's row is the + // one this `(tenantId, presetKey)` now resolves to. + const [existing] = await db + .select() + .from(routine) + .where( + and( + eq(routine.tenantId, input.tenantId), + eq(routine.presetKey, input.presetKey), + isNull(routine.deletedAt), + ), + ) + .limit(1); + if (existing === undefined) { + throw new Error( + `createRoutineIfAbsent: insert conflicted for preset key ` + + `${JSON.stringify(input.presetKey)} but no existing row was found`, + ); + } + return { row: mapRoutineRow(existing), created: false }; + }, + async getRoutine(tenantId, routineId) { const [row] = await db .select() @@ -532,11 +623,47 @@ export function createInMemoryRoutineStore(): RoutineStore { deadLetteredAt: null, createdAt: now, updatedAt: now, + presetKey: input.presetKey ?? null, }; routinesById.set(row.id, row); return row; }, + async createRoutineIfAbsent(input) { + const existing = [...routinesById.values()].find( + (row) => + row.tenantId === input.tenantId && + row.presetKey === input.presetKey && + row.deletedAt === null, + ); + if (existing !== undefined) { + return { row: existing, created: false }; + } + const now = new Date(); + const row: RoutineRow = { + id: generateId("workflowRun"), + tenantId: input.tenantId, + name: input.name, + definitionId: input.definitionId, + trigger: input.trigger, + scope: input.scope, + input: input.input, + enabled: true, + deliveryWorkbenchId: input.deliveryWorkbenchId ?? null, + createdBy: input.createdBy, + nextFireAt: computeNextFireAt(input.trigger, now), + lastFireAt: null, + deletedAt: null, + consecutiveFailures: 0, + deadLetteredAt: null, + createdAt: now, + updatedAt: now, + presetKey: input.presetKey, + }; + routinesById.set(row.id, row); + return { row, created: true }; + }, + async getRoutine(tenantId, routineId) { const row = routinesById.get(routineId); if (row === undefined || row.tenantId !== tenantId) return undefined; diff --git a/packages/routines/test/routes.test.ts b/packages/routines/test/routes.test.ts index c81d2c534..c6ef2f816 100644 --- a/packages/routines/test/routes.test.ts +++ b/packages/routines/test/routes.test.ts @@ -789,6 +789,94 @@ describe("createRoutineRoutes", () => { }); }); + // CL-6375: a template-minted routine (e.g. a seeded default preset) + // must never re-create itself or re-announce on a second seed call — + // real create-if-absent, not a check-then-insert race. + describe("presetKey (create-if-absent)", () => { + function fakeDeliverySpaceMinting() { + let createCalls = 0; + const compensatedWorkbenchIds: string[] = []; + return { + get createCalls() { + return createCalls; + }, + compensatedWorkbenchIds, + async createDeliverySpace() { + createCalls += 1; + const workbenchId = `ch_minted_${createCalls}`; + return { + workbenchId, + compensate: async () => { + compensatedWorkbenchIds.push(workbenchId); + }, + }; + }, + }; + } + + test("a second create with the same presetKey returns 200 and reuses the first row, never a second one", async () => { + const deps = buildDeps(); + const app = mountAs(createRoutineRoutes(deps), "user_1"); + const body = { ...VALID_BODY, presetKey: "workbench-digest" }; + + const first = await createRoutine(app, body); + const second = await createRoutine(app, body); + + expect(first.response.status).toBe(201); + expect(second.response.status).toBe(200); + expect(second.body["id"]).toBe(first.body["id"]); + + const rows = await deps.store.listRoutines(TENANT.id); + expect(rows.length).toBe(1); + }); + + test("only the genuine first creation posts the 'Created routine' notice", async () => { + const workbenchNotice = fakeWorkbenchNotice(); + const deps = buildDeps({ workbenchNotice }); + const app = mountAs(createRoutineRoutes(deps), "user_1"); + const body = { ...VALID_BODY, presetKey: "workbench-digest" }; + + await createRoutine(app, body); + await createRoutine(app, body); + await createRoutine(app, body); + + expect(workbenchNotice.calls.length).toBe(1); + }); + + test("a losing create-if-absent request compensates (deletes) the space it just provisioned", async () => { + const deliverySpace = fakeDeliverySpaceMinting(); + const deps = buildDeps({ deliverySpace }); + const app = mountAs(createRoutineRoutes(deps), "user_1"); + const { deliveryWorkbenchId: _drop, ...withoutWorkbench } = VALID_BODY; + const body = { ...withoutWorkbench, presetKey: "workbench-digest" }; + + const first = await createRoutine(app, body); + const second = await createRoutine(app, body); + + expect(first.response.status).toBe(201); + expect(second.response.status).toBe(200); + // The winner's own space stays bound; the loser's freshly-minted + // space is the one compensated away. + expect(second.body["deliveryWorkbenchId"]).toBe( + first.body["deliveryWorkbenchId"], + ); + expect(deliverySpace.createCalls).toBe(2); + expect(deliverySpace.compensatedWorkbenchIds).toEqual(["ch_minted_2"]); + }); + + test("a plain create (no presetKey) is unaffected — two same-named routines are both created", async () => { + const deps = buildDeps(); + const app = mountAs(createRoutineRoutes(deps), "user_1"); + + const first = await createRoutine(app, VALID_BODY); + const second = await createRoutine(app, VALID_BODY); + + expect(first.response.status).toBe(201); + expect(second.response.status).toBe(201); + expect(second.body["id"]).not.toBe(first.body["id"]); + }); + }); + describe("validateRoutineInput", () => { test("creates without validation when no port is wired (prior behavior)", async () => { const deps = buildDeps();