allow racing multiple events

in progress

Author: UfukUstali

example/convex/_generated/api.d.ts

@@ -14,6 +14,7 @@ import type * as example from "../example.js";
 import type * as inlineTest from "../inlineTest.js";
 import type * as nestedWorkflow from "../nestedWorkflow.js";
 import type * as passingSignals from "../passingSignals.js";
+import type * as raceEvents from "../raceEvents.js";
 import type * as transcription from "../transcription.js";
 import type * as userConfirmation from "../userConfirmation.js";
 
@@ -30,6 +31,7 @@ declare const fullApi: ApiFromModules<{
   inlineTest: typeof inlineTest;
   nestedWorkflow: typeof nestedWorkflow;
   passingSignals: typeof passingSignals;
+  raceEvents: typeof raceEvents;
   transcription: typeof transcription;
   userConfirmation: typeof userConfirmation;
 }>;

example/convex/raceEvents.ts

@@ -0,0 +1,127 @@
+import {
+  defineEvent,
+  vOnComplete,
+  vWorkflowId,
+  WorkflowManager,
+} from "@convex-dev/workflow";
+import { v } from "convex/values";
+import { components, internal } from "./_generated/api";
+import { internalMutation } from "./_generated/server";
+import { literals } from "convex-helpers/validators";
+
+const workflow = new WorkflowManager(components.workflow);
+
+const proposals = {
+  A: "Proposal A",
+  B: "Proposal B",
+  C: "Proposal C",
+};
+const vProposal = literals("A", "B", "C");
+export const approvalEvent = defineEvent({
+  name: "approval",
+  validator: v.object({ proposal: vProposal }),
+});
+export const rejectionEvent = defineEvent({
+  name: "rejection",
+  validator: v.object({ reason: v.string() }),
+});
+
+export const raceExample = workflow.define({
+  args: {},
+  returns: v.string(),
+  handler: async (step): Promise<string> => {
+    const result = await step.raceEvents([approvalEvent, rejectionEvent], {
+      name: "waitForDecision",
+      timeout: 15 * 60 * 1000, // 15 minutes
+    });
+
+    switch (result.name) {
+      case "approval": {
+        return `Approved. Selected: ${proposals[result.value.proposal]}`;
+      }
+      case "rejection": {
+        const val = result.value;
+        return `Rejected: ${val.reason}`;
+      }
+    }
+  },
+});
+
+export const sendApproval = internalMutation({
+  args: { workflowId: vWorkflowId, proposal: vProposal },
+  handler: async (ctx, args) => {
+    await workflow.sendEvent(ctx, {
+      ...approvalEvent,
+      workflowId: args.workflowId,
+      value: { proposal: args.proposal },
+    });
+  },
+});
+export const sendRejection = internalMutation({
+  args: { workflowId: vWorkflowId, reason: v.string() },
+  handler: async (ctx, args) => {
+    await workflow.sendEvent(ctx, {
+      ...rejectionEvent,
+      workflowId: args.workflowId,
+      value: { reason: args.reason },
+    });
+  },
+});
+
+export const startWorkflow = internalMutation({
+  args: {},
+  handler: async (ctx) => {
+    const id = await workflow.start(
+      ctx,
+      internal.raceEvents.raceExample,
+      {},
+      {
+        onComplete: internal.raceEvents.completeWorkflow,
+      },
+    );
+    await ctx.db.insert("flows", { workflowId: id, in: "", out: null });
+  },
+});
+export const completeWorkflow = internalMutation({
+  args: vOnComplete(v.any()),
+  handler: async (ctx, args): Promise<void> => {
+    const flow = await ctx.db
+      .query("flows")
+      .withIndex("workflowId", (q) => q.eq("workflowId", args.workflowId))
+      .first();
+    if (!flow) {
+      throw new Error(`Flow not found: ${args.workflowId}`);
+    }
+    await ctx.db.patch("flows", flow._id, { out: args.result });
+  },
+});
+
+export const raceWithoutValidators = workflow.define({
+  args: {},
+  returns: v.string(),
+  handler: async (step): Promise<string> => {
+    const result = await step.raceEvents([{ name: "go" }, { name: "stop" }]);
+    if (result.name === "go") {
+      return "Proceeding";
+    }
+    return "Stopped";
+  },
+});
+export const sendGo = internalMutation({
+  args: { workflowId: vWorkflowId },
+  handler: async (ctx, args) => {
+    await workflow.sendEvent(ctx, {
+      name: "go",
+      workflowId: args.workflowId,
+    });
+  },
+});
+export const sendStop = internalMutation({
+  args: { workflowId: vWorkflowId },
+  handler: async (ctx, args) => {
+    await workflow.sendEvent(ctx, {
+      name: "stop",
+      workflowId: args.workflowId,
+    });
+  },
+});

