1import * as AWS from "@/AWS";
2import * as Core from "@/Test/Core";
3import * as Test from "@/Test/Alchemy";
4import * as EC2 from "@distilled.cloud/aws/ec2";
5import * as ECS from "@distilled.cloud/aws/ecs";
6import * as S3 from "@distilled.cloud/aws/s3";
7import * as SQS from "@distilled.cloud/aws/sqs";
8import { describe, expect } from "alchemy-test";
9import * as Data from "effect/Data";
10import * as Effect from "effect/Effect";
11import * as Schedule from "effect/Schedule";
12import * as Schema from "effect/Schema";
13import * as Stream from "effect/Stream";
14import ServerEventTask, {
15 ACKNOWLEDGED_PREFIX,
16 artifactKey,
17 INCOMING_PREFIX,
18 INCOMING_SUFFIX,
19 PROCESSED_PREFIX,
20 RECEIVED_PREFIX,
21 ServerEventBucket,
22} from "./fixtures/server-event-source-task.ts";
23
24const testOptions = {
25 providers: AWS.providers(),
26 dev: false,
27 stage: `${Core.defaultStage()}_s3_server_events`,
28};
29const { test, beforeAll, afterAll } = Test.make(testOptions);
30const stackName = "S3ServerEventSource";
31const stack = Core.scratchStack(
32 testOptions,
33 stackName,
34 "test/AWS/S3/ServerEventSource.test.ts",
35);
36
37interface Deployment {
38 bucketName: string;
39 bucketArn: string;
40 clusterArn: string;
41 serviceName: string;
42 taskDefinitionArn: string;
43}
44
45let deployed: Deployment | undefined;
46let queueUrl: string | undefined;
47
48// Cold Docker/ECR/Fargate startup can exceed 120s; opt in with warm build
49// caches and an existing public default VPC. There is no ALB or HTTP probe.
50describe.skipIf(!process.env.AWS_TEST_SLOW || !!process.env.FAST).concurrent(
51 "S3 Server/SQS event source on ECS",
52 {
53 tags: [
54 "provider:aws",
55 "provider:aws:ec2",
56 "provider:aws:ecs",
57 "provider:aws:s3",
58 "provider:aws:sqs",
59 "live",
60 ],
61 },
62 () => {
63 beforeAll(Core.withProviders(stack.destroy(), testOptions, stackName), {
64 timeout: 120_000,
65 });
66 beforeAll(
67 Core.withProviders(
68 Effect.gen(function* () {
69 const network = yield* EC2.describeVpcs({
70 Filters: [{ Name: "is-default", Values: ["true"] }],
71 });
72 const vpc = network.Vpcs?.find((candidate) => candidate.IsDefault);
73 if (!vpc?.VpcId) {
74 return yield* Effect.fail(
75 new Error(
76 "Server event acceptance requires an existing default VPC",
77 ),
78 );
79 }
80 const vpcId = AWS.EC2.VpcId(vpc.VpcId);
81 const subnets = yield* EC2.describeSubnets({
82 Filters: [
83 { Name: "vpc-id", Values: [vpcId] },
84 { Name: "default-for-az", Values: ["true"] },
85 { Name: "state", Values: ["available"] },
86 ],
87 });
88 const subnetIds = (subnets.Subnets ?? [])
89 .flatMap((subnet) => (subnet.SubnetId ? [subnet.SubnetId] : []))
90 .sort();
91 if (subnetIds.length === 0) {
92 return yield* Effect.fail(
93 new Error(
94 "Server event acceptance requires an available public default subnet",
95 ),
96 );
97 }
98
99 deployed = yield* stack.deploy(
100 Effect.gen(function* () {
101 const bucket = yield* ServerEventBucket;
102 const cluster = yield* AWS.ECS.Cluster("S3ServerEventCluster");
103 const securityGroup = yield* AWS.EC2.SecurityGroup(
104 "S3ServerEventSecurityGroup",
105 {
106 vpcId,
107 description:
108 "S3 notification consumer with outbound access only",
109 ingress: [],
110 egress: [{ ipProtocol: "-1", cidrIpv4: "0.0.0.0/0" }],
111 },
112 );
113 const task = yield* ServerEventTask;
114 const service = yield* AWS.ECS.Service("S3ServerEventService", {
115 cluster,
116 task: {
117 taskDefinitionArn: task.taskDefinitionArn,
118 containerName: task.containerName,
119 port: task.port,
120 },
121 desiredCount: 1,
122 loadBalancer: false,
123 vpcId,
124 subnets: [subnetIds[0]!],
125 securityGroups: [securityGroup.groupId],
126 assignPublicIp: true,
127 deploymentStabilizationTimeout: "90 seconds",
128 });
129 return {
130 bucketName: bucket.bucketName,
131 bucketArn: bucket.bucketArn,
132 clusterArn: cluster.clusterArn,
133 serviceName: service.serviceName,
134 taskDefinitionArn: task.taskDefinitionArn,
135 };
136 }),
137 );
138
139 const notifications = yield* S3.getBucketNotificationConfiguration({
140 Bucket: deployed.bucketName,
141 });
142 expect(notifications.QueueConfigurations).toHaveLength(1);
143 const queueArn = notifications.QueueConfigurations![0]!.QueueArn;
144 const queueName = queueArn.split(":").at(-1)!;
145 const queue = yield* SQS.getQueueUrl({ QueueName: queueName });
146 expect(queue.QueueUrl).toBeTruthy();
147 queueUrl = queue.QueueUrl!;
148 }),
149 testOptions,
150 stackName,
151 ),
152 { timeout: 120_000 },
153 );
154
155 afterAll(
156 Core.withProviders(
157 Effect.gen(function* () {
158 yield* stack.destroy();
159 if (deployed) {
160 yield* S3.headBucket({ Bucket: deployed.bucketName }).pipe(
161 Effect.flatMap(() =>
162 Effect.fail(
163 new FixtureResourceStillExists({ resource: "bucket" }),
164 ),
165 ),
166 Effect.retry({
167 while: (error) => error._tag === "FixtureResourceStillExists",
168 schedule: Schedule.spaced("4 seconds"),
169 times: 9,
170 }),
171 Effect.catchTag("NotFound", () => Effect.void),
172 Effect.timeout("45 seconds"),
173 );
174 const clusters = yield* ECS.describeClusters({
175 clusters: [deployed.clusterArn],
176 });
177 expect(
178 (clusters.clusters ?? []).filter(
179 (cluster) => cluster.status !== "INACTIVE",
180 ),
181 ).toEqual([]);
182 }
183 if (queueUrl) {
184 yield* SQS.getQueueAttributes({
185 QueueUrl: queueUrl,
186 AttributeNames: ["QueueArn"],
187 }).pipe(
188 Effect.flatMap(() =>
189 Effect.fail(
190 new FixtureResourceStillExists({ resource: "queue" }),
191 ),
192 ),
193 Effect.retry({
194 while: (error) => error._tag === "FixtureResourceStillExists",
195 schedule: Schedule.spaced("4 seconds"),
196 times: 9,
197 }),
198 Effect.catchTag("QueueDoesNotExist", () => Effect.void),
199 Effect.timeout("45 seconds"),
200 );
201 }
202 }),
203 testOptions,
204 stackName,
205 ),
206 { timeout: 120_000 },
207 );
208
209 test.provider(
210 "deploys an ECS consumer with an S3 principal and both notification filters",
211 () =>
212 Effect.gen(function* () {
213 const fixture = deployed!;
214 const result = yield* ECS.describeServices({
215 cluster: fixture.clusterArn,
216 services: [fixture.serviceName],
217 });
218 expect(result.failures ?? []).toEqual([]);
219 expect(result.services).toHaveLength(1);
220 const service = result.services![0]!;
221 expect(service.taskDefinition).toBe(fixture.taskDefinitionArn);
222 expect(service.runningCount).toBe(1);
223 expect(service.pendingCount).toBe(0);
224 expect(service.loadBalancers ?? []).toEqual([]);
225
226 const notifications = yield* S3.getBucketNotificationConfiguration({
227 Bucket: fixture.bucketName,
228 });
229 expect(notifications.QueueConfigurations).toHaveLength(1);
230 expect(notifications.LambdaFunctionConfigurations ?? []).toEqual([]);
231 const configuration = notifications.QueueConfigurations![0]!;
232 expect(configuration.Events).toEqual([
233 "s3:ObjectCreated:*",
234 "s3:ObjectRemoved:*",
235 ]);
236 expect(
237 configuration.Filter?.Key?.FilterRules?.map((rule) => ({
238 ...rule,
239 Name: rule.Name?.toLowerCase(),
240 })),
241 ).toEqual([
242 { Name: "prefix", Value: INCOMING_PREFIX },
243 { Name: "suffix", Value: INCOMING_SUFFIX },
244 ]);
245
246 const attributes = yield* SQS.getQueueAttributes({
247 QueueUrl: queueUrl!,
248 AttributeNames: ["Policy", "QueueArn"],
249 });
250 expect(attributes.Attributes?.QueueArn).toBe(configuration.QueueArn);
251 const policy = yield* Effect.try(() =>
252 JSON.parse(attributes.Attributes!.Policy!),
253 );
254 expect(policy).toEqual({
255 Version: "2012-10-17",
256 Statement: [
257 {
258 Sid: "AllowS3EventsFromServerEventBucket",
259 Effect: "Allow",
260 Principal: { Service: "s3.amazonaws.com" },
261 Action: "sqs:SendMessage",
262 Resource: configuration.QueueArn,
263 Condition: {
264 ArnEquals: { "aws:SourceArn": fixture.bucketArn },
265 },
266 },
267 ],
268 });
269 }),
270 { timeout: 120_000 },
271 );
272
273 test.provider(
274 "records exact overwritten versions, delete markers, and version deletions through SQS",
275 () =>
276 Effect.gen(function* () {
277 const Bucket = deployed!.bucketName;
278 const key = `${INCOMING_PREFIX}versions/space + percent% question? hash# 雪${INCOMING_SUFFIX}`;
279 const first = yield* S3.putObject({
280 Bucket,
281 Key: key,
282 Body: "first version",
283 });
284 const second = yield* S3.putObject({
285 Bucket,
286 Key: key,
287 Body: "second version",
288 });
289 expect(first.VersionId).toBeTruthy();
290 expect(second.VersionId).toBeTruthy();
291 expect(first.VersionId).not.toBe(second.VersionId);
292
293 const created = yield* waitForArtifacts([
294 {
295 key,
296 eventName: "s3:ObjectCreated:Put",
297 versionId: first.VersionId!,
298 },
299 {
300 key,
301 eventName: "s3:ObjectCreated:Put",
302 versionId: second.VersionId!,
303 },
304 ]);
305 expect(created.map((record) => record.content)).toEqual([
306 "first version",
307 "second version",
308 ]);
309 expect(created.map((record) => record.readVersionId)).toEqual([
310 first.VersionId,
311 second.VersionId,
312 ]);
313 expect(created[0]!.size).toBe("first version".length);
314 expect(created[1]!.size).toBe("second version".length);
315 expect(created.every((record) => !!record.eTag)).toBe(true);
316 expect(created[0]!.sequencer).not.toBe(created[1]!.sequencer);
317
318 const marker = yield* S3.deleteObject({ Bucket, Key: key });
319 expect(marker.DeleteMarker).toBe(true);
320 expect(marker.VersionId).toBeTruthy();
321 const deleted = yield* S3.deleteObject({
322 Bucket,
323 Key: key,
324 VersionId: first.VersionId!,
325 });
326 expect(deleted.VersionId).toBe(first.VersionId);
327 const removed = yield* waitForArtifacts([
328 {
329 key,
330 eventName: "s3:ObjectRemoved:DeleteMarkerCreated",
331 versionId: marker.VersionId!,
332 },
333 {
334 key,
335 eventName: "s3:ObjectRemoved:Delete",
336 versionId: first.VersionId!,
337 },
338 ]);
339 expect(removed.every((record) => record.content === undefined)).toBe(
340 true,
341 );
342 expect(
343 removed.every((record) => record.readVersionId === undefined),
344 ).toBe(true);
345
346 const versions = yield* S3.listObjectVersions({
347 Bucket,
348 Prefix: key,
349 });
350 expect(
351 (versions.Versions ?? []).map((version) => version.VersionId),
352 ).toEqual([second.VersionId]);
353 expect(versions.DeleteMarkers).toHaveLength(1);
354 expect(versions.DeleteMarkers![0]!.VersionId).toBe(marker.VersionId);
355 expect(versions.DeleteMarkers![0]!.IsLatest).toBe(true);
356
357 yield* assertArtifactCount(key, 4);
358 }),
359 { timeout: 120_000 },
360 );
361
362 test.provider(
363 "records Copy notifications with the exact source and destination versions",
364 () =>
365 Effect.gen(function* () {
366 const Bucket = deployed!.bucketName;
367 const sourceKey = `copy-source/space + percent% 雪${INCOMING_SUFFIX}`;
368 const key = `${INCOMING_PREFIX}copy/exact version + 雪${INCOMING_SUFFIX}`;
369 const content = "copied version: 雪 + %";
370 const source = yield* S3.putObject({
371 Bucket,
372 Key: sourceKey,
373 Body: content,
374 });
375 expect(source.VersionId).toBeTruthy();
376 yield* S3.putObject({
377 Bucket,
378 Key: sourceKey,
379 Body: "newer source",
380 });
381 const CopySource = yield* Effect.sync(
382 () =>
383 `${Bucket}/${encodeURIComponent(sourceKey)}?versionId=${encodeURIComponent(source.VersionId!)}`,
384 );
385 const copied = yield* S3.copyObject({
386 Bucket,
387 Key: key,
388 CopySource,
389 });
390 expect(copied.CopySourceVersionId).toBe(source.VersionId);
391 expect(copied.VersionId).toBeTruthy();
392 const overwritten = yield* S3.putObject({
393 Bucket,
394 Key: key,
395 Body: "newer destination",
396 });
397 expect(overwritten.VersionId).not.toBe(copied.VersionId);
398 const records = yield* waitForArtifacts([
399 {
400 key,
401 eventName: "s3:ObjectCreated:Copy",
402 versionId: copied.VersionId!,
403 },
404 {
405 key,
406 eventName: "s3:ObjectCreated:Put",
407 versionId: overwritten.VersionId!,
408 },
409 ]);
410 yield* assertCreation(
411 records[0]!,
412 content,
413 copied.VersionId!,
414 copied.CopyObjectResult?.ETag,
415 );
416 yield* assertCreation(
417 records[1]!,
418 "newer destination",
419 overwritten.VersionId!,
420 overwritten.ETag,
421 );
422 yield* assertArtifactCount(key, 2);
423 yield* assertArtifactCount(sourceKey, 0);
424 }),
425 { timeout: 120_000 },
426 );
427
428 test.provider(
429 "records completed multipart notifications with the exact version and content",
430 () =>
431 Effect.gen(function* () {
432 const Bucket = deployed!.bucketName;
433 const key = `${INCOMING_PREFIX}multipart/completed + 雪${INCOMING_SUFFIX}`;
434 const content = "completed multipart version: 雪 + %";
435 const upload = yield* S3.createMultipartUpload({
436 Bucket,
437 Key: key,
438 });
439 expect(upload.UploadId).toBeTruthy();
440 const UploadId = upload.UploadId!;
441 yield* Effect.gen(function* () {
442 // A single final part may be smaller than S3's 5 MiB minimum.
443 const part = yield* S3.uploadPart({
444 Bucket,
445 Key: key,
446 UploadId,
447 PartNumber: 1,
448 Body: content,
449 });
450 expect(part.ETag).toBeTruthy();
451 yield* assertArtifactCount(key, 0);
452 const completed = yield* S3.completeMultipartUpload({
453 Bucket,
454 Key: key,
455 UploadId,
456 MultipartUpload: {
457 Parts: [{ PartNumber: 1, ETag: part.ETag! }],
458 },
459 });
460 expect(completed.VersionId).toBeTruthy();
461 const overwritten = yield* S3.putObject({
462 Bucket,
463 Key: key,
464 Body: "after multipart",
465 });
466 expect(overwritten.VersionId).not.toBe(completed.VersionId);
467 const records = yield* waitForArtifacts([
468 {
469 key,
470 eventName: "s3:ObjectCreated:CompleteMultipartUpload",
471 versionId: completed.VersionId!,
472 },
473 {
474 key,
475 eventName: "s3:ObjectCreated:Put",
476 versionId: overwritten.VersionId!,
477 },
478 ]);
479 yield* assertCreation(
480 records[0]!,
481 content,
482 completed.VersionId!,
483 completed.ETag,
484 );
485 yield* assertCreation(
486 records[1]!,
487 "after multipart",
488 overwritten.VersionId!,
489 overwritten.ETag,
490 );
491 yield* assertArtifactCount(key, 2);
492 }).pipe(
493 Effect.ensuring(
494 S3.abortMultipartUpload({ Bucket, Key: key, UploadId }).pipe(
495 Effect.catchTag("NoSuchUpload", () => Effect.void),
496 Effect.orDie,
497 ),
498 ),
499 );
500 }),
501 { timeout: 120_000 },
502 );
503
504 test.provider(
505 "records explicit delete-marker deletion and restores the previous data version",
506 () =>
507 Effect.gen(function* () {
508 const Bucket = deployed!.bucketName;
509 const key = `${INCOMING_PREFIX}marker/restore${INCOMING_SUFFIX}`;
510 const content = "restored after marker deletion";
511 const source = yield* S3.putObject({
512 Bucket,
513 Key: key,
514 Body: content,
515 });
516 expect(source.VersionId).toBeTruthy();
517 const marker = yield* S3.deleteObject({ Bucket, Key: key });
518 expect(marker.DeleteMarker).toBe(true);
519 expect(marker.VersionId).toBeTruthy();
520 const deleted = yield* S3.deleteObject({
521 Bucket,
522 Key: key,
523 VersionId: marker.VersionId!,
524 });
525 expect(deleted.DeleteMarker).toBe(true);
526 expect(deleted.VersionId).toBe(marker.VersionId);
527 const records = yield* waitForArtifacts([
528 {
529 key,
530 eventName: "s3:ObjectCreated:Put",
531 versionId: source.VersionId!,
532 },
533 {
534 key,
535 eventName: "s3:ObjectRemoved:DeleteMarkerCreated",
536 versionId: marker.VersionId!,
537 },
538 {
539 key,
540 eventName: "s3:ObjectRemoved:Delete",
541 versionId: marker.VersionId!,
542 },
543 ]);
544 yield* assertCreation(
545 records[0]!,
546 content,
547 source.VersionId!,
548 source.ETag,
549 );
550 for (const record of records.slice(1)) {
551 expect(record.content).toBeUndefined();
552 expect(record.readVersionId).toBeUndefined();
553 }
554 const versions = yield* S3.listObjectVersions({
555 Bucket,
556 Prefix: key,
557 });
558 expect(versions.IsTruncated).toBe(false);
559 expect(versions.DeleteMarkers ?? []).toEqual([]);
560 expect(versions.Versions).toHaveLength(1);
561 expect(versions.Versions![0]!.VersionId).toBe(source.VersionId);
562 expect(versions.Versions![0]!.IsLatest).toBe(true);
563 const restored = yield* S3.getObject({ Bucket, Key: key });
564 expect(restored.VersionId).toBe(source.VersionId);
565 expect(
566 yield* Stream.mkString(Stream.decodeText(restored.Body!)),
567 ).toBe(content);
568 yield* assertArtifactCount(key, 3);
569 }),
570 { timeout: 120_000 },
571 );
572
573 test.provider(
574 "acknowledges duplicate real creation payloads after source deletion without rewriting durable evidence",
575 () =>
576 Effect.gen(function* () {
577 const Bucket = deployed!.bucketName;
578 const key = `${INCOMING_PREFIX}replay/deleted + percent% 雪${INCOMING_SUFFIX}`;
579 const content = "durable content after source deletion";
580 const source = yield* S3.putObject({
581 Bucket,
582 Key: key,
583 Body: content,
584 });
585 expect(source.VersionId).toBeTruthy();
586 const identity = {
587 key,
588 eventName: "s3:ObjectCreated:Put",
589 versionId: source.VersionId!,
590 };
591 const [original] = yield* waitForArtifacts([identity]);
592 yield* assertCreation(
593 original!,
594 content,
595 source.VersionId!,
596 source.ETag,
597 );
598 const delivery = yield* findDelivery(identity);
599 yield* waitForAcknowledgements([delivery.messageId]);
600 const Key = yield* Effect.sync(() =>
601 artifactKey(key, identity.eventName, identity.versionId),
602 );
603 const before = yield* S3.headObject({ Bucket, Key });
604 expect(before.VersionId).toBeTruthy();
605
606 yield* S3.deleteObject({
607 Bucket,
608 Key: key,
609 VersionId: source.VersionId!,
610 });
611 yield* waitForArtifacts([
612 {
613 key,
614 eventName: "s3:ObjectRemoved:Delete",
615 versionId: source.VersionId!,
616 },
617 ]);
618 const missing = yield* S3.getObject({
619 Bucket,
620 Key: key,
621 VersionId: source.VersionId!,
622 }).pipe(
623 Effect.flatMap(({ Body }) => Stream.runDrain(Body!)),
624 Effect.as(false),
625 Effect.catchTag("NoSuchVersion", () => Effect.succeed(true)),
626 );
627 expect(missing).toBe(true);
628
629 const replay = yield* SQS.sendMessageBatch({
630 QueueUrl: queueUrl!,
631 Entries: [
632 { Id: "duplicate-one", MessageBody: delivery.body },
633 { Id: "duplicate-two", MessageBody: delivery.body },
634 ],
635 });
636 expect(replay.Failed ?? []).toEqual([]);
637 expect(replay.Successful).toHaveLength(2);
638 yield* waitForAcknowledgements(
639 replay.Successful!.map((entry) => entry.MessageId),
640 );
641 const after = yield* S3.headObject({ Bucket, Key });
642 expect(after.VersionId).toBe(before.VersionId);
643 expect(after.ETag).toBe(before.ETag);
644 const [durable] = yield* waitForArtifacts([identity]);
645 expect(durable).toEqual(original);
646 const versions = yield* S3.listObjectVersions({
647 Bucket,
648 Prefix: Key,
649 });
650 expect(versions.IsTruncated).toBe(false);
651 expect(versions.Versions).toHaveLength(1);
652 expect(versions.Versions![0]!.VersionId).toBe(before.VersionId);
653 yield* assertArtifactCount(key, 2);
654 }),
655 { timeout: 120_000 },
656 );
657
658 test.provider(
659 "ignores and acknowledges an S3 TestEvent before processing a later genuine notification",
660 () =>
661 Effect.gen(function* () {
662 const Bucket = deployed!.bucketName;
663 const key = `${INCOMING_PREFIX}test-event/later-genuine${INCOMING_SUFFIX}`;
664 const configuration = yield* S3.getBucketNotificationConfiguration({
665 Bucket,
666 });
667 yield* S3.putBucketNotificationConfiguration({
668 Bucket,
669 NotificationConfiguration: configuration,
670 });
671 const delivery = yield* findDelivery("s3:TestEvent").pipe(
672 Effect.retry({
673 while: (error) => error instanceof DeliveryNotReady,
674 schedule: Schedule.spaced("2 seconds"),
675 times: 9,
676 }),
677 Effect.timeout("45 seconds"),
678 );
679 yield* waitForAcknowledgements([delivery.messageId]);
680 const body = delivery.body;
681 const sent = yield* SQS.sendMessage({
682 QueueUrl: queueUrl!,
683 MessageBody: body,
684 });
685 expect(sent.MessageId).toBeTruthy();
686 yield* waitForAcknowledgements([sent.MessageId!]);
687 const observed = yield* readEvidence(
688 `${RECEIVED_PREFIX}${sent.MessageId!}.json`,
689 );
690 expect(observed).toBe(body);
691 yield* assertArtifactCount(key, 0);
692
693 const content = "genuine delivery after TestEvent acknowledgement";
694 const source = yield* S3.putObject({
695 Bucket,
696 Key: key,
697 Body: content,
698 });
699 expect(source.VersionId).toBeTruthy();
700 const [record] = yield* waitForArtifacts([
701 {
702 key,
703 eventName: "s3:ObjectCreated:Put",
704 versionId: source.VersionId!,
705 },
706 ]);
707 yield* assertCreation(
708 record!,
709 content,
710 source.VersionId!,
711 source.ETag,
712 );
713 yield* assertArtifactCount(key, 1);
714 }),
715 { timeout: 120_000 },
716 );
717
718 test.provider(
719 "excludes creations and removals that miss either the prefix or the suffix",
720 () =>
721 Effect.gen(function* () {
722 const Bucket = deployed!.bucketName;
723 const excludedKeys = [
724 `outside/${INCOMING_PREFIX}filters/prefix${INCOMING_SUFFIX}`,
725 `${INCOMING_PREFIX}filters/suffix${INCOMING_SUFFIX}.bin`,
726 `${INCOMING_PREFIX}filters/case.TXT`,
727 ];
728 for (const Key of excludedKeys) {
729 const source = yield* S3.putObject({
730 Bucket,
731 Key,
732 Body: "not subscribed",
733 });
734 expect(source.VersionId).toBeTruthy();
735 const marker = yield* S3.deleteObject({ Bucket, Key });
736 expect(marker.DeleteMarker).toBe(true);
737 yield* S3.deleteObject({
738 Bucket,
739 Key,
740 VersionId: source.VersionId!,
741 });
742 }
743 const key = `${INCOMING_PREFIX}filters/positive-control${INCOMING_SUFFIX}`;
744 const source = yield* S3.putObject({
745 Bucket,
746 Key: key,
747 Body: "subscribed",
748 });
749 const marker = yield* S3.deleteObject({ Bucket, Key: key });
750 expect(source.VersionId).toBeTruthy();
751 expect(marker.DeleteMarker).toBe(true);
752 expect(marker.VersionId).toBeTruthy();
753 yield* waitForArtifacts([
754 {
755 key,
756 eventName: "s3:ObjectCreated:Put",
757 versionId: source.VersionId!,
758 },
759 {
760 key,
761 eventName: "s3:ObjectRemoved:DeleteMarkerCreated",
762 versionId: marker.VersionId!,
763 },
764 ]);
765 // Check a bounded quiet window after the positive control arrives.
766 yield* Effect.forEach(excludedKeys, (excludedKey) =>
767 assertArtifactCount(excludedKey, 0),
768 ).pipe(
769 Effect.repeat({
770 schedule: Schedule.spaced("2 seconds"),
771 times: 4,
772 }),
773 );
774 yield* assertArtifactCount(key, 2);
775 }),
776 { timeout: 120_000 },
777 );
778 },
779);
780
781interface NotificationIdentity {
782 key: string;
783 eventName: string;
784 versionId: string;
785}
786
787const artifactSchema = Schema.Struct({
788 bucket: Schema.String,
789 key: Schema.String,
790 eventName: Schema.String,
791 versionId: Schema.String,
792 sequencer: Schema.String,
793 size: Schema.optional(Schema.Number),
794 eTag: Schema.optional(Schema.String),
795 content: Schema.optional(Schema.String),
796 readVersionId: Schema.optional(Schema.String),
797});
798
799const assertCreation = Effect.fn(function* (
800 record: typeof artifactSchema.Type,
801 content: string,
802 versionId: string,
803 eTag: string | undefined,
804) {
805 expect(record.content).toBe(content);
806 expect(record.readVersionId).toBe(versionId);
807 const size = yield* Effect.sync(
808 () => new TextEncoder().encode(content).byteLength,
809 );
810 expect(record.size).toBe(size);
811 expect(eTag).toBeTruthy();
812 expect(record.eTag).toBe(eTag!.replace(/^"|"$/g, ""));
813});
814
815const assertArtifactCount = Effect.fn(function* (key: string, count: number) {
816 const Prefix = yield* Effect.sync(
817 () => `${PROCESSED_PREFIX}${encodeURIComponent(key)}/`,
818 );
819 const artifacts = yield* S3.listObjectsV2({
820 Bucket: deployed!.bucketName,
821 Prefix,
822 });
823 expect(artifacts.IsTruncated).toBe(false);
824 expect(artifacts.Contents ?? []).toHaveLength(count);
825});
826
827const readEvidence = Effect.fn(function* (Key: string) {
828 const object = yield* S3.getObject({
829 Bucket: deployed!.bucketName,
830 Key,
831 }).pipe(
832 Effect.catchTag("NoSuchKey", () =>
833 Effect.fail(new EvidenceNotReady({ key: Key })),
834 ),
835 Effect.retry({
836 while: (error) => error._tag === "EvidenceNotReady",
837 schedule: Schedule.spaced("2 seconds"),
838 times: 9,
839 }),
840 );
841 return yield* Stream.mkString(Stream.decodeText(object.Body!));
842});
843
844const waitForAcknowledgements = (messageIds: string[]) =>
845 Effect.forEach(
846 messageIds,
847 (messageId) =>
848 readEvidence(`${ACKNOWLEDGED_PREFIX}${messageId}.json`).pipe(
849 Effect.tap((body) => Effect.sync(() => expect(body).toBe("{}"))),
850 ),
851 { concurrency: 2 },
852 ).pipe(Effect.timeout("25 seconds"));
853
854const deliverySchema = Schema.fromJsonString(
855 Schema.Struct({
856 Event: Schema.optional(Schema.String),
857 Service: Schema.optional(Schema.String),
858 Bucket: Schema.optional(Schema.String),
859 Records: Schema.optional(
860 Schema.Array(
861 Schema.Struct({
862 eventName: Schema.String,
863 s3: Schema.Struct({
864 bucket: Schema.Struct({ name: Schema.String }),
865 object: Schema.Struct({
866 key: Schema.String,
867 versionId: Schema.optional(Schema.String),
868 }),
869 }),
870 }),
871 ),
872 ),
873 }),
874);
875
876const findDelivery = Effect.fn(function* (
877 identity: NotificationIdentity | "s3:TestEvent",
878) {
879 const deliveries = yield* S3.listObjectsV2({
880 Bucket: deployed!.bucketName,
881 Prefix: RECEIVED_PREFIX,
882 });
883 expect(deliveries.IsTruncated).toBe(false);
884 for (const delivery of deliveries.Contents ?? []) {
885 const body = yield* readEvidence(delivery.Key!);
886 const payload = yield* Schema.decodeUnknownEffect(deliverySchema)(body);
887 const matches = yield* Effect.sync(() =>
888 identity === "s3:TestEvent"
889 ? payload.Event === identity &&
890 payload.Service === "Amazon S3" &&
891 payload.Bucket === deployed!.bucketName &&
892 payload.Records === undefined
893 : (payload.Records ?? []).some(
894 (record) =>
895 record.s3.bucket.name === deployed!.bucketName &&
896 decodeURIComponent(record.s3.object.key.replace(/\+/g, " ")) ===
897 identity.key &&
898 record.s3.object.versionId === identity.versionId &&
899 `s3:${record.eventName.replace(/^s3:/, "")}` ===
900 identity.eventName,
901 ),
902 );
903 if (matches) {
904 return {
905 messageId: delivery.Key!.slice(RECEIVED_PREFIX.length, -".json".length),
906 body,
907 };
908 }
909 }
910 return yield* Effect.fail(
911 new DeliveryNotReady({
912 eventName: identity === "s3:TestEvent" ? identity : identity.eventName,
913 }),
914 );
915});
916
917const readArtifact = Effect.fn(function* (identity: NotificationIdentity) {
918 const Key = yield* Effect.sync(() =>
919 artifactKey(identity.key, identity.eventName, identity.versionId),
920 );
921 const object = yield* S3.getObject({
922 Bucket: deployed!.bucketName,
923 Key,
924 }).pipe(
925 Effect.catchTag("NoSuchKey", () =>
926 Effect.fail(new ArtifactNotReady(identity)),
927 ),
928 Effect.retry({
929 while: (error) => error._tag === "ArtifactNotReady",
930 schedule: Schedule.spaced("4 seconds"),
931 times: 9,
932 }),
933 );
934 const body = yield* Stream.mkString(Stream.decodeText(object.Body!));
935 const record = yield* Effect.try(() => JSON.parse(body)).pipe(
936 Effect.flatMap(Schema.decodeUnknownEffect(artifactSchema)),
937 );
938 expect(record.bucket).toBe(deployed!.bucketName);
939 expect(record.key).toBe(identity.key);
940 expect(record.eventName).toBe(identity.eventName);
941 expect(record.versionId).toBe(identity.versionId);
942 expect(record.sequencer).toMatch(/^[0-9a-f]+$/i);
943 return record;
944});
945
946const waitForArtifacts = (identities: NotificationIdentity[]) =>
947 Effect.all(identities.map(readArtifact), { concurrency: 2 }).pipe(
948 Effect.timeout("45 seconds"),
949 );
950
951class ArtifactNotReady extends Data.TaggedError(
952 "ArtifactNotReady",
953)<NotificationIdentity> {}
954
955class DeliveryNotReady extends Data.TaggedError("DeliveryNotReady")<{
956 eventName: string;
957}> {}
958
959class EvidenceNotReady extends Data.TaggedError("EvidenceNotReady")<{
960 key: string;
961}> {}
962
963class FixtureResourceStillExists extends Data.TaggedError(
964 "FixtureResourceStillExists",
965)<{
966 resource: string;
967}> {}