1import * as Effect from "effect/Effect";
2import * as Layer from "effect/Layer";
3import * as Stream from "effect/Stream";
4import type { Bucket } from "../AWS/S3/Bucket.ts";
5import type {
6 BucketNotification,
7 NotificationsProps,
8} from "../AWS/S3/BucketNotifications.ts";
9import * as S3 from "../AWS/S3/index.ts";
10import type { S3EventType } from "../AWS/S3/S3Event.ts";
11import { normalizeBucketNotification } from "../AWS/S3/normalizeBucketNotification.ts";
12import * as SQS from "../AWS/SQS/index.ts";
13import { SQSQueueEventSource } from "./SQSQueueEventSource.ts";
14
15/** @binding */
16export const S3BucketEventSource = Layer.effect(
17 S3.BucketEventSource,
18 Effect.gen(function* () {
19 const Queue = yield* SQS.Queue;
20
21 return Effect.fn(function* <
22 Events extends S3EventType[],
23 StreamReq = never,
24 Req = never,
25 >(
26 bucket: Bucket,
27 props: NotificationsProps<Events>,
28 process: (
29 stream: Stream.Stream<BucketNotification, never, StreamReq>,
30 ) => Effect.Effect<void, never, Req>,
31 ) {
32 const queue = yield* Queue(`${bucket.LogicalId}-BucketEvents`);
33
34 // Deploy-time: grant the bucket sqs:SendMessage on the queue and attach the
35 // bucket's notification config. Skipped once running inside the deployed
36 // Function (the global guard); the runtime only registers the consumer below.
37 if (!globalThis.__ALCHEMY_RUNTIME__) {
38 const events = props.events ?? ["s3:ObjectCreated:*"];
39 const filterRules = [
40 ...(props.prefix !== undefined
41 ? [{ Name: "prefix" as const, Value: props.prefix }]
42 : []),
43 ...(props.suffix !== undefined
44 ? [{ Name: "suffix" as const, Value: props.suffix }]
45 : []),
46 ];
47 yield* queue.bind(`AWS.SQS.SendMessage(${bucket.LogicalId})`, {
48 policyStatements: [
49 {
50 Sid: `AllowS3EventsFrom${bucket.LogicalId}`,
51 Effect: "Allow",
52 Principal: { Service: "s3.amazonaws.com" },
53 Action: ["sqs:SendMessage"],
54 Resource: [queue.queueArn],
55 Condition: {
56 ArnEquals: {
57 "aws:SourceArn": bucket.bucketArn,
58 },
59 },
60 },
61 ],
62 });
63 yield* bucket.bind(
64 `AWS.S3.NotificationConfiguration(${queue.LogicalId})`,
65 {
66 notificationConfiguration: {
67 QueueConfigurations: [
68 {
69 QueueArn: queue.queueArn,
70 Events: events,
71 ...(filterRules.length > 0
72 ? { Filter: { Key: { FilterRules: filterRules } } }
73 : {}),
74 },
75 ],
76 },
77 },
78 );
79 }
80
81 yield* SQS.consumeQueueMessages(queue, (stream) =>
82 stream.pipe(
83 Stream.mapEffect((record) =>
84 Effect.sync(
85 () =>
86 (JSON.parse(record.body) as { Records?: S3.S3Record[] })
87 .Records ?? [],
88 ),
89 ),
90 Stream.flatMap((records) => Stream.fromArray(records)),
91 Stream.mapEffect(normalizeBucketNotification),
92 process,
93 ),
94 );
95 }) as S3.BucketEventSourceService;
96 }),
97).pipe(Layer.provideMerge(SQSQueueEventSource));