Own embedded workflow examples

.github/workflows/docs-sync.yml

@@ -4,13 +4,12 @@ name: Sync docs
 # docs/ bundle into the website's vendor/workflows and opens a PR there, which is
 # also what triggers the site rebuild.
 #
-# No extra-paths: this product owns no examples/ tree. Its <CodeSnippet> paths
-# resolve against another product's repo via snippetFrom in the website's
-# product-metadata.ts, so nothing outside docs/ needs to travel with the bundle.
+# examples/ travels with the bundle because Workflows docs embed snippets from
+# this repository by repo-root-relative path.
 on:
   push:
     branches: [main]
-    paths: ["docs/**"]
+    paths: ["docs/**", "examples/**", ".github/workflows/docs-sync.yml"]
   workflow_dispatch:
 
 jobs:
@@ -22,3 +21,4 @@ jobs:
         with:
           product: workflows
           token: ${{ secrets.RIVET_WEBSITE_TOKEN }}
+          extra-paths: examples

examples/docs/actors-workflows/approval-gate.ts

@@ -0,0 +1,48 @@
+import { queue, setup, workflow } from "@rivet-dev/workflows";
+export const approvalGateActor = workflow({
+	state: { status: "pending" as string },
+	queues: {
+		approval: queue<{
+			approved: boolean;
+		}>(),
+	},
+	run: async (ctx) => {
+		await ctx.step("validate-order", async (step) => {
+			await validateOrder("order-123");
+			step.state.status = "awaiting_approval";
+		});
+		const decision = await ctx.queue.next("wait-approval");
+		if (decision.body.approved) {
+			await ctx.step("fulfill-order", async (step) => {
+				await fulfillOrder("order-123");
+				step.state.status = "fulfilled";
+			});
+		} else {
+			await ctx.step("cancel-order", async (step) => {
+				await cancelOrder("order-123");
+				step.state.status = "cancelled";
+			});
+		}
+	},
+	actions: {
+		getState: (c) => c.state,
+	},
+});
+async function validateOrder(orderId: string): Promise<void> {
+	const res = await fetch(
+		`https://api.example.com/orders/${orderId}/validate`,
+		{ method: "POST" },
+	);
+	if (!res.ok) throw new Error("Order validation failed");
+}
+async function fulfillOrder(orderId: string): Promise<void> {
+	await fetch(`https://api.example.com/orders/${orderId}/fulfill`, {
+		method: "POST",
+	});
+}
+async function cancelOrder(orderId: string): Promise<void> {
+	await fetch(`https://api.example.com/orders/${orderId}/cancel`, {
+		method: "POST",
+	});
+}
+export const registry = setup({ use: { approvalGateActor } });

examples/docs/actors-workflows/batch-drainer.ts

@@ -0,0 +1,49 @@
+import {
+	setup,
+	type WorkflowStepContextOf,
+	workflow,
+} from "@rivet-dev/workflows";
+
+type MetricMessage = {
+	value: number;
+};
+export const batchDrainerActor = workflow({
+	state: {
+		pending: [] as number[],
+		flushedBatches: 0,
+		lastBatchTotal: 0,
+	},
+	run: async (ctx) => {
+		await ctx.loop("drain-loop", async (loopCtx) => {
+			const [message] = await loopCtx.queue.nextBatch("wait-metric", {
+				timeout: 5000,
+			});
+			const pendingCount = await loopCtx.step(
+				"buffer-message",
+				async (step) => {
+					if (message) {
+						step.state.pending.push((message.body as MetricMessage).value);
+					}
+					return step.state.pending.length;
+				},
+			);
+			if (pendingCount < 5) return;
+			await loopCtx.step("flush-batch", async (step) => flushBatch(step));
+		});
+	},
+	actions: {
+		getState: (c) => c.state,
+	},
+});
+function flushBatch(
+	ctx: WorkflowStepContextOf<typeof batchDrainerActor>,
+): void {
+	const total = ctx.state.pending.reduce(
+		(sum: number, value: number) => sum + value,
+		0,
+	);
+	ctx.state.lastBatchTotal = total;
+	ctx.state.flushedBatches += 1;
+	ctx.state.pending = [];
+}
+export const registry = setup({ use: { batchDrainerActor } });

