perf: add events.createBatch() for batch event creation

🤖 Generated with [Claude Code](https://claude.com/claude-code)

Co-Authored-By: Claude Opus 4.5 <noreply@anthropic.com>

Author: pranaygp

.changeset/batch-event-creation.md

@@ -0,0 +1,13 @@
+---
+"@workflow/core": patch
+"@workflow/world": patch
+"@workflow/world-local": patch
+"@workflow/world-postgres": patch
+"@workflow/world-vercel": patch
+---
+
+perf: add events.createBatch() for batch event creation
+
+- Add `createBatch()` method to Storage interface for creating multiple events atomically
+- Use batch event creation in suspension handler for improved performance
+- Use batch event creation for wait_completed events in runtime

packages/core/src/runtime.ts

@@ -340,11 +340,14 @@ export function workflowEntrypoint(
                   correlationId: e.correlationId,
                 }));
 
-              // Create all wait_completed events
-              for (const waitEvent of waitsToComplete) {
-                const result = await world.events.create(runId, waitEvent);
-                // Add the event to the events array so the workflow can see it
-                events.push(result.event);
+              // Batch create all wait_completed events
+              if (waitsToComplete.length > 0) {
+                const completedResults = await world.events.createBatch(
+                  runId,
+                  waitsToComplete
+                );
+                // Add the events to the events array so the workflow can see them
+                events.push(...completedResults.map((r) => r.event));
               }
 
               const result = await runWorkflow(

packages/core/src/runtime/suspension-handler.ts

@@ -30,7 +30,7 @@ export interface SuspensionHandlerResult {
 /**
  * Handles a workflow suspension by processing all pending operations (hooks, steps, waits).
  * Uses an event-sourced architecture where entities (steps, hooks) are created atomically
- * with their corresponding events via events.create().
+ * with their corresponding events via createBatch.
  *
  * Processing order:
  * 1. Hooks are processed first to prevent race conditions with webhook receivers
@@ -72,16 +72,16 @@ export async function handleSuspension({
   });
 
   // Process hooks first to prevent race conditions with webhook receivers
-  for (const hookEvent of hookEvents) {
+  if (hookEvents.length > 0) {
     try {
-      await world.events.create(runId, hookEvent);
+      await world.events.createBatch(runId, hookEvents);
     } catch (err) {
       if (WorkflowAPIError.is(err)) {
         if (err.status === 409) {
-          console.warn(`Hook already exists, continuing: ${err.message}`);
+          console.warn(`Some hooks already exist, continuing: ${err.message}`);
         } else if (err.status === 410) {
           console.warn(
-            `Workflow run "${runId}" has already completed, skipping hook: ${err.message}`
+            `Workflow run "${runId}" has already completed, skipping hooks: ${err.message}`
           );
         } else {
           throw err;
@@ -128,35 +128,36 @@ export async function handleSuspension({
       },
     }));
 
-  // Process steps and waits
+  // Process steps and waits in parallel
   const ops: Promise<void>[] = [];
 
-  // Create step events (World creates step entities atomically)
-  // Only for steps that don't already have a step_created event
-  for (const stepEvent of stepEvents) {
-    try {
-      await world.events.create(runId, stepEvent);
-    } catch (err) {
-      if (WorkflowAPIError.is(err) && err.status === 409) {
-        console.warn(`Step already exists, continuing: ${err.message}`);
-      } else {
-        throw err;
-      }
-    }
-  }
-
-  // Create wait events
-  for (const waitEvent of waitEvents) {
-    try {
-      await world.events.create(runId, waitEvent);
-    } catch (err) {
-      if (WorkflowAPIError.is(err) && err.status === 409) {
-        console.warn(`Wait already exists, continuing: ${err.message}`);
-      } else {
-        throw err;
-      }
-    }
-  }
+  await Promise.all([
+    // Create step events (World creates step entities atomically)
+    // Only for steps that don't already have a step_created event
+    stepEvents.length > 0
+      ? world.events.createBatch(runId, stepEvents).catch((err) => {
+          if (WorkflowAPIError.is(err) && err.status === 409) {
+            console.warn(
+              `Some steps already exist, continuing: ${err.message}`
+            );
+          } else {
+            throw err;
+          }
+        })
+      : Promise.resolve(),
+    // Create wait events
+    waitEvents.length > 0
+      ? world.events.createBatch(runId, waitEvents).catch((err) => {
+          if (WorkflowAPIError.is(err) && err.status === 409) {
+            console.warn(
+              `Some waits already exist, continuing: ${err.message}`
+            );
+          } else {
+            throw err;
+          }
+        })
+      : Promise.resolve(),
+  ]);
 
   // Queue step execution messages for ALL pending steps
   // (both newly created and those with existing step_created events)

packages/world-local/src/storage.ts

@@ -687,6 +687,16 @@ export function createStorage(basedir: string): Storage {
         };
       },
 
+      async createBatch(runId, data, params): Promise<EventResult[]> {
+        // createBatch is just a sequential loop over create() to ensure monotonic ULIDs
+        const results: EventResult[] = [];
+        for (const eventData of data) {
+          const result = await this.create(runId, eventData, params);
+          results.push(result);
+        }
+        return results;
+      },
+
       async list(params) {
         const { runId } = params;
         const resolveData = params.resolveData ?? DEFAULT_RESOLVE_DATA_OPTION;

packages/world-postgres/src/storage.ts

@@ -448,6 +448,228 @@ export function createEventsStorage(drizzle: Drizzle): Storage['events'] {
       const resolveData = params?.resolveData ?? 'all';
       return { event: filterEventData(parsed, resolveData), run, step, hook };
     },
+    async createBatch(runId, data, params): Promise<EventResult[]> {
+      if (data.length === 0) {
+        return [];
+      }
+
+      const resolveData = params?.resolveData ?? 'all';
+      const now = new Date();
+
+      // For run_created events, generate runId server-side if not provided
+      const hasRunCreated = data.some((e) => e.eventType === 'run_created');
+      let effectiveRunId = runId;
+      if (hasRunCreated && (!runId || runId === '')) {
+        effectiveRunId = `wrun_${ulid()}`;
+      }
+
+      // Separate events by type for entity creation
+      const runCreatedEvents = data.filter(
+        (e) => e.eventType === 'run_created'
+      );
+      const runStartedEvents = data.filter(
+        (e) => e.eventType === 'run_started'
+      );
+      const runCompletedEvents = data.filter(
+        (e) => e.eventType === 'run_completed'
+      );
+      const runFailedEvents = data.filter((e) => e.eventType === 'run_failed');
+      const runCancelledEvents = data.filter(
+        (e) => e.eventType === 'run_cancelled'
+      );
+      const runPausedEvents = data.filter((e) => e.eventType === 'run_paused');
+      const runResumedEvents = data.filter(
+        (e) => e.eventType === 'run_resumed'
+      );
+      const stepCreatedEvents = data.filter(
+        (e) => e.eventType === 'step_created'
+      );
+      const hookCreatedEvents = data.filter(
+        (e) => e.eventType === 'hook_created'
+      );
+
+      // Create run entities atomically with events
+      if (runCreatedEvents.length > 0) {
+        const runsToInsert = runCreatedEvents.map((eventData) => {
+          const runData = (eventData as any).eventData as {
+            deploymentId: string;
+            workflowName: string;
+            input: any[];
+            executionContext?: Record<string, any>;
+          };
+          return {
+            runId: effectiveRunId,
+            deploymentId: runData.deploymentId,
+            workflowName: runData.workflowName,
+            input: runData.input as SerializedContent,
+            executionContext: runData.executionContext as
+              | SerializedContent
+              | undefined,
+            status: 'pending' as const,
+          };
+        });
+        await drizzle
+          .insert(Schema.runs)
+          .values(runsToInsert)
+          .onConflictDoNothing();
+      }
+
+      // Update run status for run_started events
+      if (runStartedEvents.length > 0) {
+        await drizzle
+          .update(Schema.runs)
+          .set({
+            status: 'running',
+            startedAt: now,
+            updatedAt: now,
+          })
+          .where(eq(Schema.runs.runId, effectiveRunId));
+      }
+
+      // Update run status for run_completed events
+      if (runCompletedEvents.length > 0) {
+        const completedData = (runCompletedEvents[0] as any).eventData as {
+          output?: any;
+        };
+        await drizzle
+          .update(Schema.runs)
+          .set({
+            status: 'completed',
+            output: completedData.output as SerializedContent | undefined,
+            completedAt: now,
+            updatedAt: now,
+          })
+          .where(eq(Schema.runs.runId, effectiveRunId));
+      }
+
+      // Update run status for run_failed events
+      if (runFailedEvents.length > 0) {
+        const failedData = (runFailedEvents[0] as any).eventData as {
+          error: any;
+          errorCode?: string;
+        };
+        const errorMessage =
+          typeof failedData.error === 'string'
+            ? failedData.error
+            : (failedData.error?.message ?? 'Unknown error');
+        // Store structured error as JSON for deserializeRunError to parse
+        const errorJson = JSON.stringify({
+          message: errorMessage,
+          stack: failedData.error?.stack,
+          code: failedData.errorCode,
+        });
+        await drizzle
+          .update(Schema.runs)
+          .set({
+            status: 'failed',
+            error: errorJson,
+            completedAt: now,
+            updatedAt: now,
+          })
+          .where(eq(Schema.runs.runId, effectiveRunId));
+      }
+
+      // Update run status for run_cancelled events
+      if (runCancelledEvents.length > 0) {
+        await drizzle
+          .update(Schema.runs)
+          .set({
+            status: 'cancelled',
+            completedAt: now,
+            updatedAt: now,
+          })
+          .where(eq(Schema.runs.runId, effectiveRunId));
+      }
+
+      // Update run status for run_paused events
+      if (runPausedEvents.length > 0) {
+        await drizzle
+          .update(Schema.runs)
+          .set({
+            status: 'paused',
+            updatedAt: now,
+          })
+          .where(eq(Schema.runs.runId, effectiveRunId));
+      }
+
+      // Update run status for run_resumed events
+      if (runResumedEvents.length > 0) {
+        await drizzle
+          .update(Schema.runs)
+          .set({
+            status: 'running',
+            updatedAt: now,
+          })
+          .where(eq(Schema.runs.runId, effectiveRunId));
+      }
+
+      // Create step entities atomically with events
+      if (stepCreatedEvents.length > 0) {
+        const stepsToInsert = stepCreatedEvents.map((eventData) => {
+          const stepData = (eventData as any).eventData as {
+            stepName: string;
+            input: any;
+          };
+          return {
+            runId: effectiveRunId,
+            stepId: eventData.correlationId!,
+            stepName: stepData.stepName,
+            input: stepData.input as SerializedContent,
+            status: 'pending' as const,
+            attempt: 0,
+          };
+        });
+        await drizzle
+          .insert(Schema.steps)
+          .values(stepsToInsert)
+          .onConflictDoNothing();
+      }
+
+      // Create hook entities atomically with events
+      if (hookCreatedEvents.length > 0) {
+        const hooksToInsert = hookCreatedEvents.map((eventData) => {
+          const hookData = (eventData as any).eventData as {
+            token: string;
+            metadata?: any;
+          };
+          return {
+            runId: effectiveRunId,
+            hookId: eventData.correlationId!,
+            token: hookData.token,
+            metadata: hookData.metadata as SerializedContent,
+            ownerId: '', // TODO: get from context
+            projectId: '', // TODO: get from context
+            environment: '', // TODO: get from context
+          };
+        });
+        await drizzle
+          .insert(Schema.hooks)
+          .values(hooksToInsert)
+          .onConflictDoNothing();
+      }
+
+      // Insert all events in a single batch query
+      const eventsToInsert = data.map((eventData) => ({
+        runId: effectiveRunId,
+        eventId: `wevt_${ulid()}`,
+        correlationId: eventData.correlationId,
+        eventType: eventData.eventType,
+        eventData: 'eventData' in eventData ? eventData.eventData : undefined,
+      }));
+
+      const values = await drizzle
+        .insert(events)
+        .values(eventsToInsert)
+        .returning({ eventId: events.eventId, createdAt: events.createdAt });
+
+      // Combine input data with returned values
+      // TODO: Return actual entity data from the database after entity creation is moved here
+      return data.map((eventData, i) => {
+        const result = { ...eventData, ...values[i], runId: effectiveRunId };
+        const parsed = EventSchema.parse(result);
+        return { event: filterEventData(parsed, resolveData) };
+      });
+    },
     async list(params: ListEventsParams): Promise<PaginatedResponse<Event>> {
       const limit = params?.pagination?.limit ?? 100;
       const sortOrder = params.pagination?.sortOrder || 'asc';