Concepts

Patterns

Apply production workflow patterns for progress tracking, queue-driven cron jobs, orchestration, and reliable multi-step operations.

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 {
	event,
	queue,
	setup,
	type WorkflowStepContextOf,
	workflow,
} from "@rivet-dev/workflows";

type Progress = {
	stage: "idle" | "running" | "completed";
	completed: number;
	total: number;
};
export const progressActor = workflow({
	state: {
		progress: {
			stage: "idle",
			completed: 0,
			total: 0,
		} as Progress,
		sum: 0,
	},
	events: {
		progressUpdated: event<Progress>(),
	},
	queues: {
		jobs: queue<{
			value: number;
		}>(),
	},
	run: 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 {
	queue,
	type ScheduledFireInfo,
	setup,
	workflow,
} from "@rivet-dev/workflows";
export const cronActor = workflow({
	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: 60000,
			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: 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 { setup, workflow } from "@rivet-dev/workflows";

function openResource(): string {
	return "connected";
}
function closeResource(_resource: string): void {}
export const setupRunTeardownActor = workflow({
	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: 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", 1000);
			await loopCtx.step("tick-step", async (step) => {
				step.state.ticks += 1;
			});
		});
	},
	actions: {
		getState: (c) => c.state,
	},
});
export const registry = setup({ use: { setupRunTeardownActor } });