1import * as AWS from "@/AWS";
2import * as Core from "@/Test/Core";
3import * as Test from "@/Test/Alchemy";
4import * as Lambda from "@distilled.cloud/aws/lambda";
5import * as S3 from "@distilled.cloud/aws/s3";
6import { describe, expect } from "alchemy-test";
7import * as Data from "effect/Data";
8import * as Effect from "effect/Effect";
9import * as Schedule from "effect/Schedule";
10import * as Schema from "effect/Schema";
11import * as Stream from "effect/Stream";
12import * as HttpClient from "effect/http/HttpClient";
13import * as HttpClientRequest from "effect/http/HttpClientRequest";
14import BucketEventSourceFunctionLive, {
15 BucketEventSourceFunction,
16} from "./fixtures/event-source-handler.ts";
17
18const testOptions = { providers: AWS.providers() };
19const { test, beforeAll, afterAll } = Test.make(testOptions);
20const sharedStack = Core.scratchStack(
21 testOptions,
22 "S3EventSource",
23 "test/AWS/S3/EventSource.test.ts",
24);
25
26const readinessPolicy = Schedule.spaced("2 seconds");
27
28let baseUrl: string;
29let fixtureFunctionName: string;
30let fixtureBucketName: string | undefined;
31
32describe(
33 "S3 Bucket Event Source",
34 { tags: ["provider:aws", "provider:aws:lambda", "provider:aws:s3", "live"] },
35 () => {
36 beforeAll(
37 Effect.gen(function* () {
38 yield* Effect.logInfo(
39 "S3 EventSource test: destroying previous resources",
40 );
41 yield* sharedStack.destroy();
42
43 yield* Effect.logInfo("S3 EventSource test: deploying fixture");
44 const { functionUrl, functionName } = yield* sharedStack.deploy(
45 Effect.gen(function* () {
46 return yield* BucketEventSourceFunction;
47 }).pipe(Effect.provide(BucketEventSourceFunctionLive)),
48 );
49
50 expect(functionUrl).toBeTruthy();
51 baseUrl = functionUrl!.replace(/\/+$/, "");
52 fixtureFunctionName = functionName;
53
54 yield* Effect.logInfo(
55 `S3 EventSource test: function URL ready (${functionUrl}), probing readiness`,
56 );
57
58 const named = yield* HttpClient.get(`${baseUrl}/bucket-name`).pipe(
59 Effect.flatMap((response) =>
60 response.status === 200
61 ? response.json
62 : response.text.pipe(
63 Effect.flatMap((body) =>
64 Effect.fail(
65 new FunctionNotReady({ status: response.status, body }),
66 ),
67 ),
68 ),
69 ),
70 Effect.flatMap(
71 Schema.decodeUnknownEffect(
72 Schema.Struct({ bucketName: Schema.String }),
73 ),
74 ),
75 Effect.tapError((error) =>
76 Effect.logWarning(
77 `S3 EventSource test: fixture not ready yet (${String(error)})`,
78 ),
79 ),
80 Effect.retry({ schedule: readinessPolicy, times: 9 }),
81 );
82 fixtureBucketName = named.bucketName;
83 expect(fixtureBucketName).toBeTruthy();
84 yield* Effect.logInfo(
85 "S3 EventSource test: fixture responded successfully",
86 );
87 }),
88 { timeout: 120_000 },
89 );
90
91 afterAll(
92 Effect.gen(function* () {
93 yield* sharedStack.destroy();
94 // Prove the trailing destroy removed the fixture bucket (skip if
95 // setup never captured the name). afterAll lacks the providers layer
96 // test bodies get, so provide it for the out-of-band distilled call.
97 if (fixtureBucketName) {
98 yield* Core.withProviders(
99 assertBucketDeleted(fixtureBucketName),
100 testOptions,
101 "S3EventSource",
102 );
103 }
104 }),
105 { timeout: 120_000 },
106 );
107
108 test.provider(
109 "object created under the watched prefix triggers the subscription to write derived state",
110 (_stack) =>
111 Effect.gen(function* () {
112 const key = "e2e-object";
113
114 const content = "hello from s3 event source";
115 const response = yield* HttpClient.execute(
116 HttpClientRequest.bodyJsonUnsafe(
117 HttpClientRequest.post(`${baseUrl}/put`),
118 { key, value: content },
119 ),
120 );
121 expect(response.status).toBe(200);
122 const putResponse = yield* response.json.pipe(
123 Effect.flatMap(
124 Schema.decodeUnknownEffect(
125 Schema.Struct({ ok: Schema.Boolean, versionId: Schema.String }),
126 ),
127 ),
128 );
129 expect(putResponse.ok).toBe(true);
130 expect(putResponse.versionId).toBeTruthy();
131 const processed = yield* waitForProcessed({
132 key: `incoming/${key}`,
133 eventName: "s3:ObjectCreated:Put",
134 versionId: putResponse.versionId,
135 });
136 expect(processed.size).toBe(content.length);
137 expect(processed.eTag).toBeTruthy();
138 expect(processed.content).toBe(content);
139 expect(processed.readVersionId).toBe(putResponse.versionId);
140 }),
141 { timeout: 120_000 },
142 );
143
144 test.provider(
145 "preserves both PUT versions and records delete-marker and explicit-version removal events",
146 () =>
147 Effect.gen(function* () {
148 const Bucket = fixtureBucketName!;
149 const key = "incoming/versions/a b+%?#/雪.txt";
150 const oldContent = "historical notification content";
151 const currentContent =
152 "current notification content has different bytes";
153 const oldVersion = yield* S3.putObject({
154 Bucket,
155 Key: key,
156 Body: oldContent,
157 });
158 const currentVersion = yield* S3.putObject({
159 Bucket,
160 Key: key,
161 Body: currentContent,
162 });
163 expect(oldVersion.VersionId).toBeTruthy();
164 expect(currentVersion.VersionId).toBeTruthy();
165 expect(currentVersion.VersionId).not.toBe(oldVersion.VersionId);
166 const oldIdentity = {
167 key,
168 eventName: "s3:ObjectCreated:Put",
169 versionId: oldVersion.VersionId!,
170 };
171 const currentIdentity = {
172 key,
173 eventName: "s3:ObjectCreated:Put",
174 versionId: currentVersion.VersionId!,
175 };
176 const [oldEvent, currentEvent] = yield* Effect.all(
177 [waitForProcessed(oldIdentity), waitForProcessed(currentIdentity)],
178 { concurrency: 2 },
179 );
180 expect(oldEvent.content).toBe(oldContent);
181 expect(oldEvent.size).toBe(oldContent.length);
182 expect(oldEvent.readVersionId).toBe(oldVersion.VersionId);
183 expect(currentEvent.content).toBe(currentContent);
184 expect(currentEvent.size).toBe(currentContent.length);
185 expect(currentEvent.readVersionId).toBe(currentVersion.VersionId);
186 expect(oldEvent.sequencer).not.toBe(currentEvent.sequencer);
187
188 const marker = yield* S3.deleteObject({ Bucket, Key: key });
189 expect(marker.DeleteMarker).toBe(true);
190 expect(marker.VersionId).toBeTruthy();
191 expect(marker.VersionId).not.toBe(oldVersion.VersionId);
192 expect(marker.VersionId).not.toBe(currentVersion.VersionId);
193 const removed = yield* S3.deleteObject({
194 Bucket,
195 Key: key,
196 VersionId: oldVersion.VersionId!,
197 });
198 expect(removed.VersionId).toBe(oldVersion.VersionId);
199 const [markerEvent, removedEvent] = yield* Effect.all(
200 [
201 waitForProcessed({
202 key,
203 eventName: "s3:ObjectRemoved:DeleteMarkerCreated",
204 versionId: marker.VersionId!,
205 }),
206 waitForProcessed({
207 key,
208 eventName: "s3:ObjectRemoved:Delete",
209 versionId: removed.VersionId!,
210 }),
211 ],
212 { concurrency: 2 },
213 );
214 for (const event of [markerEvent, removedEvent]) {
215 expect(event.content).toBeUndefined();
216 expect(event.readVersionId).toBeUndefined();
217 }
218
219 // Removal records must not replace the earlier creation histories.
220 expect(yield* readProcessed(oldIdentity)).toEqual(oldEvent);
221 expect(yield* readProcessed(currentIdentity)).toEqual(currentEvent);
222 }),
223 { timeout: 120_000 },
224 );
225
226 test.provider(
227 "replays a creation event through the deployed Lambda after permanent source-version deletion",
228 () =>
229 Effect.gen(function* () {
230 const Bucket = fixtureBucketName!;
231 const key = "incoming/replay/a b+%?#/雪.txt";
232 const content = "recorded before permanent version deletion";
233 const created = yield* S3.putObject({
234 Bucket,
235 Key: key,
236 Body: content,
237 });
238 expect(created.VersionId).toBeTruthy();
239 const identity = {
240 key,
241 eventName: "s3:ObjectCreated:Put",
242 versionId: created.VersionId!,
243 };
244 const recorded = yield* waitForProcessed(identity);
245 expect(recorded.content).toBe(content);
246 expect(recorded.readVersionId).toBe(created.VersionId);
247 const evidenceKey = yield* Effect.sync(
248 () =>
249 `processed/${[key, identity.eventName, identity.versionId].map(encodeURIComponent).join("/")}.json`,
250 );
251 const before = yield* S3.headObject({ Bucket, Key: evidenceKey });
252 expect(before.VersionId).toBeTruthy();
253
254 yield* S3.deleteObject({
255 Bucket,
256 Key: key,
257 VersionId: identity.versionId,
258 });
259 const sourceExists = yield* S3.headObject({
260 Bucket,
261 Key: key,
262 VersionId: identity.versionId,
263 }).pipe(
264 Effect.as(true),
265 Effect.catchTag("NotFound", () => Effect.succeed(false)),
266 );
267 expect(sourceExists).toBe(false);
268
269 const region = yield* AWS.Region;
270 const Payload = yield* Effect.sync(() =>
271 JSON.stringify({
272 Records: [
273 {
274 eventVersion: "2.1",
275 eventSource: "aws:s3",
276 awsRegion: region,
277 eventName: "ObjectCreated:Put",
278 s3: {
279 s3SchemaVersion: "1.0",
280 bucket: { name: Bucket, arn: `arn:aws:s3:::${Bucket}` },
281 object: {
282 key: encodeURIComponent(key).replace(/%20/g, "+"),
283 versionId: identity.versionId,
284 sequencer: recorded.sequencer,
285 size: recorded.size,
286 eTag: recorded.eTag,
287 },
288 },
289 },
290 ],
291 }),
292 );
293 const replay = yield* Lambda.invoke({
294 FunctionName: fixtureFunctionName,
295 InvocationType: "RequestResponse",
296 Payload,
297 });
298 if (replay.Payload) yield* Stream.runDrain(replay.Payload);
299 expect(replay.StatusCode).toBe(200);
300 expect(replay.FunctionError).toBeUndefined();
301 expect(yield* readProcessed(identity)).toEqual(recorded);
302 const after = yield* S3.headObject({ Bucket, Key: evidenceKey });
303 expect(after.VersionId).toBe(before.VersionId);
304 }),
305 { timeout: 120_000 },
306 );
307
308 test.provider(
309 "records a copied version and reads its historical content after a later PUT",
310 () =>
311 Effect.gen(function* () {
312 const Bucket = fixtureBucketName!;
313 const sourceKey = "sources/copy.txt";
314 const key = "incoming/copied/a b+%.txt";
315 const content = "historical copy source content";
316 const source = yield* S3.putObject({
317 Bucket,
318 Key: sourceKey,
319 Body: content,
320 });
321 expect(source.VersionId).toBeTruthy();
322 yield* S3.putObject({
323 Bucket,
324 Key: sourceKey,
325 Body: "new source content must not be copied",
326 });
327 const CopySource = yield* Effect.sync(
328 () =>
329 `${Bucket}/${sourceKey}?versionId=${encodeURIComponent(source.VersionId!)}`,
330 );
331 const copied = yield* S3.copyObject({
332 Bucket,
333 Key: key,
334 CopySource,
335 });
336 expect(copied.VersionId).toBeTruthy();
337 expect(copied.CopySourceVersionId).toBe(source.VersionId);
338 const latestContent = "later destination content";
339 const latest = yield* S3.putObject({
340 Bucket,
341 Key: key,
342 Body: latestContent,
343 });
344 expect(latest.VersionId).toBeTruthy();
345 expect(latest.VersionId).not.toBe(copied.VersionId);
346 const [copyEvent, putEvent] = yield* Effect.all(
347 [
348 waitForProcessed({
349 key,
350 eventName: "s3:ObjectCreated:Copy",
351 versionId: copied.VersionId!,
352 }),
353 waitForProcessed({
354 key,
355 eventName: "s3:ObjectCreated:Put",
356 versionId: latest.VersionId!,
357 }),
358 ],
359 { concurrency: 2 },
360 );
361 expect(copyEvent.content).toBe(content);
362 expect(copyEvent.size).toBe(content.length);
363 expect(copyEvent.eTag).toBeTruthy();
364 expect(copyEvent.readVersionId).toBe(copied.VersionId);
365 expect(putEvent.content).toBe(latestContent);
366 expect(putEvent.readVersionId).toBe(latest.VersionId);
367 }),
368 { timeout: 120_000 },
369 );
370
371 test.provider(
372 "records a completed multipart version and reads its content after a later PUT",
373 () =>
374 Effect.gen(function* () {
375 const Bucket = fixtureBucketName!;
376 const key = "incoming/multipart/a b+%.txt";
377 const content = "single final multipart part";
378 const upload = yield* S3.createMultipartUpload({ Bucket, Key: key });
379 expect(upload.UploadId).toBeTruthy();
380 const part = yield* S3.uploadPart({
381 Bucket,
382 Key: key,
383 UploadId: upload.UploadId!,
384 PartNumber: 1,
385 Body: content,
386 });
387 expect(part.ETag).toBeTruthy();
388 const completed = yield* S3.completeMultipartUpload({
389 Bucket,
390 Key: key,
391 UploadId: upload.UploadId!,
392 MultipartUpload: { Parts: [{ PartNumber: 1, ETag: part.ETag! }] },
393 });
394 expect(completed.VersionId).toBeTruthy();
395 const latestContent = "later multipart destination content";
396 const latest = yield* S3.putObject({
397 Bucket,
398 Key: key,
399 Body: latestContent,
400 });
401 expect(latest.VersionId).toBeTruthy();
402 expect(latest.VersionId).not.toBe(completed.VersionId);
403 const [completedEvent, putEvent] = yield* Effect.all(
404 [
405 waitForProcessed({
406 key,
407 eventName: "s3:ObjectCreated:CompleteMultipartUpload",
408 versionId: completed.VersionId!,
409 }),
410 waitForProcessed({
411 key,
412 eventName: "s3:ObjectCreated:Put",
413 versionId: latest.VersionId!,
414 }),
415 ],
416 { concurrency: 2 },
417 );
418 expect(completedEvent.content).toBe(content);
419 expect(completedEvent.size).toBe(content.length);
420 expect(completedEvent.eTag).toBeTruthy();
421 expect(completedEvent.readVersionId).toBe(completed.VersionId);
422 expect(putEvent.content).toBe(latestContent);
423 expect(putEvent.readVersionId).toBe(latest.VersionId);
424 }),
425 { timeout: 120_000 },
426 );
427 },
428);
429
430interface NotificationIdentity {
431 key: string;
432 eventName: string;
433 versionId: string;
434}
435
436const processedRecord = Schema.Struct({
437 bucket: Schema.String,
438 key: Schema.String,
439 eventName: Schema.String,
440 versionId: Schema.String,
441 sequencer: Schema.String,
442 size: Schema.optional(Schema.Number),
443 eTag: Schema.optional(Schema.String),
444 content: Schema.optional(Schema.String),
445 readVersionId: Schema.optional(Schema.String),
446});
447
448const readProcessed = Effect.fn(function* (identity: NotificationIdentity) {
449 const path = yield* Effect.sync(
450 () => `/processed?${new URLSearchParams({ ...identity })}`,
451 );
452 const response = yield* HttpClient.get(`${baseUrl}${path}`);
453 if (response.status === 404) {
454 yield* response.text;
455 return yield* Effect.fail(new ProcessedNotReady(identity));
456 }
457 if (response.status !== 200) {
458 return yield* Effect.fail(
459 new FunctionNotReady({
460 status: response.status,
461 body: yield* response.text,
462 }),
463 );
464 }
465 const { processed } = yield* response.json.pipe(
466 Effect.flatMap(
467 Schema.decodeUnknownEffect(Schema.Struct({ processed: processedRecord })),
468 ),
469 );
470 expect(processed.bucket).toBe(fixtureBucketName);
471 expect(processed.key).toBe(identity.key);
472 expect(processed.eventName).toBe(identity.eventName);
473 expect(processed.versionId).toBe(identity.versionId);
474 expect(processed.sequencer).toMatch(/^[0-9a-f]+$/i);
475 return processed;
476});
477
478const waitForProcessed = (identity: NotificationIdentity) =>
479 readProcessed(identity).pipe(
480 Effect.retry({
481 while: (error) => error._tag === "ProcessedNotReady",
482 schedule: Schedule.spaced("5 seconds"),
483 times: 9,
484 }),
485 );
486
487class ProcessedNotReady extends Data.TaggedError(
488 "ProcessedNotReady",
489)<NotificationIdentity> {}
490
491class BucketStillExists extends Data.TaggedError("BucketStillExists") {}
492
493// Out-of-band assert-gone after the final destroy: retry while headBucket
494// still succeeds (S3 delete visibility is eventually consistent), settle on
495// the typed NotFound.
496const assertBucketDeleted = Effect.fn(function* (name: string) {
497 yield* S3.headBucket({ Bucket: name }).pipe(
498 Effect.flatMap(() => Effect.fail(new BucketStillExists())),
499 Effect.retry({
500 while: (e) => e._tag === "BucketStillExists",
501 schedule: Schedule.spaced("5 seconds"),
502 times: 9,
503 }),
504 Effect.catchTag("NotFound", () => Effect.void),
505 );
506});
507
508class FunctionNotReady extends Data.TaggedError("FunctionNotReady")<{
509 readonly status: number;
510 readonly body: string;
511}> {}