1import * as Lambda from "@/AWS/Lambda";
2import * as S3 from "@/AWS/S3";
3import * as Effect from "effect/Effect";
4import * as Layer from "effect/Layer";
5import * as Stream from "effect/Stream";
6import { HttpServerRequest } from "effect/http/HttpServerRequest";
7import * as HttpServerResponse from "effect/http/HttpServerResponse";
8
9// Derived writes stay outside the watched prefix to avoid recursive events.
10const INCOMING_PREFIX = "incoming/";
11const PROCESSED_PREFIX = "processed/";
12
13const recordKey = (key: string, eventName: string, versionId: string) => {
14 const path = [key, eventName, versionId].map(encodeURIComponent).join("/");
15 return `${PROCESSED_PREFIX}${path}.json`;
16};
17
18export class BucketEventSourceFunction extends Lambda.Function<BucketEventSourceFunction>()(
19 "BucketEventSourceFunction",
20) {}
21
22export default BucketEventSourceFunction.make(
23 {
24 main: import.meta.url,
25 functionUrl: true,
26 },
27 Effect.gen(function* () {
28 const bucket = yield* S3.Bucket("EventSourceBucket", {
29 forceDestroy: true,
30 versioning: "Enabled",
31 });
32
33 const putObject = yield* S3.PutObject(bucket);
34 const getObject = yield* S3.GetObject(bucket);
35 const BucketName = yield* bucket.bucketName;
36
37 yield* S3.consumeBucketEvents(
38 bucket,
39 {
40 events: ["s3:ObjectCreated:*", "s3:ObjectRemoved:*"],
41 prefix: INCOMING_PREFIX,
42 },
43 (stream) =>
44 stream.pipe(
45 Stream.runForEach((event) =>
46 Effect.gen(function* () {
47 const versionId = event.versionId;
48 if (!versionId) {
49 return yield* Effect.fail(
50 new Error("Versioned bucket notification omitted versionId"),
51 );
52 }
53 const key = yield* Effect.sync(() =>
54 recordKey(event.key, event.type, versionId),
55 );
56 // Redelivery can arrive after the source version was deleted.
57 const recorded = yield* getObject({ Key: key }).pipe(
58 Effect.flatMap(({ Body }) => Stream.runDrain(Body!)),
59 Effect.as(true),
60 Effect.catchTag("NoSuchKey", () => Effect.succeed(false)),
61 );
62 if (recorded) return;
63 const object = event.type.startsWith("s3:ObjectCreated:")
64 ? yield* getObject({
65 Key: event.key,
66 VersionId: versionId,
67 })
68 : undefined;
69 const content = object
70 ? yield* Stream.mkString(Stream.decodeText(object.Body!))
71 : undefined;
72 const body = yield* Effect.sync(() =>
73 JSON.stringify({
74 bucket: event.bucket,
75 key: event.key,
76 eventName: event.type,
77 versionId,
78 sequencer: event.sequencer,
79 size: event.size,
80 eTag: event.eTag,
81 content,
82 readVersionId: object?.VersionId,
83 }),
84 );
85 yield* putObject({
86 Key: key,
87 Body: body,
88 ContentType: "application/json",
89 });
90 }).pipe(Effect.orDie),
91 ),
92 ),
93 );
94
95 return {
96 fetch: Effect.gen(function* () {
97 const request = yield* HttpServerRequest;
98 const url = yield* Effect.sync(() => new URL(request.originalUrl));
99 const pathname = url.pathname;
100
101 if (request.method === "GET" && pathname === "/bucket-name") {
102 // The first event after a cold start can observe not-yet-hydrated
103 // resource Outputs — answer 503 so the test retries instead of
104 // recording "undefined" as the bucket name.
105 const bucketName = yield* BucketName;
106 if (!bucketName) {
107 return HttpServerResponse.text("Outputs not hydrated yet", {
108 status: 503,
109 });
110 }
111 return yield* HttpServerResponse.json({ bucketName });
112 }
113
114 if (request.method === "POST" && pathname === "/put") {
115 const body = (yield* request.json) as { key: string; value: string };
116 const result = yield* putObject({
117 Key: `${INCOMING_PREFIX}${body.key}`,
118 Body: body.value,
119 ContentType: "text/plain",
120 });
121 return yield* HttpServerResponse.json({
122 ok: true,
123 versionId: result.VersionId,
124 });
125 }
126
127 if (request.method === "GET" && pathname === "/processed") {
128 const key = url.searchParams.get("key");
129 const eventName = url.searchParams.get("eventName");
130 const versionId = url.searchParams.get("versionId");
131 if (!key || !eventName || !versionId) {
132 return HttpServerResponse.text(
133 "Missing key, eventName, or versionId",
134 { status: 400 },
135 );
136 }
137 const storedKey = yield* Effect.sync(() =>
138 recordKey(key, eventName, versionId),
139 );
140 return yield* getObject({ Key: storedKey }).pipe(
141 Effect.flatMap((result) =>
142 Stream.mkString(Stream.decodeText(result.Body!)),
143 ),
144 Effect.flatMap((text) => Effect.try(() => JSON.parse(text))),
145 Effect.flatMap((processed) =>
146 HttpServerResponse.json({ processed }),
147 ),
148 // Object not written yet — the test polls until it appears.
149 Effect.catchTag("NoSuchKey", () =>
150 HttpServerResponse.json({ processed: null }, { status: 404 }),
151 ),
152 );
153 }
154
155 return yield* HttpServerResponse.json(
156 { error: "Not found", pathname },
157 { status: 404 },
158 );
159 }).pipe(Effect.orDie),
160 };
161 }).pipe(
162 Effect.provide(
163 Layer.mergeAll(
164 Lambda.BucketEventSource,
165 S3.PutObjectHttp,
166 S3.GetObjectHttp,
167 ),
168 ),
169 ),
170);