1# Pipelines Patterns
2 
3## Fire-and-Forget
4 
5```typescript
6export default {
7  async fetch(request, env, ctx) {
8    const event = { user_id: '...', event_type: 'page_view', timestamp: new Date().toISOString() };
9    ctx.waitUntil(env.STREAM.send([event])); // Don't block response
10    return new Response('OK');
11  }
12};
13```
14 
15## Schema Validation with Zod
16 
17```typescript
18import { z } from 'zod';
19 
20const EventSchema = z.object({
21  user_id: z.string(),
22  event_type: z.enum(['purchase', 'view']),
23  amount: z.number().positive().optional()
24});
25 
26const validated = EventSchema.parse(rawEvent); // Throws on invalid
27await env.STREAM.send([validated]);
28```
29 
30**Why:** Structured streams drop invalid events silently. Client validation gives immediate feedback.
31 
32## SQL Transform Patterns
33 
34```sql
35-- Filter early (reduce storage)
36INSERT INTO my_sink
37SELECT user_id, event_type, amount
38FROM my_stream
39WHERE event_type = 'purchase' AND amount > 10
40 
41-- Select only needed fields
42INSERT INTO my_sink
43SELECT user_id, event_type, timestamp FROM my_stream
44 
45-- Enrich with CASE
46INSERT INTO my_sink
47SELECT user_id, amount,
48  CASE WHEN amount > 1000 THEN 'vip' ELSE 'standard' END as tier
49FROM my_stream
50```
51 
52## Pipelines + Queues Fan-out
53 
54```typescript
55await Promise.all([
56  env.ANALYTICS_STREAM.send([event]),  // Long-term storage
57  env.PROCESS_QUEUE.send(event)        // Immediate processing
58]);
59```
60 
61| Need | Use |
62|------|-----|
63| Long-term storage, SQL queries | Pipelines |
64| Immediate processing, retries | Queues |
65| Both | Fan-out pattern |
66 
67## Performance Tuning
68 
69| Goal | Config |
70|------|--------|
71| Low latency | `--roll-interval 10` |
72| Query performance | `--roll-interval 300 --roll-size 100` |
73| Cost optimal | `--compression zstd --roll-interval 300` |
74 
75## Schema Evolution
76 
77Pipelines are immutable. Use versioning:
78 
79```bash
80# Create v2 stream/sink/pipeline
81npx wrangler pipelines streams create events-v2 --schema-file v2.json
82 
83# Dual-write during transition
84await Promise.all([env.EVENTS_V1.send([event]), env.EVENTS_V2.send([event])]);
85 
86# Query across versions with UNION ALL
87```
88