1---
2title: Process DynamoDB Streams
3description: Enable a DynamoDB Stream on your table and consume change records as a typed Effect Stream from the same Lambda.
4sidebar:
5 order: 2
6---
7
8The DynamoDB table from the previous part stores records, but
9writes happen in the dark — there's no way for downstream systems
10to react. **DynamoDB Streams** fix that: every change to the
11table emits a change record (insert / modify / remove) you can
12stream into a Lambda. In Alchemy this is one helper call.
13
14## Subscribe to the stream
15
16Start with the smallest possible subscription: log just the
17event name (`INSERT` / `MODIFY` / `REMOVE`) for each change.
18Add it alongside the existing bindings:
19
20```diff lang="typescript"
21// src/api.ts
22import * as AWS from "alchemy/AWS";
23import * as DynamoDB from "alchemy/AWS/DynamoDB";
24+import * as Console from "effect/Console";
25import * as Effect from "effect/Effect";
26import * as Layer from "effect/Layer";
27+import * as Stream from "effect/Stream";
28// ...
29
30const table = yield* DynamoDB.Table("Items", { /* ... */ });
31const putItem = yield* DynamoDB.PutItem(table);
32const getItem = yield* DynamoDB.GetItem(table);
33
34+yield* DynamoDB.consumeTableChanges(table, (stream) =>
35+ stream.pipe(
36+ Stream.runForEach((record) =>
37+ Console.log(`${record.eventName}: ${record.dynamodb.Keys?.id?.S}`),
38+ ),
39+ ),
40+);
41```
42
43That single `consumeTableChanges(...)` call enables a stream on the table,
44creates a Lambda `EventSourceMapping` that polls it, and adds
45the IAM grants for `dynamodb:GetRecords`, `GetShardIterator`,
46`DescribeStream`, and `ListStreams` — all from the call site.
47
48## Pick the view type
49
50By default the stream emits keys only. Most handlers want the
51full row before and after the change — set `streamViewType` to
52`NEW_AND_OLD_IMAGES`:
53
54```diff lang="typescript"
55-yield* DynamoDB.consumeTableChanges(table, (stream) =>
56+yield* DynamoDB.consumeTableChanges(table, {
57+ streamViewType: "NEW_AND_OLD_IMAGES",
58+}, (stream) =>
59 stream.pipe(/* ... */),
60);
61```
62
63The other options (`KEYS_ONLY`, `NEW_IMAGE`, `OLD_IMAGE`) trade
64payload size for less detail — pick the smallest one your
65handler actually needs.
66
67## Control where consumption starts
68
69`startingPosition: "LATEST"` is the default and ignores history;
70the consumer only sees records that arrive after the
71event-source mapping is enabled. Switch to `"TRIM_HORIZON"` to
72replay everything currently in the stream's 24-hour retention
73window:
74
75```diff lang="typescript"
76yield* DynamoDB.consumeTableChanges(table, {
77 streamViewType: "NEW_AND_OLD_IMAGES",
78+ startingPosition: "TRIM_HORIZON",
79+ batchSize: 10,
80}, (stream) =>
81 stream.pipe(/* ... */),
82);
83```
84
85`batchSize` caps the number of records the function sees per
86invocation. Larger batches are more efficient; smaller batches
87fail in smaller blast radii.
88
89## Type the row payload
90
91DynamoDB stream records use the same attribute-value shape as
92`GetItem`/`PutItem`. Describe the row inline as a generic on
93`consumeTableChanges` and TypeScript will help you destructure
94`NewImage`/`OldImage`:
95
96```diff lang="typescript"
97-yield* DynamoDB.consumeTableChanges(table, {
98+yield* DynamoDB.consumeTableChanges<{ id: { S: string }; content: { S: string } }>(table, {
99 streamViewType: "NEW_AND_OLD_IMAGES",
100 startingPosition: "TRIM_HORIZON",
101 batchSize: 10,
102}, (stream) =>
103 stream.pipe(
104 Stream.runForEach((record) =>
105- Console.log(`${record.eventName}: ${record.dynamodb.Keys?.id?.S}`),
106+ Console.log(
107+ `${record.eventName}: ${record.dynamodb.Keys?.id?.S} -> ` +
108+ `${record.dynamodb.NewImage?.content.S ?? "(deleted)"}`,
109+ ),
110 ),
111 ),
112);
113```
114
115## Provide the runtime layer
116
117The Lambda-specific implementation of `TableEventSource` is in
118`Lambda.TableEventSource`. Add it to the merged layer:
119
120```diff lang="typescript"
121}).pipe(
122 Effect.provide(
123 Layer.mergeAll(
124 AWS.Lambda.BucketEventSource,
125+ AWS.Lambda.TableEventSource,
126 DynamoDB.GetItemHttp,
127 DynamoDB.PutItemHttp,
128 S3.PutObjectHttp,
129 S3.GetObjectHttp,
130 ),
131 ),
132),
133```
134
135`TableEventSource` is the binding that mutates the table's stream
136configuration during deploy and creates the event source mapping.
137You don't have to think about either — they're both side effects
138of `consumeTableChanges(...)`.
139
140## Deploy
141
142```sh
143bun alchemy deploy
144```
145
146Alchemy enables the stream on the table (takes a few seconds for
147the table to settle) then creates the event source mapping. Once
148the mapping is `Enabled`, every `PutItem`/`UpdateItem`/`DeleteItem`
149shows up in your function's CloudWatch logs.
150
151## Verify
152
153Reuse the `PUT /items/:id` route to publish changes, then tail
154the function logs to confirm they arrive:
155
156```sh
157bun alchemy logs Api --tail
158```
159
160In another terminal:
161
162```sh
163curl -X PUT --data 'first' "$URL/items/one"
164curl -X PUT --data 'second' "$URL/items/two"
165curl -X PUT --data 'updated' "$URL/items/one"
166```
167
168You'll see three `MODIFY`/`INSERT` log lines stream past — one
169per write. Stream records typically arrive within 1–2 seconds of
170the originating change.
171
172## Why a Stream instead of a callback?
173
174The `consumeTableChanges(table, handler)` signature mirrors the S3
175events helper for a reason — once your event source is a typed `Stream<Record>`,
176back-pressure, batching, retries, parallel side effects, and
177in-order processing are all just composition:
178
179```typescript
180stream.pipe(
181 Stream.groupedWithin(50, "5 seconds"),
182 Stream.mapEffect((batch) => writeToWarehouse(batch), { concurrency: 4 }),
183 Stream.runDrain,
184);
185```
186
187The same operators work for DynamoDB Streams, S3 events, SQS
188messages, and Kinesis records — the only thing that changes is
189the element type.
190
191Next we'll add an [SQS Queue](/aws/messaging/sqs) so the Lambda
192can fan changes (or anything else) out to a different consumer
193process and pick them up downstream.