Skip to content

Commit 0f68f3e

Browse files
NathanFlurryNathanFlurry
committed
docs(workflows): own embedded examples
1 parent 7f27dc1 commit 0f68f3e

45 files changed

Lines changed: 1690 additions & 6 deletions

Some content is hidden

Large Commits have some content hidden by default. Use the searchbox below for content that may be hidden.

‎.github/workflows/docs-sync.yml‎

Lines changed: 4 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -4,13 +4,12 @@ name: Sync docs
44
# docs/ bundle into the website's vendor/workflows and opens a PR there, which is
55
# also what triggers the site rebuild.
66
#
7-
# No extra-paths: this product owns no examples/ tree. Its <CodeSnippet> paths
8-
# resolve against another product's repo via snippetFrom in the website's
9-
# product-metadata.ts, so nothing outside docs/ needs to travel with the bundle.
7+
# examples/ travels with the bundle because Workflows docs embed snippets from
8+
# this repository by repo-root-relative path.
109
on:
1110
push:
1211
branches: [main]
13-
paths: ["docs/**"]
12+
paths: ["docs/**", "examples/**", ".github/workflows/docs-sync.yml"]
1413
workflow_dispatch:
1514

1615
jobs:
@@ -22,3 +21,4 @@ jobs:
2221
with:
2322
product: workflows
2423
token: ${{ secrets.RIVET_WEBSITE_TOKEN }}
24+
extra-paths: examples

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

Lines changed: 48 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,48 @@
1+
import { queue, setup, workflow } from "@rivet-dev/workflows";
2+
export const approvalGateActor = workflow({
3+
state: { status: "pending" as string },
4+
queues: {
5+
approval: queue<{
6+
approved: boolean;
7+
}>(),
8+
},
9+
run: async (ctx) => {
10+
await ctx.step("validate-order", async (step) => {
11+
await validateOrder("order-123");
12+
step.state.status = "awaiting_approval";
13+
});
14+
const decision = await ctx.queue.next("wait-approval");
15+
if (decision.body.approved) {
16+
await ctx.step("fulfill-order", async (step) => {
17+
await fulfillOrder("order-123");
18+
step.state.status = "fulfilled";
19+
});
20+
} else {
21+
await ctx.step("cancel-order", async (step) => {
22+
await cancelOrder("order-123");
23+
step.state.status = "cancelled";
24+
});
25+
}
26+
},
27+
actions: {
28+
getState: (c) => c.state,
29+
},
30+
});
31+
async function validateOrder(orderId: string): Promise<void> {
32+
const res = await fetch(
33+
`https://api.example.com/orders/${orderId}/validate`,
34+
{ method: "POST" },
35+
);
36+
if (!res.ok) throw new Error("Order validation failed");
37+
}
38+
async function fulfillOrder(orderId: string): Promise<void> {
39+
await fetch(`https://api.example.com/orders/${orderId}/fulfill`, {
40+
method: "POST",
41+
});
42+
}
43+
async function cancelOrder(orderId: string): Promise<void> {
44+
await fetch(`https://api.example.com/orders/${orderId}/cancel`, {
45+
method: "POST",
46+
});
47+
}
48+
export const registry = setup({ use: { approvalGateActor } });

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

Lines changed: 49 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,49 @@
1+
import {
2+
setup,
3+
type WorkflowStepContextOf,
4+
workflow,
5+
} from "@rivet-dev/workflows";
6+
7+
type MetricMessage = {
8+
value: number;
9+
};
10+
export const batchDrainerActor = workflow({
11+
state: {
12+
pending: [] as number[],
13+
flushedBatches: 0,
14+
lastBatchTotal: 0,
15+
},
16+
run: async (ctx) => {
17+
await ctx.loop("drain-loop", async (loopCtx) => {
18+
const [message] = await loopCtx.queue.nextBatch("wait-metric", {
19+
timeout: 5000,
20+
});
21+
const pendingCount = await loopCtx.step(
22+
"buffer-message",
23+
async (step) => {
24+
if (message) {
25+
step.state.pending.push((message.body as MetricMessage).value);
26+
}
27+
return step.state.pending.length;
28+
},
29+
);
30+
if (pendingCount < 5) return;
31+
await loopCtx.step("flush-batch", async (step) => flushBatch(step));
32+
});
33+
},
34+
actions: {
35+
getState: (c) => c.state,
36+
},
37+
});
38+
function flushBatch(
39+
ctx: WorkflowStepContextOf<typeof batchDrainerActor>,
40+
): void {
41+
const total = ctx.state.pending.reduce(
42+
(sum: number, value: number) => sum + value,
43+
0,
44+
);
45+
ctx.state.lastBatchTotal = total;
46+
ctx.state.flushedBatches += 1;
47+
ctx.state.pending = [];
48+
}
49+
export const registry = setup({ use: { batchDrainerActor } });

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

