1# Workflow Patterns
2 
3## Image Processing Pipeline
4 
5```typescript
6export class ImageProcessingWorkflow extends WorkflowEntrypoint<Env, Params> {
7  async run(event, step) {
8    const imageData = await step.do('fetch', async () => (await this.env.BUCKET.get(event.payload.imageKey)).arrayBuffer());
9    const description = await step.do('generate description', async () => 
10      await this.env.AI.run('@cf/llava-hf/llava-1.5-7b-hf', {image: Array.from(new Uint8Array(imageData)), prompt: 'Describe this image', max_tokens: 50})
11    );
12    await step.waitForEvent('await approval', { type: 'approved', timeout: '24h' });
13    await step.do('publish', async () => await this.env.BUCKET.put(`public/${event.payload.imageKey}`, imageData));
14  }
15}
16```
17 
18## User Lifecycle
19 
20```typescript
21export class UserLifecycleWorkflow extends WorkflowEntrypoint<Env, Params> {
22  async run(event, step) {
23    await step.do('welcome email', async () => await sendEmail(event.payload.email, 'Welcome!'));
24    await step.sleep('trial period', '7 days');
25    const hasConverted = await step.do('check conversion', async () => {
26      const user = await this.env.DB.prepare('SELECT subscription_status FROM users WHERE id = ?').bind(event.payload.userId).first();
27      return user.subscription_status === 'active';
28    });
29    if (!hasConverted) await step.do('trial expiration email', async () => await sendEmail(event.payload.email, 'Trial ending'));
30  }
31}
32```
33 
34## Data Pipeline
35 
36```typescript
37export class DataPipelineWorkflow extends WorkflowEntrypoint<Env, Params> {
38  async run(event, step) {
39    const rawData = await step.do('extract', {retries: { limit: 10, delay: '30s', backoff: 'exponential' }}, async () => {
40      const res = await fetch(event.payload.sourceUrl);
41      if (!res.ok) throw new Error('Fetch failed');
42      return res.json();
43    });
44    const transformed = await step.do('transform', async () => 
45      rawData.map(item => ({ id: item.id, normalized: normalizeData(item) }))
46    );
47    const dataRef = await step.do('store', async () => {
48      const key = `processed/${Date.now()}.json`;
49      await this.env.BUCKET.put(key, JSON.stringify(transformed));
50      return { key };
51    });
52    await step.do('load', async () => {
53      const data = await (await this.env.BUCKET.get(dataRef.key)).json();
54      for (let i = 0; i < data.length; i += 100) {
55        await this.env.DB.batch(data.slice(i, i + 100).map(item => 
56          this.env.DB.prepare('INSERT INTO records VALUES (?, ?)').bind(item.id, item.normalized)
57        ));
58      }
59    });
60  }
61}
62```
63 
64## Human-in-the-Loop Approval
65 
66```typescript
67export class ApprovalWorkflow extends WorkflowEntrypoint<Env, Params> {
68  async run(event, step) {
69    await step.do('create approval', async () => await this.env.DB.prepare('INSERT INTO approvals (id, user_id, status) VALUES (?, ?, ?)').bind(event.instanceId, event.payload.userId, 'pending').run());
70    try {
71      const approval = await step.waitForEvent<{ approved: boolean }>('wait for approval', { type: 'approval-response', timeout: '48h' });
72      if (approval.approved) { await step.do('process approval', async () => {}); } 
73      else { await step.do('handle rejection', async () => {}); }
74    } catch (e) {
75      await step.do('auto reject', async () => await this.env.DB.prepare('UPDATE approvals SET status = ? WHERE id = ?').bind('auto-rejected', event.instanceId).run());
76    }
77  }
78}
79```
80 
81## Testing Workflows
82 
83### Setup
84 
85```typescript
86// vitest.config.ts
87import { defineWorkersConfig } from '@cloudflare/vitest-pool-workers/config';
88 
89export default defineWorkersConfig({
90  test: {
91    poolOptions: {
92      workers: {
93        wrangler: { configPath: './wrangler.jsonc' }
94      }
95    }
96  }
97});
98```
99 
100### Introspection API
101 
102```typescript
103import { introspectWorkflowInstance } from 'cloudflare:test';
104 
105const instance = await env.MY_WORKFLOW.create({ params: { userId: '123' } });
106const introspector = await introspectWorkflowInstance(env.MY_WORKFLOW, instance.id);
107 
108// Wait for step completion
109const result = await introspector.waitForStepResult({ name: 'fetch user', index: 0 });
110 
111// Mock step behavior
112await introspector.modify(async (m) => {
113  await m.mockStepResult({ name: 'api call' }, { mocked: true });
114});
115```
116 
117## Best Practices
118 
119### ✅ DO
120 
1211. **Granular steps**: One API call per step (unless proving idempotency)
1222. **Idempotency**: Check-then-execute; use idempotency keys
1233. **Deterministic names**: Use static or step-output-based names
1244. **Return state**: Persist via step returns, not variables
1255. **Always await**: `await step.do()`, avoid dangling promises
1266. **Deterministic conditionals**: Base on `event.payload` or step outputs
1277. **Store large data externally**: R2/KV for data exceeding step return limit, return refs
1288. **Batch creation**: `createBatch()` for multiple instances
129 
130### ❌ DON'T
131 
1321. **One giant step**: Breaks durability & retry control
1332. **State outside steps**: Lost on hibernation
1343. **Mutate events**: Events immutable, return new state
1354. **Non-deterministic logic outside steps**: `Math.random()`, `Date.now()` must be in steps
1365. **Side effects outside steps**: May duplicate on restart
1376. **Non-deterministic step names**: Prevents caching
1387. **Ignore timeouts**: `waitForEvent` throws, use try-catch
1398. **Reuse instance IDs**: Must be unique within retention
140 
141## Orchestration Patterns
142 
143### Fan-Out (Parallel Processing)
144```typescript
145const files = await step.do('list', async () => this.env.BUCKET.list());
146await Promise.all(files.objects.map((file, i) => step.do(`process ${i}`, async () => processFile(await (await this.env.BUCKET.get(file.key)).arrayBuffer()))));
147```
148 
149### Parent-Child Workflows
150```typescript
151const child = await step.do('start child', async () => await this.env.CHILD_WORKFLOW.create({id: `child-${event.instanceId}`, params: { data: result.data }}));
152await step.do('other work', async () => console.log(`Child started: ${child.id}`));
153```
154 
155### Race Pattern
156```typescript
157const winner = await Promise.race([
158  step.do('option A', async () => slowOperation()),
159  step.do('option B', async () => fastOperation())
160]);
161```
162 
163### Scheduled Workflow Chain
164```typescript
165export default { async scheduled(event, env) { await env.DAILY_WORKFLOW.create({id: `daily-${event.scheduledTime}`, params: { timestamp: event.scheduledTime }}); }};
166export class DailyWorkflow extends WorkflowEntrypoint<Env, Params> {
167  async run(event, step) {
168    await step.do('daily task', async () => {});
169    await step.sleep('wait 7 days', '7 days');
170    await step.do('weekly followup', async () => {});
171  }
172}
173```
174 
175See: [configuration.md](./configuration.md), [api.md](./api.md), [gotchas.md](./gotchas.md)
176