examples/docs/actors-workflows/bounded-drain.ts

@@ -0,0 +1,70 @@
+import {
+	setup,
+	type WorkflowStepContextOf,
+	workflow,
+} from "@rivet-dev/workflows";
+
+type WorkMessage = {
+	id: string;
+	value: number;
+};
+const MAX_PER_ITERATION = 10;
+const CONCURRENCY_LIMIT = 3;
+async function processWork(value: number): Promise<number> {
+	return value * 2;
+}
+async function runWithLimit<T>(
+	limit: number,
+	items: T[],
+	fn: (item: T) => Promise<void>,
+): Promise<void> {
+	let nextIndex = 0;
+	const workers = Array.from({ length: limit }, async () => {
+		while (nextIndex < items.length) {
+			const current = items[nextIndex];
+			nextIndex += 1;
+			await fn(current);
+		}
+	});
+	await Promise.all(workers);
+}
+export const boundedDrainActor = workflow({
+	state: {
+		processed: 0,
+		lastWindowSize: 0,
+		lastWindowTotal: 0,
+	},
+	run: async (ctx) => {
+		await ctx.loop("bounded-drain-loop", async (loopCtx) => {
+			const window: WorkMessage[] = [];
+			for (let i = 0; i < MAX_PER_ITERATION; i += 1) {
+				const [message] = await loopCtx.queue.nextBatch("wait-work", {
+					timeout: i === 0 ? 30000 : 10,
+				});
+				if (!message) break;
+				window.push(message.body as WorkMessage);
+			}
+			if (window.length === 0) return;
+			await loopCtx.step("process-window", async (step) =>
+				processWindow(step, window),
+			);
+		});
+	},
+	actions: {
+		getState: (c) => c.state,
+	},
+});
+async function processWindow(
+	ctx: WorkflowStepContextOf<typeof boundedDrainActor>,
+	window: WorkMessage[],
+): Promise<void> {
+	let windowTotal = 0;
+	await runWithLimit(CONCURRENCY_LIMIT, window, async (work) => {
+		const result = await processWork(work.value);
+		windowTotal += result;
+	});
+	ctx.state.processed += window.length;
+	ctx.state.lastWindowSize = window.length;
+	ctx.state.lastWindowTotal = windowTotal;
+}
+export const registry = setup({ use: { boundedDrainActor } });

examples/docs/actors-workflows/checkpoint-friendly.ts

@@ -0,0 +1,47 @@
+import { setup, workflow } from "@rivet-dev/workflows";
+
+type PaymentMessage = {
+	id: string;
+	amount: number;
+};
+export const checkpointFriendlyActor = workflow({
+	state: {
+		appliedCount: 0,
+		totalAmount: 0,
+		lastPaymentId: null as string | null,
+	},
+	run: async (ctx) => {
+		await ctx.loop("payment-loop", async (loopCtx) => {
+			const [message] = await loopCtx.queue.nextBatch("wait-payment", {
+				timeout: 30000,
+			});
+			if (!message) return;
+			const payment = message.body as PaymentMessage;
+			await loopCtx.rollbackCheckpoint("apply-payment-checkpoint");
+			const plan = (await loopCtx.step("build-plan", async (_loopCtx) =>
+				buildPaymentPlan(payment),
+			)) as {
+				paymentId: string;
+				amount: number;
+			};
+			await loopCtx.step("apply-side-effects", async (step) => {
+				step.state.appliedCount += 1;
+				step.state.totalAmount += plan.amount;
+				step.state.lastPaymentId = plan.paymentId;
+			});
+		});
+	},
+	actions: {
+		getState: (c) => c.state,
+	},
+});
+function buildPaymentPlan(payment: PaymentMessage): {
+	paymentId: string;
+	amount: number;
+} {
+	return {
+		paymentId: payment.id,
+		amount: payment.amount,
+	};
+}
+export const registry = setup({ use: { checkpointFriendlyActor } });

examples/docs/actors-workflows/child-worker-orchestration.ts

