feat: pause and resume a schedule

README.md

@@ -2,7 +2,7 @@
 
 A Corbits hub module that wakes Interchange agents on five-field UTC cron schedules, mounted as `@intx/hub-api` routes on the hub (Interchange's multi-tenant control plane) and stored in its Postgres. A ticker mails each due schedule to its agent's current run, the running instance of the agent's workflow definition.
 
-Schedules support create, list and delete. There is no edit: delete and recreate. An expression must be able to fire (no Feb 30), at most 256 characters and 64 comma clauses.
+Schedules support create, list, pause, resume and delete. There is no edit: delete and recreate. An expression must be able to fire (no Feb 30), at most 256 characters and 64 comma clauses.
 
 ## Why @corbits/cron?
 
@@ -49,11 +49,13 @@ Prints the next 09:00 UTC, the next time a schedule with that expression fires.
 
 Paths are relative to where the host mounts the sub-app.
 
-| Method | Path   | Grant                         | Purpose                                                                                                                                            |
-| ------ | ------ | ----------------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------- |
-| GET    | `/`    | `cron-schedule:*` `read`      | `{ schedules: [...] }` for the caller's tenant.                                                                                                    |
-| POST   | `/`    | `cron-schedule:*` `create`    | Create from `{ expression, definitionName, subject, body }`. 201 `{ schedule }`; 400 `invalid_body`, `invalid_expression` or `unknown_definition`. |
-| DELETE | `/:id` | `cron-schedule:<id>` `manage` | Delete one schedule. `{ ok: true }`; 404 `not_found` when it is not in the caller's tenant.                                                        |
+| Method | Path          | Grant                         | Purpose                                                                                                                                            |
+| ------ | ------------- | ----------------------------- | -------------------------------------------------------------------------------------------------------------------------------------------------- |
+| GET    | `/`           | `cron-schedule:*` `read`      | `{ schedules: [...] }` for the caller's tenant.                                                                                                    |
+| POST   | `/`           | `cron-schedule:*` `create`    | Create from `{ expression, definitionName, subject, body }`. 201 `{ schedule }`; 400 `invalid_body`, `invalid_expression` or `unknown_definition`. |
+| POST   | `/:id/pause`  | `cron-schedule:<id>` `manage` | Stop the ticker firing it. `{ schedule }` with `enabled: false`; 404 `not_found`.                                                                  |
+| POST   | `/:id/resume` | `cron-schedule:<id>` `manage` | Fire again from the next matching minute after now. `{ schedule }` with `enabled: true`; 404 `not_found`.                                          |
+| DELETE | `/:id`        | `cron-schedule:<id>` `manage` | Delete one schedule. `{ ok: true }`; 404 `not_found` when it is not in the caller's tenant.                                                        |
 
 ### `createCronTicker(opts)`
 

e2e/routes.test.ts

@@ -4,8 +4,12 @@ import { afterAll, beforeAll, expect, test } from "bun:test";
 import { randomUUID } from "node:crypto";
 import { createDB } from "@intx/db";
 import type { RequireGrant, TenantEnv } from "@intx/hub-api";
+import { eq } from "drizzle-orm";
 import type { Hono } from "hono";
 
+import { cronScheduleTable } from "../src/schema.js";
+import { createCronTicker } from "../src/ticker.js";
+
 import {
   createTestDatabase,
   cronRoutesApp,
@@ -109,6 +113,66 @@ describeIfDb("createCronRoutes", () => {
     }
   });
 
+  test("a paused schedule is skipped by the ticker; resume restarts its clock", async () => {
+    const { db, close } = createDB(requireDatabase().config);
+    try {
+      const tenantId = `tnt_cron_pause_${randomUUID().slice(0, 8)}`;
+      await seedTenant(db, tenantId);
+      await seedDeployment(db, tenantId, "agent-pause-source");
+      const app = cronRoutesApp(db, tenantId, allowAll);
+      const id = `sched_pause_${randomUUID().slice(0, 8)}`;
+      await db.insert(cronScheduleTable).values({
+        id,
+        tenantId,
+        expression: "* * * * *",
+        definitionName: "agent-pause-source",
+        subject: "s",
+        body: "b",
+        createdAt: new Date(Date.now() - 5 * 60_000),
+      });
+
+      const paused = await app.request(`/cron/${id}/pause`, { method: "POST" });
+      expect(paused.status).toBe(200);
+      expect(
+        ((await paused.json()) as { schedule: { enabled: boolean } }).schedule
+          .enabled,
+      ).toBe(false);
+
+      const delivered: string[] = [];
+      const ticker = createCronTicker({
+        db,
+        intervalMs: 50,
+        deliver: (message) => {
+          delivered.push(message.subject);
+        },
+      });
+      ticker.start();
+      await new Promise((resolve) => setTimeout(resolve, 500));
+      ticker.stop();
+      expect(delivered).toEqual([]);
+
+      const resumed = await app.request(`/cron/${id}/resume`, {
+        method: "POST",
+      });
+      expect(resumed.status).toBe(200);
+      const [row] = await db
+        .select()
+        .from(cronScheduleTable)
+        .where(eq(cronScheduleTable.id, id));
+      expect(row?.enabled).toBe(true);
+      // Restarted at now, not due for the five paused minutes.
+      expect(Date.now() - (row?.lastFiredAt?.getTime() ?? 0)).toBeLessThan(
+        30_000,
+      );
+
+      expect(
+        (await app.request("/cron/missing/pause", { method: "POST" })).status,
+      ).toBe(404);
+    } finally {
+      await close();
+    }
+  });
+
   test("each route is gated by the host's requireGrant", async () => {
     const { db, close } = createDB(requireDatabase().config);
     try {

migrations/0001_schedule.sql

@@ -17,4 +17,6 @@ CREATE TABLE IF NOT EXISTS "cron"."schedule" (
 -- Tables created before waiting_since existed.
 ALTER TABLE "cron"."schedule" ADD COLUMN IF NOT EXISTS "waiting_since" timestamptz;
 --> statement-breakpoint
+ALTER TABLE "cron"."schedule" ADD COLUMN IF NOT EXISTS "enabled" boolean NOT NULL DEFAULT true;
+--> statement-breakpoint
 CREATE INDEX IF NOT EXISTS "cron_schedule_tenant_id_idx" ON "cron"."schedule" ("tenant_id");

src/routes.ts

@@ -86,5 +86,31 @@ export function createCronRoutes({
     },
   );
 
+  // Resume restarts the clock at now: a schedule is due from its last fire,
+  // so keeping the old one would fire at once for the paused minutes.
+  for (const [path, enabled] of [
+    ["pause", false],
+    ["resume", true],
+  ] as const) {
+    app.post(
+      `/:id/${path}`,
+      requireGrant(idResource("cron-schedule", "id"), "manage"),
+      async (c) => {
+        const [row] = await db
+          .update(cronScheduleTable)
+          .set(enabled ? { enabled, lastFiredAt: new Date() } : { enabled })
+          .where(
+            and(
+              eq(cronScheduleTable.tenantId, c.get("tenant").id),
+              eq(cronScheduleTable.id, c.req.param("id")),
+            ),
+          )
+          .returning();
+        if (row === undefined) return c.json({ error: "not_found" }, 404);
+        return c.json({ schedule: row });
+      },
+    );
+  }
+
   return app;
 }

src/schema.ts

@@ -1,7 +1,13 @@
 // 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, so it never collides with the host's own tables.
-import { pgTable, pgSchema, text, timestamp } from "drizzle-orm/pg-core";
+import {
+  boolean,
+  pgTable,
+  pgSchema,
+  text,
+  timestamp,
+} from "drizzle-orm/pg-core";
 
 const hostTenant = pgTable("tenant", { id: text("id").primaryKey() });
 
@@ -18,6 +24,8 @@ export const cronScheduleTable = cronSchema.table("schedule", {
   definitionName: text("definition_name").notNull(),
   subject: text("subject").notNull(),
   body: text("body").notNull(),
+  // False while paused: the ticker skips the row until it is resumed.
+  enabled: boolean("enabled").notNull().default(true),
   lastFiredAt: timestamp("last_fired_at", { withTimezone: true }),
   // Set while the agent exists but has no live run — a hub restart, say.
   // Cleared by the first tick that delivers again.

src/ticker.ts

@@ -1,6 +1,6 @@
 // Turns due cron schedules into mail. A schedule that missed several ticks
 // fires once for the most recent due minute, never once per missed tick.
-import { eq, isNull } from "drizzle-orm";
+import { and, eq, isNull } from "drizzle-orm";
 import type { PostgresJsDatabase } from "drizzle-orm/postgres-js";
 
 import { cronIsDue, parseCronExpression, parsedCronCanFire } from "./cron.js";
@@ -123,7 +123,12 @@ async function tick<TSchema extends Record<string, unknown>>(
     const candidates = await tx
       .select()
       .from(cronScheduleTable)
-      .where(isNull(cronScheduleTable.stoppedAt))
+      .where(
+        and(
+          isNull(cronScheduleTable.stoppedAt),
+          eq(cronScheduleTable.enabled, true),
+        ),
+      )
       .for("update", { skipLocked: true });
 
     for (const row of candidates) {