Add @corbits/cron: cron ticks delivered as mail (CL-8183)

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",

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<TenantEnv>();
+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();
+```

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:"
+  }
+}

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";

packages/cron/src/migrations.ts

@@ -0,0 +1 @@
+export { applyCronMigrations, type ApplyCronMigrationsReport } from "./schema";

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> | 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<E extends Env>(app: Hono<E>, opts: MountCronOpts): Hono<E> {
+  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;
+}