1---
2title: SQS
3description: Add an SQS Queue, publish messages from your Lambda, and consume them from a second consumer Lambda — all wired through Alchemy bindings.
4---
5
6**SQS** is AWS's general-purpose message queue — anything you can
7serialize, dropped into a mailbox and processed asynchronously
8somewhere else. In Alchemy it's the `Queue` resource plus
9producer bindings (`SendMessage`, `SendMessageBatch`, and the
10`Stream`-sink-shaped `QueueSink`) and a consumer side:
11`SQS.consumeQueueMessages` subscribes a Lambda to the queue as a
12typed `Stream<SQSRecord>`, creating the event source mapping and
13granting the receive/delete permissions for you.
14
15The walkthrough below builds on the [Lambda](/aws/compute/lambda) and
16[DynamoDB](/aws/data/dynamodb) pages: it adds a **Queue**, sends
17messages to it from the existing function, and stands up a
18**second Lambda** that consumes them.
19
20## Add the queue
21
22```diff lang="typescript"
23// src/api.ts
24import * as AWS from "alchemy/AWS";
25import * as DynamoDB from "alchemy/AWS/DynamoDB";
26+import * as SQS from "alchemy/AWS/SQS";
27// ...
28
29Effect.gen(function* () {
30 const bucket = yield* S3.Bucket("Blobs");
31 const table = yield* DynamoDB.Table("Items", { /* ... */ });
32+ const queue = yield* SQS.Queue("Jobs");
33
34 // ... existing bindings ...
35});
36```
37
38Calling `SQS.Queue("Jobs")` provisions a standard queue with the
39defaults (4-day retention, 30-second visibility timeout, no
40delivery delay). FIFO queues, dead-letter targets, and the rest
41of the props can be supplied as a second argument when you need
42them.
43
44## Bind SendMessage on the producer
45
46`SQS.SendMessage(queue)` returns a callable Effect — same
47shape as `S3.PutObject` or `DynamoDB.PutItem` — and quietly
48attaches `sqs:SendMessage` scoped to the queue ARN:
49
50```diff lang="typescript"
51const putItem = yield* DynamoDB.PutItem(table);
52+const sendMessage = yield* SQS.SendMessage(queue);
53```
54
55Let's hook it into the existing `PUT /items/:id` route so every
56new record also drops a `job.created` message onto the queue:
57
58```diff lang="typescript"
59if (request.method === "PUT") {
60 const content = yield* request.text;
61 yield* putItem({
62 Item: {
63 id: { S: id },
64 content: { S: content },
65 },
66 });
67+ yield* sendMessage({
68+ MessageBody: JSON.stringify({
69+ type: "job.created",
70+ id,
71+ content,
72+ }),
73+ });
74 return HttpServerResponse.empty({ status: 204 });
75}
76```
77
78The producer also needs the runtime layer for `SendMessage` —
79add it to the merged layer alongside the others:
80
81```diff lang="typescript"
82}).pipe(
83 Effect.provide(
84 Layer.mergeAll(
85 AWS.Lambda.BucketEventSource,
86 AWS.Lambda.TableEventSource,
87 DynamoDB.GetItemHttp,
88 DynamoDB.PutItemHttp,
89 S3.PutObjectHttp,
90 S3.GetObjectHttp,
91+ SQS.SendMessageHttp,
92 ),
93 ),
94),
95```
96
97## Stand up a consumer Lambda
98
99Consumers live as their own Lambda Function. Create
100`src/worker.ts` — bare shell first, no event source yet:
101
102```typescript
103// src/worker.ts
104import * as AWS from "alchemy/AWS";
105import * as Effect from "effect/Effect";
106
107export default class Worker extends AWS.Lambda.Function<Worker>()(
108 "JobsWorker",
109 { main: import.meta.url },
110 Effect.gen(function* () {
111 return {};
112 }),
113) {}
114```
115
116This consumer doesn't need a Function URL — its only entrypoint
117is the SQS message poller AWS will configure for it.
118
119## Share the queue between two functions
120
121Both producer and consumer need to reference the same queue —
122the cleanest way to do that is to lift the declaration out of
123`api.ts` into a small shared module:
124
125```typescript
126// src/queue.ts
127import * as SQS from "alchemy/AWS/SQS";
128
129export const Jobs = SQS.Queue("Jobs");
130```
131
132Now both functions can import `Jobs`. Update the producer to use
133the shared handle instead of declaring the queue inline:
134
135```diff lang="typescript"
136// src/api.ts
137+import { Jobs } from "./queue.ts";
138// ...
139-const queue = yield* SQS.Queue("Jobs");
140+const queue = yield* Jobs;
141```
142
143## Resolve the queue in the consumer
144
145Yield the same `Jobs` handle in the consumer's constructor to get
146a typed `Queue` resource:
147
148```diff lang="typescript"
149// src/worker.ts
150import * as AWS from "alchemy/AWS";
151import * as Effect from "effect/Effect";
152+import { Jobs } from "./queue.ts";
153
154export default class Worker extends AWS.Lambda.Function<Worker>()(
155 "JobsWorker",
156 { main: import.meta.url },
157 Effect.gen(function* () {
158+ const queue = yield* Jobs;
159 return {};
160 }),
161) {}
162```
163
164## Subscribe to messages
165
166`SQS.consumeQueueMessages(queue, ...)` is the consumer-side mirror
167of `consumeBucketEvents` / `stream`: a typed `Stream<SQSRecord>` you
168can compose with any operator. Add the smallest possible
169subscription — log each message body:
170
171```diff lang="typescript"
172// src/worker.ts
173import * as AWS from "alchemy/AWS";
174+import * as SQS from "alchemy/AWS/SQS";
175+import * as Console from "effect/Console";
176import * as Effect from "effect/Effect";
177+import * as Stream from "effect/Stream";
178import { Jobs } from "./queue.ts";
179
180export default class Worker extends AWS.Lambda.Function<Worker>()(
181 "JobsWorker",
182 { main: import.meta.url },
183 Effect.gen(function* () {
184 const queue = yield* Jobs;
185+
186+ yield* SQS.consumeQueueMessages(queue, { batchSize: 10 }, (stream) =>
187+ stream.pipe(
188+ Stream.runForEach((record) =>
189+ Console.log(`received: ${record.body}`),
190+ ),
191+ ),
192+ );
193
194 return {};
195 }),
196) {}
197```
198
199That single `consumeQueueMessages(...)` call creates the Lambda event
200source mapping pointing at the queue and grants the function
201`sqs:ReceiveMessage`, `DeleteMessage`, and `GetQueueAttributes`
202on the queue ARN.
203
204## Provide the runtime layer
205
206The consumer-side machinery for queue subscriptions ships in
207`Lambda.QueueEventSource`. Provide it at the bottom of the
208function:
209
210```diff lang="typescript"
211+import * as Layer from "effect/Layer";
212// ...
213 return {};
214- }),
215+ }).pipe(Effect.provide(AWS.Lambda.QueueEventSource)),
216) {}
217```
218
219## Wire the worker into the Stack
220
221The Worker needs to be yielded so Alchemy actually deploys it:
222
223```diff lang="typescript"
224// alchemy.run.ts
225import Api from "./src/api.ts";
226+import Worker from "./src/worker.ts";
227
228export default Alchemy.Stack(
229 "MyApp",
230 { providers: AWS.providers(), state: AWS.state() },
231 Effect.gen(function* () {
232 const api = yield* Api;
233+ yield* Worker;
234 return { url: api.functionUrl };
235 }),
236);
237```
238
239## Deploy
240
241```sh
242bun alchemy deploy
243```
244
245Two functions deploy in parallel — `Api` (with `SendMessage`
246permissions) and `JobsWorker` (with `ReceiveMessage` permissions
247plus an event source mapping pointing at `Jobs`).
248
249## Verify
250
251Trigger a write, then watch the consumer's logs:
252
253```sh
254curl -X PUT --data 'hello queue' "$URL/items/q1"
255bun alchemy logs JobsWorker --tail
256```
257
258Within a couple of seconds you'll see:
259
260```
261received: {"type":"job.created","id":"q1","content":"hello queue"}
262```
263
264## Bonus: process records in parallel
265
266`Stream.mapEffect` accepts a `concurrency` option, so you can
267fan out per-record work without rewriting anything:
268
269```typescript
270stream.pipe(
271 Stream.mapEffect(
272 (record) => doSomethingWith(record.body).pipe(Effect.orDie),
273 { concurrency: 4 },
274 ),
275 Stream.runDrain,
276);
277```
278
279Up to 4 records process concurrently, and the event source
280mapping won't ack them until the stream completes — which is the
281same delivery semantics SQS already gives you, just with
282controllable concurrency on top.
283
284## Bonus: forward a stream to SQS with `QueueSink`
285
286`SendMessage(queue)` is perfect for one-off sends from a
287request handler, but if you've got a `Stream` of payloads —
288DynamoDB change records, S3 notifications, anything — you don't
289want to call `sendMessage` per item. **`SQS.QueueSink`** is the
290sink-shaped counterpart: it consumes a `Stream<string>`, batches
291items into chunks of up to 10, and ships each chunk through
292`sqs:SendMessageBatch` for you.
293
294Skip the per-write `sendMessage` in `PUT /items/:id` and forward
295the existing DynamoDB stream straight into the queue instead:
296
297```diff lang="typescript"
298// src/api.ts
299import * as Stream from "effect/Stream";
300import { Jobs } from "./queue.ts";
301// ...
302
303const queue = yield* Jobs;
304const putItem = yield* DynamoDB.PutItem(table);
305-const sendMessage = yield* SQS.SendMessage(queue);
306+const sink = yield* SQS.QueueSink(queue);
307
308yield* DynamoDB.consumeTableChanges(table, {
309 streamViewType: "NEW_AND_OLD_IMAGES",
310}, (stream) =>
311 stream.pipe(
312+ Stream.map((record) =>
313+ JSON.stringify({
314+ eventName: record.eventName,
315+ keys: record.dynamodb.Keys,
316+ newImage: record.dynamodb.NewImage,
317+ }),
318+ ),
319+ Stream.run(sink),
320 ),
321);
322```
323
324`QueueSink(queue)` grants `sqs:SendMessage` and
325`sqs:SendMessageBatch` on the queue ARN, then returns a
326`Sink<void, string, readonly string[]>` you compose with
327`Stream.run`. `QueueSinkHttp` is itself implemented on top of
328`SendMessageBatch`, so split the merged layer into two tiers
329with `Layer.provideMerge` — sink + event sources on top,
330the underlying request layers + policy on the bottom:
331
332```diff lang="typescript"
333}).pipe(
334 Effect.provide(
335- Layer.mergeAll(
336- AWS.Lambda.BucketEventSource,
337- AWS.Lambda.TableEventSource,
338- DynamoDB.GetItemHttp,
339- DynamoDB.PutItemHttp,
340- S3.PutObjectHttp,
341- S3.GetObjectHttp,
342- SQS.SendMessageHttp,
343- ),
344+ Layer.mergeAll(
345+ AWS.Lambda.BucketEventSource,
346+ AWS.Lambda.TableEventSource,
347+ SQS.QueueSinkHttp,
348+ ).pipe(
349+ Layer.provideMerge(
350+ Layer.mergeAll(
351+ DynamoDB.GetItemHttp,
352+ DynamoDB.PutItemHttp,
353+ S3.PutObjectHttp,
354+ S3.GetObjectHttp,
355+ SQS.SendMessageBatchHttp,
356+ ),
357+ ),
358+ ),
359 ),
360),
361```
362
363The consumer doesn't change — `JobsWorker` still subscribes via
364`SQS.consumeQueueMessages(queue)` and now sees one record per DynamoDB
365change, batched ten-at-a-time on the producer side.
366
367## Where next
368
369- [Kinesis](/aws/messaging/kinesis) — stream higher-volume records:
370 ordered, sharded, and with the same `Stream` consumer surface.
371- [Process DynamoDB Streams](/aws/messaging/dynamodb-streams) — the
372 change-data-capture pipeline the `QueueSink` bonus above
373 builds on.
374- [`Queue` reference](/providers/aws/sqs/queue) — FIFO queues,
375 dead-letter targets, and every other prop.