feat(api): implement cron trigger management for workflows, including upsert and retrieval functionality

Author: bchapuis

apps/api/package.json

@@ -46,6 +46,7 @@
     "@hono/zod-validator": "^0.5.0",
     "@sendgrid/mail": "^8.1.5",
     "cloudflare": "^4.2.0",
+    "cron-parser": "^5.2.0",
     "drizzle-orm": "0.43.1",
     "exifreader": "^4.31.0",
     "hono": "^4.7.8",

apps/api/src/cron.ts

@@ -0,0 +1,83 @@
+import CronParser from "cron-parser"; // Or your chosen cron library
+import { and, eq, lte } from "drizzle-orm";
+import { drizzle } from "drizzle-orm/d1";
+
+import { cronTriggers } from "./db/schema";
+
+// This is a placeholder. You need to implement how workflows are triggered.
+// If executeWorkflow becomes more complex or used elsewhere, it could also be in its own file.
+async function executeWorkflow(
+  workflowId: string,
+  _env: any,
+  _ctx: any
+): Promise<void> {
+  console.log(`Executing workflow ${workflowId}`);
+  // Example: await fetch(`https://your-service.com/api/execute-workflow/${workflowId}`, { method: 'POST' });
+  // Or: directly call a function if workflow logic is bundled
+  // Remember to handle errors appropriately.
+  await Promise.resolve(); // Replace with actual workflow execution logic
+}
+
+export async function handleCronTriggers(
+  event: ScheduledEvent,
+  env: { DB: D1Database },
+  ctx: ExecutionContext
+): Promise<void> {
+  console.log(`Cron event triggered at: ${new Date(event.scheduledTime)}`);
+  const db = drizzle(env.DB, { schema: { cronTriggers } });
+
+  const now = new Date();
+
+  try {
+    const dueTriggers = await db
+      .select()
+      .from(cronTriggers)
+      .where(
+        and(eq(cronTriggers.active, true), lte(cronTriggers.nextRunAt, now))
+      )
+      .all();
+
+    if (dueTriggers.length === 0) {
+      console.log("No due cron triggers found.");
+      return;
+    }
+
+    console.log(`Found ${dueTriggers.length} due cron triggers.`);
+
+    for (const trigger of dueTriggers) {
+      console.log(`Processing trigger for workflow: ${trigger.workflowId}`);
+      try {
+        // 1. Execute the workflow
+        await executeWorkflow(trigger.workflowId, env, ctx);
+
+        // 2. Calculate next run time
+        const interval = CronParser.parse(trigger.cronExpression, {
+          currentDate: now,
+        });
+        const nextRun = interval.next().toDate();
+
+        // 3. Update the trigger in the database
+        await db
+          .update(cronTriggers)
+          .set({
+            lastRun: now,
+            nextRunAt: nextRun,
+          })
+          .where(eq(cronTriggers.workflowId, trigger.workflowId))
+          .execute();
+
+        console.log(
+          `Workflow ${trigger.workflowId} executed. Next run at: ${nextRun.toISOString()}`
+        );
+      } catch (err) {
+        console.error(
+          `Error processing trigger for workflow ${trigger.workflowId}:`,
+          err
+        );
+        // Optionally, you might want to implement retry logic or disable the trigger after too many failures.
+      }
+    }
+  } catch (dbError) {
+    console.error("Database error during scheduled task:", dbError);
+  }
+}

apps/api/src/db/migrations/0003_add_cron_triggers.sql

@@ -0,0 +1,18 @@
+CREATE TABLE `cron_triggers` (
+	`workflow_id` text PRIMARY KEY NOT NULL,
+	`cron_expression` text NOT NULL,
+	`active` integer DEFAULT true NOT NULL,
+	`last_run` integer,
+	`next_run_at` integer,
+	`created_at` integer DEFAULT CURRENT_TIMESTAMP NOT NULL,
+	`updated_at` integer DEFAULT CURRENT_TIMESTAMP NOT NULL,
+	FOREIGN KEY (`workflow_id`) REFERENCES `workflows`(`id`) ON UPDATE no action ON DELETE cascade
+);
+--> statement-breakpoint
+CREATE INDEX `cron_triggers_workflow_id_idx` ON `cron_triggers` (`workflow_id`);--> statement-breakpoint
+CREATE INDEX `cron_triggers_active_idx` ON `cron_triggers` (`active`);--> statement-breakpoint
+CREATE INDEX `cron_triggers_last_run_idx` ON `cron_triggers` (`last_run`);--> statement-breakpoint
+CREATE INDEX `cron_triggers_next_run_at_idx` ON `cron_triggers` (`next_run_at`);--> statement-breakpoint
+CREATE INDEX `cron_triggers_active_next_run_at_idx` ON `cron_triggers` (`active`,`next_run_at`);--> statement-breakpoint
+CREATE INDEX `cron_triggers_created_at_idx` ON `cron_triggers` (`created_at`);--> statement-breakpoint
+CREATE INDEX `cron_triggers_updated_at_idx` ON `cron_triggers` (`updated_at`);