1import * as lambda from "@distilled.cloud/aws/lambda";
2import * as Data from "effect/Data";
3import type * as Duration from "effect/Duration";
4import * as Effect from "effect/Effect";
5import * as Option from "effect/Option";
6import * as Schedule from "effect/Schedule";
7import * as Stream from "effect/Stream";
8import { deepEqual, isResolved } from "../../Diff.ts";
9import { toWireSeconds } from "../../Util/Duration.ts";
10import * as Provider from "../../Provider.ts";
11import { Resource } from "../../Resource.ts";
12import { createInternalTags, diffTags, hasTags } from "../../Tags.ts";
13import { AWSEnvironment } from "../Environment.ts";
14import type { Providers } from "../Providers.ts";
15
16export type StartingPosition = "TRIM_HORIZON" | "LATEST" | "AT_TIMESTAMP";
17
18/** Internal retry signal: the mapping has not reached its desired state yet. */
19class MappingNotSettled extends Data.TaggedError("MappingNotSettled")<{
20 readonly state: string | undefined;
21}> {}
22
23export type FunctionResponseType = "ReportBatchItemFailures";
24
25export interface EventSourceMappingProps {
26 /**
27 * The name or ARN of the Lambda function to invoke.
28 */
29 functionName: string;
30 /**
31 * The ARN of the event source (SQS queue, Kinesis stream, DynamoDB stream, etc.).
32 */
33 eventSourceArn: string;
34 /**
35 * The maximum number of records in each batch that Lambda pulls and sends to the function.
36 *
37 * - SQS: default 10, max 10,000 (FIFO max 10)
38 * - Kinesis: default 100, max 10,000
39 * - DynamoDB Streams: default 100, max 10,000
40 */
41 batchSize?: number;
42 /**
43 * The maximum amount of time that Lambda spends gathering records before
44 * invoking the function (e.g. `"5 seconds"`). Rounded to whole seconds on
45 * the wire.
46 * @default 0
47 */
48 maximumBatchingWindow?: Duration.Input;
49 /**
50 * Whether the event source mapping is active.
51 * @default true
52 */
53 enabled?: boolean;
54 /**
55 * The position in a stream from which to start reading. Required for Kinesis and DynamoDB Streams.
56 *
57 * - `LATEST` - Read only new records.
58 * - `TRIM_HORIZON` - Process all available records.
59 * - `AT_TIMESTAMP` - Start reading from a specific time.
60 */
61 startingPosition?: StartingPosition;
62 /**
63 * The timestamp to start reading from when `startingPosition` is `AT_TIMESTAMP`.
64 */
65 startingPositionTimestamp?: Date;
66 /**
67 * (Kinesis and DynamoDB Streams) The number of batches to process from each shard concurrently.
68 * @default 1
69 */
70 parallelizationFactor?: number;
71 /**
72 * (Kinesis and DynamoDB Streams) Split the batch in two and retry if the function returns an error.
73 * @default false
74 */
75 bisectBatchOnFunctionError?: boolean;
76 /**
77 * (Kinesis and DynamoDB Streams) Discard records older than the specified
78 * age (e.g. `"1 hour"`). Rounded to whole seconds on the wire.
79 * @default infinite (omit to never expire records)
80 */
81 maximumRecordAge?: Duration.Input;
82 /**
83 * (Kinesis and DynamoDB Streams) Discard records after the specified number of retries.
84 * @default -1 (infinite)
85 */
86 maximumRetryAttempts?: number;
87 /**
88 * (Kinesis and DynamoDB Streams) The duration of a processing window for
89 * tumbling windows (e.g. `"30 seconds"`). Rounded to whole seconds on the
90 * wire.
91 */
92 tumblingWindow?: Duration.Input;
93 /**
94 * A list of current response type enums applied to the event source mapping.
95 * @default ["ReportBatchItemFailures"]
96 */
97 functionResponseTypes?: FunctionResponseType[];
98 /**
99 * (SQS) Scaling configuration for the event source.
100 */
101 scalingConfig?: lambda.ScalingConfig;
102 /**
103 * (Kinesis and DynamoDB Streams) A destination for records that failed processing.
104 */
105 destinationConfig?: lambda.DestinationConfig;
106 /**
107 * Filter criteria to control which records are sent to the function.
108 */
109 filterCriteria?: lambda.FilterCriteria;
110 /**
111 * The ARN of an AWS KMS key to encrypt the filter criteria.
112 */
113 kmsKeyArn?: string;
114 /**
115 * Metrics configuration for the event source mapping.
116 * @default { Metrics: ["EventCount"] }
117 */
118 metricsConfig?: lambda.EventSourceMappingMetricsConfig;
119 /**
120 * (SQS, MSK, self-managed Kafka) Provisioned poller configuration.
121 */
122 provisionedPollerConfig?: lambda.ProvisionedPollerConfig;
123 /**
124 * (Amazon MSK) Configuration for an Amazon Managed Streaming for Apache Kafka event source.
125 */
126 amazonManagedKafkaEventSourceConfig?: lambda.AmazonManagedKafkaEventSourceConfig;
127 /**
128 * (Self-managed Kafka) Configuration for a self-managed Apache Kafka event source.
129 */
130 selfManagedKafkaEventSourceConfig?: lambda.SelfManagedKafkaEventSourceConfig;
131 /**
132 * (Self-managed Kafka) The self-managed Apache Kafka cluster for the event source.
133 */
134 selfManagedEventSource?: lambda.SelfManagedEventSource;
135 /**
136 * (Amazon MQ, MSK, self-managed Kafka) Source access configuration for VPC, authentication, etc.
137 */
138 sourceAccessConfigurations?: lambda.SourceAccessConfiguration[];
139 /**
140 * (Amazon MSK, self-managed Kafka) The Kafka topic name(s).
141 */
142 topics?: string[];
143 /**
144 * (Amazon MQ) The name of the Amazon MQ broker destination queue to consume.
145 */
146 queues?: string[];
147 /**
148 * (Amazon DocumentDB) Configuration for a DocumentDB event source.
149 */
150 documentDBEventSourceConfig?: lambda.DocumentDBEventSourceConfig;
151 /**
152 * (Amazon MSK and self-managed Apache Kafka) The logging configuration for the event source.
153 */
154 loggingConfig?: lambda.LoggingConfig;
155 /**
156 * Tags to associate with the event source mapping.
157 */
158 tags?: Record<string, string>;
159}
160
161export interface EventSourceMapping extends Resource<
162 "AWS.Lambda.EventSourceMapping",
163 EventSourceMappingProps,
164 {
165 /**
166 * The UUID of the event source mapping.
167 */
168 uuid: string;
169 /**
170 * The ARN of the event source mapping.
171 */
172 eventSourceMappingArn: string;
173 /**
174 * The ARN of the Lambda function.
175 */
176 functionArn: string;
177 /**
178 * The current state of the event source mapping.
179 */
180 state: string;
181 },
182 never,
183 Providers
184> {}
185
186/**
187 * Connects an event source — an SQS queue, Kinesis stream, DynamoDB stream,
188 * Amazon MQ broker, or Kafka topic — to a Lambda function so that records are
189 * polled from the source and delivered to the function in batches.
190 *
191 * Most stacks create mappings indirectly through the higher-level event-source
192 * helpers (`SQS.consumeQueueMessages(queue, ...)`,
193 * `Kinesis.consumeStreamRecords(stream, ...)`, `DynamoDB.consumeTableChanges(table, ...)`),
194 * which wire up the matching IAM permissions automatically. Use this resource
195 * directly when you need full control over batching, starting position, retry
196 * behavior, or filtering.
197 *
198 * ### Polling an SQS Queue
199 * SQS is the simplest source: no `startingPosition` is needed because there is
200 * no stream cursor. Lambda long-polls the queue and invokes the function with
201 * up to `batchSize` messages, and `functionName` plus `eventSourceArn` are the
202 * only required props.
203 *
204 * **Example:** Subscribe a function to a queue
205 * ```typescript
206 * import * as AWS from "alchemy/AWS";
207 *
208 * const queue = yield* AWS.SQS.Queue("Jobs", {});
209 * const worker = yield* AWS.Lambda.Function("Worker", {
210 * main: "./src/worker.ts",
211 * });
212 *
213 * const mapping = yield* AWS.Lambda.EventSourceMapping("JobsToWorker", {
214 * functionName: worker.functionName,
215 * eventSourceArn: queue.queueArn,
216 * batchSize: 10,
217 * maximumBatchingWindow: "5 seconds",
218 * });
219 * ```
220 *
221 * This delivers up to 10 messages per invocation, waiting up to 5 seconds to
222 * fill a batch before invoking. Increasing the batching window trades latency
223 * for fewer, larger invocations — useful for amortizing cold starts or
224 * downstream write costs on bursty queues.
225 *
226 * ### Streaming from Kinesis & DynamoDB
227 * Stream sources (Kinesis and DynamoDB Streams) deliver records in shard order
228 * and therefore require a `startingPosition` that tells Lambda where in the
229 * shard to begin reading. These sources also unlock the stream-only tuning
230 * knobs covered in the next sections.
231 *
232 * **Example:** Process a Kinesis stream from the latest records
233 * ```typescript
234 * import * as AWS from "alchemy/AWS";
235 *
236 * const stream = yield* AWS.Kinesis.Stream("Events", {});
237 * const consumer = yield* AWS.Lambda.Function("Consumer", {
238 * main: "./src/consumer.ts",
239 * });
240 *
241 * const mapping = yield* AWS.Lambda.EventSourceMapping("EventsToConsumer", {
242 * functionName: consumer.functionName,
243 * eventSourceArn: stream.streamArn,
244 * startingPosition: "LATEST",
245 * batchSize: 100,
246 * });
247 * ```
248 *
249 * `startingPosition: "LATEST"` skips any backlog and only processes records
250 * written after the mapping is created — the right choice for live event
251 * pipelines where replaying history would be wasteful or incorrect.
252 *
253 * **Example:** Replay a DynamoDB stream from the beginning
254 * ```typescript
255 * import * as AWS from "alchemy/AWS";
256 *
257 * const table = yield* AWS.DynamoDB.Table("Orders", {
258 * partitionKey: { name: "id", type: "S" },
259 * });
260 * const handler = yield* AWS.Lambda.Function("OrdersStream", {
261 * main: "./src/orders.ts",
262 * });
263 *
264 * const mapping = yield* AWS.Lambda.EventSourceMapping("OrdersToHandler", {
265 * functionName: handler.functionName,
266 * eventSourceArn: table.latestStreamArn!,
267 * startingPosition: "TRIM_HORIZON",
268 * });
269 * ```
270 *
271 * `TRIM_HORIZON` starts at the oldest record still in the stream, so the
272 * function processes the full available history before catching up to new
273 * writes — use it when every change matters (e.g. building a projection).
274 *
275 * **Example:** Start reading from a specific timestamp
276 * ```typescript
277 * const mapping = yield* AWS.Lambda.EventSourceMapping("EventsFromTime", {
278 * functionName: consumer.functionName,
279 * eventSourceArn: stream.streamArn,
280 * startingPosition: "AT_TIMESTAMP",
281 * startingPositionTimestamp: new Date("2026-01-01T00:00:00Z"),
282 * });
283 * ```
284 *
285 * `AT_TIMESTAMP` (Kinesis only) begins at the first record on or after
286 * `startingPositionTimestamp`, letting you reprocess a known time range without
287 * replaying the entire stream.
288 *
289 * ### Tuning Throughput
290 * For stream sources, throughput is governed by how records are batched and how
291 * many batches run in parallel per shard. These knobs let you balance
292 * end-to-end latency against invocation count and downstream load.
293 *
294 * **Example:** Increase parallelism and use tumbling windows
295 * ```typescript
296 * const mapping = yield* AWS.Lambda.EventSourceMapping("HighThroughput", {
297 * functionName: consumer.functionName,
298 * eventSourceArn: stream.streamArn,
299 * startingPosition: "LATEST",
300 * batchSize: 500,
301 * maximumBatchingWindow: "10 seconds",
302 * parallelizationFactor: 5,
303 * tumblingWindow: "30 seconds",
304 * });
305 * ```
306 *
307 * `parallelizationFactor` runs up to 5 concurrent batches per shard (records
308 * with the same partition key still stay in order), while
309 * `tumblingWindow` aggregates results across sequential batches for
310 * windowed stream processing. Raising `batchSize`/`maximumBatchingWindow`
311 * favors fewer, larger invocations.
312 *
313 * ### Error Handling & Retries
314 * For stream sources a single poison-pill record can block a shard forever.
315 * These props bound retries, split failing batches, expire stale records, and
316 * route failures elsewhere instead of stalling the stream.
317 *
318 * **Example:** Bisect on error, cap retries, and expire old records
319 * ```typescript
320 * const dlq = yield* AWS.SQS.Queue("StreamFailures", {});
321 *
322 * const mapping = yield* AWS.Lambda.EventSourceMapping("ResilientStream", {
323 * functionName: consumer.functionName,
324 * eventSourceArn: stream.streamArn,
325 * startingPosition: "LATEST",
326 * bisectBatchOnFunctionError: true,
327 * maximumRetryAttempts: 3,
328 * maximumRecordAge: "1 hour",
329 * destinationConfig: {
330 * OnFailure: { Destination: dlq.queueArn },
331 * },
332 * });
333 * ```
334 *
335 * On a function error, `bisectBatchOnFunctionError` splits the batch in two and
336 * retries each half to isolate the bad record; after `maximumRetryAttempts` (or
337 * once a record is older than `maximumRecordAge`) the record is
338 * discarded and its metadata is sent to the `destinationConfig.OnFailure`
339 * target so it is never silently lost.
340 *
341 * **Example:** Report partial batch failures
342 * ```typescript
343 * const mapping = yield* AWS.Lambda.EventSourceMapping("PartialFailures", {
344 * functionName: handler.functionName,
345 * eventSourceArn: table.latestStreamArn!,
346 * startingPosition: "TRIM_HORIZON",
347 * functionResponseTypes: ["ReportBatchItemFailures"],
348 * });
349 * ```
350 *
351 * `functionResponseTypes: ["ReportBatchItemFailures"]` lets the function return
352 * only the IDs of records it failed to process, so Lambda retries just those
353 * instead of the whole batch — avoiding redundant reprocessing of records that
354 * already succeeded.
355 *
356 * ### Filtering Records
357 * Attach `filterCriteria` so the function is only invoked for records matching
358 * an event pattern. Filtering happens before invocation, so it cuts both cost
359 * and unnecessary cold starts. Encrypt the patterns with `kmsKeyArn` when they
360 * contain sensitive values.
361 *
362 * **Example:** Only deliver records where `type` is `"order"`
363 * ```typescript
364 * const mapping = yield* AWS.Lambda.EventSourceMapping("OrdersOnly", {
365 * functionName: worker.functionName,
366 * eventSourceArn: queue.queueArn,
367 * filterCriteria: {
368 * Filters: [{ Pattern: JSON.stringify({ body: { type: ["order"] } }) }],
369 * },
370 * kmsKeyArn:
371 * "arn:aws:kms:us-east-1:111122223333:key/abcd1234-...",
372 * });
373 * ```
374 *
375 * Each `Pattern` is a JSON event-pattern string; messages that don't match are
376 * dropped without invoking the function. The optional `kmsKeyArn` encrypts the
377 * stored filter criteria with your own KMS key instead of an AWS-managed one.
378 *
379 * ### Enabling & Disabling
380 * The `enabled` flag controls whether Lambda actively polls the source without
381 * deleting the mapping, so you can pause and resume delivery in place.
382 *
383 * **Example:** Create a paused mapping
384 * ```typescript
385 * const mapping = yield* AWS.Lambda.EventSourceMapping("PausedConsumer", {
386 * functionName: consumer.functionName,
387 * eventSourceArn: stream.streamArn,
388 * startingPosition: "LATEST",
389 * enabled: false,
390 * });
391 * ```
392 *
393 * With `enabled: false` the mapping exists but pulls no records — flip it back
394 * to `true` to resume. This is handy for maintenance windows or for staging a
395 * consumer before turning on traffic.
396 *
397 * ### Scaling & Provisioned Pollers
398 * Cap concurrency for SQS sources with `scalingConfig`, or reserve dedicated
399 * polling capacity (for Kafka/MSK and SQS) with `provisionedPollerConfig` to
400 * keep latency predictable under load.
401 *
402 * **Example:** Limit SQS concurrency and provision pollers
403 * ```typescript
404 * const mapping = yield* AWS.Lambda.EventSourceMapping("BoundedConsumer", {
405 * functionName: worker.functionName,
406 * eventSourceArn: queue.queueArn,
407 * scalingConfig: { MaximumConcurrency: 10 },
408 * provisionedPollerConfig: {
409 * MinimumPollers: 1,
410 * MaximumPollers: 20,
411 * },
412 * });
413 * ```
414 *
415 * `scalingConfig.MaximumConcurrency` caps how many function instances Lambda
416 * runs for this queue (protecting downstream systems), while
417 * `provisionedPollerConfig` keeps a pool of dedicated event pollers warm so
418 * throughput doesn't lag behind sudden spikes.
419 *
420 * ### Kafka, MQ & DocumentDB Sources
421 * Beyond AWS-native streams, an event source mapping can poll Amazon MSK,
422 * self-managed Apache Kafka, Amazon MQ brokers, and Amazon DocumentDB change
423 * streams. These sources use `topics`/`queues` to select what to consume,
424 * `sourceAccessConfigurations` for VPC and authentication wiring, and
425 * source-specific config props.
426 *
427 * **Example:** Consume a self-managed Kafka topic
428 * ```typescript
429 * const mapping = yield* AWS.Lambda.EventSourceMapping("KafkaConsumer", {
430 * functionName: consumer.functionName,
431 * eventSourceArn: stream.streamArn,
432 * topics: ["orders"],
433 * selfManagedEventSource: {
434 * Endpoints: { KAFKA_BOOTSTRAP_SERVERS: ["broker1:9092", "broker2:9092"] },
435 * },
436 * selfManagedKafkaEventSourceConfig: { ConsumerGroupId: "orders-consumer" },
437 * sourceAccessConfigurations: [
438 * { Type: "SASL_SCRAM_512_AUTH", URI: "arn:aws:secretsmanager:...:secret:kafka" },
439 * ],
440 * loggingConfig: { LogFormat: "JSON" },
441 * });
442 * ```
443 *
444 * `topics` names the Kafka topic(s) to read; `selfManagedEventSource.Endpoints`
445 * points at the brokers; `sourceAccessConfigurations` supplies the SASL/VPC
446 * credentials; and `selfManagedKafkaEventSourceConfig.ConsumerGroupId` pins the
447 * consumer group. For Amazon MSK use `amazonManagedKafkaEventSourceConfig`
448 * instead.
449 *
450 * **Example:** Consume an Amazon MQ queue and a DocumentDB change stream
451 * ```typescript
452 * const mqMapping = yield* AWS.Lambda.EventSourceMapping("MqConsumer", {
453 * functionName: worker.functionName,
454 * eventSourceArn: stream.streamArn,
455 * queues: ["orders-queue"],
456 * sourceAccessConfigurations: [
457 * { Type: "BASIC_AUTH", URI: "arn:aws:secretsmanager:...:secret:mq" },
458 * ],
459 * });
460 *
461 * const docDbMapping = yield* AWS.Lambda.EventSourceMapping("DocDbConsumer", {
462 * functionName: worker.functionName,
463 * eventSourceArn: stream.streamArn,
464 * documentDBEventSourceConfig: {
465 * DatabaseName: "shop",
466 * CollectionName: "orders",
467 * FullDocument: "UpdateLookup",
468 * },
469 * });
470 * ```
471 *
472 * For Amazon MQ, `queues` names the broker destination to consume and
473 * `sourceAccessConfigurations` carries the broker credentials; for DocumentDB,
474 * `documentDBEventSourceConfig` selects the database/collection and whether full
475 * documents are delivered on updates.
476 *
477 * ### Metrics & Tags
478 * Opt into per-mapping CloudWatch metrics with `metricsConfig` and brand the
479 * mapping with your own `tags` (Alchemy also applies its internal ownership
480 * tags automatically).
481 *
482 * **Example:** Enable event metrics and add tags
483 * ```typescript
484 * const mapping = yield* AWS.Lambda.EventSourceMapping("ObservedConsumer", {
485 * functionName: worker.functionName,
486 * eventSourceArn: queue.queueArn,
487 * metricsConfig: { Metrics: ["EventCount"] },
488 * tags: { team: "payments", env: "prod" },
489 * });
490 * ```
491 *
492 * `metricsConfig.Metrics` turns on the named CloudWatch metrics (e.g.
493 * `EventCount`) for this mapping, and `tags` attaches arbitrary key/value pairs
494 * for cost allocation and discovery.
495 *
496 * @resource
497 */
498export const EventSourceMapping = Resource<EventSourceMapping>(
499 "AWS.Lambda.EventSourceMapping",
500);
501
502export const EventSourceMappingProvider = () =>
503 Provider.effect(
504 EventSourceMapping,
505 Effect.gen(function* () {
506 const createEventSourceMappingTags = Effect.fn(function* (id: string) {
507 const internalTags = yield* createInternalTags(id);
508 return {
509 ...internalTags,
510 "alchemy::id": sanitizeAwsTagValue(internalTags["alchemy::id"]),
511 };
512 });
513
514 const toCreateRequest = (
515 props: EventSourceMappingProps,
516 tags: Record<string, string>,
517 ): lambda.CreateEventSourceMappingRequest => ({
518 FunctionName: props.functionName as string,
519 EventSourceArn: props.eventSourceArn as string,
520 Enabled: props.enabled ?? true,
521 BatchSize: props.batchSize,
522 MaximumBatchingWindowInSeconds: toWireSeconds(
523 props.maximumBatchingWindow,
524 ),
525 StartingPosition: props.startingPosition,
526 StartingPositionTimestamp: props.startingPositionTimestamp,
527 ParallelizationFactor: props.parallelizationFactor,
528 BisectBatchOnFunctionError: props.bisectBatchOnFunctionError,
529 MaximumRecordAgeInSeconds: toWireSeconds(props.maximumRecordAge),
530 MaximumRetryAttempts: props.maximumRetryAttempts,
531 TumblingWindowInSeconds: toWireSeconds(props.tumblingWindow),
532 FunctionResponseTypes: props.functionResponseTypes ?? [
533 "ReportBatchItemFailures",
534 ],
535 ScalingConfig: props.scalingConfig,
536 DestinationConfig: props.destinationConfig,
537 FilterCriteria: props.filterCriteria,
538 KMSKeyArn: props.kmsKeyArn,
539 MetricsConfig: props.metricsConfig ?? { Metrics: ["EventCount"] },
540 ProvisionedPollerConfig: props.provisionedPollerConfig,
541 AmazonManagedKafkaEventSourceConfig:
542 props.amazonManagedKafkaEventSourceConfig,
543 SelfManagedKafkaEventSourceConfig:
544 props.selfManagedKafkaEventSourceConfig,
545 SelfManagedEventSource: props.selfManagedEventSource,
546 SourceAccessConfigurations: props.sourceAccessConfigurations,
547 Topics: props.topics,
548 Queues: props.queues,
549 DocumentDBEventSourceConfig: props.documentDBEventSourceConfig,
550 LoggingConfig: props.loggingConfig,
551 Tags: tags,
552 });
553
554 const toUpdateRequest = (
555 uuid: string,
556 props: EventSourceMappingProps,
557 ): lambda.UpdateEventSourceMappingRequest => ({
558 UUID: uuid,
559 FunctionName: props.functionName as string,
560 Enabled: props.enabled ?? true,
561 BatchSize: props.batchSize,
562 MaximumBatchingWindowInSeconds: toWireSeconds(
563 props.maximumBatchingWindow,
564 ),
565 BisectBatchOnFunctionError: props.bisectBatchOnFunctionError,
566 MaximumRecordAgeInSeconds: toWireSeconds(props.maximumRecordAge),
567 MaximumRetryAttempts: props.maximumRetryAttempts,
568 TumblingWindowInSeconds: toWireSeconds(props.tumblingWindow),
569 FunctionResponseTypes: props.functionResponseTypes ?? [
570 "ReportBatchItemFailures",
571 ],
572 ScalingConfig: props.scalingConfig,
573 DestinationConfig: props.destinationConfig,
574 FilterCriteria: props.filterCriteria,
575 KMSKeyArn: props.kmsKeyArn,
576 MetricsConfig: props.metricsConfig ?? { Metrics: ["EventCount"] },
577 ProvisionedPollerConfig: props.provisionedPollerConfig,
578 AmazonManagedKafkaEventSourceConfig:
579 props.amazonManagedKafkaEventSourceConfig,
580 SelfManagedKafkaEventSourceConfig:
581 props.selfManagedKafkaEventSourceConfig,
582 SourceAccessConfigurations: props.sourceAccessConfigurations,
583 DocumentDBEventSourceConfig: props.documentDBEventSourceConfig,
584 LoggingConfig: props.loggingConfig,
585 });
586
587 const configToAttrs = (
588 config: lambda.EventSourceMappingConfiguration,
589 ): EventSourceMapping["Attributes"] => ({
590 uuid: config.UUID!,
591 eventSourceMappingArn: config.EventSourceMappingArn!,
592 functionArn: config.FunctionArn!,
593 state: config.State!,
594 });
595
596 return {
597 stables: ["uuid", "eventSourceMappingArn"],
598 diff: Effect.fn(function* ({ news, olds }) {
599 if (!isResolved(news)) return;
600 if (
601 (news.eventSourceArn as string) !== (olds.eventSourceArn as string)
602 ) {
603 return { action: "replace" } as const;
604 }
605 if (news.startingPosition !== olds.startingPosition) {
606 return { action: "replace" } as const;
607 }
608 if (
609 news.startingPositionTimestamp?.getTime() !==
610 olds.startingPositionTimestamp?.getTime()
611 ) {
612 return { action: "replace" } as const;
613 }
614 if (
615 !deepEqual(news.selfManagedEventSource, olds.selfManagedEventSource)
616 ) {
617 return { action: "replace" } as const;
618 }
619 }),
620 reconcile: Effect.fn(function* ({ id, news, output, session }) {
621 const { accountId, region } = yield* AWSEnvironment.current;
622 const expectedInternalTags = yield* createEventSourceMappingTags(id);
623 const desiredTags = { ...expectedInternalTags, ...news.tags };
624
625 const functionName = news.functionName as string;
626 const eventSourceArn = news.eventSourceArn as string;
627
628 // Observe — find the existing mapping. UUIDs are server-assigned
629 // so we either trust `output.uuid` (fast path) or scan
630 // `listEventSourceMappings` and confirm ownership via tags
631 // (recovery from a state-persistence failure or adoption).
632 let config: lambda.EventSourceMappingConfiguration | undefined;
633 if (output?.uuid) {
634 config = yield* lambda
635 .getEventSourceMapping({ UUID: output.uuid })
636 .pipe(
637 Effect.catchTag("ResourceNotFoundException", () =>
638 Effect.succeed(undefined),
639 ),
640 );
641 }
642 if (!config?.UUID) {
643 config = yield* lambda.listEventSourceMappings
644 .pages({ FunctionName: functionName })
645 .pipe(
646 Stream.mapEffect(
647 Effect.fn(function* (page) {
648 const mapping = page.EventSourceMappings?.find(
649 (m) => m.EventSourceArn === eventSourceArn,
650 );
651 if (mapping?.UUID) {
652 const { Tags } = yield* lambda
653 .listTags({
654 Resource: `arn:aws:lambda:${region}:${accountId}:event-source-mapping:${mapping.UUID}`,
655 })
656 .pipe(retryTransient);
657 if (hasTags(expectedInternalTags, Tags)) {
658 return mapping;
659 }
660 }
661 }),
662 ),
663 Stream.filter((item) => item !== undefined),
664 Stream.runHead,
665 Effect.map(Option.getOrUndefined),
666 );
667 }
668
669 // Ensure — create if no live mapping exists. Tolerate
670 // `ResourceConflictException` (peer reconciler raced ahead) by
671 // re-scanning to find the mapping by tag ownership.
672 if (!config?.UUID) {
673 config = yield* lambda
674 .createEventSourceMapping(toCreateRequest(news, desiredTags))
675 .pipe(
676 Effect.catchTags({
677 ResourceConflictException: () =>
678 lambda.listEventSourceMappings
679 .pages({ FunctionName: functionName })
680 .pipe(
681 Stream.mapEffect(
682 Effect.fn(function* (page) {
683 const mapping = page.EventSourceMappings?.find(
684 (m) => m.EventSourceArn === eventSourceArn,
685 );
686 if (mapping?.UUID) {
687 const { Tags } = yield* lambda
688 .listTags({
689 Resource: `arn:aws:lambda:${region}:${accountId}:event-source-mapping:${mapping.UUID}`,
690 })
691 .pipe(retryTransient);
692 if (hasTags(expectedInternalTags, Tags)) {
693 return mapping;
694 }
695 }
696 }),
697 ),
698 Stream.filter((item) => item !== undefined),
699 Stream.runHead,
700 Effect.map(Option.getOrUndefined),
701 Effect.flatMap((mapping) =>
702 mapping
703 ? Effect.succeed(mapping)
704 : Effect.die(
705 new Error(
706 `EventSourceMapping(${id}) not found on function ${functionName}`,
707 ),
708 ),
709 ),
710 ),
711 }),
712 retryPermissionsPropagation,
713 retryTransient,
714 );
715 }
716
717 if (!config?.UUID) {
718 return yield* Effect.die(
719 new Error(`EventSourceMapping(${id}) could not be reconciled`),
720 );
721 }
722
723 const uuid = config.UUID;
724 const mappingArn = `arn:aws:lambda:${region}:${accountId}:event-source-mapping:${uuid}`;
725
726 // Sync configuration — `updateEventSourceMapping` is a full PUT
727 // for mutable fields. We always send the full desired config so
728 // observed state converges. Retry `ResourceInUseException`
729 // (mapping is transitioning) and known IAM-propagation errors.
730 config = yield* lambda
731 .updateEventSourceMapping(toUpdateRequest(uuid, news))
732 .pipe(
733 Effect.retry({
734 while: (e: any) =>
735 e._tag === "ResourceInUseException" ||
736 e._tag === "ResourceConflictException",
737 schedule: Schedule.max([
738 Schedule.exponential(100),
739 Schedule.recurs(20),
740 ]),
741 }),
742 retryPermissionsPropagation,
743 retryTransient,
744 );
745
746 // Wait for the mapping to settle into its desired terminal state.
747 // A fresh mapping sits in `Creating` for up to ~a minute before
748 // Lambda actually starts polling — and for stream sources with
749 // `startingPosition: LATEST`, records written before polling
750 // starts are silently skipped. Returning early would hand the
751 // caller a mapping that looks deployed but drops events.
752 const desiredState = (news.enabled ?? true) ? "Enabled" : "Disabled";
753 if (config.State !== desiredState) {
754 config = yield* lambda.getEventSourceMapping({ UUID: uuid }).pipe(
755 Effect.flatMap((observed) =>
756 observed.State === desiredState
757 ? Effect.succeed(observed)
758 : Effect.fail(
759 new MappingNotSettled({ state: observed.State }),
760 ),
761 ),
762 Effect.retry({
763 while: (e) => e._tag === "MappingNotSettled",
764 schedule: Schedule.max([
765 Schedule.spaced("2 seconds"),
766 Schedule.recurs(45),
767 ]).pipe(
768 Schedule.tap(({ attempt }) =>
769 session.note(
770 `EventSourceMapping ${uuid}: waiting to become ${desiredState} (${(attempt + 1) * 2}s elapsed)`,
771 ),
772 ),
773 ),
774 }),
775 // Never wedge a deploy on a mapping stuck in transition —
776 // surface the observed state in the attributes instead.
777 Effect.catchTag("MappingNotSettled", () =>
778 lambda.getEventSourceMapping({ UUID: uuid }),
779 ),
780 );
781 }
782
783 // Sync tags — diff observed cloud tags against desired so
784 // adoption rewrites ownership tags correctly.
785 const observedTagsResp = yield* lambda
786 .listTags({ Resource: mappingArn })
787 .pipe(retryTransient);
788 const observedTags: Record<string, string> = Object.fromEntries(
789 Object.entries(observedTagsResp.Tags ?? {}).filter(
790 (entry): entry is [string, string] =>
791 typeof entry[1] === "string",
792 ),
793 );
794 const { removed, upsert } = diffTags(observedTags, desiredTags);
795
796 if (removed.length > 0) {
797 yield* lambda
798 .untagResource({ Resource: mappingArn, TagKeys: removed })
799 .pipe(retryTransient);
800 }
801 if (upsert.length > 0) {
802 const tagsToAdd: Record<string, string> = {};
803 for (const { Key, Value } of upsert) {
804 tagsToAdd[Key] = Value;
805 }
806 yield* lambda
807 .tagResource({ Resource: mappingArn, Tags: tagsToAdd })
808 .pipe(retryTransient);
809 }
810
811 yield* session.note(config.EventSourceMappingArn ?? uuid);
812
813 return configToAttrs(config);
814 }),
815 delete: Effect.fn(function* ({ output }) {
816 yield* lambda.deleteEventSourceMapping({ UUID: output.uuid }).pipe(
817 Effect.retry({
818 while: (e: any) =>
819 e._tag === "ResourceInUseException" ||
820 e._tag === "ResourceConflictException",
821 schedule: Schedule.max([
822 Schedule.exponential(100),
823 Schedule.recurs(20),
824 ]),
825 }),
826 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
827 );
828 }),
829 // `list()` — AWS account/region collection (§4a). Exhaustively
830 // paginate `listEventSourceMappings` with no `FunctionName` filter to
831 // enumerate every mapping in the current region, hydrating each into
832 // the exact `Attributes` shape `reconcile`/`configToAttrs` returns.
833 // The list response carries full `EventSourceMappingConfiguration`
834 // objects, so there is no per-item `getEventSourceMapping` to issue —
835 // hence no bounded fan-out or per-item `ResourceNotFoundException`
836 // handling is required. We drop any partial entry missing a required
837 // identifier field rather than emitting a malformed `Attributes`.
838 list: () =>
839 lambda.listEventSourceMappings.pages({}).pipe(
840 Stream.runCollect,
841 Effect.map((chunk) =>
842 Array.from(chunk).flatMap((page) =>
843 (page.EventSourceMappings ?? [])
844 .filter(
845 (
846 config,
847 ): config is lambda.EventSourceMappingConfiguration & {
848 UUID: string;
849 EventSourceMappingArn: string;
850 FunctionArn: string;
851 State: string;
852 } =>
853 config.UUID != null &&
854 config.EventSourceMappingArn != null &&
855 config.FunctionArn != null &&
856 config.State != null,
857 )
858 .map(configToAttrs),
859 ),
860 ),
861 ),
862 };
863 }),
864 );
865
866const retryTransient: <A, R, Err>(
867 self: Effect.Effect<A, Err, R>,
868) => Effect.Effect<A, Err, R> = Effect.retry({
869 while: (e: any) =>
870 e._tag === "InternalFailure" ||
871 e._tag === "RequestExpired" ||
872 e._tag === "ServiceException" ||
873 e._tag === "ServiceUnavailable" ||
874 e._tag === "ThrottlingException" ||
875 e._tag === "TooManyRequestsException" ||
876 e._tag === "RequestLimitExceeded" ||
877 e._tag === "ResourceInUseException",
878 schedule: Schedule.max([Schedule.exponential(100), Schedule.recurs(30)]),
879});
880
881const retryPermissionsPropagation = Effect.retry({
882 while: (e: any) =>
883 e._tag === "InvalidParameterValueException" &&
884 (e.message?.includes(
885 "The function execution role does not have permissions to call",
886 ) ||
887 e.message?.includes("cannot be assumed by Lambda") ||
888 e.message?.includes("Please add Lambda as a Trusted Entity") ||
889 e.message?.includes("Cannot access stream") ||
890 e.message?.includes("Please ensure the role can perform the GetRecords")),
891 schedule: Schedule.max([Schedule.exponential(100), Schedule.recurs(30)]),
892}) as <A, R, Err>(self: Effect.Effect<A, Err, R>) => Effect.Effect<A, Err, R>;
893
894const sanitizeAwsTagValue = (value: string) =>
895 value.replace(/[^\p{L}\p{Z}\p{N}_.:/=+\-@]/gu, "-");