1import * as AWS from "@/AWS";
2import { Stream } from "@/AWS/KinesisVideo";
3import * as Test from "@/Test/Alchemy";
4import * as kv from "@distilled.cloud/aws/kinesis-video";
5import { expect } from "alchemy-test";
6import * as Data from "effect/Data";
7import * as Effect from "effect/Effect";
8import * as Schedule from "effect/Schedule";
9
10const { test } = Test.make({ providers: AWS.providers() });
11
12const findStream = (streamName: string) =>
13 kv.describeStream({ StreamName: streamName }).pipe(
14 Effect.map((r) => r.StreamInfo),
15 Effect.catchTag("ResourceNotFoundException", () =>
16 Effect.succeed(undefined),
17 ),
18 );
19
20class StreamStillExists extends Data.TaggedError("StreamStillExists")<{
21 readonly streamName: string;
22}> {}
23
24// DeleteStream is asynchronous: the stream lingers in DELETING (data already
25// inaccessible) for a while — treat DELETING as deleted.
26const assertStreamDeleted = (streamName: string) =>
27 findStream(streamName).pipe(
28 Effect.flatMap((info) =>
29 info === undefined || info.Status === "DELETING"
30 ? Effect.void
31 : Effect.fail(new StreamStillExists({ streamName })),
32 ),
33 Effect.retry({
34 while: (e) => e._tag === "StreamStillExists",
35 schedule: Schedule.max([
36 Schedule.spaced("3 seconds"),
37 Schedule.recurs(20),
38 ]),
39 }),
40 );
41
42const fetchTags = (streamArn: string) =>
43 kv
44 .listTagsForStream({ StreamARN: streamArn })
45 .pipe(Effect.map((r) => r.Tags ?? {}));
46
47test.provider(
48 "create, update retention/mediaType/tags, delete video stream",
49 (stack) =>
50 Effect.gen(function* () {
51 yield* stack.destroy();
52
53 const stream = yield* stack.deploy(
54 Effect.gen(function* () {
55 return yield* Stream("TestVideoStream", {
56 dataRetention: "24 hours",
57 tags: { Environment: "test" },
58 });
59 }),
60 );
61
62 expect(stream.streamName).toBeDefined();
63 expect(stream.streamArn).toContain(":stream/");
64
65 // out-of-band verification via distilled
66 const created = yield* findStream(stream.streamName);
67 expect(created?.Status).toBe("ACTIVE");
68 expect(created?.DataRetentionInHours).toBe(24);
69 const tags = yield* fetchTags(stream.streamArn);
70 expect(tags.Environment).toBe("test");
71 expect(tags["alchemy::id"]).toBe("TestVideoStream");
72
73 // update: increase retention, set mediaType + deviceName, extra tag
74 const updated = yield* stack.deploy(
75 Effect.gen(function* () {
76 return yield* Stream("TestVideoStream", {
77 dataRetention: "48 hours",
78 mediaType: "video/h264",
79 deviceName: "camera-1",
80 tags: { Environment: "test", Extra: "yes" },
81 });
82 }),
83 );
84 expect(updated.streamName).toBe(stream.streamName);
85 expect(updated.streamArn).toBe(stream.streamArn);
86
87 const afterUpdate = yield* findStream(stream.streamName);
88 expect(afterUpdate?.DataRetentionInHours).toBe(48);
89 expect(afterUpdate?.MediaType).toBe("video/h264");
90 expect(afterUpdate?.DeviceName).toBe("camera-1");
91 const tagsAfter = yield* fetchTags(stream.streamArn);
92 expect(tagsAfter.Extra).toBe("yes");
93
94 // decrease retention back down (DECREASE_DATA_RETENTION path)
95 const decreased = yield* stack.deploy(
96 Effect.gen(function* () {
97 return yield* Stream("TestVideoStream", {
98 dataRetention: "24 hours",
99 mediaType: "video/h264",
100 deviceName: "camera-1",
101 tags: { Environment: "test" },
102 });
103 }),
104 );
105 expect(decreased.streamArn).toBe(stream.streamArn);
106 const afterDecrease = yield* findStream(stream.streamName);
107 expect(afterDecrease?.DataRetentionInHours).toBe(24);
108 const tagsAfterRemoval = yield* fetchTags(stream.streamArn);
109 expect(tagsAfterRemoval.Extra).toBeUndefined();
110
111 yield* stack.destroy();
112 yield* assertStreamDeleted(stream.streamName);
113 }),
114 {
115 tags: ["provider:aws", "provider:aws:kinesisvideo", "live"],
116 timeout: 240_000,
117 },
118);
119
120test.provider(
121 "explicit name and replacement on rename",
122 (stack) =>
123 Effect.gen(function* () {
124 yield* stack.destroy();
125
126 const first = yield* stack.deploy(
127 Effect.gen(function* () {
128 return yield* Stream("NamedVideoStream", {
129 streamName: "alchemy-test-kvs-stream-a",
130 });
131 }),
132 );
133 expect(first.streamName).toBe("alchemy-test-kvs-stream-a");
134
135 // renaming triggers a replacement: new physical stream, old gone
136 const second = yield* stack.deploy(
137 Effect.gen(function* () {
138 return yield* Stream("NamedVideoStream", {
139 streamName: "alchemy-test-kvs-stream-b",
140 });
141 }),
142 );
143 expect(second.streamName).toBe("alchemy-test-kvs-stream-b");
144 expect(second.streamArn).not.toBe(first.streamArn);
145
146 yield* assertStreamDeleted(first.streamName);
147
148 yield* stack.destroy();
149 yield* assertStreamDeleted(second.streamName);
150 }),
151 {
152 tags: ["provider:aws", "provider:aws:kinesisvideo", "live"],
153 timeout: 240_000,
154 },
155);