Durable workflows

import { workflow, step, sleep, waitFor } from "mainsdk/workflows";
import { s } from "mainsdk/schema";

// Your own code. Steps return JSON.
declare function loadUser(userId: string): Promise<{ id: string; email: string }>;
declare function sendWelcome(email: string): Promise<{ sent: boolean }>;
declare function sendNudge(email: string): Promise<{ sent: boolean }>;

export const onboarding = workflow(
  "onboarding",
  { version: 1, input: s.object({ userId: s.string() }) },
  async ({ userId }) => {
    const user = await step("load-user", () => loadUser(userId));        // your code; returns JSON
    await step("welcome", () => sendWelcome(user.email));                // runs once, even after a crash
    await sleep("3d");                                                   // parks the run; holds no worker
    const activated = await waitFor("user.activated", { match: { userId }, timeout: "7d" });
    if (!activated) await step("nudge", () => sendNudge(user.email));
    return { activated: activated !== null };
  },
);

import { workflows, type WorkflowDef } from "mainsdk/workflows";

declare const onboarding: WorkflowDef<{ userId: string }>;   // exported from app/workflows/onboarding.ts
declare const user: { id: string };                          // the person who just signed up

const { runId } = await workflows.start(onboarding, { userId: user.id }, { idempotencyKey: `onboarding:${user.id}` });

workflows:
  driver: db                    # db | file | local
  retries: { max: 3, backoff: exponential, initial: 1s }
  max-step-result: 256KB        # JSON size of one step result and of a run's input
  lease: 30s                    # how long a worker holds a run before another may take it
  keep: 30d                     # finished runs are deleted after this

import { step } from "mainsdk/workflows";

declare function grantCredits(userId: string, credits: number): Promise<{ granted: number }>;
declare const user: { id: string };

await step("grant", () => grantCredits(user.id, 5), { retries: { max: 2 } });

import { approve, step } from "mainsdk/workflows";

declare function grantCredits(userId: string, credits: number): Promise<{ granted: number }>;
declare const user: { id: string };

const ok = await approve({ action: "credits.grant", risk: "write", spend: "$5", reason: "Welcome credits", timeout: "2d" });
if (ok.approved) await step("grant", () => grantCredits(user.id, 5));

import { events } from "mainsdk";

await events.publish("user.activated", { userId: "u_1" });   // resumes the run that waits for this user

import { workflows } from "mainsdk/workflows";

declare const runId: string;                                 // from workflows.start()

const run = await workflows.get(runId);                      // state, currentStep, history, output, error
const waiting = await workflows.list({ name: "onboarding", state: "waiting" });
await workflows.cancel(runId, { reason: "user closed the account" });

import { workflow, workflows } from "mainsdk/workflows";

export const digestV1 = workflow("digest", { version: 1 }, async () => { /* old steps */ });
export const digestV2 = workflow("digest", { version: 2 }, async () => { /* new steps */ });
workflows.register(digestV1);   // `start` registers a definition too
workflows.register(digestV2);

import { describe, it } from "node:test";
import assert from "node:assert/strict";
import { createTestApp } from "@mainsdk/testing";
import { workflow, step, sleep, waitFor } from "mainsdk/workflows";
import { s } from "mainsdk/schema";

const sent: string[] = [];
const onboarding = workflow(
  "onboarding",
  { version: 1, input: s.object({ userId: s.string() }) },
  async ({ userId }) => {
    await step("welcome", () => { sent.push(`welcome:${userId}`); return { sent: true }; });
    await sleep("3d");
    const activated = await waitFor("user.activated", { match: { userId }, timeout: "7d" });
    return { activated: activated !== null };
  },
);

describe("onboarding", () => {
  it("workflows AC1 sends the welcome once, sleeps, then finishes when the user activates", async () => {
    const app = createTestApp();
    const { runId } = await app.workflows.start(onboarding, { userId: "u1" }, { idempotencyKey: "onboarding:u1" });
    await app.clock.tick("1m");
    assert.equal((await app.workflows.get(runId)).state, "sleeping");
    await app.clock.tick("3d");
    assert.equal((await app.workflows.get(runId)).state, "waiting");
    await app.events.publish("user.activated", { userId: "u1" });
    await app.clock.tick("1m");
    const run = await app.workflows.get(runId);
    assert.equal(run.state, "completed");
    assert.deepEqual(run.output, { activated: true });
    assert.deepEqual(sent, ["welcome:u1"]);
  });
});