1# Pipelines API Reference
2 
3## Pipeline Binding Interface
4 
5```typescript
6// From @cloudflare/workers-types
7interface Pipeline {
8  send(data: object | object[]): Promise<void>;
9}
10 
11interface Env {
12  STREAM: Pipeline;
13}
14 
15export default {
16  async fetch(request: Request, env: Env): Promise<Response> {
17    // send() returns Promise<void> - no result data
18    await env.STREAM.send([event]);
19    return new Response('OK');
20  }
21} satisfies ExportedHandler<Env>;
22```
23 
24**Key points:**
25- `send()` accepts single object or array
26- Always returns `Promise<void>` (no confirmation data)
27- Throws on network/validation errors (wrap in try/catch)
28- Use `ctx.waitUntil()` for fire-and-forget pattern
29 
30## Writing Events
31 
32### Single Event
33 
34```typescript
35await env.STREAM.send([{
36  user_id: "12345",
37  event_type: "purchase",
38  product_id: "widget-001",
39  amount: 29.99
40}]);
41```
42 
43### Batch Events
44 
45```typescript
46const events = [
47  { user_id: "user1", event_type: "view" },
48  { user_id: "user2", event_type: "purchase", amount: 50 }
49];
50await env.STREAM.send(events);
51```
52 
53**Limits:**
54- Max 1 MB per request
55- 5 MB/s per stream
56 
57### Fire-and-Forget Pattern
58 
59```typescript
60export default {
61  async fetch(request: Request, env: Env, ctx: ExecutionContext): Promise<Response> {
62    const event = { /* ... */ };
63    
64    // Don't block response on send
65    ctx.waitUntil(env.STREAM.send([event]));
66    
67    return new Response('OK');
68  }
69};
70```
71 
72### Error Handling
73 
74```typescript
75try {
76  await env.STREAM.send([event]);
77} catch (error) {
78  console.error('Pipeline send failed:', error);
79  // Log to another system, retry, or return error response
80  return new Response('Failed to track event', { status: 500 });
81}
82```
83 
84## HTTP Ingest API
85 
86### Endpoint Format
87 
88```
89https://{stream-id}.ingest.cloudflare.com
90```
91 
92Get `{stream-id}` from: `npx wrangler pipelines streams list`
93 
94### Request Format
95 
96**CRITICAL:** Must send array, not single object
97 
98```bash
99# ✅ Correct
100curl -X POST https://{stream-id}.ingest.cloudflare.com \
101  -H "Content-Type: application/json" \
102  -d '[{"user_id": "123", "event_type": "purchase"}]'
103 
104# ❌ Wrong - will fail
105curl -X POST https://{stream-id}.ingest.cloudflare.com \
106  -H "Content-Type: application/json" \
107  -d '{"user_id": "123", "event_type": "purchase"}'
108```
109 
110### Authentication
111 
112```bash
113curl -X POST https://{stream-id}.ingest.cloudflare.com \
114  -H "Content-Type: application/json" \
115  -H "Authorization: Bearer YOUR_API_TOKEN" \
116  -d '[{"event": "data"}]'
117```
118 
119**Required permission:** Workers Pipeline Send
120 
121Create token: Dashboard → Workers → API tokens → Create with Pipeline Send permission
122 
123### Response Codes
124 
125| Code | Meaning | Action |
126|------|---------|--------|
127| 200 | Accepted | Success |
128| 400 | Invalid format | Check JSON array, schema match |
129| 401 | Auth failed | Verify token valid |
130| 413 | Payload too large | Split into smaller batches (<1 MB) |
131| 429 | Rate limited | Back off, retry with delay |
132| 5xx | Server error | Retry with exponential backoff |
133 
134## SQL Functions Quick Reference
135 
136Available in `INSERT INTO sink SELECT ... FROM stream` transformations:
137 
138| Function | Example | Use Case |
139|----------|---------|----------|
140| `UPPER(s)` | `UPPER(event_type)` | Normalize strings |
141| `LOWER(s)` | `LOWER(email)` | Case-insensitive matching |
142| `CONCAT(...)` | `CONCAT(user_id, '_', product_id)` | Generate composite keys |
143| `CASE WHEN ... THEN ... END` | `CASE WHEN amount > 100 THEN 'high' ELSE 'low' END` | Conditional enrichment |
144| `CAST(x AS type)` | `CAST(timestamp AS string)` | Type conversion |
145| `COALESCE(x, y)` | `COALESCE(amount, 0.0)` | Default values |
146| Math operators | `amount * 1.1`, `price / quantity` | Calculations |
147| Comparison | `amount > 100`, `status IN ('active', 'pending')` | Filtering |
148 
149**String types for CAST:** `string`, `int32`, `int64`, `float32`, `float64`, `bool`, `timestamp`
150 
151Full reference: [Pipelines SQL Reference](https://developers.cloudflare.com/pipelines/sql-reference/)
152 
153## SQL Transform Examples
154 
155### Filter Events
156 
157```sql
158INSERT INTO my_sink
159SELECT * FROM my_stream
160WHERE event_type = 'purchase' AND amount > 100
161```
162 
163### Select Specific Fields
164 
165```sql
166INSERT INTO my_sink
167SELECT user_id, event_type, timestamp, amount
168FROM my_stream
169```
170 
171### Transform and Enrich
172 
173```sql
174INSERT INTO my_sink
175SELECT
176  user_id,
177  UPPER(event_type) as event_type,
178  timestamp,
179  amount * 1.1 as amount_with_tax,
180  CONCAT(user_id, '_', product_id) as unique_key,
181  CASE
182    WHEN amount > 1000 THEN 'high_value'
183    WHEN amount > 100 THEN 'medium_value'
184    ELSE 'low_value'
185  END as customer_tier
186FROM my_stream
187WHERE event_type IN ('purchase', 'refund')
188```
189 
190## Querying Results (R2 Data Catalog)
191 
192```bash
193export WRANGLER_R2_SQL_AUTH_TOKEN=YOUR_CATALOG_TOKEN
194 
195npx wrangler r2 sql query "warehouse_name" "
196SELECT 
197  event_type,
198  COUNT(*) as event_count,
199  SUM(amount) as total_revenue
200FROM default.my_table
201WHERE event_type = 'purchase'
202  AND timestamp >= '2025-01-01'
203GROUP BY event_type
204ORDER BY total_revenue DESC
205LIMIT 100"
206```
207 
208**Note:** Iceberg tables support standard SQL queries with GROUP BY, JOINs, WHERE, ORDER BY, etc.
209