src/client/index.ts

@@ -40,10 +40,12 @@ export {
   vEventId,
   vWorkflowId,
   vWorkflowStep,
+  vOnComplete,
   type EventId,
   type WorkflowId,
   type WorkflowStep,
 } from "../types.js";
+export type { RaceResult } from "./workflowContext.js";
 export type { RunOptions, WorkflowCtx } from "./workflowContext.js";
 
 export type CallbackOptions = {

src/client/step.ts

@@ -44,6 +44,14 @@ export type StepRequest = {
     | {
         kind: "sleep";
         args: Record<string, never>;
+      }
+    | {
+        kind: "race";
+        args: {
+          events: Array<{ name: string }>;
+          timeout?: number;
+          failure?: "fail" | "retry" | "discard";
+        };
       };
   retry: RetryBehavior | boolean | undefined;
   inline: boolean;
@@ -189,31 +197,56 @@ export class StepExecutor {
           startedAt: this.now,
           completedAt: runResult ? this.now : undefined,
         } satisfies Omit<Step, "kind">;
-        const step =
-          target.kind === "function"
-            ? {
-                kind: "function" as const,
-                functionType: target.functionType,
-                handle: await createFunctionHandle(target.function),
-                ...commonFields,
-              }
-            : target.kind === "workflow"
-              ? {
-                  kind: "workflow" as const,
-                  handle: await createFunctionHandle(target.function),
-                  ...commonFields,
-                }
-              : target.kind === "event"
-              ? {
-                  kind: "event" as const,
-                  eventId: target.args.eventId,
-                  ...commonFields,
-                  args: target.args,
-                }
-              : {
-                  kind: "sleep" as const,
-                  ...commonFields,
-                };
+        let step;
+        switch (target.kind) {
+          case "function": {
+            step = {
+              kind: "function" as const,
+              functionType: target.functionType,
+              handle: await createFunctionHandle(target.function),
+              ...commonFields,
+            };
+            break;
+          }
+          case "workflow": {
+            step = {
+              kind: "workflow" as const,
+              handle: await createFunctionHandle(target.function),
+              ...commonFields,
+            };
+            break;
+          }
+          case "event": {
+            step = {
+              kind: "event" as const,
+              eventId: target.args.eventId,
+              ...commonFields,
+              args: target.args,
+            };
+            break;
+          }
+          case "sleep": {
+            step = {
+              kind: "sleep" as const,
+              ...commonFields,
+            };
+            break;
+          }
+          case "race": {
+            step = {
+              kind: "race" as const,
+              events: target.args.events,
+              timeout:
+                target.args.timeout !== undefined
+                  ? { ms: target.args.timeout }
+                  : undefined,
+              failure: target.args.failure,
+              ...commonFields,
+              args: target.args,
+            };
+            break;
+          }
+        }
         return {
           retry: message.retry,
           schedulerOptions: message.schedulerOptions,

src/client/workflowContext.ts

@@ -12,6 +12,23 @@ import type { Validator } from "convex/values";
 import type { EventId, SchedulerOptions, WorkflowId } from "../types.js";
 import { safeFunctionName } from "./safeFunctionName.js";
 import type { StepRequest } from "./step.js";
+import { assert } from "convex-helpers";
+
+export type RaceResult<
+  T extends ReadonlyArray<{
+    name: string;
+    validator?: Validator<any, any, any>;
+  }>,
+> = {
+  [K in keyof T]: T[K] extends {
+    name: infer N extends string;
+    validator: Validator<infer V, any, any>;
+  }
+    ? { name: N; value: V }
+    : T[K] extends { name: infer N extends string }
+      ? { name: N; value: unknown }
+      : never;
+}[number];
 
 export type RunOptions = {
   /**
@@ -125,6 +142,33 @@ export type WorkflowCtx = {
    * @param opts - Optionally name the step. Default: "sleep"
    */
   sleep(duration: number, opts?: { name?: string }): Promise<void>;
+
+  /**
+   * Waits for any one of multiple events, returning the first that fires.
+   *
+   * Each event in the array must have a unique name. The workflow blocks
+   * until an event with one of the given names is sent, then returns the
+   * matched event's name and value.
+   *
+   * @param events - Array of event definitions, each with a unique `name`
+   *   and an optional `validator` to parse the event's payload.
+   * @param opts - Optional options, including a custom step `name` for
+   *   observability (defaults to `"race(name1, name2, ...)"`) and a
+   *   `timeout` in milliseconds after which the race rejects with an error.
+   */
+  raceEvents<
+    const T extends ReadonlyArray<{
+      name: string;
+      validator?: Validator<any, any, any>;
+    }>,
+  >(
+    events: T,
+    opts?: {
+      name?: string;
+      timeout?: number;
+      failure?: "fail" | "retry" | "discard";
+    },
+  ): Promise<RaceResult<T>>;
 };
 
 export type OptionalRestArgs<
@@ -197,6 +241,58 @@ export function createWorkflowCtx(
       }
       return result as any;
     },
+
+    raceEvents: async <
+      T extends ReadonlyArray<{
+        name: string;
+        validator?: Validator<any, any, any>;
+      }>,
+    >(
+      events: T,
+      opts?: {
+        name?: string;
+        timeout?: number;
+        failure?: "fail" | "retry" | "discard";
+      },
+    ) => {
+      assert(events.length > 0, "At least one event must be specified.");
+      assert(
+        new Set(events.map((e) => e.name)).size === events.length,
+        "All events must have unique names.",
+      );
+      assert(
+        opts?.timeout === undefined || opts.timeout > 0,
+        "Timeout must be a positive number.",
+      );
+      const result = await run(sender, {
+        name: opts?.name ?? `race(${events.map((e) => e.name).join(", ")})`,
+        target: {
+          kind: "race",
+          args: {
+            events: events.map((e) => ({ name: e.name })),
+            timeout: opts?.timeout,
+            failure: opts?.failure,
+          },
+        },
+        retry: undefined,
+        inline: false,
+        schedulerOptions: {},
+      });
+      const winner = events.find((e) => e.name === (result as any).eventName);
+      if (winner?.validator) {
+        return {
+          name: winner.name as RaceResult<T>["name"],
+          value: parse(
+            winner.validator,
+            (result as any).value,
+          ) as RaceResult<T>["value"],
+        } as RaceResult<T>;
+      }
+      return {
+        name: (result as any).eventName,
+        value: (result as any).value,
+      } as any;
+    },
   } satisfies WorkflowCtx;
 }
 

src/component/_generated/api.ts

@@ -13,6 +13,7 @@ import type * as journal from "../journal.js";
 import type * as logging from "../logging.js";
 import type * as model from "../model.js";
 import type * as pool from "../pool.js";
+import type * as race from "../race.js";
 import type * as utils from "../utils.js";
 import type * as workflow from "../workflow.js";
 
@@ -29,6 +30,7 @@ const fullApi: ApiFromModules<{
   logging: typeof logging;
   model: typeof model;
   pool: typeof pool;
+  race: typeof race;
   utils: typeof utils;
   workflow: typeof workflow;
 }> = anyApi as any;