Concepts

Patterns

Common workflow shapes used in production systems.

These are common workflow shapes used in production systems.

Store workflow progress in state + broadcast

Store progress in state so replay and recovery always restore it. Broadcast state changes so clients can render progress in realtime.

import { actor, event, queue, setup } from "rivetkit";
import { type WorkflowStepContextOf, workflow } from "rivetkit/workflow";

type Progress = {
  stage: "idle" | "running" | "completed";
  completed: number;
  total: number;
};

export const progressActor = actor({
  state: {
    progress: {
      stage: "idle",
      completed: 0,
      total: 0,
    } as Progress,
    sum: 0,
  },
  events: {
    progressUpdated: event<Progress>(),
  },
  queues: {
    jobs: queue<{ value: number }>(),
  },
  run: workflow(async (ctx) => {
    await ctx.loop("progress-loop", async (loopCtx) => {
        const message = await loopCtx.queue.next("wait-job");

        await loopCtx.step("mark-running", async (step) =>
          markProgressRunning(step),
        );

        await loopCtx.step("apply-job", async (step) =>
          applyProgressJob(step, message.body.value),
        );

      });
  }),
  actions: {
    getState: (c) => c.state,
  },
});

function markProgressRunning(ctx: WorkflowStepContextOf<typeof progressActor>): void {
  ctx.state.progress = {
    stage: "running",
    completed: ctx.state.progress.completed,
    total: ctx.state.progress.total + 1,
  };
  ctx.broadcast("progressUpdated", ctx.state.progress);
}

function applyProgressJob(
  ctx: WorkflowStepContextOf<typeof progressActor>,
  value: number,
): void {
  ctx.state.sum += value;
  ctx.state.progress = {
    stage: "completed",
    completed: ctx.state.progress.completed + 1,
    total: ctx.state.progress.total,
  };
  ctx.broadcast("progressUpdated", ctx.state.progress);
}

export const registry = setup({ use: { progressActor } });

import { createClient } from "rivetkit/client";
import type { registry } from "./index";

const client = createClient<typeof registry>("http://localhost:6420");
const handle = client.progressActor.getOrCreate(["main"]);
const conn = handle.connect();

conn.on("progressUpdated", (progress) => {
  console.log("progress", progress);
});

await handle.send("jobs", { value: 5 });
await handle.send("jobs", { value: 7 });

console.log(await handle.getState());

Cron (queue-driven)

Rivet scheduling triggers actions. For cron-like workflows, use a small scheduled action as a bridge that enqueues work, then process that work in the workflow loop.

import { actor, queue, setup, type ScheduledFireInfo } from "rivetkit";
import { type WorkflowStepContextOf, workflow } from "rivetkit/workflow";

export const cronActor = actor({
  state: {
    runs: 0,
    lastRunAt: null as number | null,
  },
  queues: {
    "cron-tick": queue<{ scheduledAt: number }>(),
  },
  onCreate: async (c) => {
    await c.cron.every({
      name: "workflow-tick",
      interval: 60_000,
      action: "enqueueCronTick",
      args: [],
      maxHistory: 100,
    });
  },
  actions: {
    enqueueCronTick: async (c, fire: ScheduledFireInfo) => {
      await c.queue.send("cron-tick", { scheduledAt: fire.scheduledAt });
    },
    getState: (c) => c.state,
  },
  run: workflow(async (ctx) => {
    await ctx.loop("cron-loop", async (loopCtx) => {
        const message = await loopCtx.queue.next("wait-cron-tick");

        await loopCtx.step("run-cron-job", async (step) => {
          step.state.runs += 1;
          step.state.lastRunAt = message.body.scheduledAt;
        });

      });
  }),
});

export const registry = setup({ use: { cronActor } });

Setup & teardown

Use this when you need one-time initialization before a long-lived loop, plus cleanup when the actor stops sleeping or is destroyed.

import { actor, setup } from "rivetkit";
import { type WorkflowStepContextOf, workflow } from "rivetkit/workflow";

function openResource(): string {
  return "connected";
}

function closeResource(_resource: string): void {}

export const setupRunTeardownActor = actor({
  vars: {
    resource: null as string | null,
  },
  state: {
    initialized: false,
    ticks: 0,
  },
  onWake: (c) => {
    c.vars.resource = openResource();
  },
  onSleep: (c) => {
    if (!c.vars.resource) return;
    closeResource(c.vars.resource);
    c.vars.resource = null;
  },
  run: workflow(async (ctx) => {
    await ctx.step("setup", async (step) => {
      if (!step.vars.resource) step.vars.resource = openResource();
      step.state.initialized = true;
    });

    await ctx.loop("main-loop", async (loopCtx) => {
        await loopCtx.sleep("tick", 1_000);
        await loopCtx.step("tick-step", async (step) => {
          step.state.ticks += 1;
        });
      });
  }),
  actions: {
    getState: (c) => c.state,
  },
});

export const registry = setup({ use: { setupRunTeardownActor } });