1# Queues Patterns & Best Practices
2 
3## Async Task Processing
4 
5```typescript
6// Producer: Accept request, queue work
7export default {
8  async fetch(request: Request, env: Env): Promise<Response> {
9    const { userId, reportType } = await request.json();
10    await env.REPORT_QUEUE.send({ userId, reportType, requestedAt: Date.now() });
11    return Response.json({ message: 'Report queued', status: 'pending' });
12  }
13};
14 
15// Consumer: Process reports
16export default {
17  async queue(batch: MessageBatch, env: Env): Promise<void> {
18    for (const msg of batch.messages) {
19      const { userId, reportType } = msg.body;
20      const report = await generateReport(userId, reportType, env);
21      await env.REPORTS_BUCKET.put(`${userId}/${reportType}.pdf`, report);
22      msg.ack();
23    }
24  }
25};
26```
27 
28## Buffering API Calls
29 
30```typescript
31// Producer: Queue log entries
32ctx.waitUntil(env.LOGS_QUEUE.send({
33  method: request.method,
34  url: request.url,
35  timestamp: Date.now()
36}));
37 
38// Consumer: Batch write to external API
39async queue(batch: MessageBatch, env: Env): Promise<void> {
40  const logs = batch.messages.map(m => m.body);
41  await fetch(env.LOG_ENDPOINT, { method: 'POST', body: JSON.stringify({ logs }) });
42  batch.ackAll();
43}
44```
45 
46## Rate Limiting Upstream
47 
48```typescript
49async queue(batch: MessageBatch, env: Env): Promise<void> {
50  for (const msg of batch.messages) {
51    try {
52      await callRateLimitedAPI(msg.body);
53      msg.ack();
54    } catch (error) {
55      if (error.status === 429) {
56        const retryAfter = parseInt(error.headers.get('Retry-After') || '60');
57        msg.retry({ delaySeconds: retryAfter });
58      } else throw error;
59    }
60  }
61}
62```
63 
64## Event-Driven Workflows
65 
66```typescript
67// R2 event → Queue → Worker
68export default {
69  async queue(batch: MessageBatch, env: Env): Promise<void> {
70    for (const msg of batch.messages) {
71      const event = msg.body;
72      if (event.action === 'PutObject') {
73        await processNewFile(event.object.key, env);
74      } else if (event.action === 'DeleteObject') {
75        await cleanupReferences(event.object.key, env);
76      }
77      msg.ack();
78    }
79  }
80};
81```
82 
83## Dead Letter Queue Pattern
84 
85```typescript
86// Main queue: After max_retries, goes to DLQ automatically
87export default {
88  async queue(batch: MessageBatch, env: Env): Promise<void> {
89    for (const msg of batch.messages) {
90      try {
91        await riskyOperation(msg.body);
92        msg.ack();
93      } catch (error) {
94        console.error(`Failed after ${msg.attempts} attempts:`, error);
95      }
96    }
97  }
98};
99 
100// DLQ consumer: Log and store failed messages
101export default {
102  async queue(batch: MessageBatch, env: Env): Promise<void> {
103    for (const msg of batch.messages) {
104      await env.FAILED_KV.put(msg.id, JSON.stringify(msg.body));
105      msg.ack();
106    }
107  }
108};
109```
110 
111## Priority Queues
112 
113High priority: `max_batch_size: 5, max_batch_timeout: 1`. Low priority: `max_batch_size: 100, max_batch_timeout: 30`.
114 
115## Delayed Job Processing
116 
117```typescript
118await env.EMAIL_QUEUE.send({ to, template, userId }, { delaySeconds: 3600 });
119```
120 
121## Fan-out Pattern
122 
123```typescript
124async fetch(request: Request, env: Env): Promise<Response> {
125  const event = await request.json();
126  
127  // Send to multiple queues for parallel processing
128  await Promise.all([
129    env.ANALYTICS_QUEUE.send(event),
130    env.NOTIFICATIONS_QUEUE.send(event),
131    env.AUDIT_LOG_QUEUE.send(event)
132  ]);
133  
134  return Response.json({ status: 'processed' });
135}
136```
137 
138## Idempotency Pattern
139 
140```typescript
141async queue(batch: MessageBatch, env: Env): Promise<void> {
142  for (const msg of batch.messages) {
143    // Check if already processed
144    const processed = await env.PROCESSED_KV.get(msg.id);
145    if (processed) {
146      msg.ack();
147      continue;
148    }
149    
150    await processMessage(msg.body);
151    await env.PROCESSED_KV.put(msg.id, '1', { expirationTtl: 86400 });
152    msg.ack();
153  }
154}
155```
156 
157## Integration: D1 Batch Writes
158 
159```typescript
160async queue(batch: MessageBatch, env: Env): Promise<void> {
161  // Collect all inserts for single D1 batch
162  const statements = batch.messages.map(msg => 
163    env.DB.prepare('INSERT INTO events (id, data, created) VALUES (?, ?, ?)')
164      .bind(msg.id, JSON.stringify(msg.body), Date.now())
165  );
166  
167  try {
168    await env.DB.batch(statements);
169    batch.ackAll();
170  } catch (error) {
171    console.error('D1 batch failed:', error);
172    batch.retryAll({ delaySeconds: 60 });
173  }
174}
175```
176 
177## Integration: Workflows
178 
179```typescript
180// Queue triggers Workflow for long-running tasks
181async queue(batch: MessageBatch, env: Env): Promise<void> {
182  for (const msg of batch.messages) {
183    try {
184      const instance = await env.MY_WORKFLOW.create({
185        id: msg.id,
186        params: msg.body
187      });
188      console.log('Workflow started:', instance.id);
189      msg.ack();
190    } catch (error) {
191      msg.retry({ delaySeconds: 30 });
192    }
193  }
194}
195```
196 
197## Integration: Durable Objects
198 
199```typescript
200// Queue distributes work to Durable Objects by ID
201async queue(batch: MessageBatch, env: Env): Promise<void> {
202  for (const msg of batch.messages) {
203    const { userId, action } = msg.body;
204    
205    // Route to user-specific DO
206    const id = env.USER_DO.idFromName(userId);
207    const stub = env.USER_DO.get(id);
208    
209    try {
210      await stub.fetch(new Request('https://do/process', {
211        method: 'POST',
212        body: JSON.stringify({ action, messageId: msg.id })
213      }));
214      msg.ack();
215    } catch (error) {
216      msg.retry({ delaySeconds: 60 });
217    }
218  }
219}
220```
221