Lines changed: 70 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,70 @@
1+
import {
2+
setup,
3+
type WorkflowStepContextOf,
4+
workflow,
5+
} from "@rivet-dev/workflows";
6+
7+
type WorkMessage = {
8+
id: string;
9+
value: number;
10+
};
11+
const MAX_PER_ITERATION = 10;
12+
const CONCURRENCY_LIMIT = 3;
13+
async function processWork(value: number): Promise<number> {
14+
return value * 2;
15+
}
16+
async function runWithLimit<T>(
17+
limit: number,
18+
items: T[],
19+
fn: (item: T) => Promise<void>,
20+
): Promise<void> {
21+
let nextIndex = 0;
22+
const workers = Array.from({ length: limit }, async () => {
23+
while (nextIndex < items.length) {
24+
const current = items[nextIndex];
25+
nextIndex += 1;
26+
await fn(current);
27+
}
28+
});
29+
await Promise.all(workers);
30+
}
31+
export const boundedDrainActor = workflow({
32+
state: {
33+
processed: 0,
34+
lastWindowSize: 0,
35+
lastWindowTotal: 0,
36+
},
37+
run: async (ctx) => {
38+
await ctx.loop("bounded-drain-loop", async (loopCtx) => {
39+
const window: WorkMessage[] = [];
40+
for (let i = 0; i < MAX_PER_ITERATION; i += 1) {
41+
const [message] = await loopCtx.queue.nextBatch("wait-work", {
42+
timeout: i === 0 ? 30000 : 10,
43+
});
44+
if (!message) break;
45+
window.push(message.body as WorkMessage);
46+
}
47+
if (window.length === 0) return;
48+
await loopCtx.step("process-window", async (step) =>
49+
processWindow(step, window),
50+
);
51+
});
52+
},
53+
actions: {
54+
getState: (c) => c.state,
55+
},
56+
});
57+
async function processWindow(
58+
ctx: WorkflowStepContextOf<typeof boundedDrainActor>,
59+
window: WorkMessage[],
60+
): Promise<void> {
61+
let windowTotal = 0;
62+
await runWithLimit(CONCURRENCY_LIMIT, window, async (work) => {
63+
const result = await processWork(work.value);
64+
windowTotal += result;
65+
});
66+
ctx.state.processed += window.length;
67+
ctx.state.lastWindowSize = window.length;
68+
ctx.state.lastWindowTotal = windowTotal;
69+
}
70+
export const registry = setup({ use: { boundedDrainActor } });

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

Lines changed: 47 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,47 @@
1+
import { setup, workflow } from "@rivet-dev/workflows";
2+
3+
type PaymentMessage = {
4+
id: string;
5+
amount: number;
6+
};
7+
export const checkpointFriendlyActor = workflow({
8+
state: {
9+
appliedCount: 0,
10+
totalAmount: 0,
11+
lastPaymentId: null as string | null,
12+
},
13+
run: async (ctx) => {
14+
await ctx.loop("payment-loop", async (loopCtx) => {
15+
const [message] = await loopCtx.queue.nextBatch("wait-payment", {
16+
timeout: 30000,
17+
});
18+
if (!message) return;
19+
const payment = message.body as PaymentMessage;
20+
await loopCtx.rollbackCheckpoint("apply-payment-checkpoint");
21+
const plan = (await loopCtx.step("build-plan", async (_loopCtx) =>
22+
buildPaymentPlan(payment),
23+
)) as {
24+
paymentId: string;
25+
amount: number;
26+
};
27+
await loopCtx.step("apply-side-effects", async (step) => {
28+
step.state.appliedCount += 1;
29+
step.state.totalAmount += plan.amount;
30+
step.state.lastPaymentId = plan.paymentId;
31+
});
32+
});
33+
},
34+
actions: {
35+
getState: (c) => c.state,
36+
},
37+
});
38+
function buildPaymentPlan(payment: PaymentMessage): {
39+
paymentId: string;
40+
amount: number;
41+
} {
42+
return {
43+
paymentId: payment.id,
44+
amount: payment.amount,
45+
};
46+
}
47+
export const registry = setup({ use: { checkpointFriendlyActor } });

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

