1import type lambda from "aws-lambda";
2import * as Effect from "effect/Effect";
3import * as Layer from "effect/Layer";
4import * as Stream from "effect/Stream";
5
6import * as Namespace from "../../Namespace.ts";
7import type { Bucket } from "../S3/Bucket.ts";
8import {
9 BucketEventSource as S3BucketEventSource,
10 type BucketEventSourceService,
11} from "../S3/BucketEventSource.ts";
12import type {
13 BucketNotification,
14 NotificationsProps,
15} from "../S3/BucketNotifications.ts";
16import type { S3EventType } from "../S3/S3Event.ts";
17import { normalizeBucketNotification } from "../S3/normalizeBucketNotification.ts";
18import * as Lambda from "./Function.ts";
19import { Permission as LambdaPermission } from "./Permission.ts";
20
21/**
22 * Connects an S3 bucket notification stream to the current Lambda function.
23 *
24 * This layer listens for bucket notifications routed through the Lambda runtime
25 * and exposes them as an `Effect.Stream`, while the companion policy configures
26 * the invoke permission and bucket notification binding during deployment.
27 * ### Wiring Events
28 * **Example:** Listen for Object Created Events
29 * ```typescript
30 * yield* AWS.Lambda.BucketEventSource(
31 * bucket,
32 * { events: ["s3:ObjectCreated:*"] },
33 * (events) => Stream.runForEach(events, (event) => Effect.log(event.key)),
34 * );
35 * ```
36 *
37 * @binding
38 */
39export const BucketEventSource = Layer.effect(
40 S3BucketEventSource,
41 Effect.gen(function* () {
42 // this layer can only be used in a Lambda Function
43 const func = yield* Lambda.Function;
44 const Permission = yield* LambdaPermission;
45
46 return Effect.fn(function* <
47 Events extends S3EventType[],
48 StreamReq = never,
49 Req = never,
50 >(
51 bucket: Bucket,
52 props: NotificationsProps<Events>,
53 process: (
54 stream: Stream.Stream<BucketNotification, never, StreamReq>,
55 ) => Effect.Effect<void, never, Req>,
56 ) {
57 // this adds it to the Lambda Function's environment variables
58 const BucketName = yield* bucket.bucketName;
59
60 // Deploy-time: grant the invoke permission and attach the bucket
61 // notification config. Skipped once running inside the deployed Function
62 // (the global guard), where the only work is registering the runtime
63 // handler below. Namespaced under the host so the sub-resources' logical
64 // identity matches the previous Binding.Policy.
65 if (!globalThis.__ALCHEMY_RUNTIME__) {
66 yield* Namespace.push(
67 func.LogicalId,
68 Effect.gen(function* () {
69 const {
70 events: Events = ["s3:ObjectCreated:*"],
71 prefix,
72 suffix,
73 } = props ?? {};
74 const filterRules = [
75 ...(prefix !== undefined
76 ? [{ Name: "prefix" as const, Value: prefix }]
77 : []),
78 ...(suffix !== undefined
79 ? [{ Name: "suffix" as const, Value: suffix }]
80 : []),
81 ];
82 yield* Permission(
83 `AWS.Lambda.InvokeFunction(${bucket.LogicalId})`,
84 {
85 action: "lambda:InvokeFunction",
86 functionName: func.functionName,
87 principal: "s3.amazonaws.com",
88 sourceArn: bucket.bucketArn,
89 },
90 );
91 yield* bucket.bind(`AWS.S3.Notifications(${bucket.LogicalId})`, {
92 notificationConfiguration: {
93 LambdaFunctionConfigurations: [
94 {
95 LambdaFunctionArn: func.functionArn,
96 Events,
97 ...(filterRules.length > 0
98 ? { Filter: { Key: { FilterRules: filterRules } } }
99 : {}),
100 },
101 ],
102 },
103 });
104 }),
105 );
106 }
107
108 yield* func.listen(
109 Effect.gen(function* () {
110 // this accesses it
111 const bucketName = yield* BucketName;
112 return (event: any) => {
113 if (isS3Event(event)) {
114 const events = event.Records.filter(
115 (record) => record.s3.bucket.name === bucketName,
116 );
117 if (events.length > 0) {
118 return process(
119 Stream.fromArray(events).pipe(
120 Stream.mapEffect(normalizeBucketNotification),
121 ),
122 // TODO(sam): don't die?
123 ).pipe(Effect.orDie);
124 }
125 }
126 };
127 }),
128 );
129 }) as BucketEventSourceService;
130 }),
131);
132
133const isS3Event = (event: any): event is lambda.S3Event =>
134 Array.isArray(event.Records) &&
135 event.Records.some((record: any) => record.s3);