pass in spread object and reorganize workflowContext

Author: ianmacartney

example/convex/_generated/api.d.ts

@@ -11,9 +11,9 @@
 import type * as admin from "../admin.js";
 import type * as example from "../example.js";
 import type * as nestedWorkflow from "../nestedWorkflow.js";
+import type * as passingSignals from "../passingSignals.js";
 import type * as transcription from "../transcription.js";
-import type * as userConfirmation_steps from "../userConfirmation/steps.js";
-import type * as userConfirmation_workflow from "../userConfirmation/workflow.js";
+import type * as userConfirmation from "../userConfirmation.js";
 
 import type {
   ApiFromModules,
@@ -33,9 +33,9 @@ declare const fullApi: ApiFromModules<{
   admin: typeof admin;
   example: typeof example;
   nestedWorkflow: typeof nestedWorkflow;
+  passingSignals: typeof passingSignals;
   transcription: typeof transcription;
-  "userConfirmation/steps": typeof userConfirmation_steps;
-  "userConfirmation/workflow": typeof userConfirmation_workflow;
+  userConfirmation: typeof userConfirmation;
 }>;
 declare const fullApiWithMounts: typeof fullApi;
 

example/convex/passingSignals.ts

@@ -0,0 +1,45 @@
+import { vWorkflowId, WorkflowManager } from "@convex-dev/workflow";
+import { components, internal } from "./_generated/api";
+import { internalMutation } from "./_generated/server";
+import { vEventId } from "../../src/types";
+
+const workflow = new WorkflowManager(components.workflow);
+
+export const signalBasedWorkflow = workflow.define({
+  args: {},
+  handler: async (ctx) => {
+    console.log("Starting signal based  workflow");
+    for (let i = 0; i < 3; i++) {
+      const signalId = await ctx.runMutation(
+        internal.passingSignals.createSignal,
+        { workflowId: ctx.workflowId },
+      );
+      await ctx.awaitEvent({ id: signalId });
+      console.log("Signal received", signalId);
+    }
+    console.log("All signals received");
+  },
+});
+
+export const createSignal = internalMutation({
+  args: { workflowId: vWorkflowId },
+  handler: async (ctx, args) => {
+    const eventId = await workflow.createEvent(ctx, {
+      name: "signal",
+      workflowId: args.workflowId,
+    });
+    // You would normally store this eventId somewhere to be able to send the
+    // signal later.
+    await ctx.scheduler.runAfter(1000, internal.passingSignals.sendSignal, {
+      eventId,
+    });
+    return eventId;
+  },
+});
+
+export const sendSignal = internalMutation({
+  args: { eventId: vEventId("signal") },
+  handler: async (ctx, args) => {
+    await workflow.sendEvent(ctx, { id: args.eventId });
+  },
+});

example/convex/userConfirmation.ts

@@ -0,0 +1,59 @@
+import {
+  defineEvent,
+  vWorkflowId,
+  WorkflowManager,
+} from "@convex-dev/workflow";
+import { v } from "convex/values";
+import { components, internal } from "./_generated/api";
+import { internalAction, internalMutation } from "./_generated/server";
+
+export const approvalEvent = defineEvent({
+  name: "approval" as const,
+  validator: v.union(
+    v.object({ approved: v.literal(true), choice: v.number() }),
+    v.object({ approved: v.literal(false), reason: v.string() }),
+  ),
+});
+
+const workflow = new WorkflowManager(components.workflow);
+
+export const confirmationWorkflow = workflow.define({
+  args: { prompt: v.string() },
+  returns: v.string(),
+  handler: async (ctx, args): Promise<string> => {
+    console.log("Starting confirmation workflow");
+    const proposals = await ctx.runAction(
+      internal.userConfirmation.generateProposals,
+      { prompt: args.prompt },
+      { retry: true },
+    );
+    console.log("Proposals generated", proposals);
+    const approval = await ctx.awaitEvent(approvalEvent);
+    if (!approval.approved) {
+      return "rejected: " + approval.reason;
+    }
+    const choice = proposals[approval.choice];
+    console.log("Choice selected", choice);
+    return choice;
+  },
+});
+
+export const generateProposals = internalAction({
+  args: { prompt: v.string() },
+  handler: async (_ctx, _args) => {
+    // imagine this is a call to an LLM
+    return ["proposal1", "proposal2", "proposal3"];
+  },
+});
+
+export const chooseProposal = internalMutation({
+  args: { workflowId: vWorkflowId, choice: v.number() },
+  handler: async (ctx, args) => {
+    await workflow.sendEvent(ctx, {
+      ...approvalEvent,
+      workflowId: args.workflowId,
+      value: { approved: true, choice: args.choice },
+    });
+    return true;
+  },
+});

example/convex/userConfirmation/steps.ts

@@ -3,6 +3,7 @@ import type {
   WorkpoolOptions,
   WorkpoolRetryOptions,
 } from "@convex-dev/workpool";
+import { parse } from "convex-helpers/validators";
 import {
   createFunctionHandle,
   type FunctionArgs,
@@ -25,24 +26,22 @@ import type {
 import type { Step } from "../component/schema.js";
 import type {
   EventId,
-  EventSpec,
   OnCompleteArgs,
   WorkflowId,
   WorkflowStep,
 } from "../types.js";
 import { safeFunctionName } from "./safeFunctionName.js";
-import type { OpaqueIds, WorkflowComponent, WorkflowCtx } from "./types.js";
+import type { OpaqueIds, WorkflowComponent } from "./types.js";
+import type { WorkflowCtx } from "./workflowContext.js";
 import { workflowMutation } from "./workflowMutation.js";
-import { parse } from "convex-helpers/validators";
 
 export {
   vWorkflowId,
-  type WorkflowId,
   vWorkflowStep,
+  type WorkflowId,
   type WorkflowStep,
 } from "../types.js";
-export type { RunOptions, WorkflowCtx } from "./types.js";
-export { defineEvent } from "./events.js";
+export type { RunOptions, WorkflowCtx } from "./workflowContext.js";
 
 export type CallbackOptions = {
   /**
@@ -263,51 +262,93 @@ export class WorkflowManager {
   /**
    * Send an event to a workflow.
    *
-   * @param ctx - Either ctx from a mutation/action or a workflow step.
-   * @param args - The event arguments.
+   * @param ctx - From a mutation, action or workflow step.
+   * @param args - Either send an event by its ID, or by name and workflow ID.
+   *   If you have a validator, you must provide a value.
+   *   If you provide an error string, awaiting the event will throw an error.
    */
   async sendEvent<T = null, Name extends string = string>(
     ctx: RunMutationCtx,
-    {
-      workflowId,
-      event,
-      value,
-    }: (
-      | { workflowId: WorkflowId; event: EventSpec<Name, T> }
-      | { workflowId?: undefined; event: EventSpec<Name, T> & { id: string } }
+    args: (
+      | { workflowId: WorkflowId; name: Name; id?: EventId<Name> }
+      | { workflowId?: undefined; name?: Name; id: EventId<Name> }
     ) &
       (
-        | (T extends null
-            ? { value?: null; error?: undefined }
-            : { value: T; error?: undefined })
+        | { validator?: undefined; value?: T }
+        | { validator: Validator<T, any, any>; value: T }
         | { error: string; value?: undefined }
       ),
   ): Promise<EventId<Name>> {
-    let result = {
-      kind: "success" as const,
-      returnValue: event.validator ? parse(event.validator, value) : value,
-    } satisfies RunResult;
+    let result: RunResult =
+      "error" in args
+        ? {
+            kind: "failed",
+            error: args.error,
+          }
+        : {
+            kind: "success" as const,
+            returnValue: args.validator
+              ? parse(args.validator, args.value)
+              : "value" in args
+                ? args.value
+                : null,
+          };
     return (await ctx.runMutation(this.component.event.send, {
-      eventId: event.id,
+      eventId: args.id,
       result,
-      name: event.name,
-      workflowId: workflowId,
+      name: args.name,
+      workflowId: args.workflowId,
       workpoolOptions: this.options?.workpoolOptions,
     })) as EventId<Name>;
   }
 
+  /**
+   * Create an event ahead of time, enabling awaiting a specific event by ID.
+   * @param ctx - From an action, mutation or workflow step.
+   * @param args - The name of the event and what workflow it belongs to.
+   * @returns The event ID, which can be used to send the event or await it.
+   */
   async createEvent<Name extends string>(
     ctx: RunMutationCtx,
-    component: WorkflowComponent,
     args: { name: Name; workflowId: WorkflowId },
   ): Promise<EventId<Name>> {
-    return (await ctx.runMutation(component.event.create, {
+    return (await ctx.runMutation(this.component.event.create, {
       name: args.name,
       workflowId: args.workflowId,
     })) as EventId<Name>;
   }
 }
 
+/**
+ * Define an event specification: a name and a validator.
+ * This helps share definitions between workflow.sendEvent and ctx.awaitEvent.
+ * e.g.
+ * ```ts
+ * const approvalEvent = defineEvent({
+ *   name: "approval",
+ *   validator: v.object({ approved: v.boolean() }),
+ * });
+ * ```
+ * Then you can await it in a workflow:
+ * ```ts
+ * const result = await ctx.awaitEvent(approvalEvent);
+ * ```
+ * And send from somewhere else:
+ * ```ts
+ * await workflow.sendEvent(ctx, {
+ *   ...approvalEvent,
+ *   workflowId,
+ *   value: { approved: true },
+ * });
+ * ```
+ */
+export function defineEvent<
+  Name extends string,
+  V extends Validator<unknown, "required", string>,
+>(spec: { name: Name; validator: V }) {
+  return spec;
+}
+
 type RunQueryCtx = {
   runQuery: GenericQueryCtx<GenericDataModel>["runQuery"];
 };

example/convex/userConfirmation/workflow.ts

@@ -17,9 +17,9 @@ import {
   journalEntrySize,
   valueSize,
 } from "../component/schema.js";
-import type { SchedulerOptions, WorkflowComponent } from "./types.js";
+import type { WorkflowComponent } from "./types.js";
 import { MAX_JOURNAL_SIZE } from "../shared.js";
-import type { EventId } from "../types.js";
+import type { EventId, SchedulerOptions } from "../types.js";
 
 export type WorkerResult =
   | { type: "handlerDone"; runResult: RunResult }