clover's git

the-knife

1	---
2	title: Event Sources
3	description: An event source is a binding that runs your Function when something happens on a resource — one call wires the event-source mapping, the permissions, and a typed handler.
4	---
5
6	An **Event Source** is a [Binding](/infrastructure-as-effects/binding)
7	that runs your Function when something happens on a resource. A message
8	lands on a queue, an object lands in a bucket, a row changes in a table.
9	You name the resource and a handler, and the messages arrive as an
10	Effect `Stream`:
11
12	```typescript
13	import * as AWS from "alchemy/AWS";
14	import * as Effect from "effect/Effect";
15	import * as Stream from "effect/Stream";
16
17	export const Orders = AWS.SQS.Queue("Orders");
18
19	export default AWS.Lambda.Function(
20	"OrderProcessor",
21	{ main: import.meta.url },
22	Effect.gen(function* () {
23	yield* AWS.SQS.consumeQueueMessages(Orders, (records) =>
24	records.pipe(
25	Stream.map((record) => record.body),
26	Stream.runForEach((body) => Effect.log(`order: ${body}`)),
27	),
28	);
29	}).pipe(Effect.provide(AWS.Lambda.QueueEventSource)),
30	);
31	```
32
33	A pure consumer returns nothing from its constructor. It only declares
34	what it listens to.
35
36	## What one call wires
37
38	In traditional IaC, running code when a message lands takes three
39	pieces: permission to receive and delete messages, a mapping that tells
40	the queue to invoke the Function, and a handler that unpacks the raw
41	event:
42
43	```typescript
44	// infrastructure: grant access and wire the trigger
45	policy: { Action: ["sqs:ReceiveMessage", "sqs:DeleteMessage", "sqs:GetQueueAttributes"], Resource: [queue.arn] },
46	eventSourceMapping: { eventSourceArn: queue.arn, functionName: fn.name },
47	```
48
49	```typescript
50	// handler: unpack the batch by hand
51	export const handler = async (event) => {
52	for (const record of event.Records) {
53	console.log(record.body);
54	}
55	};
56	```
57
58	`consumeQueueMessages` does all three. At deploy time it attaches that
59	statement to the Function's role, scoped to the queue's ARN, and
60	creates the event source mapping that points the queue at this
61	Function. At runtime it registers the listener that turns each
62	invocation's batch into `records`.
63
64	`consumeQueueMessages` is the contract and `AWS.Lambda.QueueEventSource`
65	is the Layer that satisfies it, the same split every binding has
66	([a contract and a Layer](/infrastructure-as-effects/binding#a-contract-and-a-layer)).
67
68	## It's a real Effect Stream
69
70	`records` is the same `Stream` from any Effect program, so every
71	combinator composes. Map table changes to JSON and run them straight
72	into a queue:
73
74	```typescript
75	const sink = yield* AWS.SQS.QueueSink(Outbound);
76
77	yield* AWS.DynamoDB.consumeTableChanges(
78	Jobs,
79	{ streamViewType: "NEW_AND_OLD_IMAGES", startingPosition: "TRIM_HORIZON" },
80	(stream) =>
81	stream.pipe(
82	Stream.map((record) =>
83	JSON.stringify({
84	eventName: record.eventName,
85	keys: record.dynamodb.Keys,
86	}),
87	),
88	Stream.run(sink),
89	),
90	);
91	```
92
93	`Stream.throttle` or `Stream.groupedWithin` drop in anywhere along the
94	chain. That `sink` is a [Sink](/infrastructure-as-effects/sinks), the
95	write-side dual of an event source. Its permissions are generated the
96	same way.
97
98	## The same shape on Cloudflare
99
100	Consume a Cloudflare queue the same way. The consumer settings go on
101	the call, and the handler receives a `Stream` of typed messages:
102
103	```typescript
104	import * as Cloudflare from "alchemy/Cloudflare";
105	import * as Effect from "effect/Effect";
106	import * as Stream from "effect/Stream";
107	import * as HttpServerResponse from "effect/http/HttpServerResponse";
108
109	export const Orders = Cloudflare.Queues.Queue("Orders");
110
111	export default Cloudflare.Worker(
112	"OrderProcessor",
113	{ main: import.meta.url },
114	Effect.gen(function* () {
115	yield* Cloudflare.Queues.consumeQueueMessages<{ text: string }>(
116	Orders,
117	{ batchSize: 10, maxRetries: 3, retryDelay: "1 second" },
118	(stream) =>
119	Stream.runForEach(stream, (msg) => Effect.log(msg.body.text)),
120	);
121
122	return {
123	fetch: Effect.gen(function* () {
124	return HttpServerResponse.text("ok");
125	}),
126	};
127	}).pipe(Effect.provide(Cloudflare.Queues.EventSourceLive)),
128	);
129	```
130
131	There is no IAM here. At deploy time the call creates the queue
132	consumer that dispatches batches to this Worker. At runtime it
133	registers the `queue` listener.
134
135	## Batching and retries
136
137	Set `batchSize` and `maximumBatchingWindow` on the call to shape each
138	invocation's batch:
139
140	```typescript
141	yield* AWS.SQS.consumeQueueMessages(
142	Orders,
143	{ batchSize: 10, maximumBatchingWindow: "5 seconds" },
144	(records) => Stream.runForEach(records, (record) => Effect.log(record.body)),
145	);
146	```
147
148	If your handler fails, the queue redelivers the batch. Stream sources
149	like Kinesis and DynamoDB add settings that bound a poison record's
150	blast radius:
151
152	```typescript
153	yield* AWS.DynamoDB.consumeTableChanges(
154	Jobs,
155	{
156	streamViewType: "NEW_AND_OLD_IMAGES",
157	startingPosition: "TRIM_HORIZON",
158	bisectBatchOnFunctionError: true,
159	maximumRetryAttempts: 3,
160	},
161	(stream) => Stream.runForEach(stream, (record) => Effect.log(record.eventName)),
162	);
163	```
164
165	On Cloudflare, set the retry policy on the call. A batch is acked when
166	your handler succeeds, retried `maxRetries` times when it fails, then
167	dead-lettered:
168
169	```typescript
170	yield* Cloudflare.Queues.consumeQueueMessages<{ text: string }>(
171	Orders,
172	{ maxRetries: 3, retryDelay: "1 second", deadLetterQueue: "orders-dlq" },
173	(stream) => Stream.runForEach(stream, (msg) => Effect.log(msg.body.text)),
174	);
175	```
176
177	Call `msg.ack()` or `msg.retry()` inside the handler to decide per
178	message.
179
180	## Not every source is a Stream
181
182	A source that delivers one event at a time hands your handler the event
183	directly. A cron trigger passes one `controller` per fire:
184
185	```typescript
186	export default Cloudflare.Worker(
187	"Nightly",
188	{ main: import.meta.url },
189	Effect.gen(function* () {
190	yield* Cloudflare.Workers.cron("0 12 * * *", (controller) =>
191	Effect.log(`scheduled at ${controller.scheduledTime}`),
192	);
193
194	return {
195	fetch: Effect.gen(function* () {
196	return HttpServerResponse.text("ok");
197	}),
198	};
199	}).pipe(Effect.provide(Cloudflare.Workers.CronEventSourceLive)),
200	);
201	```
202
203	At deploy time the cron expression is attached to the Worker. At runtime
204	the `scheduled` listener is registered. GitHub works the same way:
205	`consumeRepositoryEvents` creates a repository webhook pointing at the
206	Worker's URL, and the runtime listener verifies each delivery's HMAC
207	signature before calling your handler.
208
209	## Every event source
210
211	The shape never changes. Only the `consume*` callable, the record type,
212	and the Layer you provide differ:
213
214	```typescript
215	// AWS: handlers receive a Stream per batch
216	yield* AWS.SQS.consumeQueueMessages(queue, fn);
217	yield* AWS.Kinesis.consumeStreamRecords(stream, props, fn);
218	yield* AWS.DynamoDB.consumeTableChanges(table, props, fn);
219	yield* AWS.S3.consumeBucketEvents(bucket, fn);
220	yield* AWS.SNS.consumeTopicNotifications(topic, fn);
221	yield* AWS.EventBridge.consumeBusEvents(bus, pattern, fn);
222
223	// Cloudflare
224	yield* Cloudflare.Queues.consumeQueueMessages(queue, fn);
225	yield* Cloudflare.Workers.cron(expression, fn);
226	yield* GitHub.consumeRepositoryEvents(props, fn);
227	```
228
229	EventBridge can also route matching events to another resource instead
230	of consuming them locally:
231
232	```typescript
233	yield* AWS.EventBridge.events(bus, { source: ["my.app"] }).toQueue(queue);
234	```
235
236	Each source has its own guide. On AWS: [SQS](/aws/messaging/sqs),
237	[Kinesis](/aws/messaging/kinesis),
238	[DynamoDB Streams](/aws/messaging/dynamodb-streams),
239	[S3 Events](/aws/messaging/s3-events), [SNS](/aws/messaging/sns), and
240	[EventBridge](/aws/messaging/eventbridge). On Cloudflare:
241	[Queues](/cloudflare/messaging/queues), [Cron](/cloudflare/messaging/cron),
242	and [GitHub Events](/cloudflare/messaging/github-events).
243
244	## Where next
245
246	- [Sinks](/infrastructure-as-effects/sinks) — the write-side dual:
247	resources as Effect `Sink`s, with batching and IAM generated the
248	same way.
249	- [Bindings](/infrastructure-as-effects/binding) — the deploy-time
250	mechanics every event source is built on.
251	- [Runtime](/infrastructure-as-effects/runtime) —
252	the Effectful Constructor these examples live inside.