1---
2title: React to S3 Events
3description: Subscribe a Lambda Function to S3 bucket notifications, process them as an Effect Stream, and let Alchemy wire up the IAM and event-source plumbing.
4sidebar:
5 order: 3
6---
7
8Reading and writing S3 objects on demand is one half of the story.
9The other half is **reacting** to objects as they arrive — for
10indexing, virus scanning, thumbnail generation, you name it. In
11this part you'll subscribe the same Lambda from the previous part
12to bucket notifications and process them as an Effect `Stream`.
13
14## Subscribe to bucket notifications
15
16Bucket notifications live behind an Alchemy helper:
17`S3.consumeBucketEvents(bucket, process)`. Add the smallest
18possible subscription — log every new object's key — alongside
19the existing bindings:
20
21```diff lang="typescript"
22// src/api.ts
23import * as AWS from "alchemy/AWS";
24import * as S3 from "alchemy/AWS/S3";
25+import * as Console from "effect/Console";
26import * as Effect from "effect/Effect";
27import * as Layer from "effect/Layer";
28+import * as Stream from "effect/Stream";
29// ...
30
31Effect.gen(function* () {
32 const bucket = yield* S3.Bucket("Blobs");
33 const putObject = yield* S3.PutObject(bucket);
34 const getObject = yield* S3.GetObject(bucket);
35
36+ yield* S3.consumeBucketEvents(bucket, (stream) =>
37+ stream.pipe(
38+ Stream.runForEach((event) =>
39+ Console.log(`new object: ${event.key}`),
40+ ),
41+ ),
42+ );
43
44 return {
45 fetch: /* ... unchanged ... */,
46 };
47});
48```
49
50That single `yield*` does three things in one shot: it
51configures the bucket's notification settings to push events
52into the function, adds the `lambda:InvokeFunction` permission
53S3 needs to call you, and registers the stream handler that
54runs every time an event batch lands.
55
56## Filter the event types
57
58Without a filter you get every notification S3 emits — object
59created, deleted, restored, replicated, the lot. Constrain the
60subscription to a specific subset with `events`:
61
62```diff lang="typescript"
63-yield* S3.consumeBucketEvents(bucket, (stream) =>
64+yield* S3.consumeBucketEvents(
65+ bucket,
66+ { events: ["s3:ObjectCreated:*"] },
67+ (stream) =>
68 stream.pipe(
69 Stream.runForEach((event) =>
70 Console.log(`new object: ${event.key}`),
71 ),
72 ),
73);
74```
75
76The `events` filter is type-checked against the AWS S3 event
77catalog, so typos surface at the call site rather than at deploy
78time.
79
80## What the stream actually carries
81
82`stream` is a `Stream.Stream<BucketNotification>` — a normalized
83shape with `type`, `bucket`, `key`, `size`, and `eTag`. Each
84Lambda invocation delivers one batch as one stream emission
85sequence, then completes; the next invocation starts a fresh
86stream. Treating events as a `Stream` means the usual operators
87(`mapEffect`, `groupedWithin`, `mapAccum`) work without thinking
88about the Lambda payload envelope.
89
90## Provide the runtime layer
91
92The runtime side of the subscription lives in
93`Lambda.BucketEventSource`. It's the layer that turns the abstract
94`BucketEventSource` capability into a concrete Lambda event-source
95mapping plus IAM grant. Add it alongside the S3 bindings:
96
97```diff lang="typescript"
98}).pipe(
99 Effect.provide(
100- Layer.mergeAll(S3.PutObjectHttp, S3.GetObjectHttp),
101+ Layer.mergeAll(
102+ AWS.Lambda.BucketEventSource,
103+ S3.PutObjectHttp,
104+ S3.GetObjectHttp,
105+ ),
106 ),
107),
108```
109
110`BucketEventSource` and the `PutObject`/`GetObject` lives are all
111peers here — none of them require each other — so a flat
112`Layer.mergeAll(...)` is enough.
113
114## Run an Effect per event
115
116Logging is a good smoke test, but a real handler usually wants
117to do something async per event. Swap `runForEach` for
118`mapEffect` so each event runs through a generator block:
119
120```diff lang="typescript"
121yield* S3.consumeBucketEvents(
122 bucket,
123 { events: ["s3:ObjectCreated:*"] },
124 (stream) =>
125 stream.pipe(
126- Stream.runForEach((event) =>
127- Console.log(`new object: ${event.key}`),
128- ),
129+ Stream.mapEffect((event) =>
130+ Console.log(`new object: ${event.key}`),
131+ ),
132+ Stream.runDrain,
133+ ),
134);
135```
136
137`Stream.runDrain` consumes the stream without collecting — we
138only care about the side effects, not the return values.
139
140## Read the object back
141
142`getObject` is already bound to the bucket, so the per-event
143Effect can fetch the new object's contents inline:
144
145```diff lang="typescript"
146stream.pipe(
147 Stream.mapEffect((event) =>
148- Console.log(`new object: ${event.key}`),
149+ Effect.gen(function* () {
150+ const result = yield* getObject({ Key: event.key });
151+ const body = yield* result.Body!.pipe(
152+ Stream.decodeText,
153+ Stream.mkString,
154+ );
155+ yield* Console.log(`${event.key} -> ${body}`);
156+ }).pipe(Effect.orDie),
157 ),
158 Stream.runDrain,
159),
160```
161
162`Stream.mapEffect` is back-pressured by default — events flow
163through one at a time. Pass a `concurrency` option (we'll see it
164in later parts) when you want fan-out.
165
166## Deploy and verify
167
168```sh
169bun alchemy deploy
170```
171
172The CLI applies the new bucket notification configuration and
173redeploys the function with the broader IAM policy. Then trigger
174the subscription end-to-end by `PUT`ing an object via the HTTP
175route from the previous part:
176
177```typescript
178// test/integ.test.ts
179test(
180 "PUT triggers notification handler",
181 Effect.gen(function* () {
182 const { url } = yield* stack;
183 yield* HttpClient.put(`${url}/notify-me.txt`, {
184 body: HttpBody.text("payload"),
185 });
186 // ...optionally tail CloudWatch logs to assert the message landed
187 }),
188);
189```
190
191You'll see the `notify-me.txt -> payload` line in the function's
192log group within a couple of seconds.
193
194S3 events typically arrive in batches of up to 10 records — the
195same `Stream` is fed each batch and completes between invocations,
196so you get the same composition story whether the batch is one
197event or a hundred.
198
199Next we'll trade blob storage for structured records and
200[add a DynamoDB table](/aws/data/dynamodb).