From 74c5105c82cb4b663cfc85ddeb3e481f11e1258f Mon Sep 17 00:00:00 2001 From: Sawyer Date: Thu, 17 Sep 2026 14:54:34 -0700 Subject: [PATCH 1/4] Add tests for @corbits/cron (CL-8183) Cron expression tests moved from the retired @corbits/workflow-schedule, plus a DB-gated ticker test proving a due row fires once, a not-yet-due row doesn't, and SKIP LOCKED means two concurrent tickers never double-fire the same row. --- packages/cron/src/cron.test.ts | 230 ++++++++++++++++++++++ packages/cron/test/ticker.drizzle.test.ts | 142 +++++++++++++ 2 files changed, 372 insertions(+) create mode 100644 packages/cron/src/cron.test.ts create mode 100644 packages/cron/test/ticker.drizzle.test.ts diff --git a/packages/cron/src/cron.test.ts b/packages/cron/src/cron.test.ts new file mode 100644 index 0000000000..6840184235 --- /dev/null +++ b/packages/cron/src/cron.test.ts @@ -0,0 +1,230 @@ +// Table-driven cron correctness: step-from-min, DOM/DOW OR, Sunday=7, +// bounded lookahead, and timezone / DST round-trips. +import { describe, expect, test } from "bun:test"; + +import { + cronExpressionCanFire, + cronMatchesMinute, + isValidCronExpression, + isValidTimeZone, + nextCronFireAfter, + zonedParts, +} from "./cron"; + +describe("isValidCronExpression", () => { + test("accepts standard 5-field expressions", () => { + expect(isValidCronExpression("*/5 * * * *")).toBe(true); + expect(isValidCronExpression("0 9 * * 1")).toBe(true); + expect(isValidCronExpression("15,45 * * * *")).toBe(true); + expect(isValidCronExpression("0 0 13 * 5")).toBe(true); + }); + + test("accepts 7 as Sunday on day-of-week", () => { + expect(isValidCronExpression("* * * * 7")).toBe(true); + expect(isValidCronExpression("0 0 * * 0,7")).toBe(true); + }); + + test("rejects wrong field counts and garbage", () => { + expect(isValidCronExpression("* * * *")).toBe(false); + expect(isValidCronExpression("* * * * * *")).toBe(false); + expect(isValidCronExpression("not a cron string at all")).toBe(false); + expect(isValidCronExpression("")).toBe(false); + }); + + test("rejects every field out of range", () => { + expect(isValidCronExpression("99 99 99 99 99")).toBe(false); + expect(isValidCronExpression("0 0 32 * *")).toBe(false); + expect(isValidCronExpression("0 0 * 13 *")).toBe(false); + expect(isValidCronExpression("60 * * * *")).toBe(false); + expect(isValidCronExpression("* * * * 8")).toBe(false); + }); + + test("rejects a reversed range", () => { + expect(isValidCronExpression("10-5 * * * *")).toBe(false); + expect(isValidCronExpression("10-5/2 * * * *")).toBe(false); + }); + + test("accepts the standard range-then-step idiom", () => { + expect(isValidCronExpression("5-10/2 * * * *")).toBe(true); + }); +}); + +describe("step fields offset from field minimum", () => { + // `*/2` on day-of-month (min=1) → 1,3,5… not 2,4,6… + const cases: { + name: string; + expression: string; + at: string; + matches: boolean; + }[] = [ + { + name: "DOM */2 matches day 1", + expression: "0 0 */2 * *", + at: "2026-01-01T00:00:00Z", + matches: true, + }, + { + name: "DOM */2 does not match day 2", + expression: "0 0 */2 * *", + at: "2026-01-02T00:00:00Z", + matches: false, + }, + { + name: "DOM */2 matches day 3", + expression: "0 0 */2 * *", + at: "2026-01-03T00:00:00Z", + matches: true, + }, + { + name: "minute */2 still matches 0 (min=0)", + expression: "*/2 * * * *", + at: "2026-01-01T00:00:00Z", + matches: true, + }, + { + name: "minute */2 matches 2", + expression: "*/2 * * * *", + at: "2026-01-01T00:02:00Z", + matches: true, + }, + { + name: "minute */2 does not match 1", + expression: "*/2 * * * *", + at: "2026-01-01T00:01:00Z", + matches: false, + }, + { + name: "month */2 matches January (min=1)", + expression: "0 0 1 */2 *", + at: "2026-01-01T00:00:00Z", + matches: true, + }, + { + name: "month */2 does not match February", + expression: "0 0 1 */2 *", + at: "2026-02-01T00:00:00Z", + matches: false, + }, + ]; + + for (const c of cases) { + test(c.name, () => { + expect(cronMatchesMinute(c.expression, new Date(c.at))).toBe(c.matches); + }); + } +}); + +describe("DOM / DOW OR when both restricted (POSIX/Vixie)", () => { + // `0 0 13 * 5` = midnight on the 13th OR any Friday. + const expression = "0 0 13 * 5"; + + test("matches the 13th even when it is not Friday", () => { + // 2026-01-13 is a Tuesday. + expect(cronMatchesMinute(expression, new Date("2026-01-13T00:00:00Z"))).toBe(true); + }); + + test("matches a Friday that is not the 13th", () => { + // 2026-01-16 is a Friday. + expect(cronMatchesMinute(expression, new Date("2026-01-16T00:00:00Z"))).toBe(true); + }); + + test("does not match a day that is neither the 13th nor Friday", () => { + // 2026-01-14 is a Wednesday. + expect(cronMatchesMinute(expression, new Date("2026-01-14T00:00:00Z"))).toBe(false); + }); + + test("when only DOM is restricted, DOW is ignored (AND with *)", () => { + // Only the 15th, any weekday. + expect(cronMatchesMinute("0 0 15 * *", new Date("2026-01-15T00:00:00Z"))).toBe(true); + expect(cronMatchesMinute("0 0 15 * *", new Date("2026-01-16T00:00:00Z"))).toBe(false); + }); + + test("when only DOW is restricted, DOM is ignored", () => { + // Every Monday at midnight. + expect(cronMatchesMinute("0 0 * * 1", new Date("2026-01-12T00:00:00Z"))).toBe(true); + expect(cronMatchesMinute("0 0 * * 1", new Date("2026-01-13T00:00:00Z"))).toBe(false); + }); +}); + +describe("day-of-week 0 and 7 are both Sunday", () => { + test("expression with 7 matches a Sunday Date (day 0)", () => { + // 2026-01-04 is a Sunday. + expect(cronMatchesMinute("0 0 * * 7", new Date("2026-01-04T00:00:00Z"))).toBe(true); + }); + + test("expression with 0 matches the same Sunday", () => { + expect(cronMatchesMinute("0 0 * * 0", new Date("2026-01-04T00:00:00Z"))).toBe(true); + }); + + test("neither matches a Monday", () => { + expect(cronMatchesMinute("0 0 * * 7", new Date("2026-01-05T00:00:00Z"))).toBe(false); + }); +}); + +describe("nextCronFireAfter + canFire bounds", () => { + test("finds the next matching minute", () => { + const next = nextCronFireAfter("0-5 * * * *", new Date("2026-01-01T00:00:00Z")); + expect(next.toISOString()).toBe("2026-01-01T00:01:00.000Z"); + }); + + test("impossible Feb 31 fails canFire and nextCronFireAfter", () => { + expect(cronExpressionCanFire("0 0 31 2 *")).toBe(false); + expect(() => nextCronFireAfter("0 0 31 2 *", new Date("2026-01-01T00:00:00Z"))).toThrow( + /no fire time within the lookahead window/, + ); + }); + + test("a once-a-year expression still finds its fire", () => { + // Jan 1 at 00:00 — after Dec 31, next is next year. + const next = nextCronFireAfter("0 0 1 1 *", new Date("2026-06-01T00:00:00Z")); + expect(next.toISOString()).toBe("2027-01-01T00:00:00.000Z"); + }); +}); + +describe("timezone matching and DST", () => { + test("isValidTimeZone accepts IANA names and rejects garbage", () => { + expect(isValidTimeZone("UTC")).toBe(true); + expect(isValidTimeZone("America/Los_Angeles")).toBe(true); + expect(isValidTimeZone("Europe/London")).toBe(true); + expect(isValidTimeZone("Not/A_Zone")).toBe(false); + expect(isValidTimeZone("")).toBe(false); + }); + + test("daily 09:00 America/Los_Angeles matches the correct UTC instant (PST)", () => { + // 2026-01-15 is winter — PST = UTC-8, so 09:00 local = 17:00 UTC. + const at = new Date("2026-01-15T17:00:00Z"); + expect(zonedParts(at, "America/Los_Angeles")).toMatchObject({ + hour: 9, + minute: 0, + day: 15, + month: 1, + }); + expect(cronMatchesMinute("0 9 * * *", at, "America/Los_Angeles")).toBe(true); + expect(cronMatchesMinute("0 9 * * *", at, "UTC")).toBe(false); + }); + + test("next fire for 09:00 America/Los_Angeles across a DST spring-forward", () => { + // US Pacific spring forward 2026: 2026-03-08 02:00 → 03:00 local. + // Before the transition (March 7 12:00 UTC = March 7 04:00 PST): + // next 09:00 local is March 7 09:00 PST = March 7 17:00 UTC. + const before = new Date("2026-03-07T12:00:00Z"); + const nextBefore = nextCronFireAfter("0 9 * * *", before, "America/Los_Angeles"); + expect(nextBefore.toISOString()).toBe("2026-03-07T17:00:00.000Z"); + expect(zonedParts(nextBefore, "America/Los_Angeles").hour).toBe(9); + + // After spring-forward, 09:00 PDT = UTC-7 → 16:00 UTC. + const afterTransition = new Date("2026-03-09T12:00:00Z"); + const nextAfter = nextCronFireAfter("0 9 * * *", afterTransition, "America/Los_Angeles"); + expect(nextAfter.toISOString()).toBe("2026-03-09T16:00:00.000Z"); + expect(zonedParts(nextAfter, "America/Los_Angeles").hour).toBe(9); + }); + + test("next fire for 09:00 America/Los_Angeles across a DST fall-back", () => { + // US Pacific fall back 2026: 2026-11-01 02:00 → 01:00 local. + // After the transition, 09:00 PST = UTC-8 → 17:00 UTC. + const afterFallback = new Date("2026-11-02T12:00:00Z"); + const next = nextCronFireAfter("0 9 * * *", afterFallback, "America/Los_Angeles"); + expect(next.toISOString()).toBe("2026-11-02T17:00:00.000Z"); + expect(zonedParts(next, "America/Los_Angeles").hour).toBe(9); + }); +}); diff --git a/packages/cron/test/ticker.drizzle.test.ts b/packages/cron/test/ticker.drizzle.test.ts new file mode 100644 index 0000000000..bf402e1fba --- /dev/null +++ b/packages/cron/test/ticker.drizzle.test.ts @@ -0,0 +1,142 @@ +// DB-gated: skipped when no DATABASE_URL is reachable, mirroring +// `packages/agent-directory/test/tool-package-version.drizzle.test.ts`. +// Platform migrations run into a scratch schema so this test never +// touches a developer's real tenant table; `applyCronMigrations` is told +// that scratch schema so its `tenant_id` FK targets it. +import { afterAll, beforeAll, expect, test } from "bun:test"; +import { randomUUID } from "node:crypto"; +import { createDB, dropSchema, runMigrations, schema } from "@intx/db"; +import { eq } from "drizzle-orm"; + +import { dbTargetFromUrl } from "../../../scripts/db-setup"; +import { e2eDatabaseUrl } from "../../../scripts/e2e/database-url"; +import { dbGate } from "../../../scripts/e2e/db-gate"; +import { applyCronMigrations, cronScheduleTable } from "../src/schema"; +import { createCronTicker } from "../src/ticker"; + +const databaseUrl = e2eDatabaseUrl(); +const describeIfDb = dbGate(databaseUrl, import.meta.path); + +const SCHEMA = "cron_ticker_test"; + +describeIfDb("createCronTicker", () => { + const target = dbTargetFromUrl(databaseUrl ?? "postgres://localhost:5432/unused"); + + beforeAll(async () => { + await runMigrations(target, { schema: SCHEMA }); + await applyCronMigrations(databaseUrl ?? "", { tenantSchema: SCHEMA }); + }); + + afterAll(async () => { + await dropSchema(target, { schema: SCHEMA }); + }); + + async function seedTenant(db: Awaited>["db"], id: string) { + await db.insert(schema.tenant).values({ + id, + name: id, + slug: id.replace(/_/g, "-"), + domain: `${id.replace(/_/g, "-")}.workbench.test`, + }); + } + + test("fires a due row once and leaves a not-yet-due row alone", async () => { + const { db, close } = createDB({ ...target, schema: SCHEMA }); + try { + const tenantId = `tnt_cron_${randomUUID().slice(0, 8)}`; + await seedTenant(db, tenantId); + + const dueId = `sched_due_${randomUUID().slice(0, 8)}`; + const notDueId = `sched_not_due_${randomUUID().slice(0, 8)}`; + const twoMinutesAgo = new Date(Date.now() - 2 * 60_000); + await db.insert(cronScheduleTable).values([ + { + id: dueId, + tenantId, + expression: "* * * * *", + toAddress: "ops@example.com", + subject: "due", + body: "fire me", + createdAt: twoMinutesAgo, + }, + { + id: notDueId, + tenantId, + expression: "0 0 1 1 *", + toAddress: "ops@example.com", + subject: "not due", + body: "never yet", + createdAt: twoMinutesAgo, + }, + ]); + + const delivered: string[] = []; + const ticker = createCronTicker({ + db, + intervalMs: 50, + deliver: (message) => { + delivered.push(message.subject); + }, + }); + ticker.start(); + await new Promise((resolve) => setTimeout(resolve, 200)); + ticker.stop(); + + expect(delivered).toEqual(["due"]); + + const [firedRow] = await db + .select() + .from(cronScheduleTable) + .where(eq(cronScheduleTable.id, dueId)); + expect(firedRow?.lastFiredAt).not.toBeNull(); + } finally { + await close(); + } + }); + + test("SKIP LOCKED means two concurrent tickers never double-fire a row", async () => { + const { db: dbA, close: closeA } = createDB({ ...target, schema: SCHEMA }); + const { db: dbB, close: closeB } = createDB({ ...target, schema: SCHEMA }); + try { + const tenantId = `tnt_cron_race_${randomUUID().slice(0, 8)}`; + await seedTenant(dbA, tenantId); + + const id = `sched_race_${randomUUID().slice(0, 8)}`; + await dbA.insert(cronScheduleTable).values({ + id, + tenantId, + expression: "* * * * *", + toAddress: "ops@example.com", + subject: "race", + body: "fire once", + createdAt: new Date(Date.now() - 2 * 60_000), + }); + + let fireCount = 0; + const tickerA = createCronTicker({ + db: dbA, + intervalMs: 20, + deliver: () => { + fireCount++; + }, + }); + const tickerB = createCronTicker({ + db: dbB, + intervalMs: 20, + deliver: () => { + fireCount++; + }, + }); + tickerA.start(); + tickerB.start(); + await new Promise((resolve) => setTimeout(resolve, 300)); + tickerA.stop(); + tickerB.stop(); + + expect(fireCount).toBe(1); + } finally { + await closeA(); + await closeB(); + } + }); +}); From 4a0983f097234106234326157fac9a58f9bc0537 Mon Sep 17 00:00:00 2001 From: Sawyer Date: Thu, 17 Sep 2026 14:54:49 -0700 Subject: [PATCH 2/4] Add @corbits/cron: cron ticks delivered as mail (CL-8183) MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit The vendored Interchange workflow-trigger grammar (vendor/intx/workflow/src/definition/triggers.ts) has no `schedule` trigger and no scheduler, so a saved cron expression can never launch a workflow natively. @corbits/cron is a small experimental bridge instead: a tenant saves a cron expression plus a mail (to/subject/body) on its own `cron.schedule` table, and `createCronTicker` polls for due schedules and delivers each as plain mail through the host's transport — a schedule-triggered workflow becomes a mail-triggered one addressed at itself. - Moves the 5-field cron parser out of the retired @corbits/workflow-schedule (deleted entirely; its only other consumer, @corbits/workflows' scheduleCronFromProjection, is deleted alongside it since nothing outside that package read the re-export). - One drizzle table, its own `cron` Postgres schema, FK'd to Interchange's native `tenant` table. - createCronTicker claims due rows with SELECT ... FOR UPDATE SKIP LOCKED so concurrent tickers never double-fire, and only the latest missed tick fires. - mountCron adds CRUD at /api/tenants/:tenantId/cron. --- apps/hub/package.json | 2 +- bun.lock | 36 ++- packages/{workflow-schedule => cron}/LICENSE | 0 packages/cron/README.md | 45 ++++ packages/cron/package.json | 27 ++ .../{workflow-schedule => cron}/src/cron.ts | 0 packages/cron/src/index.ts | 21 ++ packages/cron/src/migrations.ts | 1 + packages/cron/src/mount.ts | 80 ++++++ packages/cron/src/schema.ts | 118 +++++++++ packages/cron/src/ticker.ts | 120 +++++++++ .../{workflow-schedule => cron}/tsconfig.json | 2 +- .../tsconfig.src.json | 2 +- packages/workflow-schedule/package.json | 21 -- packages/workflow-schedule/src/cron.test.ts | 230 ------------------ .../workflow-schedule/src/emitter.test.ts | 92 ------- packages/workflow-schedule/src/emitter.ts | 182 -------------- packages/workflow-schedule/src/index.ts | 13 - packages/workflow-schedule/src/timer.test.ts | 40 --- packages/workflow-schedule/src/timer.ts | 56 ----- packages/workflows/package.json | 1 - packages/workflows/src/index.ts | 12 - .../src/schedule/from-projection.test.ts | 46 ---- .../workflows/src/schedule/from-projection.ts | 25 -- packages/workflows/tsconfig.json | 3 - packages/workflows/tsconfig.src.json | 3 - scripts/checks/no-product-tenancy.ts | 9 + scripts/db-setup.ts | 2 + tsconfig.build.json | 6 +- 29 files changed, 453 insertions(+), 742 deletions(-) rename packages/{workflow-schedule => cron}/LICENSE (100%) create mode 100644 packages/cron/README.md create mode 100644 packages/cron/package.json rename packages/{workflow-schedule => cron}/src/cron.ts (100%) create mode 100644 packages/cron/src/index.ts create mode 100644 packages/cron/src/migrations.ts create mode 100644 packages/cron/src/mount.ts create mode 100644 packages/cron/src/schema.ts create mode 100644 packages/cron/src/ticker.ts rename packages/{workflow-schedule => cron}/tsconfig.json (91%) rename packages/{workflow-schedule => cron}/tsconfig.src.json (87%) delete mode 100644 packages/workflow-schedule/package.json delete mode 100644 packages/workflow-schedule/src/cron.test.ts delete mode 100644 packages/workflow-schedule/src/emitter.test.ts delete mode 100644 packages/workflow-schedule/src/emitter.ts delete mode 100644 packages/workflow-schedule/src/index.ts delete mode 100644 packages/workflow-schedule/src/timer.test.ts delete mode 100644 packages/workflow-schedule/src/timer.ts delete mode 100644 packages/workflows/src/schedule/from-projection.test.ts delete mode 100644 packages/workflows/src/schedule/from-projection.ts diff --git a/apps/hub/package.json b/apps/hub/package.json index 5b1eda8f8e..af65b309b6 100644 --- a/apps/hub/package.json +++ b/apps/hub/package.json @@ -26,6 +26,7 @@ "@corbits/commands": "workspace:*", "@corbits/connections": "workspace:*", "@corbits/credential-providers": "workspace:*", + "@corbits/cron": "workspace:*", "@corbits/diligence-brief-workflow": "workspace:*", "@corbits/docker-provisioner": "workspace:*", "@corbits/e2b-sandbox-sidecar": "workspace:*", @@ -52,7 +53,6 @@ "@corbits/url-path": "workspace:*", "@corbits/webhook-triggers": "workspace:*", "@corbits/workbench-digest-workflow": "workspace:*", - "@corbits/workflow-schedule": "workspace:*", "@corbits/workflows": "workspace:*", "@corbits/xai-provider": "workspace:*", "@intx/authz": "0.3.0", diff --git a/bun.lock b/bun.lock index d9f3eaa70e..0a4da98167 100644 --- a/bun.lock +++ b/bun.lock @@ -104,6 +104,7 @@ "@corbits/commands": "workspace:*", "@corbits/connections": "workspace:*", "@corbits/credential-providers": "workspace:*", + "@corbits/cron": "workspace:*", "@corbits/diligence-brief-workflow": "workspace:*", "@corbits/docker-provisioner": "workspace:*", "@corbits/e2b-sandbox-sidecar": "workspace:*", @@ -130,7 +131,6 @@ "@corbits/url-path": "workspace:*", "@corbits/webhook-triggers": "workspace:*", "@corbits/workbench-digest-workflow": "workspace:*", - "@corbits/workflow-schedule": "workspace:*", "@corbits/workflows": "workspace:*", "@corbits/xai-provider": "workspace:*", "@intx/authz": "0.3.0", @@ -579,6 +579,21 @@ "typescript": "catalog:", }, }, + "packages/cron": { + "name": "@corbits/cron", + "version": "0.0.1", + "dependencies": { + "arktype": "catalog:", + "drizzle-orm": "catalog:", + "hono": "catalog:", + "postgres": "catalog:", + }, + "devDependencies": { + "@intx/db": "workspace:*", + "@types/bun": "catalog:", + "typescript": "catalog:", + }, + }, "packages/docker-provisioner": { "name": "@corbits/docker-provisioner", "version": "0.0.1", @@ -1008,20 +1023,11 @@ "typescript": "catalog:", }, }, - "packages/workflow-schedule": { - "name": "@corbits/workflow-schedule", - "version": "0.0.1", - "devDependencies": { - "@types/bun": "catalog:", - "typescript": "catalog:", - }, - }, "packages/workflows": { "name": "@corbits/workflows", "version": "0.0.1", "dependencies": { "@corbits/error-sink": "workspace:*", - "@corbits/workflow-schedule": "workspace:*", "@intx/authz": "0.3.0", "@intx/db": "workspace:*", "@intx/hub-api": "workspace:*", @@ -1988,6 +1994,8 @@ "@corbits/credential-providers": ["@corbits/credential-providers@workspace:packages/credential-providers"], + "@corbits/cron": ["@corbits/cron@workspace:packages/cron"], + "@corbits/diligence-brief-workflow": ["@corbits/diligence-brief-workflow@workspace:workflows/diligence-brief"], "@corbits/docker-provisioner": ["@corbits/docker-provisioner@workspace:packages/docker-provisioner"], @@ -2100,8 +2108,6 @@ "@corbits/workflow-authoring-tools": ["@corbits/workflow-authoring-tools@workspace:tools/workflow-authoring"], - "@corbits/workflow-schedule": ["@corbits/workflow-schedule@workspace:packages/workflow-schedule"], - "@corbits/workflows": ["@corbits/workflows@workspace:packages/workflows"], "@corbits/xai-provider": ["@corbits/xai-provider@workspace:packages/xai-provider"], @@ -3278,8 +3284,14 @@ "@babel/helper-compilation-targets/semver": ["semver@6.3.1", "", { "bin": { "semver": "bin/semver.js" } }, "sha512-BR7VvDCVHO+q2xBEWskxS6DJE1qRnb7DxzUrogb71CWoSficBxYsiAGd+Kl0mmq/MprG9yArRkyrQxTO6XjMzA=="], + "@corbits/artifact-ui/@corbits/react-ui": ["@corbits/react-ui@github:corbitsdev/react-ui#3b12281", { "dependencies": { "class-variance-authority": "^0.7.1", "clsx": "^2.1.1", "tailwind-merge": "^3.3.1" }, "peerDependencies": { "@radix-ui/react-dialog": "^1.1.15", "@radix-ui/react-dropdown-menu": "^2.1.16", "@radix-ui/react-slot": "^1.2.3", "@radix-ui/react-tooltip": "^1.2.8", "@tanstack/react-query": "^5.90.2", "lucide-react": "^0.545.0 || ^1.0.0", "react": "^18.2.0 || ^19.0.0", "react-dom": "^18.2.0 || ^19.0.0", "sonner": "^2.0.7" }, "optionalPeers": ["@tanstack/react-query"] }, "corbitsdev-react-ui-3b12281", "sha512-Abvm/DO0Gqg0ITHGT9355ZxyKRPMVJLSSQSjpd3a8qt4JPrSMOLIOS4sX8ZMNNaArIbnY9F+VKrOWkUJUyO4Nw=="], + + "@corbits/assistant-workflow/@corbits/memory": ["@corbits/memory@github:corbitsdev/corbits-memory#9e6f213", { "dependencies": { "@intx/agent": "0.2.2", "@intx/authz": "0.2.2", "@intx/hub-api": "0.2.2", "@intx/log": "0.2.2", "@intx/workflow": "0.2.2", "arktype": "^2.1.29", "drizzle-orm": "^0.45.1", "hono": "^4.9.0", "hono-openapi": "^1.3.1", "postgres": "^3.4.7" } }, "corbitsdev-corbits-memory-9e6f213", "sha512-utnM4ZT2zmslcPXYWAAqxlDNLcpGsXFiTOtj8h7+OXnhCP0Eaw8yl25+yCTyHpvt3jcdeG4h5uFsSj7ou0BZCA=="], + "@corbits/mailbox/hono-openapi": ["hono-openapi@1.3.1", "", { "peerDependencies": { "@hono/standard-validator": "^0.2.0", "@standard-community/standard-json": "^0.3.5", "@standard-community/standard-openapi": "^0.2.9", "@types/json-schema": "^7.0.15", "hono": "^4.11.2", "openapi-types": "^12.1.3" }, "optionalPeers": ["@hono/standard-validator", "hono"] }, "sha512-NLVeVkhKZ3drmQNEIPac8HX8Y54uf1hJAgIM/7MfDsaeVVmB+QILWQxx5x3R3NvRHgedcbEbOCGY2uR7WQYyMw=="], + "@corbits/scout-agent/@corbits/memory": ["@corbits/memory@github:corbitsdev/corbits-memory#9e6f213", { "dependencies": { "@intx/agent": "0.2.2", "@intx/authz": "0.2.2", "@intx/hub-api": "0.2.2", "@intx/log": "0.2.2", "@intx/workflow": "0.2.2", "arktype": "^2.1.29", "drizzle-orm": "^0.45.1", "hono": "^4.9.0", "hono-openapi": "^1.3.1", "postgres": "^3.4.7" } }, "corbitsdev-corbits-memory-9e6f213", "sha512-utnM4ZT2zmslcPXYWAAqxlDNLcpGsXFiTOtj8h7+OXnhCP0Eaw8yl25+yCTyHpvt3jcdeG4h5uFsSj7ou0BZCA=="], + "@esbuild-kit/core-utils/esbuild": ["esbuild@0.18.20", "", { "optionalDependencies": { "@esbuild/android-arm": "0.18.20", "@esbuild/android-arm64": "0.18.20", "@esbuild/android-x64": "0.18.20", "@esbuild/darwin-arm64": "0.18.20", "@esbuild/darwin-x64": "0.18.20", "@esbuild/freebsd-arm64": "0.18.20", "@esbuild/freebsd-x64": "0.18.20", "@esbuild/linux-arm": "0.18.20", "@esbuild/linux-arm64": "0.18.20", "@esbuild/linux-ia32": "0.18.20", "@esbuild/linux-loong64": "0.18.20", "@esbuild/linux-mips64el": "0.18.20", "@esbuild/linux-ppc64": "0.18.20", "@esbuild/linux-riscv64": "0.18.20", "@esbuild/linux-s390x": "0.18.20", "@esbuild/linux-x64": "0.18.20", "@esbuild/netbsd-x64": "0.18.20", "@esbuild/openbsd-x64": "0.18.20", "@esbuild/sunos-x64": "0.18.20", "@esbuild/win32-arm64": "0.18.20", "@esbuild/win32-ia32": "0.18.20", "@esbuild/win32-x64": "0.18.20" }, "bin": { "esbuild": "bin/esbuild" } }, "sha512-ceqxoedUrcayh7Y7ZX6NdbbDzGROiyVBgC4PriJThBKSVPWnnFHZAkfI1lJT8QFkOwH4qOS2SJkS4wvpGl8BpA=="], "@tailwindcss/oxide-wasm32-wasi/@emnapi/core": ["@emnapi/core@1.11.3", "", { "dependencies": { "@emnapi/wasi-threads": "1.2.3", "tslib": "^2.4.0" }, "bundled": true }, "sha512-zLpS5asjEb7lq8jYLq37N6XKaE41DIexlY1rF/z4/tIl3wo13Sqm28fRyfIsKZD+NZ8mM5RoKkpW/rBcuoSZSg=="], diff --git a/packages/workflow-schedule/LICENSE b/packages/cron/LICENSE similarity index 100% rename from packages/workflow-schedule/LICENSE rename to packages/cron/LICENSE diff --git a/packages/cron/README.md b/packages/cron/README.md new file mode 100644 index 0000000000..27842302c2 --- /dev/null +++ b/packages/cron/README.md @@ -0,0 +1,45 @@ +# @corbits/cron + +Experimental. The vendored Interchange workflow-trigger grammar +(`vendor/intx/workflow/src/definition/triggers.ts`) has no `schedule` +trigger and no scheduler — a workflow can only be launched by `mail`. This +package is the tiny bridge: a tenant saves a cron expression plus a mail +(`to`/`subject`/`body`); `createCronTicker` polls for due schedules and +delivers each as mail through the host's own transport, so a +schedule-triggered workflow is just a `mail`-triggered one addressed at +itself. + +- `src/cron.ts` — the 5-field cron grammar (moved from the retired + `@corbits/workflow-schedule`, unchanged). +- `src/schema.ts` — the `cron.schedule` table (its own Postgres schema, FK'd + to Interchange's `tenant` table) and its migration. +- `src/ticker.ts` — `createCronTicker({ db, deliver, intervalMs })`: claims + due rows with `SELECT ... FOR UPDATE SKIP LOCKED` so concurrent tickers + never double-fire, and only the latest missed tick fires. +- `src/mount.ts` — `mountCron(app, { db, requireTenantMember })`: CRUD at + `/api/tenants/:tenantId/cron`. + +## Mounting in the hub + +```ts +const cronApp = new Hono(); +mountCron(cronApp, { + db, + requireTenantMember: (ctx, tenantId) => { + const c = ctx as { get(key: "tenant"): { id: string } }; + return c.get("tenant").id === tenantId; + }, +}); +app.route("/", cronApp); + +createCronTicker({ + db, + intervalMs: 60_000, + deliver: (message) => + lookups.persistMail({ + senderAddress: message.from, + recipients: message.to, + raw: buildRawMessage(message), + }), +}).start(); +``` diff --git a/packages/cron/package.json b/packages/cron/package.json new file mode 100644 index 0000000000..9f2e4af3ff --- /dev/null +++ b/packages/cron/package.json @@ -0,0 +1,27 @@ +{ + "name": "@corbits/cron", + "version": "0.0.1", + "private": true, + "description": "Experimental cron-to-mail bridge: fires a tenant's saved cron schedules as plain `mail` deliveries, since the vendored Interchange workflow-trigger grammar has no native `schedule` trigger", + "license": "LGPL-2.1-or-later", + "type": "module", + "exports": { + ".": "./src/index.ts", + "./migrations": "./src/migrations.ts" + }, + "scripts": { + "typecheck": "tsc --noEmit", + "test": "bun test" + }, + "dependencies": { + "arktype": "catalog:", + "drizzle-orm": "catalog:", + "hono": "catalog:", + "postgres": "catalog:" + }, + "devDependencies": { + "@intx/db": "workspace:*", + "@types/bun": "catalog:", + "typescript": "catalog:" + } +} diff --git a/packages/workflow-schedule/src/cron.ts b/packages/cron/src/cron.ts similarity index 100% rename from packages/workflow-schedule/src/cron.ts rename to packages/cron/src/cron.ts diff --git a/packages/cron/src/index.ts b/packages/cron/src/index.ts new file mode 100644 index 0000000000..21d41d6238 --- /dev/null +++ b/packages/cron/src/index.ts @@ -0,0 +1,21 @@ +export { + CRON_FIELD_RANGES, + cronExpressionCanFire, + isValidCronExpression, + isValidTimeZone, + MAX_LOOKAHEAD_MINUTES, + nextCronFireAfter, + zonedParts, + type CronField, + type ZonedParts, +} from "./cron"; +export { cronScheduleTable } from "./schema"; +export { + createCronTicker, + cronSenderAddress, + type CronDb, + type CronSenderAddress, + type CronTicker, + type DeliverCronMail, +} from "./ticker"; +export { mountCron, type MountCronOpts, type RequireTenantMember } from "./mount"; diff --git a/packages/cron/src/migrations.ts b/packages/cron/src/migrations.ts new file mode 100644 index 0000000000..0fec612e01 --- /dev/null +++ b/packages/cron/src/migrations.ts @@ -0,0 +1 @@ +export { applyCronMigrations, type ApplyCronMigrationsReport } from "./schema"; diff --git a/packages/cron/src/mount.ts b/packages/cron/src/mount.ts new file mode 100644 index 0000000000..6d491d6992 --- /dev/null +++ b/packages/cron/src/mount.ts @@ -0,0 +1,80 @@ +// CRUD over a tenant's saved cron schedules. Mirrors `@corbits/mailbox`'s +// `mountMailbox` shape: absolute routes registered directly on the host's +// app (never a sub-router mounted under a prefix), a caller-supplied +// resolver for "is the caller allowed to act on this tenant," and arktype +// parsing at the one trust boundary that matters (the request body). +import { randomUUID } from "node:crypto"; +import { type } from "arktype"; +import { eq, and } from "drizzle-orm"; +import type { Env, Hono } from "hono"; + +import { isValidCronExpression } from "./cron"; +import { cronScheduleTable } from "./schema"; +import type { CronDb } from "./ticker"; + +export type RequireTenantMember = (ctx: unknown, tenantId: string) => Promise | boolean; + +export type MountCronOpts = { + db: CronDb; + requireTenantMember: RequireTenantMember; +}; + +const CreateScheduleBody = type({ + expression: "string", + toAddress: "string", + subject: "string", + body: "string", +}); + +/** Mount `/api/tenants/:tenantId/cron` CRUD onto the host's app. */ +export function mountCron(app: Hono, opts: MountCronOpts): Hono { + const { db, requireTenantMember } = opts; + + app.get("/api/tenants/:tenantId/cron", async (c) => { + const tenantId = c.req.param("tenantId"); + if (!(await requireTenantMember(c, tenantId))) return c.json({ error: "forbidden" }, 403); + const rows = await db + .select() + .from(cronScheduleTable) + .where(eq(cronScheduleTable.tenantId, tenantId)); + return c.json({ schedules: rows }); + }); + + app.post("/api/tenants/:tenantId/cron", async (c) => { + const tenantId = c.req.param("tenantId"); + if (!(await requireTenantMember(c, tenantId))) return c.json({ error: "forbidden" }, 403); + const parsed = CreateScheduleBody(await c.req.json().catch(() => undefined)); + if (parsed instanceof type.errors) { + return c.json({ error: "invalid_body", detail: parsed.summary }, 400); + } + if (!isValidCronExpression(parsed.expression)) { + return c.json({ error: "invalid_expression" }, 400); + } + const [row] = await db + .insert(cronScheduleTable) + .values({ + id: randomUUID(), + tenantId, + expression: parsed.expression, + toAddress: parsed.toAddress, + subject: parsed.subject, + body: parsed.body, + }) + .returning(); + return c.json({ schedule: row }, 201); + }); + + app.delete("/api/tenants/:tenantId/cron/:id", async (c) => { + const tenantId = c.req.param("tenantId"); + if (!(await requireTenantMember(c, tenantId))) return c.json({ error: "forbidden" }, 403); + const id = c.req.param("id"); + const [deleted] = await db + .delete(cronScheduleTable) + .where(and(eq(cronScheduleTable.tenantId, tenantId), eq(cronScheduleTable.id, id))) + .returning({ id: cronScheduleTable.id }); + if (deleted === undefined) return c.json({ error: "not_found" }, 404); + return c.json({ ok: true }); + }); + + return app; +} diff --git a/packages/cron/src/schema.ts b/packages/cron/src/schema.ts new file mode 100644 index 0000000000..f082f3e550 --- /dev/null +++ b/packages/cron/src/schema.ts @@ -0,0 +1,118 @@ +// The `cron` schema's one table: a tenant's saved cron schedules. Kept on +// its own Postgres schema, with a real FK back to Interchange's `tenant` +// table, per this repo's "custom tables live on their own schema" rule. +import { pgSchema, text, timestamp } from "drizzle-orm/pg-core"; +import postgres from "postgres"; + +export const cronSchema = pgSchema("cron"); + +export const cronScheduleTable = cronSchema.table("schedule", { + id: text("id").primaryKey(), + tenantId: text("tenant_id").notNull(), + expression: text("expression").notNull(), + toAddress: text("to_address").notNull(), + subject: text("subject").notNull(), + body: text("body").notNull(), + lastFiredAt: timestamp("last_fired_at", { withTimezone: true }), + createdAt: timestamp("created_at", { withTimezone: true }).notNull().defaultNow(), +}); + +const LEDGER_TABLE = "cron_migrations"; + +export interface CronMigration { + name: string; + sql: string; +} + +function quoteIdentifier(name: string): string { + return `"${name.replace(/"/g, '""')}"`; +} + +/** + * Migration list building the FK against `tenantSchema.tenant` — + * `"public"` in every real deployment; a test harness that runs the + * platform migrations into a scratch schema passes that schema instead so + * the FK targets its own `tenant` row, not the real one. + */ +function buildCronMigrations(tenantSchema: string): readonly CronMigration[] { + const tenantTable = `${quoteIdentifier(tenantSchema)}."tenant"`; + return [ + { + name: "0001_schedule", + sql: ` + CREATE SCHEMA IF NOT EXISTS "cron"; + CREATE TABLE IF NOT EXISTS "cron"."schedule" ( + "id" text PRIMARY KEY, + "tenant_id" text NOT NULL REFERENCES ${tenantTable}("id") ON DELETE CASCADE, + "expression" text NOT NULL, + "to_address" text NOT NULL, + "subject" text NOT NULL, + "body" text NOT NULL, + "last_fired_at" timestamptz, + "created_at" timestamptz NOT NULL DEFAULT now() + ); + CREATE INDEX IF NOT EXISTS "cron_schedule_tenant_id_idx" ON "cron"."schedule" ("tenant_id"); + `, + }, + ]; +} + +export const cronMigrations: readonly CronMigration[] = buildCronMigrations("public"); + +export interface ApplyCronMigrationsReport { + applied: string[]; + alreadyApplied: string[]; +} + +/** + * Apply `cronMigrations` against `databaseUrl`, idempotently: a migration + * already recorded in this package's own ledger is skipped, never re-run. + * Mirrors `@corbits/inbox`'s `applyInboxMigrations` shape. + */ +export async function applyCronMigrations( + databaseUrl: string, + options?: { tenantSchema?: string }, +): Promise { + const client = postgres(databaseUrl, { max: 1, onnotice: () => undefined }); + try { + await client.unsafe(`CREATE SCHEMA IF NOT EXISTS "cron"`); + await client.unsafe( + `CREATE TABLE IF NOT EXISTS "cron"."${LEDGER_TABLE}" (` + + `name text PRIMARY KEY, applied_at timestamptz NOT NULL DEFAULT now())`, + ); + + const applied: string[] = []; + const alreadyApplied: string[] = []; + const migrations = buildCronMigrations(options?.tenantSchema ?? "public"); + + for (const migration of migrations) { + const existing = await client.unsafe( + `SELECT 1 FROM "cron"."${LEDGER_TABLE}" WHERE name = $1`, + [migration.name], + ); + if (existing.length > 0) { + alreadyApplied.push(migration.name); + continue; + } + try { + await client.begin(async (tx) => { + await tx.unsafe(migration.sql); + await tx.unsafe(`INSERT INTO "cron"."${LEDGER_TABLE}" (name) VALUES ($1)`, [ + migration.name, + ]); + }); + applied.push(migration.name); + } catch (error) { + throw new Error( + `@corbits/cron migration ${JSON.stringify(migration.name)} failed: ` + + `${error instanceof Error ? error.message : String(error)}`, + { cause: error }, + ); + } + } + + return { applied, alreadyApplied }; + } finally { + await client.end({ timeout: 5 }); + } +} diff --git a/packages/cron/src/ticker.ts b/packages/cron/src/ticker.ts new file mode 100644 index 0000000000..25c7e6881f --- /dev/null +++ b/packages/cron/src/ticker.ts @@ -0,0 +1,120 @@ +// Turns due cron schedules into mail. Every tick opens one transaction, +// claims whatever rows are due with `SELECT ... FOR UPDATE SKIP LOCKED` (so +// two tickers racing the same table never double-fire a row), delivers +// each, and stamps `last_fired_at`. A schedule that missed several ticks +// (the process was down, the poll was slow) fires once for the most recent +// due minute, never once per missed tick — `last_fired_at` is only ever +// "did this fire since its last fire," not a queue to drain. +import { eq } from "drizzle-orm"; +import type { PostgresJsDatabase } from "drizzle-orm/postgres-js"; + +import { nextCronFireAfter } from "./cron"; +import { cronScheduleTable } from "./schema"; + +export type CronDb = Record> = + PostgresJsDatabase; + +export type DeliverCronMail = (message: { + to: string[]; + subject: string; + body: string; + from: string; +}) => Promise | void; + +export type CronSenderAddress = (tenantId: string) => string; + +/** The system sender address a tenant's cron mail comes from. */ +export const cronSenderAddress: CronSenderAddress = (tenantId) => `cron@${tenantId}.internal`; + +export type CreateCronTickerOpts< + TSchema extends Record = Record, +> = { + db: CronDb; + deliver: DeliverCronMail; + intervalMs: number; + senderAddressFor?: CronSenderAddress; +}; + +export type CronTicker = { + start(): void; + stop(): void; +}; + +/** + * A schedule's next fire time: `nextCronFireAfter` from its `last_fired_at` + * when it has fired before, or from `created_at` when it never has — a + * freshly saved schedule is due at its first matching minute after + * creation, not retroactively for every minute since the epoch. + */ +function isDue( + row: { expression: string; lastFiredAt: Date | null; createdAt: Date }, + now: Date, +): boolean { + const after = row.lastFiredAt ?? row.createdAt; + try { + return nextCronFireAfter(row.expression, after) <= now; + } catch { + // report-error-ignore: an expression with no fire in the lookahead + // window is simply never due, not an operational failure. + return false; + } +} + +async function tick>( + db: CronDb, + deliver: DeliverCronMail, + senderAddressFor: CronSenderAddress, +) { + await db.transaction(async (tx) => { + const now = new Date(); + // Every row, locked against a concurrent ticker (SKIP LOCKED means two + // tickers racing this table split the due rows rather than double-fire + // any of them); which ones are actually due is a JS-side check because + // "due" depends on parsing each row's own cron expression. + const candidates = await tx + .select() + .from(cronScheduleTable) + .for("update", { skipLocked: true }); + + for (const row of candidates.filter((row) => isDue(row, now))) { + await deliver({ + to: [row.toAddress], + subject: row.subject, + body: row.body, + from: senderAddressFor(row.tenantId), + }); + await tx + .update(cronScheduleTable) + .set({ lastFiredAt: now }) + .where(eq(cronScheduleTable.id, row.id)); + } + }); +} + +/** Ticks every `intervalMs`, delivering each due schedule as mail. */ +export function createCronTicker>( + opts: CreateCronTickerOpts, +): CronTicker { + const senderAddressFor = opts.senderAddressFor ?? cronSenderAddress; + let timer: ReturnType | undefined; + let inFlight: Promise | undefined; + + const runTick = () => { + if (inFlight !== undefined) return; + inFlight = tick(opts.db, opts.deliver, senderAddressFor).finally(() => { + inFlight = undefined; + }); + }; + + return { + start() { + if (timer !== undefined) return; + timer = setInterval(runTick, opts.intervalMs); + }, + stop() { + if (timer === undefined) return; + clearInterval(timer); + timer = undefined; + }, + }; +} diff --git a/packages/workflow-schedule/tsconfig.json b/packages/cron/tsconfig.json similarity index 91% rename from packages/workflow-schedule/tsconfig.json rename to packages/cron/tsconfig.json index e125e90614..a8c2b9966d 100644 --- a/packages/workflow-schedule/tsconfig.json +++ b/packages/cron/tsconfig.json @@ -10,6 +10,6 @@ "tsBuildInfoFile": "tsconfig.tsbuildinfo", "rootDir": "../.." }, - "include": ["src"], + "include": ["src", "test"], "exclude": [] } diff --git a/packages/workflow-schedule/tsconfig.src.json b/packages/cron/tsconfig.src.json similarity index 87% rename from packages/workflow-schedule/tsconfig.src.json rename to packages/cron/tsconfig.src.json index 50abef4ff0..20d913b3f6 100644 --- a/packages/workflow-schedule/tsconfig.src.json +++ b/packages/cron/tsconfig.src.json @@ -3,8 +3,8 @@ "include": ["src", "package.json", "src/**/*.json"], "exclude": ["src/**/*.test.ts", "src/**/*.test.tsx"], "compilerOptions": { - "lib": ["ESNext", "DOM", "DOM.Iterable"], "types": ["bun"], + "customConditions": ["bun"], "composite": true, "emitDeclarationOnly": true, "outDir": "dist", diff --git a/packages/workflow-schedule/package.json b/packages/workflow-schedule/package.json deleted file mode 100644 index e98f8f2d08..0000000000 --- a/packages/workflow-schedule/package.json +++ /dev/null @@ -1,21 +0,0 @@ -{ - "name": "@corbits/workflow-schedule", - "version": "0.0.1", - "private": true, - "description": "Cron grammar and the native TimerSet re-arm helper a schedule-triggered workflow uses to own its own cron: the next fire time for a ScheduleTrigger expression, and the cron-flagged TimerSet the @intx/workflow-host scheduler seam fires and the run re-arms on every tick", - "license": "LGPL-2.1-or-later", - "type": "module", - "exports": { - ".": "./src/index.ts", - "./timer": "./src/timer.ts", - "./emitter": "./src/emitter.ts" - }, - "scripts": { - "typecheck": "tsc --noEmit", - "test": "bun test" - }, - "devDependencies": { - "@types/bun": "catalog:", - "typescript": "catalog:" - } -} diff --git a/packages/workflow-schedule/src/cron.test.ts b/packages/workflow-schedule/src/cron.test.ts deleted file mode 100644 index 6840184235..0000000000 --- a/packages/workflow-schedule/src/cron.test.ts +++ /dev/null @@ -1,230 +0,0 @@ -// Table-driven cron correctness: step-from-min, DOM/DOW OR, Sunday=7, -// bounded lookahead, and timezone / DST round-trips. -import { describe, expect, test } from "bun:test"; - -import { - cronExpressionCanFire, - cronMatchesMinute, - isValidCronExpression, - isValidTimeZone, - nextCronFireAfter, - zonedParts, -} from "./cron"; - -describe("isValidCronExpression", () => { - test("accepts standard 5-field expressions", () => { - expect(isValidCronExpression("*/5 * * * *")).toBe(true); - expect(isValidCronExpression("0 9 * * 1")).toBe(true); - expect(isValidCronExpression("15,45 * * * *")).toBe(true); - expect(isValidCronExpression("0 0 13 * 5")).toBe(true); - }); - - test("accepts 7 as Sunday on day-of-week", () => { - expect(isValidCronExpression("* * * * 7")).toBe(true); - expect(isValidCronExpression("0 0 * * 0,7")).toBe(true); - }); - - test("rejects wrong field counts and garbage", () => { - expect(isValidCronExpression("* * * *")).toBe(false); - expect(isValidCronExpression("* * * * * *")).toBe(false); - expect(isValidCronExpression("not a cron string at all")).toBe(false); - expect(isValidCronExpression("")).toBe(false); - }); - - test("rejects every field out of range", () => { - expect(isValidCronExpression("99 99 99 99 99")).toBe(false); - expect(isValidCronExpression("0 0 32 * *")).toBe(false); - expect(isValidCronExpression("0 0 * 13 *")).toBe(false); - expect(isValidCronExpression("60 * * * *")).toBe(false); - expect(isValidCronExpression("* * * * 8")).toBe(false); - }); - - test("rejects a reversed range", () => { - expect(isValidCronExpression("10-5 * * * *")).toBe(false); - expect(isValidCronExpression("10-5/2 * * * *")).toBe(false); - }); - - test("accepts the standard range-then-step idiom", () => { - expect(isValidCronExpression("5-10/2 * * * *")).toBe(true); - }); -}); - -describe("step fields offset from field minimum", () => { - // `*/2` on day-of-month (min=1) → 1,3,5… not 2,4,6… - const cases: { - name: string; - expression: string; - at: string; - matches: boolean; - }[] = [ - { - name: "DOM */2 matches day 1", - expression: "0 0 */2 * *", - at: "2026-01-01T00:00:00Z", - matches: true, - }, - { - name: "DOM */2 does not match day 2", - expression: "0 0 */2 * *", - at: "2026-01-02T00:00:00Z", - matches: false, - }, - { - name: "DOM */2 matches day 3", - expression: "0 0 */2 * *", - at: "2026-01-03T00:00:00Z", - matches: true, - }, - { - name: "minute */2 still matches 0 (min=0)", - expression: "*/2 * * * *", - at: "2026-01-01T00:00:00Z", - matches: true, - }, - { - name: "minute */2 matches 2", - expression: "*/2 * * * *", - at: "2026-01-01T00:02:00Z", - matches: true, - }, - { - name: "minute */2 does not match 1", - expression: "*/2 * * * *", - at: "2026-01-01T00:01:00Z", - matches: false, - }, - { - name: "month */2 matches January (min=1)", - expression: "0 0 1 */2 *", - at: "2026-01-01T00:00:00Z", - matches: true, - }, - { - name: "month */2 does not match February", - expression: "0 0 1 */2 *", - at: "2026-02-01T00:00:00Z", - matches: false, - }, - ]; - - for (const c of cases) { - test(c.name, () => { - expect(cronMatchesMinute(c.expression, new Date(c.at))).toBe(c.matches); - }); - } -}); - -describe("DOM / DOW OR when both restricted (POSIX/Vixie)", () => { - // `0 0 13 * 5` = midnight on the 13th OR any Friday. - const expression = "0 0 13 * 5"; - - test("matches the 13th even when it is not Friday", () => { - // 2026-01-13 is a Tuesday. - expect(cronMatchesMinute(expression, new Date("2026-01-13T00:00:00Z"))).toBe(true); - }); - - test("matches a Friday that is not the 13th", () => { - // 2026-01-16 is a Friday. - expect(cronMatchesMinute(expression, new Date("2026-01-16T00:00:00Z"))).toBe(true); - }); - - test("does not match a day that is neither the 13th nor Friday", () => { - // 2026-01-14 is a Wednesday. - expect(cronMatchesMinute(expression, new Date("2026-01-14T00:00:00Z"))).toBe(false); - }); - - test("when only DOM is restricted, DOW is ignored (AND with *)", () => { - // Only the 15th, any weekday. - expect(cronMatchesMinute("0 0 15 * *", new Date("2026-01-15T00:00:00Z"))).toBe(true); - expect(cronMatchesMinute("0 0 15 * *", new Date("2026-01-16T00:00:00Z"))).toBe(false); - }); - - test("when only DOW is restricted, DOM is ignored", () => { - // Every Monday at midnight. - expect(cronMatchesMinute("0 0 * * 1", new Date("2026-01-12T00:00:00Z"))).toBe(true); - expect(cronMatchesMinute("0 0 * * 1", new Date("2026-01-13T00:00:00Z"))).toBe(false); - }); -}); - -describe("day-of-week 0 and 7 are both Sunday", () => { - test("expression with 7 matches a Sunday Date (day 0)", () => { - // 2026-01-04 is a Sunday. - expect(cronMatchesMinute("0 0 * * 7", new Date("2026-01-04T00:00:00Z"))).toBe(true); - }); - - test("expression with 0 matches the same Sunday", () => { - expect(cronMatchesMinute("0 0 * * 0", new Date("2026-01-04T00:00:00Z"))).toBe(true); - }); - - test("neither matches a Monday", () => { - expect(cronMatchesMinute("0 0 * * 7", new Date("2026-01-05T00:00:00Z"))).toBe(false); - }); -}); - -describe("nextCronFireAfter + canFire bounds", () => { - test("finds the next matching minute", () => { - const next = nextCronFireAfter("0-5 * * * *", new Date("2026-01-01T00:00:00Z")); - expect(next.toISOString()).toBe("2026-01-01T00:01:00.000Z"); - }); - - test("impossible Feb 31 fails canFire and nextCronFireAfter", () => { - expect(cronExpressionCanFire("0 0 31 2 *")).toBe(false); - expect(() => nextCronFireAfter("0 0 31 2 *", new Date("2026-01-01T00:00:00Z"))).toThrow( - /no fire time within the lookahead window/, - ); - }); - - test("a once-a-year expression still finds its fire", () => { - // Jan 1 at 00:00 — after Dec 31, next is next year. - const next = nextCronFireAfter("0 0 1 1 *", new Date("2026-06-01T00:00:00Z")); - expect(next.toISOString()).toBe("2027-01-01T00:00:00.000Z"); - }); -}); - -describe("timezone matching and DST", () => { - test("isValidTimeZone accepts IANA names and rejects garbage", () => { - expect(isValidTimeZone("UTC")).toBe(true); - expect(isValidTimeZone("America/Los_Angeles")).toBe(true); - expect(isValidTimeZone("Europe/London")).toBe(true); - expect(isValidTimeZone("Not/A_Zone")).toBe(false); - expect(isValidTimeZone("")).toBe(false); - }); - - test("daily 09:00 America/Los_Angeles matches the correct UTC instant (PST)", () => { - // 2026-01-15 is winter — PST = UTC-8, so 09:00 local = 17:00 UTC. - const at = new Date("2026-01-15T17:00:00Z"); - expect(zonedParts(at, "America/Los_Angeles")).toMatchObject({ - hour: 9, - minute: 0, - day: 15, - month: 1, - }); - expect(cronMatchesMinute("0 9 * * *", at, "America/Los_Angeles")).toBe(true); - expect(cronMatchesMinute("0 9 * * *", at, "UTC")).toBe(false); - }); - - test("next fire for 09:00 America/Los_Angeles across a DST spring-forward", () => { - // US Pacific spring forward 2026: 2026-03-08 02:00 → 03:00 local. - // Before the transition (March 7 12:00 UTC = March 7 04:00 PST): - // next 09:00 local is March 7 09:00 PST = March 7 17:00 UTC. - const before = new Date("2026-03-07T12:00:00Z"); - const nextBefore = nextCronFireAfter("0 9 * * *", before, "America/Los_Angeles"); - expect(nextBefore.toISOString()).toBe("2026-03-07T17:00:00.000Z"); - expect(zonedParts(nextBefore, "America/Los_Angeles").hour).toBe(9); - - // After spring-forward, 09:00 PDT = UTC-7 → 16:00 UTC. - const afterTransition = new Date("2026-03-09T12:00:00Z"); - const nextAfter = nextCronFireAfter("0 9 * * *", afterTransition, "America/Los_Angeles"); - expect(nextAfter.toISOString()).toBe("2026-03-09T16:00:00.000Z"); - expect(zonedParts(nextAfter, "America/Los_Angeles").hour).toBe(9); - }); - - test("next fire for 09:00 America/Los_Angeles across a DST fall-back", () => { - // US Pacific fall back 2026: 2026-11-01 02:00 → 01:00 local. - // After the transition, 09:00 PST = UTC-8 → 17:00 UTC. - const afterFallback = new Date("2026-11-02T12:00:00Z"); - const next = nextCronFireAfter("0 9 * * *", afterFallback, "America/Los_Angeles"); - expect(next.toISOString()).toBe("2026-11-02T17:00:00.000Z"); - expect(zonedParts(next, "America/Los_Angeles").hour).toBe(9); - }); -}); diff --git a/packages/workflow-schedule/src/emitter.test.ts b/packages/workflow-schedule/src/emitter.test.ts deleted file mode 100644 index afa6ef5cd6..0000000000 --- a/packages/workflow-schedule/src/emitter.test.ts +++ /dev/null @@ -1,92 +0,0 @@ -// The emitter's own logic worth proving: a tick re-arms after it fires, -// at the next instant after the tick that ran — never at the one it just -// ran, and never at a tick that fell during a slow fire. -import { describe, expect, test } from "bun:test"; - -import { createCronEmitter, timerIdForTick, type CronDeployment } from "./emitter"; - -const deployment: CronDeployment = { - deploymentId: "def_digest", - cron: "* * * * *", -}; - -describe("createCronEmitter re-arm", () => { - test("re-arms at the tick after the one that fired, skipping ticks a slow fire ran through", async () => { - // 20ms before the tick, so the armed `setTimeout` really elapses. - let now = new Date("2026-01-01T09:00:59.980Z"); - const fired: string[] = []; - const emitter = createCronEmitter({ - listCronDeployments: () => Promise.resolve([deployment]), - fire: (_deployment, tick) => { - fired.push(tick.fireAt); - // The fire takes three minutes of wall clock. - now = new Date("2026-01-01T09:04:10.000Z"); - return Promise.resolve(); - }, - onError: (error) => { - throw error; - }, - clock: () => now, - rescanIntervalMs: 60_000, - }); - - await emitter.rescan(); - expect(emitter.armed()).toEqual([ - { - deploymentId: "def_digest", - tick: { - timerId: timerIdForTick("def_digest", new Date("2026-01-01T09:01:00.000Z")), - fireAt: "2026-01-01T09:01:00.000Z", - cron: "* * * * *", - }, - }, - ]); - - await Bun.sleep(40); - expect(fired).toEqual(["2026-01-01T09:01:00.000Z"]); - // Re-armed strictly after the slow fire's end, not at 09:02 (a tick - // that already passed — the scheduler seam drops a past-due cron - // TimerSet, so arming one would stall the schedule for good). - expect(emitter.armed()[0]?.tick.fireAt).toBe("2026-01-01T09:05:00.000Z"); - emitter.stop(); - }); - - test("a second rescan does not re-arm an already-armed tick", async () => { - const now = new Date("2026-01-01T09:00:30.000Z"); - const emitter = createCronEmitter({ - listCronDeployments: () => Promise.resolve([deployment]), - fire: () => Promise.resolve(), - onError: (error) => { - throw error; - }, - clock: () => now, - rescanIntervalMs: 60_000, - }); - await emitter.rescan(); - const first = emitter.armed()[0]?.tick.timerId; - await emitter.rescan(); - expect(emitter.armed()).toHaveLength(1); - expect(emitter.armed()[0]?.tick.timerId).toBe(first); - emitter.stop(); - }); - - test("a deployment that left the list is disarmed", async () => { - const now = new Date("2026-01-01T09:00:30.000Z"); - let listed: readonly CronDeployment[] = [deployment]; - const emitter = createCronEmitter({ - listCronDeployments: () => Promise.resolve(listed), - fire: () => Promise.resolve(), - onError: (error) => { - throw error; - }, - clock: () => now, - rescanIntervalMs: 60_000, - }); - await emitter.rescan(); - expect(emitter.armed()).toHaveLength(1); - listed = []; - await emitter.rescan(); - expect(emitter.armed()).toHaveLength(0); - emitter.stop(); - }); -}); diff --git a/packages/workflow-schedule/src/emitter.ts b/packages/workflow-schedule/src/emitter.ts deleted file mode 100644 index 56bb4db21d..0000000000 --- a/packages/workflow-schedule/src/emitter.ts +++ /dev/null @@ -1,182 +0,0 @@ -// The cron emitter: the half of a schedule-triggered deployment that owns -// the clock. `armCronTimer` says which instant the next tick falls on; this -// module holds that tick, fires it, and re-arms from the instant it fired. -// -// Why the fire is a trigger and not a `TimerFired` read off the scheduler -// seam: the seam (`vendor/intx/workflow-host/src/seams/scheduler.ts`) is -// constructed per workflow-run child, scoped to that child's own run repo -// (`apps/sidecar/src/workflow-substrate-factory/index.ts` builds it with -// `listActiveDeployments: () => [workflowRunRepoId]`), and the supervisor -// is the sole writer of that ref. A `TimerFired` there unblocks a -// `waitForTimer` inside a run that is already executing; it cannot start a -// fresh one, and a definition cannot re-arm its own cron (`sleep` is -// literal-only and banned in loops). So a tick that must launch a run fires -// through the deployment's own native trigger, which the caller supplies as -// `fire`. -// -// A tick's identity is derived, never minted: `timerIdForTick` is a pure -// function of the deployment id and the tick's instant, so two emitters -// arming the same tick name it the same thing and a sink that dedupes on -// timer identity collapses them. -import { armCronTimer, type CronTimerSet } from "./timer"; - -export type CronDeployment = { - readonly deploymentId: string; - /** The definition's `ScheduleTrigger.cron`, 5-field. */ - readonly cron: string; - /** IANA zone the expression is read in; the instant is always UTC. */ - readonly timeZone?: string; -}; - -export type CronEmitterDeps = { - /** - * Every deployment whose definition carries a `ScheduleTrigger`, as of - * now. Re-read on each rescan, so a deploy, an undeploy, or a cron edit - * takes effect without a restart. - */ - readonly listCronDeployments: () => Promise; - /** - * Fire one tick. Called once per armed tick; a throw is reported by the - * caller's `onError` and the schedule re-arms regardless, so a single bad - * tick never stalls the cadence. - */ - readonly fire: (deployment: D, tick: CronTimerSet) => Promise; - readonly onError: (error: unknown, deployment: D | CronDeployment) => void; - readonly clock: () => Date; - /** How often the deployment list is re-read. */ - readonly rescanIntervalMs?: number; -}; - -export type CronEmitterHandle = { - /** Read the deployment list and arm every deployment's next tick. */ - rescan(): Promise; - /** Currently armed ticks, one per deployment. Tests assert on this. */ - armed(): readonly { deploymentId: string; tick: CronTimerSet }[]; - /** The deployment armed for `deploymentId`, if any. */ - armedDeployment(deploymentId: string): D | undefined; - stop(): void; -}; - -export const DEFAULT_CRON_RESCAN_INTERVAL_MS = 60_000; - -/** - * The identity of the tick `deploymentId` fires at `fireAt`. Deterministic: - * the same deployment and the same instant always name the same tick. - */ -export function timerIdForTick(deploymentId: string, fireAt: Date): string { - return `cron:${deploymentId}:${fireAt.toISOString()}`; -} - -type ArmedTick = { - deployment: D; - tick: CronTimerSet; - timeout: ReturnType; -}; - -export function createCronEmitter( - deps: CronEmitterDeps, -): CronEmitterHandle { - const armedTicks = new Map>(); - let stopped = false; - - function arm(deployment: D, firedAt?: Date): void { - if (stopped) return; - const now = deps.clock(); - let tick: CronTimerSet; - try { - tick = armCronTimer({ - cron: deployment.cron, - now, - ...(firedAt !== undefined ? { firedAt } : {}), - ...(deployment.timeZone !== undefined ? { timeZone: deployment.timeZone } : {}), - newTimerId: (fireAt) => timerIdForTick(deployment.deploymentId, fireAt), - }); - } catch (error) { - // report-error-ignore: this package takes no dependency on - // @corbits/error-sink so it stays publishable and host-agnostic; - // `onError` is the reporting seam and the hub wires `reportError` - // straight into it. - deps.onError(error, deployment); - return; - } - const existing = armedTicks.get(deployment.deploymentId); - if (existing !== undefined) { - if (existing.tick.timerId === tick.timerId) return; - clearTimeout(existing.timeout); - } - const armedTick: ArmedTick = { - deployment, - tick, - timeout: setTimeout( - () => { - void fireTick(deployment.deploymentId); - }, - Math.max(0, Date.parse(tick.fireAt) - now.getTime()), - ), - }; - if (typeof armedTick.timeout.unref === "function") { - armedTick.timeout.unref(); - } - armedTicks.set(deployment.deploymentId, armedTick); - } - - async function fireTick(deploymentId: string): Promise { - const entry = armedTicks.get(deploymentId); - if (entry === undefined || stopped) return; - armedTicks.delete(deploymentId); - const firedAt = new Date(entry.tick.fireAt); - try { - await deps.fire(entry.deployment, entry.tick); - } catch (error) { - // report-error-ignore: reported through the caller's `onError`, which - // the hub wires to `reportError` — see the sibling catch in `arm`. - deps.onError(error, entry.deployment); - } - // Re-arm from the tick's own instant, not from `now`: a fire observed - // late must still land on the next tick after the one it just ran, - // and `armCronTimer` skips whatever fell in between rather than - // replaying it. - arm(entry.deployment, firedAt); - } - - async function rescan(): Promise { - const deployments = await deps.listCronDeployments(); - const live = new Set(); - for (const deployment of deployments) { - live.add(deployment.deploymentId); - arm(deployment); - } - for (const [deploymentId, entry] of armedTicks) { - if (live.has(deploymentId)) continue; - clearTimeout(entry.timeout); - armedTicks.delete(deploymentId); - } - } - - const rescanInterval = setInterval(() => { - void rescan().catch((error: unknown) => { - deps.onError(error, { deploymentId: "*", cron: "*" }); - }); - }, deps.rescanIntervalMs ?? DEFAULT_CRON_RESCAN_INTERVAL_MS); - if (typeof rescanInterval.unref === "function") rescanInterval.unref(); - - return { - rescan, - armedDeployment(deploymentId) { - return armedTicks.get(deploymentId)?.deployment; - }, - armed() { - return [...armedTicks.values()].map((entry) => ({ - deploymentId: entry.deployment.deploymentId, - tick: entry.tick, - })); - }, - stop() { - if (stopped) return; - stopped = true; - clearInterval(rescanInterval); - for (const entry of armedTicks.values()) clearTimeout(entry.timeout); - armedTicks.clear(); - }, - }; -} diff --git a/packages/workflow-schedule/src/index.ts b/packages/workflow-schedule/src/index.ts deleted file mode 100644 index 09a3fc7abf..0000000000 --- a/packages/workflow-schedule/src/index.ts +++ /dev/null @@ -1,13 +0,0 @@ -// The cron grammar a `ScheduleTrigger` speaks, shared by save-time -// validation, UI next-run display, and the timer re-arm helper. Browser -// safe: this entry point pulls in no server-only dependency. -export { - CRON_FIELD_RANGES, - MAX_LOOKAHEAD_MINUTES, - cronExpressionCanFire, - isValidCronExpression, - isValidTimeZone, - nextCronFireAfter, - zonedParts, -} from "./cron"; -export type { CronField, ZonedParts } from "./cron"; diff --git a/packages/workflow-schedule/src/timer.test.ts b/packages/workflow-schedule/src/timer.test.ts deleted file mode 100644 index f077aec311..0000000000 --- a/packages/workflow-schedule/src/timer.test.ts +++ /dev/null @@ -1,40 +0,0 @@ -import { describe, expect, test } from "bun:test"; - -import { armCronTimer } from "./timer"; - -const newTimerId = () => "timer_1"; - -describe("armCronTimer", () => { - test("arms the next tick strictly after now", () => { - const armed = armCronTimer({ - cron: "0 * * * *", - now: new Date("2026-01-01T09:00:00.000Z"), - newTimerId, - }); - expect(armed).toEqual({ - timerId: "timer_1", - fireAt: "2026-01-01T10:00:00.000Z", - cron: "0 * * * *", - }); - }); - - test("a tick observed late skips the missed ticks rather than arming in the past", () => { - const armed = armCronTimer({ - cron: "0 * * * *", - firedAt: new Date("2026-01-01T09:00:00.000Z"), - now: new Date("2026-01-01T12:30:00.000Z"), - newTimerId, - }); - expect(armed.fireAt).toBe("2026-01-01T13:00:00.000Z"); - }); - - test("reads the expression in the given zone", () => { - const armed = armCronTimer({ - cron: "30 9 * * *", - now: new Date("2026-01-01T09:00:00.000Z"), - timeZone: "America/New_York", - newTimerId, - }); - expect(armed.fireAt).toBe("2026-01-01T14:30:00.000Z"); - }); -}); diff --git a/packages/workflow-schedule/src/timer.ts b/packages/workflow-schedule/src/timer.ts deleted file mode 100644 index 47cd87a160..0000000000 --- a/packages/workflow-schedule/src/timer.ts +++ /dev/null @@ -1,56 +0,0 @@ -// The cron half of `@intx/workflow-host`'s scheduler seam. That seam is -// the single writer of `TimerFired`; a `TimerSet` carrying a `cron` -// discriminator is never replayed when its `fireAt` has passed, because -// "the next cron tick is committed by whoever owns the cron" — the run. -// This module is that owner's arithmetic: which instant the next tick -// falls on, and what the re-armed `TimerSet` looks like. -import { nextCronFireAfter } from "./cron"; - -/** - * The cron-flagged `TimerSet` payload the scheduler seam ingests. The - * on-disk workflow-event envelope is flat (`{ seq, type, ...fields }`); - * `seq` is minted by the committing writer, so it is not part of this - * shape. - */ -export type CronTimerSet = { - readonly timerId: string; - readonly fireAt: string; - readonly cron: string; -}; - -export type CronTimerArmOpts = { - /** The definition's `ScheduleTrigger.cron`, 5-field. */ - readonly cron: string; - /** Wall clock the run is arming at. */ - readonly now: Date; - /** - * When the tick this arm re-arms after fired. A `TimerFired` observed - * late — the run was suspended, or the host restarted — must not - * re-arm a tick at or before `now`: the seam drops a past-due cron - * `TimerSet` as a missed tick and the schedule would stall forever. - */ - readonly firedAt?: Date; - /** IANA zone the expression is read in; the instant is always UTC. */ - readonly timeZone?: string; - /** Names the tick. Given the instant it falls on, so an identity can be - * derived from it rather than minted. */ - readonly newTimerId: (fireAt: Date) => string; -}; - -/** - * The next cron tick as a `TimerSet`, armed strictly in the future. - * Missed ticks between `firedAt` and `now` are skipped rather than - * replayed, matching the seam's recovery semantics. - */ -export function armCronTimer(opts: CronTimerArmOpts): CronTimerSet { - const from = - opts.firedAt !== undefined && opts.firedAt.getTime() > opts.now.getTime() - ? opts.firedAt - : opts.now; - const fireAt = nextCronFireAfter(opts.cron, from, opts.timeZone ?? "UTC"); - return { - timerId: opts.newTimerId(fireAt), - fireAt: fireAt.toISOString(), - cron: opts.cron, - }; -} diff --git a/packages/workflows/package.json b/packages/workflows/package.json index 3c19a94217..02dd8df731 100644 --- a/packages/workflows/package.json +++ b/packages/workflows/package.json @@ -16,7 +16,6 @@ }, "dependencies": { "@corbits/error-sink": "workspace:*", - "@corbits/workflow-schedule": "workspace:*", "@intx/authz": "0.3.0", "@intx/db": "workspace:*", "@intx/hub-api": "workspace:*", diff --git a/packages/workflows/src/index.ts b/packages/workflows/src/index.ts index 23bbf4dd2a..b7701590e3 100644 --- a/packages/workflows/src/index.ts +++ b/packages/workflows/src/index.ts @@ -28,18 +28,6 @@ export { type WorkflowCatalogEntry, type TriggerFieldsValidation, } from "./catalog"; -export { - CRON_FIELD_RANGES, - cronExpressionCanFire, - isValidCronExpression, - isValidTimeZone, - MAX_LOOKAHEAD_MINUTES, - nextCronFireAfter, - zonedParts, - type CronField, - type ZonedParts, -} from "@corbits/workflow-schedule"; -export { scheduleCronFromProjection } from "./schedule/from-projection"; export { ensureRunSession, recordAgentSessionAtProvision, diff --git a/packages/workflows/src/schedule/from-projection.test.ts b/packages/workflows/src/schedule/from-projection.test.ts deleted file mode 100644 index b923e91139..0000000000 --- a/packages/workflows/src/schedule/from-projection.test.ts +++ /dev/null @@ -1,46 +0,0 @@ -import { describe, expect, test } from "bun:test"; - -import { scheduleCronFromProjection } from "./from-projection"; - -const baseProjection = { - id: "wf_digest", - stepOrder: ["digest"], - steps: { digest: { kind: "step" } }, -}; - -describe("scheduleCronFromProjection", () => { - test("returns the first schedule cron", () => { - expect( - scheduleCronFromProjection({ - ...baseProjection, - triggers: [ - { type: "mail", to: "digest@example.com" }, - { type: "schedule", cron: "0 9 * * *" }, - { type: "schedule", cron: "0 18 * * *" }, - ], - }), - ).toBe("0 9 * * *"); - }); - - test("returns undefined when there is no schedule trigger", () => { - expect( - scheduleCronFromProjection({ - ...baseProjection, - triggers: [{ type: "mail", to: "digest@example.com" }], - }), - ).toBeUndefined(); - }); - - test("returns undefined for an unparseable projection", () => { - expect(scheduleCronFromProjection({ triggers: [] })).toBeUndefined(); - }); - - test("returns undefined when the first schedule cron is invalid", () => { - expect( - scheduleCronFromProjection({ - ...baseProjection, - triggers: [{ type: "schedule", cron: "99 99 99 99 99" }], - }), - ).toBeUndefined(); - }); -}); diff --git a/packages/workflows/src/schedule/from-projection.ts b/packages/workflows/src/schedule/from-projection.ts deleted file mode 100644 index 18716317a1..0000000000 --- a/packages/workflows/src/schedule/from-projection.ts +++ /dev/null @@ -1,25 +0,0 @@ -// First `{ type: "schedule", cron }` on a frozen inert projection — the -// hashed cadence the hub poller ticks. Later schedule triggers are ignored -// on purpose: one definition, one cadence. An unparseable projection or an -// invalid cron is not a schedule (fail closed). -import { type } from "arktype"; -import { WorkflowProjectionDefinition } from "@intx/types/sidecar"; - -import { isValidCronExpression } from "@corbits/workflow-schedule"; - -const ScheduleTrigger = type({ - type: "'schedule'", - cron: "string", -}); - -export function scheduleCronFromProjection(wireProjection: unknown): string | undefined { - const projection = WorkflowProjectionDefinition(wireProjection); - if (projection instanceof type.errors) return undefined; - for (const trigger of projection.triggers) { - const parsed = ScheduleTrigger(trigger); - if (parsed instanceof type.errors) continue; - if (!isValidCronExpression(parsed.cron)) return undefined; - return parsed.cron; - } - return undefined; -} diff --git a/packages/workflows/tsconfig.json b/packages/workflows/tsconfig.json index 54fd3da3be..0896db0dac 100644 --- a/packages/workflows/tsconfig.json +++ b/packages/workflows/tsconfig.json @@ -30,9 +30,6 @@ }, { "path": "../error-sink/tsconfig.src.json" - }, - { - "path": "../workflow-schedule/tsconfig.src.json" } ] } diff --git a/packages/workflows/tsconfig.src.json b/packages/workflows/tsconfig.src.json index 93fe7de3e0..8b23588e09 100644 --- a/packages/workflows/tsconfig.src.json +++ b/packages/workflows/tsconfig.src.json @@ -28,9 +28,6 @@ }, { "path": "../error-sink/tsconfig.src.json" - }, - { - "path": "../workflow-schedule/tsconfig.src.json" } ] } diff --git a/scripts/checks/no-product-tenancy.ts b/scripts/checks/no-product-tenancy.ts index 1f55967206..dd39f2b268 100644 --- a/scripts/checks/no-product-tenancy.ts +++ b/scripts/checks/no-product-tenancy.ts @@ -137,6 +137,15 @@ const ALLOWLIST: readonly { maxOccurrences: 1, tables: ["evals.run"], }, + { + // CL-8183: the vendored Interchange workflow-trigger grammar has no + // `schedule` trigger, so `@corbits/cron` fires a tenant's saved cron + // expressions as plain mail instead. Product-owned schedule state, + // never tenancy — FK'd back to the native `tenant` table. + relPath: "packages/cron/src/schema.ts", + maxOccurrences: 1, + tables: ["cron.schedule"], + }, ]; export async function scanFiles(root: string, dirs: readonly string[]): Promise { diff --git a/scripts/db-setup.ts b/scripts/db-setup.ts index 8869be707c..797919a3af 100644 --- a/scripts/db-setup.ts +++ b/scripts/db-setup.ts @@ -34,6 +34,7 @@ import { applyNotifyMigrations } from "../packages/notify/src/migrations"; import { applyInboxMigrations, applyMailboxMigrations } from "../packages/inbox/src/migrations"; import { applyAgentDirectoryMigrations } from "../packages/agent-directory/src/migrations"; import { applyOnboardingMigrations } from "../packages/onboarding/src/migrations"; +import { applyCronMigrations } from "../packages/cron/src/migrations"; const repoRoot = path.resolve(import.meta.dir, ".."); const HUB_DIR = path.join(repoRoot, "apps", "hub"); @@ -59,6 +60,7 @@ const INSTALLED_PACKAGE_MIGRATIONS: readonly { { name: "@corbits/inbox", apply: applyInboxMigrations }, { name: "@corbits/agent-directory", apply: applyAgentDirectoryMigrations }, { name: "@workbench/onboarding", apply: applyOnboardingMigrations }, + { name: "@corbits/cron", apply: applyCronMigrations }, ]; /** diff --git a/tsconfig.build.json b/tsconfig.build.json index 3a6eb1e90a..72a171f0fb 100644 --- a/tsconfig.build.json +++ b/tsconfig.build.json @@ -52,6 +52,9 @@ { "path": "./packages/credential-providers/tsconfig.src.json" }, + { + "path": "./packages/cron/tsconfig.src.json" + }, { "path": "./packages/docker-provisioner/tsconfig.src.json" }, @@ -112,9 +115,6 @@ { "path": "./packages/url-path/tsconfig.src.json" }, - { - "path": "./packages/workflow-schedule/tsconfig.src.json" - }, { "path": "./packages/workflows/tsconfig.src.json" }, From 26f773a10c9b5c90edfff2fb7b3754c6c2aac062 Mon Sep 17 00:00:00 2001 From: Sawyer Date: Thu, 17 Sep 2026 14:59:33 -0700 Subject: [PATCH 3/4] cron: reference Interchange tenant id with a real foreign key (CL-8210) --- packages/cron/src/schema.ts | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/packages/cron/src/schema.ts b/packages/cron/src/schema.ts index f082f3e550..d434b11516 100644 --- a/packages/cron/src/schema.ts +++ b/packages/cron/src/schema.ts @@ -1,14 +1,18 @@ // The `cron` schema's one table: a tenant's saved cron schedules. Kept on // its own Postgres schema, with a real FK back to Interchange's `tenant` // table, per this repo's "custom tables live on their own schema" rule. -import { pgSchema, text, timestamp } from "drizzle-orm/pg-core"; +import { pgTable, pgSchema, text, timestamp } from "drizzle-orm/pg-core"; import postgres from "postgres"; +const hostTenant = pgTable("tenant", { id: text("id").primaryKey() }); + export const cronSchema = pgSchema("cron"); export const cronScheduleTable = cronSchema.table("schedule", { id: text("id").primaryKey(), - tenantId: text("tenant_id").notNull(), + tenantId: text("tenant_id") + .notNull() + .references(() => hostTenant.id, { onDelete: "cascade" }), expression: text("expression").notNull(), toAddress: text("to_address").notNull(), subject: text("subject").notNull(), From da20f612650080e91a5b6da69352607ed28a5d6f Mon Sep 17 00:00:00 2001 From: Sawyer Date: Thu, 17 Sep 2026 14:59:55 -0700 Subject: [PATCH 4/4] cron: allowlist the tenant FK stub --- scripts/checks/no-product-tenancy.ts | 6 ++++-- 1 file changed, 4 insertions(+), 2 deletions(-) diff --git a/scripts/checks/no-product-tenancy.ts b/scripts/checks/no-product-tenancy.ts index dd39f2b268..367fb6192e 100644 --- a/scripts/checks/no-product-tenancy.ts +++ b/scripts/checks/no-product-tenancy.ts @@ -143,8 +143,10 @@ const ALLOWLIST: readonly { // expressions as plain mail instead. Product-owned schedule state, // never tenancy — FK'd back to the native `tenant` table. relPath: "packages/cron/src/schema.ts", - maxOccurrences: 1, - tables: ["cron.schedule"], + maxOccurrences: 2, + // CL-8210: Interchange's `tenant` re-declared (id only, never migrated) + // to carry the FK from cron.schedule. + tables: ["tenant", "cron.schedule"], }, ];