1import { adopt } from "@/AdoptPolicy";
2import * as AWS from "@/AWS";
3import { Stream } from "@/AWS/Kinesis";
4import * as Provider from "@/Provider";
5import { State } from "@/State";
6import * as Test from "@/Test/Alchemy";
7import * as Kinesis from "@distilled.cloud/aws/kinesis";
8import { describe, expect } from "alchemy-test";
9import * as Data from "effect/Data";
10import * as Effect from "effect/Effect";
11import * as Schedule from "effect/Schedule";
12
13const { test } = Test.make({ providers: AWS.providers() });
14
15describe.skipIf(!!process.env.FAST)(
16 "AWS.Kinesis.Stream",
17 { tags: ["provider:aws", "provider:aws:kinesis", "live"] },
18 () => {
19 test.provider(
20 "create and delete stream with default props",
21 (stack) =>
22 Effect.gen(function* () {
23 const stream = yield* stack.deploy(
24 Effect.gen(function* () {
25 return yield* Stream("DefaultStream");
26 }),
27 );
28
29 expect(stream.streamName).toBeDefined();
30 expect(stream.streamArn).toBeDefined();
31 expect(stream.streamStatus).toEqual("ACTIVE");
32
33 const streamDescription = yield* Kinesis.describeStreamSummary({
34 StreamName: stream.streamName,
35 });
36 expect(
37 streamDescription.StreamDescriptionSummary.StreamStatus,
38 ).toEqual("ACTIVE");
39 expect(
40 streamDescription.StreamDescriptionSummary.StreamModeDetails
41 ?.StreamMode,
42 ).toEqual("ON_DEMAND");
43
44 yield* stack.destroy();
45
46 yield* assertStreamDeleted(stream.streamName);
47 }),
48 { timeout: 180_000 },
49 );
50
51 test.provider(
52 "create, update, delete on-demand stream with tags",
53 (stack) =>
54 Effect.gen(function* () {
55 const stream = yield* stack.deploy(
56 Effect.gen(function* () {
57 return yield* Stream("TestStream", {
58 streamMode: "ON_DEMAND",
59 tags: { Environment: "test" },
60 });
61 }),
62 );
63
64 // Verify the stream was created
65 const streamDescription = yield* Kinesis.describeStreamSummary({
66 StreamName: stream.streamName,
67 });
68 expect(
69 streamDescription.StreamDescriptionSummary.StreamStatus,
70 ).toEqual("ACTIVE");
71 expect(
72 streamDescription.StreamDescriptionSummary.StreamModeDetails
73 ?.StreamMode,
74 ).toEqual("ON_DEMAND");
75 expect(
76 streamDescription.StreamDescriptionSummary.RetentionPeriodHours,
77 ).toEqual(24);
78
79 // Verify tags
80 const tagging = yield* Kinesis.listTagsForStream({
81 StreamName: stream.streamName,
82 });
83 expect(tagging.Tags).toContainEqual({
84 Key: "Environment",
85 Value: "test",
86 });
87
88 // Update the stream - increase retention period and update tags
89 yield* stack.deploy(
90 Effect.gen(function* () {
91 return yield* Stream("TestStream", {
92 streamMode: "ON_DEMAND",
93 retentionPeriod: "48 hours",
94 tags: { Environment: "production", Team: "platform" },
95 });
96 }),
97 );
98
99 // Verify the retention period was updated
100 const updatedDescription = yield* Kinesis.describeStreamSummary({
101 StreamName: stream.streamName,
102 });
103 expect(
104 updatedDescription.StreamDescriptionSummary.RetentionPeriodHours,
105 ).toEqual(48);
106
107 // Verify tags were updated
108 const updatedTagging = yield* Kinesis.listTagsForStream({
109 StreamName: stream.streamName,
110 });
111 expect(updatedTagging.Tags).toContainEqual({
112 Key: "Environment",
113 Value: "production",
114 });
115 expect(updatedTagging.Tags).toContainEqual({
116 Key: "Team",
117 Value: "platform",
118 });
119
120 yield* stack.destroy();
121
122 yield* assertStreamDeleted(stream.streamName);
123 }),
124 { timeout: 180_000 },
125 );
126
127 test.provider(
128 "create provisioned stream with shards",
129 (stack) =>
130 Effect.gen(function* () {
131 const stream = yield* stack.deploy(
132 Effect.gen(function* () {
133 return yield* Stream("ProvisionedStream", {
134 streamMode: "PROVISIONED",
135 shardCount: 2,
136 });
137 }),
138 );
139
140 // Verify the stream was created with shards
141 const streamDescription = yield* Kinesis.describeStreamSummary({
142 StreamName: stream.streamName,
143 });
144 expect(
145 streamDescription.StreamDescriptionSummary.StreamStatus,
146 ).toEqual("ACTIVE");
147 expect(
148 streamDescription.StreamDescriptionSummary.StreamModeDetails
149 ?.StreamMode,
150 ).toEqual("PROVISIONED");
151 expect(
152 streamDescription.StreamDescriptionSummary.OpenShardCount,
153 ).toEqual(2);
154
155 yield* stack.destroy();
156
157 yield* assertStreamDeleted(stream.streamName);
158 }),
159 { timeout: 180_000 },
160 );
161
162 test.provider(
163 "update provisioned stream shard count",
164 (stack) =>
165 Effect.gen(function* () {
166 const stream = yield* stack.deploy(
167 Effect.gen(function* () {
168 return yield* Stream("ShardStream", {
169 streamMode: "PROVISIONED",
170 shardCount: 1,
171 });
172 }),
173 );
174
175 // Verify initial shard count
176 const streamDescription = yield* Kinesis.describeStreamSummary({
177 StreamName: stream.streamName,
178 });
179 expect(
180 streamDescription.StreamDescriptionSummary.OpenShardCount,
181 ).toEqual(1);
182
183 // Update shard count
184 yield* stack.deploy(
185 Effect.gen(function* () {
186 return yield* Stream("ShardStream", {
187 streamMode: "PROVISIONED",
188 shardCount: 2,
189 });
190 }),
191 );
192
193 // Verify shard count was updated
194 const updatedDescription = yield* Kinesis.describeStreamSummary({
195 StreamName: stream.streamName,
196 });
197 expect(
198 updatedDescription.StreamDescriptionSummary.OpenShardCount,
199 ).toEqual(2);
200
201 yield* stack.destroy();
202
203 yield* assertStreamDeleted(stream.streamName);
204 }),
205 { timeout: 300_000 },
206 );
207
208 test.provider(
209 "create stream with custom name",
210 (stack) =>
211 Effect.gen(function* () {
212 const customName = `test-custom-kinesis-stream-custom-name-stream`;
213
214 const stream = yield* stack.deploy(
215 Effect.gen(function* () {
216 return yield* Stream("CustomNameStream", {
217 streamName: customName,
218 });
219 }),
220 );
221
222 expect(stream.streamName).toEqual(customName);
223 expect(stream.streamArn).toContain(customName);
224
225 // Verify the stream exists
226 const streamDescription = yield* Kinesis.describeStreamSummary({
227 StreamName: customName,
228 });
229 expect(streamDescription.StreamDescriptionSummary.StreamName).toEqual(
230 customName,
231 );
232
233 yield* stack.destroy();
234
235 yield* assertStreamDeleted(customName);
236 }),
237 { timeout: 180_000 },
238 );
239
240 test.provider(
241 "create stream with encryption",
242 (stack) =>
243 Effect.gen(function* () {
244 const stream = yield* stack.deploy(
245 Effect.gen(function* () {
246 return yield* Stream("EncryptedStream", {
247 encryption: true,
248 });
249 }),
250 );
251
252 // Verify the stream has encryption enabled
253 const streamDescription = yield* Kinesis.describeStreamSummary({
254 StreamName: stream.streamName,
255 });
256 expect(
257 streamDescription.StreamDescriptionSummary.EncryptionType,
258 ).toEqual("KMS");
259
260 // Update to disable encryption
261 yield* stack.deploy(
262 Effect.gen(function* () {
263 return yield* Stream("EncryptedStream", {
264 encryption: false,
265 });
266 }),
267 );
268
269 // Verify encryption is disabled
270 const updatedDescription = yield* Kinesis.describeStreamSummary({
271 StreamName: stream.streamName,
272 });
273 expect(
274 updatedDescription.StreamDescriptionSummary.EncryptionType,
275 ).toEqual("NONE");
276
277 yield* stack.destroy();
278
279 yield* assertStreamDeleted(stream.streamName);
280 }),
281 { timeout: 180_000 },
282 );
283
284 test.provider(
285 "create stream with enhanced monitoring and update metrics",
286 (stack) =>
287 Effect.gen(function* () {
288 const stream = yield* stack.deploy(
289 Effect.gen(function* () {
290 return yield* Stream("MonitoredStream", {
291 shardLevelMetrics: ["IncomingBytes", "OutgoingRecords"],
292 });
293 }),
294 );
295
296 // Verify enhanced monitoring is enabled
297 const streamDescription = yield* Kinesis.describeStreamSummary({
298 StreamName: stream.streamName,
299 });
300 const metrics =
301 streamDescription.StreamDescriptionSummary.EnhancedMonitoring?.[0]
302 ?.ShardLevelMetrics ?? [];
303 expect(metrics).toContain("IncomingBytes");
304 expect(metrics).toContain("OutgoingRecords");
305
306 // Update metrics - add some, remove some
307 yield* stack.deploy(
308 Effect.gen(function* () {
309 return yield* Stream("MonitoredStream", {
310 shardLevelMetrics: [
311 "IncomingBytes",
312 "IncomingRecords",
313 "IteratorAgeMilliseconds",
314 ],
315 });
316 }),
317 );
318
319 // Verify metrics were updated
320 const updatedDescription = yield* Kinesis.describeStreamSummary({
321 StreamName: stream.streamName,
322 });
323 const updatedMetrics =
324 updatedDescription.StreamDescriptionSummary.EnhancedMonitoring?.[0]
325 ?.ShardLevelMetrics ?? [];
326 expect(updatedMetrics).toContain("IncomingBytes");
327 expect(updatedMetrics).toContain("IncomingRecords");
328 expect(updatedMetrics).toContain("IteratorAgeMilliseconds");
329 expect(updatedMetrics).not.toContain("OutgoingRecords");
330
331 yield* stack.destroy();
332
333 yield* assertStreamDeleted(stream.streamName);
334 }),
335 { timeout: 180_000 },
336 );
337
338 test.provider(
339 "idempotent create - stream already exists",
340 (stack) =>
341 Effect.gen(function* () {
342 // First create
343 const stream1 = yield* stack.deploy(
344 Effect.gen(function* () {
345 return yield* Stream("IdempotentStream", {});
346 }),
347 );
348 const streamName = stream1.streamName;
349
350 // Second create (should be idempotent)
351 const stream2 = yield* stack.deploy(
352 Effect.gen(function* () {
353 return yield* Stream("IdempotentStream", {});
354 }),
355 );
356 expect(stream2.streamName).toEqual(streamName);
357
358 yield* stack.destroy();
359
360 yield* assertStreamDeleted(streamName);
361 }),
362 { timeout: 180_000 },
363 );
364
365 test.provider(
366 "switch stream mode from provisioned to on-demand",
367 (stack) =>
368 Effect.gen(function* () {
369 const stream = yield* stack.deploy(
370 Effect.gen(function* () {
371 return yield* Stream("ModeChangeStream", {
372 streamMode: "PROVISIONED",
373 shardCount: 1,
374 });
375 }),
376 );
377
378 // Verify provisioned mode
379 const streamDescription = yield* Kinesis.describeStreamSummary({
380 StreamName: stream.streamName,
381 });
382 expect(
383 streamDescription.StreamDescriptionSummary.StreamModeDetails
384 ?.StreamMode,
385 ).toEqual("PROVISIONED");
386
387 // Update to on-demand mode
388 yield* stack.deploy(
389 Effect.gen(function* () {
390 return yield* Stream("ModeChangeStream", {
391 streamMode: "ON_DEMAND",
392 });
393 }),
394 );
395
396 // Verify on-demand mode
397 const updatedDescription = yield* Kinesis.describeStreamSummary({
398 StreamName: stream.streamName,
399 });
400 expect(
401 updatedDescription.StreamDescriptionSummary.StreamModeDetails
402 ?.StreamMode,
403 ).toEqual("ON_DEMAND");
404
405 yield* stack.destroy();
406
407 yield* assertStreamDeleted(stream.streamName);
408 }),
409 { timeout: 300_000 },
410 );
411
412 test.provider(
413 "decrease retention period",
414 (stack) =>
415 Effect.gen(function* () {
416 const stream = yield* stack.deploy(
417 Effect.gen(function* () {
418 return yield* Stream("RetentionStream", {
419 retentionPeriod: "48 hours",
420 });
421 }),
422 );
423
424 // Verify initial retention period
425 const streamDescription = yield* Kinesis.describeStreamSummary({
426 StreamName: stream.streamName,
427 });
428 expect(
429 streamDescription.StreamDescriptionSummary.RetentionPeriodHours,
430 ).toEqual(48);
431
432 // Decrease retention period back to default
433 yield* stack.deploy(
434 Effect.gen(function* () {
435 return yield* Stream("RetentionStream", {
436 retentionPeriod: "24 hours",
437 });
438 }),
439 );
440
441 // Verify retention period was decreased
442 const updatedDescription = yield* Kinesis.describeStreamSummary({
443 StreamName: stream.streamName,
444 });
445 expect(
446 updatedDescription.StreamDescriptionSummary.RetentionPeriodHours,
447 ).toEqual(24);
448
449 yield* stack.destroy();
450
451 yield* assertStreamDeleted(stream.streamName);
452 }),
453 { timeout: 180_000 },
454 );
455
456 test.provider(
457 "update stream resource policy and max record size",
458 (stack) =>
459 Effect.gen(function* () {
460 const stream = yield* stack.deploy(
461 Effect.gen(function* () {
462 return yield* Stream("PolicyStream", {
463 streamMode: "PROVISIONED",
464 shardCount: 1,
465 });
466 }),
467 );
468
469 const policy = JSON.stringify({
470 Version: "2012-10-17",
471 Statement: [
472 {
473 Sid: "AllowSameAccountDescribe",
474 Effect: "Allow",
475 Principal: {
476 AWS: `arn:aws:iam::${stream.streamArn.split(":")[4]}:root`,
477 },
478 Action: ["kinesis:DescribeStreamSummary"],
479 Resource: stream.streamArn,
480 },
481 ],
482 });
483
484 const updated = yield* stack.deploy(
485 Effect.gen(function* () {
486 return yield* Stream("PolicyStream", {
487 streamMode: "PROVISIONED",
488 shardCount: 1,
489 resourcePolicy: policy,
490 maxRecordSizeInKiB: 2048,
491 });
492 }),
493 );
494
495 expect(updated.resourcePolicy).toContain("AllowSameAccountDescribe");
496 expect(updated.maxRecordSizeInKiB).toEqual(2048);
497
498 const policyResponse = yield* Kinesis.getResourcePolicy({
499 ResourceARN: stream.streamArn,
500 });
501 expect(policyResponse.Policy).toContain("AllowSameAccountDescribe");
502
503 const summary = yield* Kinesis.describeStreamSummary({
504 StreamName: stream.streamName,
505 });
506 expect(summary.StreamDescriptionSummary.MaxRecordSizeInKiB).toEqual(
507 2048,
508 );
509
510 yield* stack.destroy();
511 yield* assertStreamDeleted(stream.streamName);
512 }),
513 { timeout: 240_000 },
514 );
515
516 test.provider(
517 "update warm throughput when account supports it",
518 (stack) =>
519 Effect.gen(function* () {
520 const accountSettings = yield* Kinesis.describeAccountSettings({});
521 const status =
522 accountSettings.MinimumThroughputBillingCommitment?.Status ??
523 "DISABLED";
524
525 if (status === "DISABLED") {
526 return;
527 }
528
529 const stream = yield* stack.deploy(
530 Effect.gen(function* () {
531 return yield* Stream("WarmThroughputStream");
532 }),
533 );
534
535 const updated = yield* stack.deploy(
536 Effect.gen(function* () {
537 return yield* Stream("WarmThroughputStream", {
538 warmThroughputMiBps: 10,
539 });
540 }),
541 );
542
543 expect(updated.warmThroughput?.targetMiBps).toEqual(10);
544
545 const summary = yield* Kinesis.describeStreamSummary({
546 StreamName: stream.streamName,
547 });
548 expect(
549 summary.StreamDescriptionSummary.WarmThroughput?.TargetMiBps,
550 ).toEqual(10);
551
552 yield* stack.destroy();
553 yield* assertStreamDeleted(stream.streamName);
554 }),
555 { timeout: 240_000 },
556 );
557
558 test.provider(
559 "owned stream (matching alchemy tags) is silently adopted without --adopt",
560 (stack) =>
561 Effect.gen(function* () {
562 yield* stack.destroy();
563
564 // Deterministic name so a previously-killed run's orphan is reclaimed
565 // by the pre-clean below instead of leaking forever.
566 const streamName = "alchemy-test-kinesis-adopt";
567
568 // Idempotent pre-clean: reclaim any leftover from an interrupted run.
569 yield* deleteStreamIfExists(streamName);
570 yield* assertStreamDeleted(streamName);
571
572 // The state wipe below detaches the stream from the scratch stack, so
573 // a crash in that window would orphan it — guarantee deletion.
574 yield* Effect.gen(function* () {
575 const initial = yield* stack.deploy(
576 Effect.gen(function* () {
577 return yield* Stream("AdoptableStream", { streamName });
578 }),
579 );
580 expect(initial.streamName).toEqual(streamName);
581
582 // Wipe state — the stream stays in Kinesis.
583 yield* Effect.gen(function* () {
584 const state = yield* yield* State;
585 yield* state.delete({
586 stack: stack.name,
587 stage: stack.stage,
588 fqn: "AdoptableStream",
589 });
590 }).pipe(Effect.provide(stack.state));
591
592 const adopted = yield* stack.deploy(
593 Effect.gen(function* () {
594 return yield* Stream("AdoptableStream", { streamName });
595 }),
596 );
597
598 expect(adopted.streamArn).toEqual(initial.streamArn);
599
600 yield* stack.destroy();
601 yield* assertStreamDeleted(streamName);
602 }).pipe(Effect.ensuring(deleteStreamIfExists(streamName)));
603 }),
604 { timeout: 240_000 },
605 );
606
607 test.provider(
608 "foreign-tagged stream requires adopt(true) to take over",
609 (stack) =>
610 Effect.gen(function* () {
611 yield* stack.destroy();
612
613 // Deterministic name so a previously-killed run's orphan is reclaimed
614 // by the pre-clean below instead of leaking forever.
615 const streamName = "alchemy-test-kinesis-takeover";
616
617 // Idempotent pre-clean: reclaim any leftover from an interrupted run.
618 yield* deleteStreamIfExists(streamName);
619 yield* assertStreamDeleted(streamName);
620
621 // The state wipe below detaches the stream from the scratch stack, so
622 // a crash in that window would orphan it — guarantee deletion.
623 yield* Effect.gen(function* () {
624 yield* stack.deploy(
625 Effect.gen(function* () {
626 return yield* Stream("Original", { streamName });
627 }),
628 );
629
630 yield* Effect.gen(function* () {
631 const state = yield* yield* State;
632 yield* state.delete({
633 stack: stack.name,
634 stage: stack.stage,
635 fqn: "Original",
636 });
637 }).pipe(Effect.provide(stack.state));
638
639 const takenOver = yield* stack
640 .deploy(
641 Effect.gen(function* () {
642 return yield* Stream("Different", { streamName });
643 }),
644 )
645 .pipe(adopt(true));
646
647 expect(takenOver.streamName).toEqual(streamName);
648
649 const tagsResp = yield* Kinesis.listTagsForResource({
650 ResourceARN: takenOver.streamArn,
651 });
652 const tagMap = Object.fromEntries(
653 (tagsResp.Tags ?? [])
654 .filter(
655 (t): t is { Key: string; Value: string } =>
656 typeof t.Value === "string",
657 )
658 .map((t) => [t.Key, t.Value]),
659 );
660 expect(tagMap["alchemy::id"]).toEqual("Different");
661
662 yield* stack.destroy();
663 yield* assertStreamDeleted(streamName);
664 }).pipe(Effect.ensuring(deleteStreamIfExists(streamName)));
665 }),
666 { timeout: 240_000 },
667 );
668
669 // Canonical `list()` test (AWS account/region-scoped collection): deploy a
670 // real stream, resolve the typed provider via `Provider.findProvider`, call
671 // `list()`, and assert the deployed stream appears in the exhaustively-
672 // paginated, fully-hydrated result.
673 test.provider(
674 "list enumerates the deployed stream",
675 (stack) =>
676 Effect.gen(function* () {
677 yield* stack.destroy();
678
679 const stream = yield* stack.deploy(
680 Effect.gen(function* () {
681 return yield* Stream("ListStream", {
682 streamName: "alchemy-test-kinesis-stream-list",
683 tags: { Environment: "test" },
684 });
685 }),
686 );
687
688 const provider = yield* Provider.findProvider(Stream);
689 const all = yield* provider.list();
690
691 const found = all.find((s) => s.streamName === stream.streamName);
692 expect(found).toBeDefined();
693 // The hydrated element must match the full `read` Attributes shape.
694 expect(found?.streamArn).toEqual(stream.streamArn);
695 expect(found?.streamStatus).toEqual("ACTIVE");
696 expect(found?.tags?.Environment).toEqual("test");
697
698 yield* stack.destroy();
699
700 yield* assertStreamDeleted(stream.streamName);
701 }),
702 { timeout: 240_000 },
703 );
704
705 // Idempotent out-of-band delete: safe when the stream never existed or was
706 // already deleted (tolerates ResourceNotFoundException), retries the
707 // transient in-use/limit errors, and dies loudly on anything else so a
708 // cleanup failure is never silent. Error channel is `never` so it can be
709 // used as an `Effect.ensuring` finalizer.
710 const deleteStreamIfExists = Effect.fn(function* (streamName: string) {
711 yield* Kinesis.deleteStream({
712 StreamName: streamName,
713 EnforceConsumerDeletion: true,
714 }).pipe(
715 Effect.retry({
716 while: (e) =>
717 e._tag === "ResourceInUseException" ||
718 e._tag === "LimitExceededException",
719 schedule: Schedule.spaced("5 seconds"),
720 times: 8,
721 }),
722 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
723 Effect.orDie,
724 );
725 });
726
727 class StreamStillExists extends Data.TaggedError("StreamStillExists") {}
728
729 const assertStreamDeleted = Effect.fn(function* (streamName: string) {
730 yield* Kinesis.describeStreamSummary({
731 StreamName: streamName,
732 }).pipe(
733 Effect.flatMap(() => Effect.fail(new StreamStillExists())),
734 Effect.retry({
735 while: (e: { _tag: string }) =>
736 e._tag === "StreamStillExists" ||
737 // During stream deletion, AWS may return incomplete responses that fail parsing
738 e._tag === "ParseError",
739 schedule: Schedule.max([
740 Schedule.exponential(500),
741 Schedule.recurs(8),
742 ]),
743 }),
744 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
745 );
746 });
747 },
748);