Lines changed: 73 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,73 @@
1+
import {
2+
actor,
3+
setup,
4+
type WorkflowStepContextOf,
5+
workflow,
6+
} from "@rivet-dev/workflows";
7+
8+
type BatchMessage = {
9+
payload: number;
10+
};
11+
export const childWorkerActor = actor({
12+
actions: {
13+
process: async (_c, payload: number) => payload * 3,
14+
},
15+
});
16+
export const orchestratorActor = workflow({
17+
state: {
18+
lastTotal: 0,
19+
},
20+
run: async (ctx) => {
21+
await ctx.step("start-children", async (step) => startChildren(step));
22+
await ctx.loop("orchestrate-loop", async (loopCtx) => {
23+
const [message] = await loopCtx.queue.nextBatch("wait-batch", {
24+
timeout: 30000,
25+
});
26+
if (!message) return;
27+
const batch = message.body as BatchMessage;
28+
const results = await loopCtx.join("collect-updates", {
29+
a: {
30+
run: async (joinCtx) =>
31+
await joinCtx.step("run-child-a", async (step) =>
32+
runChildWorker(step, "child-a", batch.payload),
33+
),
34+
},
35+
b: {
36+
run: async (joinCtx) =>
37+
await joinCtx.step("run-child-b", async (step) =>
38+
runChildWorker(step, "child-b", batch.payload),
39+
),
40+
},
41+
c: {
42+
run: async (joinCtx) =>
43+
await joinCtx.step("run-child-c", async (step) =>
44+
runChildWorker(step, "child-c", batch.payload),
45+
),
46+
},
47+
});
48+
await loopCtx.step("reconcile", async (step) => {
49+
step.state.lastTotal = results.a + results.b + results.c;
50+
});
51+
});
52+
},
53+
actions: {
54+
getState: (c) => c.state,
55+
},
56+
});
57+
async function startChildren(
58+
ctx: WorkflowStepContextOf<typeof orchestratorActor>,
59+
): Promise<void> {
60+
const client = ctx.client();
61+
await client.childWorkerActor.getOrCreate(["child-a"]).process(0);
62+
await client.childWorkerActor.getOrCreate(["child-b"]).process(0);
63+
await client.childWorkerActor.getOrCreate(["child-c"]).process(0);
64+
}
65+
async function runChildWorker(
66+
ctx: WorkflowStepContextOf<typeof orchestratorActor>,
67+
workerId: "child-a" | "child-b" | "child-c",
68+
payload: number,
69+
): Promise<number> {
70+
const client = ctx.client();
71+
return await client.childWorkerActor.getOrCreate([workerId]).process(payload);
72+
}
73+
export const registry = setup({ use: { orchestratorActor, childWorkerActor } });

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

Lines changed: 51 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,51 @@
1+
import {
2+
actor,
3+
setup,
4+
type WorkflowStepContextOf,
5+
workflow,
6+
} from "@rivet-dev/workflows";
7+
8+
type TaskMessage = {
9+
taskId: string;
10+
workerId: string;
11+
value: number;
12+
};
13+
export const workerActor = actor({
14+
actions: {
15+
runTask: async (_c, value: number) => value * 2,
16+
},
17+
});
18+
export const coordinatorActor = workflow({
19+
state: {
20+
lastTaskId: null as string | null,
21+
lastResult: 0,
22+
},
23+
run: async (ctx) => {
24+
await ctx.loop("orchestrator-loop", async (loopCtx) => {
25+
const [message] = await loopCtx.queue.nextBatch("wait-task", {
26+
timeout: 30000,
27+
});
28+
if (!message) return;
29+
const task = message.body as TaskMessage;
30+
const result = await loopCtx.step("dispatch-rpc", async (step) =>
31+
dispatchTask(step, task),
32+
);
33+
await loopCtx.step("record-result", async (step) => {
34+
step.state.lastTaskId = task.taskId;
35+
step.state.lastResult = result as number;
36+
});
37+
});
38+
},
39+
actions: {
40+
getState: (c) => c.state,
41+
},
42+
});
43+
async function dispatchTask(
44+
ctx: WorkflowStepContextOf<typeof coordinatorActor>,
45+
task: TaskMessage,
46+
): Promise<number> {
47+
const client = ctx.client();
48+
const worker = client.workerActor.getOrCreate([task.workerId]);
49+
return await worker.runTask(task.value);
50+
}
51+
export const registry = setup({ use: { coordinatorActor, workerActor } });

0 commit comments

Comments
 (0)