Skip to content

Commit e960bf8

Browse files
Add @corbits/cron: cron ticks delivered as mail (CL-8183) (#800)
* 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. * Add @corbits/cron: cron ticks delivered as mail (CL-8183) 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. * cron: reference Interchange tenant id with a real foreign key (CL-8210) * cron: allowlist the tenant FK stub
1 parent e0b1001 commit e960bf8

30 files changed

Lines changed: 601 additions & 512 deletions

‎apps/hub/package.json‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -26,6 +26,7 @@
2626
"@corbits/commands": "workspace:*",
2727
"@corbits/connections": "workspace:*",
2828
"@corbits/credential-providers": "workspace:*",
29+
"@corbits/cron": "workspace:*",
2930
"@corbits/diligence-brief-workflow": "workspace:*",
3031
"@corbits/docker-provisioner": "workspace:*",
3132
"@corbits/e2b-sandbox-sidecar": "workspace:*",
@@ -52,7 +53,6 @@
5253
"@corbits/url-path": "workspace:*",
5354
"@corbits/webhook-triggers": "workspace:*",
5455
"@corbits/workbench-digest-workflow": "workspace:*",
55-
"@corbits/workflow-schedule": "workspace:*",
5656
"@corbits/workflows": "workspace:*",
5757
"@corbits/xai-provider": "workspace:*",
5858
"@intx/authz": "0.3.0",

‎bun.lock‎

Lines changed: 24 additions & 12 deletions
Some generated files are not rendered by default. Learn more about customizing how changed files appear on GitHub.

‎packages/cron/README.md‎

Lines changed: 45 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,45 @@
1+
# @corbits/cron
2+
3+
Experimental. The vendored Interchange workflow-trigger grammar
4+
(`vendor/intx/workflow/src/definition/triggers.ts`) has no `schedule`
5+
trigger and no scheduler — a workflow can only be launched by `mail`. This
6+
package is the tiny bridge: a tenant saves a cron expression plus a mail
7+
(`to`/`subject`/`body`); `createCronTicker` polls for due schedules and
8+
delivers each as mail through the host's own transport, so a
9+
schedule-triggered workflow is just a `mail`-triggered one addressed at
10+
itself.
11+
12+
- `src/cron.ts` — the 5-field cron grammar (moved from the retired
13+
`@corbits/workflow-schedule`, unchanged).
14+
- `src/schema.ts` — the `cron.schedule` table (its own Postgres schema, FK'd
15+
to Interchange's `tenant` table) and its migration.
16+
- `src/ticker.ts` — `createCronTicker({ db, deliver, intervalMs })`: claims
17+
due rows with `SELECT ... FOR UPDATE SKIP LOCKED` so concurrent tickers
18+
never double-fire, and only the latest missed tick fires.
19+
- `src/mount.ts` — `mountCron(app, { db, requireTenantMember })`: CRUD at
20+
`/api/tenants/:tenantId/cron`.
21+
22+
## Mounting in the hub
23+
24+
```ts
25+
const cronApp = new Hono<TenantEnv>();
26+
mountCron(cronApp, {
27+
db,
28+
requireTenantMember: (ctx, tenantId) => {
29+
const c = ctx as { get(key: "tenant"): { id: string } };
30+
return c.get("tenant").id === tenantId;
31+
},
32+
});
33+
app.route("/", cronApp);
34+
35+
createCronTicker({
36+
db,
37+
intervalMs: 60_000,
38+
deliver: (message) =>
39+
lookups.persistMail({
40+
senderAddress: message.from,
41+
recipients: message.to,
42+
raw: buildRawMessage(message),
43+
}),
44+
}).start();
45+
```

‎packages/cron/package.json‎

Lines changed: 27 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,27 @@
1+
{
2+
"name": "@corbits/cron",
3+
"version": "0.0.1",
4+
"private": true,
5+
"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",
6+
"license": "LGPL-2.1-or-later",
7+
"type": "module",
8+
"exports": {
9+
".": "./src/index.ts",
10+
"./migrations": "./src/migrations.ts"
11+
},
12+
"scripts": {
13+
"typecheck": "tsc --noEmit",
14+
"test": "bun test"
15+
},
16+
"dependencies": {
17+
"arktype": "catalog:",
18+
"drizzle-orm": "catalog:",
19+
"hono": "catalog:",
20+
"postgres": "catalog:"
21+
},
22+
"devDependencies": {
23+
"@intx/db": "workspace:*",
24+
"@types/bun": "catalog:",
25+
"typescript": "catalog:"
26+
}
27+
}
File renamed without changes.

‎packages/cron/src/index.ts‎

Lines changed: 21 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,21 @@
1+
export {
2+
CRON_FIELD_RANGES,
3+
cronExpressionCanFire,
4+
isValidCronExpression,
5+
isValidTimeZone,
6+
MAX_LOOKAHEAD_MINUTES,
7+
nextCronFireAfter,
8+
zonedParts,
9+
type CronField,
10+
type ZonedParts,
11+
} from "./cron";
12+
export { cronScheduleTable } from "./schema";
13+
export {
14+
createCronTicker,
15+
cronSenderAddress,
16+
type CronDb,
17+
type CronSenderAddress,
18+
type CronTicker,
19+
type DeliverCronMail,
20+
} from "./ticker";
21+
export { mountCron, type MountCronOpts, type RequireTenantMember } from "./mount";

‎packages/cron/src/migrations.ts‎

Lines changed: 1 addition & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1 @@
1+
export { applyCronMigrations, type ApplyCronMigrationsReport } from "./schema";

‎packages/cron/src/mount.ts‎

Lines changed: 80 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,80 @@
1+
// CRUD over a tenant's saved cron schedules. Mirrors `@corbits/mailbox`'s
2+
// `mountMailbox` shape: absolute routes registered directly on the host's
3+
// app (never a sub-router mounted under a prefix), a caller-supplied
4+
// resolver for "is the caller allowed to act on this tenant," and arktype
5+
// parsing at the one trust boundary that matters (the request body).
6+
import { randomUUID } from "node:crypto";
7+
import { type } from "arktype";
8+
import { eq, and } from "drizzle-orm";
9+
import type { Env, Hono } from "hono";
10+
11+
import { isValidCronExpression } from "./cron";
12+
import { cronScheduleTable } from "./schema";
13+
import type { CronDb } from "./ticker";
14+
15+
export type RequireTenantMember = (ctx: unknown, tenantId: string) => Promise<boolean> | boolean;
16+
17+
export type MountCronOpts = {
18+
db: CronDb;
19+
requireTenantMember: RequireTenantMember;
20+
};
21+
22+
const CreateScheduleBody = type({
23+
expression: "string",
24+
toAddress: "string",
25+
subject: "string",
26+
body: "string",
27+
});
28+
29+
/** Mount `/api/tenants/:tenantId/cron` CRUD onto the host's app. */
30+
export function mountCron<E extends Env>(app: Hono<E>, opts: MountCronOpts): Hono<E> {
31+
const { db, requireTenantMember } = opts;
32+
33+
app.get("/api/tenants/:tenantId/cron", async (c) => {
34+
const tenantId = c.req.param("tenantId");
35+
if (!(await requireTenantMember(c, tenantId))) return c.json({ error: "forbidden" }, 403);
36+
const rows = await db
37+
.select()
38+
.from(cronScheduleTable)
39+
.where(eq(cronScheduleTable.tenantId, tenantId));
40+
return c.json({ schedules: rows });
41+
});
42+
43+
app.post("/api/tenants/:tenantId/cron", async (c) => {
44+
const tenantId = c.req.param("tenantId");
45+
if (!(await requireTenantMember(c, tenantId))) return c.json({ error: "forbidden" }, 403);
46+
const parsed = CreateScheduleBody(await c.req.json().catch(() => undefined));
47+
if (parsed instanceof type.errors) {
48+
return c.json({ error: "invalid_body", detail: parsed.summary }, 400);
49+
}
50+
if (!isValidCronExpression(parsed.expression)) {
51+
return c.json({ error: "invalid_expression" }, 400);
52+
}
53+
const [row] = await db
54+
.insert(cronScheduleTable)
55+
.values({
56+
id: randomUUID(),
57+
tenantId,
58+
expression: parsed.expression,
59+
toAddress: parsed.toAddress,
60+
subject: parsed.subject,
61+
body: parsed.body,
62+
})
63+
.returning();
64+
return c.json({ schedule: row }, 201);
65+
});
66+
67+
app.delete("/api/tenants/:tenantId/cron/:id", async (c) => {
68+
const tenantId = c.req.param("tenantId");
69+
if (!(await requireTenantMember(c, tenantId))) return c.json({ error: "forbidden" }, 403);
70+
const id = c.req.param("id");
71+
const [deleted] = await db
72+
.delete(cronScheduleTable)
73+
.where(and(eq(cronScheduleTable.tenantId, tenantId), eq(cronScheduleTable.id, id)))
74+
.returning({ id: cronScheduleTable.id });
75+
if (deleted === undefined) return c.json({ error: "not_found" }, 404);
76+
return c.json({ ok: true });
77+
});
78+
79+
return app;
80+
}

0 commit comments

Comments
 (0)