1import * as AWS from "@/AWS";
2import { normalizePolicyDocument } from "@/AWS/IAM/Policy";
3import * as Output from "@/Output";
4import { isResourceState, State } from "@/State";
5import * as Test from "@/Test/Alchemy";
6import * as Lambda from "@distilled.cloud/aws/lambda";
7import * as S3 from "@distilled.cloud/aws/s3";
8import * as SNS from "@distilled.cloud/aws/sns";
9import * as SQS from "@distilled.cloud/aws/sqs";
10import { expect } from "alchemy-test";
11import * as Effect from "effect/Effect";
12import * as Schedule from "effect/Schedule";
13
14const { test } = Test.make({ providers: AWS.providers() });
15const waitPolicy = {
16 until: (gone: boolean) => gone,
17 schedule: Schedule.spaced("2 seconds"),
18 times: 9,
19};
20
21const filter = (
22 prefix: string,
23 suffix = ".json",
24 reversed = false,
25): S3.NotificationConfigurationFilter => {
26 const rules: S3.FilterRule[] = [
27 { Name: "prefix", Value: prefix },
28 { Name: "suffix", Value: suffix },
29 ];
30 return { Key: { FilterRules: reversed ? rules.reverse() : rules } };
31};
32
33const canonicalTargets = <
34 T extends {
35 Id?: string;
36 Events: S3.Event[];
37 Filter?: S3.NotificationConfigurationFilter;
38 },
39>(
40 targets: T[] = [],
41) =>
42 targets
43 .map((target) => ({
44 ...target,
45 Events: [...target.Events].sort(),
46 Filter: target.Filter
47 ? {
48 Key: {
49 FilterRules: (target.Filter.Key?.FilterRules ?? [])
50 .map((rule) => ({ ...rule, Name: rule.Name?.toLowerCase() }))
51 .sort((a, b) => (a.Name ?? "").localeCompare(b.Name ?? "")),
52 },
53 }
54 : undefined,
55 }))
56 .sort((a, b) => (a.Id ?? "").localeCompare(b.Id ?? ""));
57
58const canonical = (configuration: S3.NotificationConfiguration) => ({
59 QueueConfigurations: canonicalTargets(configuration.QueueConfigurations),
60 TopicConfigurations: canonicalTargets(configuration.TopicConfigurations),
61 LambdaFunctionConfigurations: canonicalTargets(
62 configuration.LambdaFunctionConfigurations,
63 ),
64 EventBridgeConfiguration: configuration.EventBridgeConfiguration,
65});
66
67const readNotifications = (bucketName: string) =>
68 S3.getBucketNotificationConfiguration({ Bucket: bucketName });
69
70const readEventBridgeOwner = (bucketName: string) =>
71 S3.getBucketTagging({ Bucket: bucketName }).pipe(
72 Effect.map(
73 (response) =>
74 response.TagSet.find(
75 (tag) => tag.Key === "alchemy:notifications:eventbridge",
76 )?.Value,
77 ),
78 Effect.catchTag("NoSuchTagSet", () => Effect.succeed(undefined)),
79 );
80
81const notificationQueue = Effect.fn(function* (
82 id: string,
83 bucket: AWS.S3.Bucket,
84) {
85 const queue = yield* AWS.SQS.Queue(id);
86 // Keep both the self-reference and bucket dependency in the binding graph.
87 yield* queue.bind("AllowBucketNotifications", {
88 policyStatements: [
89 {
90 Sid: "AllowBucketNotifications",
91 Effect: "Allow",
92 Principal: { Service: "s3.amazonaws.com" },
93 Action: ["sqs:SendMessage"],
94 Resource: [queue.queueArn],
95 Condition: { ArnEquals: { "aws:SourceArn": bucket.bucketArn } },
96 },
97 ],
98 });
99 return queue;
100});
101
102const assertBucketDeleted = Effect.fn(function* (bucketName: string) {
103 const gone = yield* readNotifications(bucketName).pipe(
104 Effect.as(false),
105 Effect.catchTag("NoSuchBucket", () => Effect.succeed(true)),
106 Effect.repeat(waitPolicy),
107 );
108 expect(gone).toBe(true);
109});
110
111const assertQueueDeleted = Effect.fn(function* (queueUrl: string) {
112 const gone = yield* SQS.getQueueAttributes({
113 QueueUrl: queueUrl,
114 AttributeNames: ["QueueArn"],
115 }).pipe(
116 Effect.as(false),
117 Effect.catchTag("QueueDoesNotExist", () => Effect.succeed(true)),
118 Effect.repeat(waitPolicy),
119 );
120 expect(gone).toBe(true);
121});
122
123test.provider(
124 "queue-only notification bindings create, update filters and events, switch destinations, and remove",
125 (stack) =>
126 Effect.gen(function* () {
127 yield* stack.destroy();
128 const definition = (
129 updated = false,
130 destination: "first" | "second" = "first",
131 enabled = true,
132 ) =>
133 Effect.gen(function* () {
134 const bucket = yield* AWS.S3.Bucket("Bucket", {});
135 const first = yield* notificationQueue("FirstQueue", bucket);
136 const second = yield* notificationQueue("SecondQueue", bucket);
137 if (enabled) {
138 yield* bucket.bind("QueueNotifications", {
139 notificationConfiguration: {
140 QueueConfigurations: [
141 {
142 QueueArn:
143 destination === "first"
144 ? first.queueArn
145 : second.queueArn,
146 Events: [
147 updated
148 ? "s3:ObjectRemoved:Delete"
149 : "s3:ObjectCreated:Put",
150 ],
151 Filter: updated
152 ? filter("updated/")
153 : filter("initial/", ".txt"),
154 },
155 ],
156 },
157 });
158 }
159 return { bucket, first, second };
160 });
161 const { bucket, first, second } = yield* stack.deploy(definition());
162 const initial = yield* readNotifications(bucket.bucketName);
163 expect(canonical(initial)).toEqual(
164 canonical({
165 QueueConfigurations: [
166 {
167 Id: expect.any(String),
168 QueueArn: first.queueArn,
169 Events: ["s3:ObjectCreated:Put"],
170 Filter: filter("initial/", ".txt"),
171 },
172 ],
173 }),
174 );
175 const expectedPolicy = normalizePolicyDocument({
176 Version: "2012-10-17",
177 Statement: [
178 {
179 Sid: "AllowBucketNotifications",
180 Effect: "Allow",
181 Principal: { Service: "s3.amazonaws.com" },
182 Action: ["sqs:SendMessage"],
183 Resource: [first.queueArn],
184 Condition: { ArnEquals: { "aws:SourceArn": bucket.bucketArn } },
185 },
186 ],
187 });
188 const observedPolicy = yield* SQS.getQueueAttributes({
189 QueueUrl: first.queueUrl,
190 AttributeNames: ["Policy"],
191 }).pipe(
192 Effect.map((response) =>
193 normalizePolicyDocument(response.Attributes?.Policy ?? ""),
194 ),
195 Effect.repeat({
196 until: (policy) => policy === expectedPolicy,
197 schedule: Schedule.spaced("2 seconds"),
198 times: 9,
199 }),
200 );
201 expect(observedPolicy).toEqual(expectedPolicy);
202
203 yield* stack.deploy(definition(true));
204 const updated = yield* readNotifications(bucket.bucketName);
205 expect(canonical(updated)).toEqual(
206 canonical({
207 QueueConfigurations: [
208 {
209 Id: expect.any(String),
210 QueueArn: first.queueArn,
211 Events: ["s3:ObjectRemoved:Delete"],
212 Filter: filter("updated/"),
213 },
214 ],
215 }),
216 );
217
218 // Retain both destinations while switching the notification reference.
219 yield* stack.deploy(definition(true, "second"));
220 const switched = yield* readNotifications(bucket.bucketName);
221 expect(canonical(switched)).toEqual(
222 canonical({
223 QueueConfigurations: [
224 {
225 Id: expect.any(String),
226 QueueArn: second.queueArn,
227 Events: ["s3:ObjectRemoved:Delete"],
228 Filter: filter("updated/"),
229 },
230 ],
231 }),
232 );
233 yield* S3.putBucketNotificationConfiguration({
234 Bucket: bucket.bucketName,
235 NotificationConfiguration: {},
236 });
237 yield* stack.deploy(definition(true, "second"));
238 expect(canonical(yield* readNotifications(bucket.bucketName))).toEqual(
239 canonical(switched),
240 );
241 yield* stack.deploy(definition(true, "second"));
242 expect(canonical(yield* readNotifications(bucket.bucketName))).toEqual(
243 canonical(switched),
244 );
245 yield* stack.deploy(definition(true, "second", false));
246 expect(canonical(yield* readNotifications(bucket.bucketName))).toEqual(
247 canonical({}),
248 );
249
250 yield* stack.destroy();
251 yield* Effect.all(
252 [
253 assertBucketDeleted(bucket.bucketName),
254 assertQueueDeleted(first.queueUrl),
255 assertQueueDeleted(second.queueUrl),
256 ],
257 { concurrency: 3 },
258 );
259 }),
260 {
261 tags: [
262 "provider:aws",
263 "provider:aws:iam",
264 "provider:aws:s3",
265 "provider:aws:sqs",
266 "live",
267 ],
268 timeout: 120_000,
269 },
270);
271
272test.provider(
273 "mixed notification bindings keep stable IDs across reordering and preserve foreign configurations on removal",
274 (stack) =>
275 Effect.gen(function* () {
276 yield* stack.destroy();
277 const definition = (
278 mode: "all" | "without-queue" | "none" = "all",
279 reversed = false,
280 revision = "initial",
281 ) =>
282 Effect.gen(function* () {
283 const bucket = yield* AWS.S3.Bucket("Bucket", { tags: { revision } });
284 const queue = yield* notificationQueue("Queue", bucket);
285 const topic = yield* AWS.SNS.Topic("Topic", {
286 attributes: {
287 Policy: bucket.bucketArn.pipe(
288 Output.mapEffect((bucketArn) =>
289 Effect.sync(() =>
290 JSON.stringify({
291 Version: "2012-10-17",
292 Statement: [
293 {
294 Effect: "Allow",
295 Principal: { Service: "s3.amazonaws.com" },
296 Action: "sns:Publish",
297 Resource: "*",
298 Condition: {
299 ArnEquals: { "aws:SourceArn": bucketArn },
300 },
301 },
302 ],
303 }),
304 ),
305 ),
306 ),
307 },
308 });
309 const fn = yield* AWS.Lambda.Function("Target", {
310 main: new URL("./fixtures/notification-target.ts", import.meta.url)
311 .pathname,
312 isExternal: true,
313 handler: "handler",
314 functionUrl: false,
315 });
316 yield* AWS.Lambda.Permission("AllowBucket", {
317 action: "lambda:InvokeFunction",
318 functionName: fn.functionName,
319 principal: "s3.amazonaws.com",
320 sourceArn: bucket.bucketArn,
321 });
322 const events: S3.Event[] = [
323 "s3:ObjectCreated:Put",
324 "s3:ObjectCreated:Post",
325 ];
326 if (reversed) events.reverse();
327 const bindings = [
328 ...(mode === "all"
329 ? [
330 bucket.bind("QueueNotifications", {
331 notificationConfiguration: {
332 QueueConfigurations: (reversed
333 ? ["b", "a"]
334 : ["a", "b"]
335 ).map((key) => ({
336 QueueArn: queue.queueArn,
337 Events: events,
338 Filter: filter(
339 `owned/queue-${key}/`,
340 ".json",
341 reversed,
342 ),
343 })),
344 },
345 }),
346 ]
347 : []),
348 ...(mode !== "none"
349 ? [
350 bucket.bind("TopicNotifications", {
351 notificationConfiguration: {
352 TopicConfigurations: [
353 {
354 TopicArn: topic.topicArn,
355 Events: events,
356 Filter: filter("owned/topic/", ".json", reversed),
357 },
358 ],
359 },
360 }),
361 bucket.bind("LambdaNotifications", {
362 notificationConfiguration: {
363 LambdaFunctionConfigurations: [
364 {
365 LambdaFunctionArn: fn.functionArn,
366 Events: events,
367 Filter: filter("owned/lambda/", ".json", reversed),
368 },
369 ],
370 },
371 }),
372 ]
373 : []),
374 ];
375 yield* Effect.all(reversed ? bindings.reverse() : bindings);
376 return { bucket, queue, topic, fn };
377 });
378 const { bucket, queue, topic, fn } = yield* stack.deploy(definition());
379 const owned = yield* readNotifications(bucket.bucketName);
380 expect(owned.QueueConfigurations).toHaveLength(2);
381 expect(owned.TopicConfigurations).toHaveLength(1);
382 expect(owned.LambdaFunctionConfigurations).toHaveLength(1);
383 const ids = [
384 ...owned.QueueConfigurations!,
385 ...owned.TopicConfigurations!,
386 ...owned.LambdaFunctionConfigurations!,
387 ].map((target) => target.Id);
388 expect(ids.every((id) => typeof id === "string" && id.length > 0)).toBe(
389 true,
390 );
391 expect(new Set(ids).size).toBe(4);
392 const actual = canonical(owned);
393 const events = ["s3:ObjectCreated:Post", "s3:ObjectCreated:Put"];
394 expect(actual.QueueConfigurations).toEqual(
395 expect.arrayContaining(
396 ["a", "b"].map((key) => ({
397 Id: expect.any(String),
398 QueueArn: queue.queueArn,
399 Events: events,
400 Filter: filter(`owned/queue-${key}/`),
401 })),
402 ),
403 );
404 expect(actual.TopicConfigurations).toEqual([
405 {
406 Id: expect.any(String),
407 TopicArn: topic.topicArn,
408 Events: events,
409 Filter: filter("owned/topic/"),
410 },
411 ]);
412 expect(actual.LambdaFunctionConfigurations).toEqual([
413 {
414 Id: expect.any(String),
415 LambdaFunctionArn: fn.functionArn,
416 Events: events,
417 Filter: filter("owned/lambda/"),
418 },
419 ]);
420 expect(actual.EventBridgeConfiguration).toBeUndefined();
421 const foreign: S3.NotificationConfiguration = {
422 QueueConfigurations: [
423 {
424 Id: "external-queue",
425 QueueArn: queue.queueArn,
426 Events: ["s3:ObjectCreated:Put"],
427 Filter: filter("external/queue/"),
428 },
429 ],
430 TopicConfigurations: [
431 {
432 Id: "external-topic",
433 TopicArn: topic.topicArn,
434 Events: ["s3:ObjectCreated:Put"],
435 Filter: filter("external/topic/"),
436 },
437 ],
438 LambdaFunctionConfigurations: [
439 {
440 Id: "external-lambda",
441 LambdaFunctionArn: fn.functionArn,
442 Events: ["s3:ObjectCreated:Put"],
443 Filter: filter("external/lambda/"),
444 },
445 ],
446 EventBridgeConfiguration: {},
447 };
448 const combined: S3.NotificationConfiguration = {
449 QueueConfigurations: [
450 ...owned.QueueConfigurations!,
451 ...foreign.QueueConfigurations!,
452 ],
453 TopicConfigurations: [
454 ...owned.TopicConfigurations!,
455 ...foreign.TopicConfigurations!,
456 ],
457 LambdaFunctionConfigurations: [
458 ...owned.LambdaFunctionConfigurations!,
459 ...foreign.LambdaFunctionConfigurations!,
460 ],
461 EventBridgeConfiguration: {},
462 };
463 // Add foreign entries only after deploy; AWS validates all real destinations.
464 yield* S3.putBucketNotificationConfiguration({
465 Bucket: bucket.bucketName,
466 NotificationConfiguration: combined,
467 });
468 expect(canonical(yield* readNotifications(bucket.bucketName))).toEqual(
469 canonical(combined),
470 );
471
472 yield* stack.deploy(definition("all", true));
473 expect(canonical(yield* readNotifications(bucket.bucketName))).toEqual(
474 canonical(combined),
475 );
476 // Exercise the provider's no-change notification sync, not only an engine noop.
477 yield* stack.deploy(definition("all", true, "noop"));
478 expect(canonical(yield* readNotifications(bucket.bucketName))).toEqual(
479 canonical(combined),
480 );
481 yield* stack.deploy(definition("without-queue", true, "noop"));
482 expect(canonical(yield* readNotifications(bucket.bucketName))).toEqual(
483 canonical({
484 ...combined,
485 QueueConfigurations: foreign.QueueConfigurations,
486 }),
487 );
488 yield* stack.deploy(definition("none", true, "noop"));
489 expect(canonical(yield* readNotifications(bucket.bucketName))).toEqual(
490 canonical(foreign),
491 );
492
493 yield* stack.destroy();
494 yield* Effect.all(
495 [
496 assertBucketDeleted(bucket.bucketName),
497 assertQueueDeleted(queue.queueUrl),
498 SNS.getTopicAttributes({ TopicArn: topic.topicArn }).pipe(
499 Effect.as(false),
500 Effect.catchTag("NotFoundException", () => Effect.succeed(true)),
501 Effect.repeat(waitPolicy),
502 Effect.tap((gone) => Effect.sync(() => expect(gone).toBe(true))),
503 ),
504 Lambda.getFunction({ FunctionName: fn.functionName }).pipe(
505 Effect.as(false),
506 Effect.catchTag("ResourceNotFoundException", () =>
507 Effect.succeed(true),
508 ),
509 Effect.repeat(waitPolicy),
510 Effect.tap((gone) => Effect.sync(() => expect(gone).toBe(true))),
511 ),
512 ],
513 { concurrency: 4 },
514 );
515 }),
516 {
517 tags: [
518 "provider:aws",
519 "provider:aws:lambda",
520 "provider:aws:s3",
521 "provider:aws:sns",
522 "provider:aws:sqs",
523 "live",
524 ],
525 timeout: 120_000,
526 },
527);
528
529test.provider(
530 "managed EventBridge remains until its final binding is removed after a no-change reconcile",
531 (stack) =>
532 Effect.gen(function* () {
533 yield* stack.destroy();
534 const definition = (bindings: string[], revision = "initial") =>
535 Effect.gen(function* () {
536 const bucket = yield* AWS.S3.Bucket("Bucket", { tags: { revision } });
537 for (const id of bindings) {
538 yield* bucket.bind(id, {
539 notificationConfiguration: { EventBridgeConfiguration: {} },
540 });
541 }
542 return bucket;
543 });
544 const bucket = yield* stack.deploy(definition(["First", "Second"]));
545 const enabled = canonical({ EventBridgeConfiguration: {} });
546 expect(canonical(yield* readNotifications(bucket.bucketName))).toEqual(
547 enabled,
548 );
549 const owner = yield* readEventBridgeOwner(bucket.bucketName);
550 expect(owner).toBeTruthy();
551 yield* stack.deploy(definition(["Second"]));
552 expect(canonical(yield* readNotifications(bucket.bucketName))).toEqual(
553 enabled,
554 );
555 // Force tag sync while keeping the remaining notification binding unchanged.
556 yield* stack.deploy(definition(["Second"], "tag-sync"));
557 expect(canonical(yield* readNotifications(bucket.bucketName))).toEqual(
558 enabled,
559 );
560 expect(yield* readEventBridgeOwner(bucket.bucketName)).toBe(owner);
561 yield* stack.deploy(definition([], "tag-sync"));
562 expect(canonical(yield* readNotifications(bucket.bucketName))).toEqual(
563 canonical({}),
564 );
565 expect(yield* readEventBridgeOwner(bucket.bucketName)).toBeUndefined();
566 yield* stack.destroy();
567 yield* assertBucketDeleted(bucket.bucketName);
568 }),
569 { tags: ["provider:aws", "provider:aws:s3", "live"], timeout: 120_000 },
570);
571
572test.provider(
573 "adding and removing identical bindings does not claim a foreign queue target or EventBridge",
574 (stack) =>
575 Effect.gen(function* () {
576 yield* stack.destroy();
577 const definition = (enabled: boolean) =>
578 Effect.gen(function* () {
579 const bucket = yield* AWS.S3.Bucket("Bucket", {});
580 const queue = yield* notificationQueue("Queue", bucket);
581 if (enabled) {
582 yield* bucket.bind("MatchingNotifications", {
583 notificationConfiguration: {
584 QueueConfigurations: [
585 {
586 QueueArn: queue.queueArn,
587 Events: ["s3:ObjectCreated:Put"],
588 Filter: filter("shared/"),
589 },
590 ],
591 EventBridgeConfiguration: {},
592 },
593 });
594 }
595 return { bucket, queue };
596 });
597 const { bucket, queue } = yield* stack.deploy(definition(false));
598 expect(canonical(yield* readNotifications(bucket.bucketName))).toEqual(
599 canonical({}),
600 );
601 const foreign: S3.NotificationConfiguration = {
602 QueueConfigurations: [
603 {
604 Id: "external-identical-target",
605 QueueArn: queue.queueArn,
606 Events: ["s3:ObjectCreated:Put"],
607 Filter: filter("shared/"),
608 },
609 ],
610 EventBridgeConfiguration: {},
611 };
612 yield* S3.putBucketNotificationConfiguration({
613 Bucket: bucket.bucketName,
614 NotificationConfiguration: foreign,
615 });
616 expect(canonical(yield* readNotifications(bucket.bucketName))).toEqual(
617 canonical(foreign),
618 );
619 expect(yield* readEventBridgeOwner(bucket.bucketName)).toBeUndefined();
620 yield* stack.deploy(definition(true));
621 expect(canonical(yield* readNotifications(bucket.bucketName))).toEqual(
622 canonical(foreign),
623 );
624 expect(yield* readEventBridgeOwner(bucket.bucketName)).toBeUndefined();
625 yield* stack.deploy(definition(false));
626 expect(canonical(yield* readNotifications(bucket.bucketName))).toEqual(
627 canonical(foreign),
628 );
629 expect(yield* readEventBridgeOwner(bucket.bucketName)).toBeUndefined();
630 yield* stack.destroy();
631 yield* Effect.all(
632 [
633 assertBucketDeleted(bucket.bucketName),
634 assertQueueDeleted(queue.queueUrl),
635 ],
636 { concurrency: 2 },
637 );
638 }),
639 {
640 tags: ["provider:aws", "provider:aws:s3", "provider:aws:sqs", "live"],
641 timeout: 120_000,
642 },
643);
644
645test.provider(
646 "legacy Lambda notifications migrate from persisted bindings without claiming new identical external bindings",
647 (stack) =>
648 Effect.gen(function* () {
649 yield* stack.destroy();
650 const definition = (mode: "legacy" | "updated" | "none") =>
651 Effect.gen(function* () {
652 const bucket = yield* AWS.S3.Bucket("Bucket", {});
653 const fn = yield* AWS.Lambda.Function("Target", {
654 main: new URL("./fixtures/notification-target.ts", import.meta.url)
655 .pathname,
656 isExternal: true,
657 handler: "handler",
658 functionUrl: false,
659 });
660 yield* AWS.Lambda.Permission("AllowBucket", {
661 action: "lambda:InvokeFunction",
662 functionName: fn.functionName,
663 principal: "s3.amazonaws.com",
664 sourceArn: bucket.bucketArn,
665 });
666 if (mode !== "none") {
667 yield* bucket.bind("ChangingNotifications", {
668 notificationConfiguration: {
669 LambdaFunctionConfigurations: [
670 {
671 LambdaFunctionArn: fn.functionArn,
672 Events: ["s3:ObjectCreated:Put"],
673 Filter: filter(
674 mode === "legacy" ? "owned/original/" : "owned/updated/",
675 ),
676 },
677 ],
678 },
679 });
680 yield* bucket.bind(
681 mode === "legacy"
682 ? "RemovedNotifications"
683 : "ExternalNotifications",
684 {
685 notificationConfiguration: {
686 LambdaFunctionConfigurations: [
687 {
688 LambdaFunctionArn: fn.functionArn,
689 Events: ["s3:ObjectCreated:Put"],
690 Filter: filter(
691 mode === "legacy" ? "owned/removed/" : "external/",
692 ),
693 },
694 ],
695 },
696 },
697 );
698 }
699 return { bucket, fn };
700 });
701 const { bucket, fn } = yield* stack.deploy(definition("legacy"));
702 const legacyTargets: S3.LambdaFunctionConfiguration[] = [
703 "owned/original/",
704 "owned/removed/",
705 ].map((prefix) => ({
706 LambdaFunctionArn: fn.functionArn,
707 Events: ["s3:ObjectCreated:Put"],
708 Filter: filter(prefix),
709 }));
710 const foreign: S3.LambdaFunctionConfiguration = {
711 Id: "external-lambda",
712 LambdaFunctionArn: fn.functionArn,
713 Events: ["s3:ObjectCreated:Put"],
714 Filter: filter("external/"),
715 };
716 // The legacy provider omitted IDs; S3 assigned them outside Alchemy's namespace.
717 yield* S3.putBucketNotificationConfiguration({
718 Bucket: bucket.bucketName,
719 NotificationConfiguration: {
720 LambdaFunctionConfigurations: [...legacyTargets, foreign],
721 },
722 });
723 const legacy = yield* readNotifications(bucket.bucketName);
724 expect(legacy.LambdaFunctionConfigurations).toHaveLength(3);
725 const legacyIds = legacy
726 .LambdaFunctionConfigurations!.filter(
727 (target) => target.Id !== foreign.Id,
728 )
729 .map((target) => target.Id!);
730 expect(legacyIds).toHaveLength(2);
731 for (const id of legacyIds) {
732 expect(id).toBeTruthy();
733 expect(id.startsWith("alchemy-notification-")).toBe(false);
734 }
735 yield* Effect.gen(function* () {
736 const state = yield* yield* State;
737 const key = { stack: stack.name, stage: stack.stage, fqn: "Bucket" };
738 const row = yield* state.get(key);
739 if (!row || !isResourceState(row) || !row.attr) {
740 return yield* Effect.fail(
741 new Error("Expected persisted bucket attributes"),
742 );
743 }
744 const bindings = row.bindings.map((binding) => ({
745 ...binding,
746 data: {
747 notificationConfiguration: {
748 LambdaFunctionConfigurations:
749 binding.sid === "ChangingNotifications"
750 ? [legacyTargets[0]!]
751 : [legacyTargets[1]!],
752 },
753 },
754 }));
755 expect(bindings).toHaveLength(2);
756 expect(bindings.map((binding) => binding.sid).sort()).toEqual([
757 "ChangingNotifications",
758 "RemovedNotifications",
759 ]);
760 const attr = { ...row.attr };
761 delete attr.managedNotificationConfiguration;
762 const persisted = yield* state.set({
763 ...key,
764 value: { ...row, attr, bindings },
765 });
766 expect(persisted.attr.managedNotificationConfiguration).toBeUndefined();
767 }).pipe(Effect.provide(stack.state));
768
769 yield* stack.deploy(definition("updated"));
770 const updated = yield* readNotifications(bucket.bucketName);
771 const managedId = updated.LambdaFunctionConfigurations?.find(
772 (target) => target.Id !== foreign.Id,
773 )?.Id;
774 expect(managedId).toMatch(/^alchemy-notification-/);
775 expect(canonical(updated)).toEqual(
776 canonical({
777 LambdaFunctionConfigurations: [
778 {
779 Id: managedId,
780 LambdaFunctionArn: fn.functionArn,
781 Events: ["s3:ObjectCreated:Put"],
782 Filter: filter("owned/updated/"),
783 },
784 foreign,
785 ],
786 }),
787 );
788 for (const target of updated.LambdaFunctionConfigurations!) {
789 expect(legacyIds).not.toContain(target.Id);
790 }
791 yield* stack.deploy(definition("updated"));
792 expect(canonical(yield* readNotifications(bucket.bucketName))).toEqual(
793 canonical(updated),
794 );
795 yield* stack.deploy(definition("none"));
796 expect(canonical(yield* readNotifications(bucket.bucketName))).toEqual(
797 canonical({ LambdaFunctionConfigurations: [foreign] }),
798 );
799 yield* stack.destroy();
800 yield* assertBucketDeleted(bucket.bucketName);
801 const gone = yield* Lambda.getFunction({
802 FunctionName: fn.functionName,
803 }).pipe(
804 Effect.as(false),
805 Effect.catchTag("ResourceNotFoundException", () =>
806 Effect.succeed(true),
807 ),
808 Effect.repeat(waitPolicy),
809 );
810 expect(gone).toBe(true);
811 }),
812 {
813 tags: ["provider:aws", "provider:aws:lambda", "provider:aws:s3", "live"],
814 timeout: 120_000,
815 },
816);
817
818test.provider(
819 "clears a stale EventBridge ownership marker when notifications and withdrawn bindings already match",
820 (stack) =>
821 Effect.gen(function* () {
822 yield* stack.destroy();
823 const definition = (enabled: boolean) =>
824 Effect.gen(function* () {
825 const bucket = yield* AWS.S3.Bucket("Bucket", {
826 tags: { purpose: "eventbridge-marker-recovery" },
827 });
828 if (enabled) {
829 yield* bucket.bind("EventBridge", {
830 notificationConfiguration: { EventBridgeConfiguration: {} },
831 });
832 }
833 return bucket;
834 });
835 const bucket = yield* stack.deploy(definition(true));
836 const owner = yield* readEventBridgeOwner(bucket.bucketName);
837 expect(owner).toBeTruthy();
838 yield* S3.putBucketNotificationConfiguration({
839 Bucket: bucket.bucketName,
840 NotificationConfiguration: {},
841 });
842 // Model withdrawn bindings with a lost snapshot and an uncleared cloud marker.
843 yield* Effect.gen(function* () {
844 const state = yield* yield* State;
845 const key = { stack: stack.name, stage: stack.stage, fqn: "Bucket" };
846 const row = yield* state.get(key);
847 if (!row || !isResourceState(row) || !row.attr) {
848 return yield* Effect.fail(
849 new Error("Expected persisted bucket attributes"),
850 );
851 }
852 const attr = { ...row.attr };
853 delete attr.managedNotificationConfiguration;
854 yield* state.set({ ...key, value: { ...row, attr, bindings: [] } });
855 }).pipe(Effect.provide(stack.state));
856 expect(canonical(yield* readNotifications(bucket.bucketName))).toEqual(
857 canonical({}),
858 );
859 expect(yield* readEventBridgeOwner(bucket.bucketName)).toBe(owner);
860 const retained = yield* stack.deploy(definition(false));
861 expect(retained.bucketName).toBe(bucket.bucketName);
862 expect(canonical(yield* readNotifications(bucket.bucketName))).toEqual(
863 canonical({}),
864 );
865 expect(yield* readEventBridgeOwner(bucket.bucketName)).toBeUndefined();
866 expect(
867 (yield* S3.getBucketTagging({ Bucket: bucket.bucketName })).TagSet,
868 ).toContainEqual({
869 Key: "purpose",
870 Value: "eventbridge-marker-recovery",
871 });
872 yield* stack.destroy();
873 yield* assertBucketDeleted(bucket.bucketName);
874 }),
875 { tags: ["provider:aws", "provider:aws:s3", "live"], timeout: 120_000 },
876);
877
878test.provider(
879 "EventBridge binding removal recovers ownership after losing the committed notification snapshot",
880 (stack) =>
881 Effect.gen(function* () {
882 yield* stack.destroy();
883 const definition = (enabled: boolean) =>
884 Effect.gen(function* () {
885 const bucket = yield* AWS.S3.Bucket("Bucket", {
886 tags: { purpose: "notifications-crash-recovery" },
887 });
888 if (enabled) {
889 yield* bucket.bind("EventBridge", {
890 notificationConfiguration: { EventBridgeConfiguration: {} },
891 });
892 }
893 return bucket;
894 });
895 const bucket = yield* stack.deploy(definition(true));
896 const enabled = canonical({ EventBridgeConfiguration: {} });
897 expect(canonical(yield* readNotifications(bucket.bucketName))).toEqual(
898 enabled,
899 );
900 const owner = yield* readEventBridgeOwner(bucket.bucketName);
901 expect(owner).toBeTruthy();
902
903 // Model a lost attributes commit without deleting the live bucket or its state row.
904 yield* Effect.gen(function* () {
905 const state = yield* yield* State;
906 const key = { stack: stack.name, stage: stack.stage, fqn: "Bucket" };
907 const row = yield* state.get(key);
908 if (!row || !isResourceState(row) || !row.attr) {
909 return yield* Effect.fail(
910 new Error("Expected persisted bucket attributes"),
911 );
912 }
913 expect(
914 row.attr.managedNotificationConfiguration?.EventBridgeConfiguration,
915 ).toEqual({});
916 const attr = { ...row.attr };
917 delete attr.managedNotificationConfiguration;
918 const persisted = yield* state.set({ ...key, value: { ...row, attr } });
919 expect(persisted.attr.managedNotificationConfiguration).toBeUndefined();
920 expect(persisted.attr.bucketName).toBe(bucket.bucketName);
921 }).pipe(Effect.provide(stack.state));
922 expect(canonical(yield* readNotifications(bucket.bucketName))).toEqual(
923 enabled,
924 );
925 expect(yield* readEventBridgeOwner(bucket.bucketName)).toBe(owner);
926
927 const retained = yield* stack.deploy(definition(false));
928 expect(retained.bucketName).toBe(bucket.bucketName);
929 expect(canonical(yield* readNotifications(bucket.bucketName))).toEqual(
930 canonical({}),
931 );
932 expect(yield* readEventBridgeOwner(bucket.bucketName)).toBeUndefined();
933 expect(
934 (yield* S3.getBucketTagging({ Bucket: bucket.bucketName })).TagSet,
935 ).toContainEqual({
936 Key: "purpose",
937 Value: "notifications-crash-recovery",
938 });
939 yield* stack.destroy();
940 yield* assertBucketDeleted(bucket.bucketName);
941 }),
942 { tags: ["provider:aws", "provider:aws:s3", "live"], timeout: 120_000 },
943);