feat(api): enhance cron job execution with workflow data handling and improved trigger management

Author: bchapuis

apps/api/src/cron.ts

@@ -1,41 +1,87 @@
 import CronParser from "cron-parser"; // Or your chosen cron library
-import { and, eq, lte } from "drizzle-orm";
 import { drizzle } from "drizzle-orm/d1";
+import { Node, Workflow as WorkflowType } from "@dafthunk/types"; // Using Workflow as WorkflowType from types
+import { ExecutionContext } from "@cloudflare/workers-types";
 
-import { cronTriggers } from "./db/schema";
+import * as schema from "./db/schema";
+import { WorkflowRow } from "./db/schema";
+import { getDueCronTriggers, updateCronTriggerRunTimes, saveExecution } from "./db/queries";
+import { createDatabase, ExecutionStatusType } from "./db"; // Added ExecutionStatusType
+import { Bindings } from "./context"; // For the full env type
 
-// 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.
+// This function will now handle the actual execution triggering and initial record saving
 async function executeWorkflow(
-  workflowId: string,
-  _env: any,
-  _ctx: any
+  workflowRow: WorkflowRow,
+  db: ReturnType<typeof createDatabase>,
+  env: Bindings,
+  _ctx: ExecutionContext
 ): 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
+  console.log(`Attempting to execute workflow ${workflowRow.id} via cron.`);
+
+  // Assuming workflowRow.data is of type WorkflowType from @dafthunk/types
+  // which includes type, nodes, and edges.
+  const workflowData: WorkflowType = workflowRow.data;
+
+  try {
+    const executionInstance = await env.EXECUTE.create({
+      params: {
+        userId: "cron_trigger",
+        organizationId: workflowRow.organizationId,
+        workflow: {
+          id: workflowRow.id,
+          name: workflowRow.name,
+          handle: workflowRow.handle,
+          type: workflowData.type, // Reinstated: type from workflow data
+          nodes: workflowData.nodes, // Reinstated: nodes from workflow data
+          edges: workflowData.edges, // Reinstated: edges from workflow data
+        },
+        monitorProgress: false,
+        deploymentId: undefined,
+      },
+    });
+
+    const executionId = executionInstance.id;
+    console.log(`Workflow ${workflowRow.id} started with execution ID: ${executionId}`);
+
+    const nodeExecutions = workflowData.nodes.map((node: Node) => ({
+      nodeId: node.id,
+      status: "idle" as const,
+    }));
+
+    await saveExecution(db, {
+      id: executionId,
+      workflowId: workflowRow.id,
+      deploymentId: undefined,
+      userId: "cron_trigger",
+      organizationId: workflowRow.organizationId,
+      status: "executing" as ExecutionStatusType,
+      visibility: "private",
+      nodeExecutions,
+      createdAt: new Date(),
+      updatedAt: new Date(),
+      startedAt: new Date(),
+    });
+    console.log(`Initial execution record saved for ${executionId}`);
+
+  } catch (execError) {
+    console.error(`Error executing workflow ${workflowRow.id} or saving initial record:`, execError);
+  }
 }
 
 export async function handleCronTriggers(
   event: ScheduledEvent,
-  env: { DB: D1Database },
+  env: Bindings, // Updated to use the full Bindings type
   ctx: ExecutionContext
 ): Promise<void> {
   console.log(`Cron event triggered at: ${new Date(event.scheduledTime)}`);
-  const db = drizzle(env.DB, { schema: { cronTriggers } });
+  // Ensure db is initialized correctly if createDatabase is the standard way
+  // drizzle can also be initialized with just env.DB and schema
+  const db = createDatabase(env.DB); 
 
   const now = new Date();
 
   try {
-    const dueTriggers = await db
-      .select()
-      .from(cronTriggers)
-      .where(
-        and(eq(cronTriggers.active, true), lte(cronTriggers.nextRunAt, now))
-      )
-      .all();
+    const dueTriggers = await getDueCronTriggers(db, now);
 
     if (dueTriggers.length === 0) {
       console.log("No due cron triggers found.");
@@ -44,34 +90,35 @@ export async function handleCronTriggers(
 
     console.log(`Found ${dueTriggers.length} due cron triggers.`);
 
-    for (const trigger of dueTriggers) {
-      console.log(`Processing trigger for workflow: ${trigger.workflowId}`);
+    for (const item of dueTriggers) {
+      // item contains { cronTrigger: CronTriggerRow, workflow: WorkflowRow }
+      console.log(`Processing trigger for workflow: ${item.workflow.id}`);
       try {
-        // 1. Execute the workflow
-        await executeWorkflow(trigger.workflowId, env, ctx);
+        // 1. Execute the workflow (asynchronously, no await here if we don't wait for termination)
+        // However, executeWorkflow itself is async due to EXECUTE.create and saveExecution.
+        // We should await it to ensure the initial saveExecution completes before updating run times.
+        await executeWorkflow(item.workflow, db, env, ctx);
 
         // 2. Calculate next run time
-        const interval = CronParser.parse(trigger.cronExpression, {
+        const interval = CronParser.parse(item.cronTrigger.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();
+        await updateCronTriggerRunTimes(
+          db,
+          item.cronTrigger.workflowId,
+          nextRun,
+          now
+        );
 
         console.log(
-          `Workflow ${trigger.workflowId} executed. Next run at: ${nextRun.toISOString()}`
+          `Workflow ${item.workflow.id} processing initiated. Next run at: ${nextRun.toISOString()}`
         );
       } catch (err) {
         console.error(
-          `Error processing trigger for workflow ${trigger.workflowId}:`,
+          `Error processing trigger for workflow ${item.workflow.id}:`,
           err
         );
         // Optionally, you might want to implement retry logic or disable the trigger after too many failures.

apps/api/src/db/queries.ts

@@ -6,7 +6,7 @@ import {
   WorkflowExecutionStatus,
 } from "@dafthunk/types";
 import * as crypto from "crypto";
-import { and, desc, eq, inArray, SQL, sql } from "drizzle-orm";
+import { and, desc, eq, inArray, SQL, sql, lte } from "drizzle-orm";
 import { alias } from "drizzle-orm/sqlite-core";
 import { v7 as uuidv7 } from "uuid";
 
@@ -1080,12 +1080,12 @@ export async function getExecutionWithVisibility(
 }
 
 /**
- * Get a cron trigger for a workflow
+ * Get a cron trigger.
  *
  * @param db Database instance
  * @param workflowId Workflow ID
  * @param organizationIdOrHandle Organization ID or handle
- * @returns The cron trigger record or undefined if not found
+ * @returns Cron trigger record or undefined.
  */
 export async function getCronTrigger(
   db: ReturnType<typeof createDatabase>,
@@ -1110,34 +1110,79 @@ export async function getCronTrigger(
 }
 
 /**
- * Upsert a cron trigger for a workflow
+ * Create or update a cron trigger.
  *
  * @param db Database instance
- * @param values Values to insert or update
- * @returns The upserted cron trigger record
+ * @param values Cron trigger data.
+ * @returns The created or updated cron trigger record.
  */
 export async function upsertCronTrigger(
   db: ReturnType<typeof createDatabase>,
-  values: CronTriggerInsert & { updatedAt?: Date } // Allow updatedAt for explicit update
+  values: CronTriggerInsert
 ): Promise<CronTriggerRow> {
-  // Separate updatedAt for the 'set' part of onConflictDoUpdate
-  const { updatedAt, ...insertValues } = values;
-
-  const updateValues: Partial<CronTriggerInsert> & { updatedAt: Date } = {
-    cronExpression: values.cronExpression,
-    active: values.active,
-    nextRunAt: values.nextRunAt,
-    updatedAt: updatedAt || new Date(), // Use provided updatedAt or current time
+  const updateSet = {
+    ...values,
+    updatedAt: new Date(),
   };
-
   const [upsertedCron] = await db
     .insert(cronTriggers)
-    .values(insertValues) // createdAt will use default from schema
+    .values(values)
     .onConflictDoUpdate({
-      target: cronTriggers.workflowId, // Primary key for conflict
-      set: updateValues,
+      target: cronTriggers.workflowId,
+      set: updateSet,
     })
     .returning();
 
   return upsertedCron;
 }
+
+/**
+ * Update run times for a cron trigger.
+ *
+ * @param db Database instance
+ * @param workflowId Workflow ID
+ * @param nextRunAt New next run time
+ * @param lastRun Current execution time
+ * @returns Updated cron trigger record or undefined.
+ */
+export async function updateCronTriggerRunTimes(
+  db: ReturnType<typeof createDatabase>,
+  workflowId: string,
+  nextRunAt: Date,
+  lastRun: Date
+): Promise<CronTriggerRow | undefined> {
+  const [updatedTrigger] = await db
+    .update(cronTriggers)
+    .set({
+      lastRun: lastRun,
+      nextRunAt: nextRunAt,
+      updatedAt: new Date(),
+    })
+    .where(eq(cronTriggers.workflowId, workflowId))
+    .returning();
+  return updatedTrigger;
+}
+
+/**
+ * Get due cron triggers with workflow data.
+ *
+ * @param db Database instance
+ * @param now Current time
+ * @returns Array of due cron triggers with workflow data.
+ */
+export async function getDueCronTriggers(
+  db: ReturnType<typeof createDatabase>,
+  now: Date
+): Promise<{ cronTrigger: CronTriggerRow; workflow: WorkflowRow }[]> {
+  return db
+    .select({
+      cronTrigger: cronTriggers,
+      workflow: workflows
+    })
+    .from(cronTriggers)
+    .innerJoin(workflows, eq(cronTriggers.workflowId, workflows.id))
+    .where(
+      and(eq(cronTriggers.active, true), lte(cronTriggers.nextRunAt, now))
+    )
+    .all();
+}