1import * as AWS from "@/AWS";
2import { DeliveryStream } from "@/AWS/Firehose";
3import * as Test from "@/Test/Alchemy";
4import * as Firehose from "@distilled.cloud/aws/firehose";
5import * as iam from "@distilled.cloud/aws/iam";
6import { describe, expect } from "alchemy-test";
7import * as Data from "effect/Data";
8import * as Effect from "effect/Effect";
9import * as Result from "effect/Result";
10import * as Schedule from "effect/Schedule";
11
12const { test } = Test.make({ providers: AWS.providers() });
13
14describe.skipIf(!!process.env.FAST)(
15 "AWS.Firehose.DeliveryStream",
16 {
17 tags: ["provider:aws", "provider:aws:firehose", "provider:aws:s3", "live"],
18 },
19 () => {
20 test.provider(
21 "create DirectPut stream to S3, update destination settings, destroy",
22 (stack) =>
23 Effect.gen(function* () {
24 // Reconcile away any partial deployment left by a crashed prior run.
25 yield* stack.destroy();
26
27 const initial = yield* stack.deploy(
28 Effect.gen(function* () {
29 const bucket = yield* AWS.S3.Bucket("FirehoseTestBucket", {
30 forceDestroy: true,
31 });
32 const stream = yield* DeliveryStream("TestDeliveryStream", {
33 destination: {
34 bucketArn: bucket.bucketArn,
35 },
36 tags: { Environment: "test" },
37 });
38 return { bucket, stream };
39 }),
40 );
41
42 expect(initial.stream.deliveryStreamName).toBeDefined();
43 expect(initial.stream.deliveryStreamStatus).toEqual("ACTIVE");
44 expect(initial.stream.deliveryStreamType).toEqual("DirectPut");
45 expect(initial.stream.bucketArn).toEqual(initial.bucket.bucketArn);
46 expect(initial.stream.roleArn).toContain(":role/");
47 expect(initial.stream.roleName).toBeDefined();
48 // AWS defaults applied when no buffering hints are provided.
49 expect(initial.stream.bufferingIntervalInSeconds).toEqual(300);
50 expect(initial.stream.bufferingSizeInMBs).toEqual(5);
51 expect(initial.stream.compressionFormat).toEqual("UNCOMPRESSED");
52 // No SSE configured — a never-encrypted stream reports DISABLED (or
53 // no encryption configuration at all).
54 expect(initial.stream.encryptionStatus ?? "DISABLED").toEqual(
55 "DISABLED",
56 );
57
58 // Out-of-band verification via distilled.
59 const described = yield* Firehose.describeDeliveryStream({
60 DeliveryStreamName: initial.stream.deliveryStreamName,
61 });
62 const description = described.DeliveryStreamDescription;
63 expect(description.DeliveryStreamStatus).toEqual("ACTIVE");
64 expect(description.DeliveryStreamType).toEqual("DirectPut");
65 expect(
66 description.Destinations[0]?.ExtendedS3DestinationDescription
67 ?.BucketARN,
68 ).toEqual(initial.bucket.bucketArn);
69
70 // Ownership + user tags.
71 const tags = yield* Firehose.listTagsForDeliveryStream({
72 DeliveryStreamName: initial.stream.deliveryStreamName,
73 });
74 const tagKeys = tags.Tags.map((tag) => tag.Key);
75 expect(tagKeys).toContain("alchemy::stack");
76 expect(tagKeys).toContain("alchemy::stage");
77 expect(tagKeys).toContain("alchemy::id");
78 expect(tags.Tags).toContainEqual({
79 Key: "Environment",
80 Value: "test",
81 });
82
83 // Update destination settings in place (UpdateDestination path) and
84 // enable SSE (StartDeliveryStreamEncryption path) in the same step.
85 const updated = yield* stack.deploy(
86 Effect.gen(function* () {
87 const bucket = yield* AWS.S3.Bucket("FirehoseTestBucket", {
88 forceDestroy: true,
89 });
90 const stream = yield* DeliveryStream("TestDeliveryStream", {
91 destination: {
92 bucketArn: bucket.bucketArn,
93 prefix: "events/",
94 errorOutputPrefix: "errors/",
95 bufferingInterval: "60 seconds",
96 bufferingSizeInMBs: 1,
97 compressionFormat: "GZIP",
98 },
99 encryption: { keyType: "AWS_OWNED_CMK" },
100 tags: { Environment: "production", Team: "platform" },
101 });
102 return { bucket, stream };
103 }),
104 );
105
106 // Same physical stream — updated in place, not replaced.
107 expect(updated.stream.deliveryStreamName).toEqual(
108 initial.stream.deliveryStreamName,
109 );
110 expect(updated.stream.prefix).toEqual("events/");
111 expect(updated.stream.errorOutputPrefix).toEqual("errors/");
112 expect(updated.stream.bufferingIntervalInSeconds).toEqual(60);
113 expect(updated.stream.bufferingSizeInMBs).toEqual(1);
114 expect(updated.stream.compressionFormat).toEqual("GZIP");
115 // SSE converged to ENABLED with the AWS-owned CMK.
116 expect(updated.stream.encryptionStatus).toEqual("ENABLED");
117 expect(updated.stream.encryptionKeyType).toEqual("AWS_OWNED_CMK");
118
119 const updatedDescription = yield* Firehose.describeDeliveryStream({
120 DeliveryStreamName: initial.stream.deliveryStreamName,
121 });
122 const s3Description =
123 updatedDescription.DeliveryStreamDescription.Destinations[0]
124 ?.ExtendedS3DestinationDescription;
125 expect(s3Description?.Prefix).toEqual("events/");
126 expect(s3Description?.BufferingHints.IntervalInSeconds).toEqual(60);
127 expect(s3Description?.BufferingHints.SizeInMBs).toEqual(1);
128 expect(s3Description?.CompressionFormat).toEqual("GZIP");
129
130 // Updated tags converged.
131 const updatedTags = yield* Firehose.listTagsForDeliveryStream({
132 DeliveryStreamName: initial.stream.deliveryStreamName,
133 });
134 expect(updatedTags.Tags).toContainEqual({
135 Key: "Environment",
136 Value: "production",
137 });
138 expect(updatedTags.Tags).toContainEqual({
139 Key: "Team",
140 Value: "platform",
141 });
142
143 // Out-of-band SSE verification via distilled.
144 expect(
145 updatedDescription.DeliveryStreamDescription
146 .DeliveryStreamEncryptionConfiguration?.Status,
147 ).toEqual("ENABLED");
148
149 // Remove encryption — the StopDeliveryStreamEncryption path.
150 const decrypted = yield* stack.deploy(
151 Effect.gen(function* () {
152 const bucket = yield* AWS.S3.Bucket("FirehoseTestBucket", {
153 forceDestroy: true,
154 });
155 const stream = yield* DeliveryStream("TestDeliveryStream", {
156 destination: {
157 bucketArn: bucket.bucketArn,
158 prefix: "events/",
159 errorOutputPrefix: "errors/",
160 bufferingInterval: "60 seconds",
161 bufferingSizeInMBs: 1,
162 compressionFormat: "GZIP",
163 },
164 tags: { Environment: "production", Team: "platform" },
165 });
166 return { bucket, stream };
167 }),
168 );
169 // Same physical stream — SSE disabled in place.
170 expect(decrypted.stream.deliveryStreamName).toEqual(
171 initial.stream.deliveryStreamName,
172 );
173 expect(decrypted.stream.encryptionStatus ?? "DISABLED").toEqual(
174 "DISABLED",
175 );
176 expect(decrypted.stream.encryptionKeyType).toBeUndefined();
177
178 yield* stack.destroy();
179
180 yield* assertDeliveryStreamDeleted(initial.stream.deliveryStreamName);
181
182 // The synthesized IAM role is cleaned up with the stream.
183 const roleResult = yield* iam
184 .getRole({ RoleName: initial.stream.roleName! })
185 .pipe(Effect.result);
186 expect(Result.isFailure(roleResult)).toBe(true);
187 }),
188 { tags: ["provider:aws:iam"], timeout: 420_000 },
189 );
190
191 test.provider(
192 "replaces the stream when a Kinesis source is added",
193 (stack) =>
194 Effect.gen(function* () {
195 // Reconcile away any partial deployment left by a crashed prior run.
196 yield* stack.destroy();
197
198 // Keep the bucket AND the kinesis stream deployed across both steps —
199 // replacing a resource while simultaneously removing a dependency
200 // deadlocks the engine.
201 const initial = yield* stack.deploy(
202 Effect.gen(function* () {
203 const bucket = yield* AWS.S3.Bucket("FirehoseSourceBucket", {
204 forceDestroy: true,
205 });
206 const source = yield* AWS.Kinesis.Stream("FirehoseSourceStream");
207 const stream = yield* DeliveryStream("SourcedDeliveryStream", {
208 destination: { bucketArn: bucket.bucketArn },
209 });
210 return { bucket, source, stream };
211 }),
212 );
213
214 expect(initial.stream.deliveryStreamType).toEqual("DirectPut");
215
216 const replaced = yield* stack.deploy(
217 Effect.gen(function* () {
218 const bucket = yield* AWS.S3.Bucket("FirehoseSourceBucket", {
219 forceDestroy: true,
220 });
221 const source = yield* AWS.Kinesis.Stream("FirehoseSourceStream");
222 const stream = yield* DeliveryStream("SourcedDeliveryStream", {
223 source: { kinesisStreamArn: source.streamArn },
224 destination: { bucketArn: bucket.bucketArn },
225 });
226 return { bucket, source, stream };
227 }),
228 );
229
230 // Source change is a replacement — new physical stream.
231 expect(replaced.stream.deliveryStreamName).not.toEqual(
232 initial.stream.deliveryStreamName,
233 );
234 expect(replaced.stream.deliveryStreamType).toEqual(
235 "KinesisStreamAsSource",
236 );
237 expect(replaced.stream.kinesisStreamArn).toEqual(
238 replaced.source.streamArn,
239 );
240
241 const described = yield* Firehose.describeDeliveryStream({
242 DeliveryStreamName: replaced.stream.deliveryStreamName,
243 });
244 expect(
245 described.DeliveryStreamDescription.Source
246 ?.KinesisStreamSourceDescription?.KinesisStreamARN,
247 ).toEqual(replaced.source.streamArn);
248
249 // The replaced (old) stream is deleted by the engine.
250 yield* assertDeliveryStreamDeleted(initial.stream.deliveryStreamName);
251
252 yield* stack.destroy();
253
254 yield* assertDeliveryStreamDeleted(
255 replaced.stream.deliveryStreamName,
256 );
257 }),
258 { tags: ["provider:aws:kinesis"], timeout: 420_000 },
259 );
260
261 class DeliveryStreamStillExists extends Data.TaggedError(
262 "DeliveryStreamStillExists",
263 ) {}
264
265 const assertDeliveryStreamDeleted = Effect.fn(function* (
266 deliveryStreamName: string,
267 ) {
268 yield* Firehose.describeDeliveryStream({
269 DeliveryStreamName: deliveryStreamName,
270 }).pipe(
271 Effect.flatMap(() => Effect.fail(new DeliveryStreamStillExists())),
272 Effect.retry({
273 while: (e: { _tag: string }) =>
274 e._tag === "DeliveryStreamStillExists",
275 schedule: Schedule.max([
276 Schedule.fixed("3 seconds"),
277 Schedule.recurs(40),
278 ]),
279 }),
280 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
281 );
282 });
283 },
284);