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