@@ -0,0 +1,73 @@
+import {
+	actor,
+	setup,
+	type WorkflowStepContextOf,
+	workflow,
+} from "@rivet-dev/workflows";
+
+type BatchMessage = {
+	payload: number;
+};
+export const childWorkerActor = actor({
+	actions: {
+		process: async (_c, payload: number) => payload * 3,
+	},
+});
+export const orchestratorActor = workflow({
+	state: {
+		lastTotal: 0,
+	},
+	run: async (ctx) => {
+		await ctx.step("start-children", async (step) => startChildren(step));
+		await ctx.loop("orchestrate-loop", async (loopCtx) => {
+			const [message] = await loopCtx.queue.nextBatch("wait-batch", {
+				timeout: 30000,
+			});
+			if (!message) return;
+			const batch = message.body as BatchMessage;
+			const results = await loopCtx.join("collect-updates", {
+				a: {
+					run: async (joinCtx) =>
+						await joinCtx.step("run-child-a", async (step) =>
+							runChildWorker(step, "child-a", batch.payload),
+						),
+				},
+				b: {
+					run: async (joinCtx) =>
+						await joinCtx.step("run-child-b", async (step) =>
+							runChildWorker(step, "child-b", batch.payload),
+						),
+				},
+				c: {
+					run: async (joinCtx) =>
+						await joinCtx.step("run-child-c", async (step) =>
+							runChildWorker(step, "child-c", batch.payload),
+						),
+				},
+			});
+			await loopCtx.step("reconcile", async (step) => {
+				step.state.lastTotal = results.a + results.b + results.c;
+			});
+		});
+	},
+	actions: {
+		getState: (c) => c.state,
+	},
+});
+async function startChildren(
+	ctx: WorkflowStepContextOf<typeof orchestratorActor>,
+): Promise<void> {
+	const client = ctx.client();
+	await client.childWorkerActor.getOrCreate(["child-a"]).process(0);
+	await client.childWorkerActor.getOrCreate(["child-b"]).process(0);
+	await client.childWorkerActor.getOrCreate(["child-c"]).process(0);
+}
+async function runChildWorker(
+	ctx: WorkflowStepContextOf<typeof orchestratorActor>,
+	workerId: "child-a" | "child-b" | "child-c",
+	payload: number,
+): Promise<number> {
+	const client = ctx.client();
+	return await client.childWorkerActor.getOrCreate([workerId]).process(payload);
+}
+export const registry = setup({ use: { orchestratorActor, childWorkerActor } });

examples/docs/actors-workflows/coordinator-worker.ts

@@ -0,0 +1,51 @@
+import {
+	actor,
+	setup,
+	type WorkflowStepContextOf,
+	workflow,
+} from "@rivet-dev/workflows";
+
+type TaskMessage = {
+	taskId: string;
+	workerId: string;
+	value: number;
+};
+export const workerActor = actor({
+	actions: {
+		runTask: async (_c, value: number) => value * 2,
+	},
+});
+export const coordinatorActor = workflow({
+	state: {
+		lastTaskId: null as string | null,
+		lastResult: 0,
+	},
+	run: async (ctx) => {
+		await ctx.loop("orchestrator-loop", async (loopCtx) => {
+			const [message] = await loopCtx.queue.nextBatch("wait-task", {
+				timeout: 30000,
+			});
+			if (!message) return;
+			const task = message.body as TaskMessage;
+			const result = await loopCtx.step("dispatch-rpc", async (step) =>
+				dispatchTask(step, task),
+			);
+			await loopCtx.step("record-result", async (step) => {
+				step.state.lastTaskId = task.taskId;
+				step.state.lastResult = result as number;
+			});
+		});
+	},
+	actions: {
+		getState: (c) => c.state,
+	},
+});
+async function dispatchTask(
+	ctx: WorkflowStepContextOf<typeof coordinatorActor>,
+	task: TaskMessage,
+): Promise<number> {
+	const client = ctx.client();
+	const worker = client.workerActor.getOrCreate([task.workerId]);
+	return await worker.runTask(task.value);
+}
+export const registry = setup({ use: { coordinatorActor, workerActor } });