Orchestrator

Copy codenpm i @akshatgiri/convex-orchestrator

Copy code// convex/convex.config.ts
import { defineApp } from "convex/server";
import convexOrchestrator from "@akshatgiri/convex-orchestrator/convex.config.js";

const app = defineApp();
app.use(convexOrchestrator);

export default app;

Copy code// convex/orchestrator.ts
import { components } from "./_generated/api.js";
import { exposeApi, exposeApiWithWorker } from "@akshatgiri/convex-orchestrator";

// Safe to expose to clients (dashboard + starters)
export const {
  startWorkflow,
  getWorkflow,
  listWorkflows,
  getWorkflowSteps,
  signalWorkflow,
} = exposeApi(components.convexOrchestrator);

// Worker operations (DO NOT expose without auth in production)
export const {
  claimWorkflow,
  heartbeat,
  completeWorkflow,
  failWorkflow,
  getOrCreateStep,
  scheduleSleep,
  waitForSignal,
  completeStep,
  failStep,
  subscribePendingWorkflows,
} = exposeApiWithWorker(components.convexOrchestrator, {
  authorize: async (ctx) => {
    // TODO: implement real auth (service token / identity / secret)
    // Returning true is fine for local dev, unsafe for production.
    void ctx;
    return true;
  },
});

Copy codeimport { workflow } from "@akshatgiri/convex-orchestrator";

const greet = workflow("greet", async (ctx, input: { name: string }) => {
  const greeting = await ctx.step("greeting", async () => `Hello, ${input.name}!`);
  return { greeting };
});

Copy codeimport { ConvexClient } from "convex/browser";
import { createWorker } from "@akshatgiri/convex-orchestrator";
import { api } from "./convex/_generated/api.js";

const client = new ConvexClient(process.env.CONVEX_URL!);

const worker = createWorker(client, api.orchestrator, {
  workflows: [greet],
  maxConcurrentWorkflows: 4,
});
await worker.start();

Copy codeawait client.mutation(api.orchestrator.startWorkflow, {
  name: "greet",
  input: { name: "World" },
});

Copy codeconst reminder = workflow("reminder", async (ctx, input: { email: string }) => {
  await ctx.step("send-welcome", () => activities.sendEmail(input.email, "Welcome!"));

  // Replay-safe: uses the marker to persist sleepUntil durably.
  await ctx.sleep("followup-delay", 24 * 60 * 60 * 1000);

  await ctx.step("send-followup", () => activities.sendEmail(input.email, "How's it going?"));
});

Copy codeconst approval = workflow("approval", async (ctx) => {
  await ctx.step("request-approval", () => activities.sendApprovalEmail());

  const decision = await ctx.waitForSignal<{ approved: boolean }>(
    "approval-decision",
    "approved",
  );

  if (decision.approved) {
    await ctx.step("provision", () => activities.provisionAccess());
  }
});

Copy codeawait client.mutation(api.orchestrator.signalWorkflow, {
  workflowId,
  signal: "approved",
  payload: { approved: true },
});

Copy codenpm i
npm run dev

Copy codebun example/worker.ts
bun example/trigger.ts greet