1import * as AWS from "@/AWS";
2import * as Core from "@/Test/Core";
3import * as Test from "@/Test/Alchemy";
4import * as kv from "@distilled.cloud/aws/kinesis-video";
5import { describe, expect } from "alchemy-test";
6import * as Data from "effect/Data";
7import * as Effect from "effect/Effect";
8import * as Schedule from "effect/Schedule";
9import * as HttpClient from "effect/http/HttpClient";
10import * as HttpClientRequest from "effect/http/HttpClientRequest";
11import KinesisVideoTestFunctionLive, {
12 KinesisVideoTestFunction,
13} from "./handler";
14
15const testOptions = { providers: AWS.providers() };
16const { test, beforeAll, afterAll } = Test.make(testOptions);
17const sharedStack = Core.scratchStack(testOptions, "KinesisVideoBindings");
18
19const readinessPolicy = Schedule.max([
20 Schedule.fixed("2 seconds"),
21 Schedule.recurs(75),
22]);
23
24let baseUrl: string;
25
26class TransientUpstream extends Data.TaggedError("TransientUpstream")<{
27 readonly status: number;
28 readonly body: string;
29}> {}
30
31class FixtureStillExists extends Data.TaggedError("FixtureStillExists")<{
32 readonly streams: string[];
33 readonly channels: string[];
34}> {}
35
36// Retry transient 5xx from the shared fixture under parallel-suite load;
37// genuine 4xx failures surface immediately.
38const send = (request: HttpClientRequest.HttpClientRequest) =>
39 HttpClient.execute(request).pipe(
40 Effect.flatMap((response) =>
41 response.status >= 500
42 ? response.text.pipe(
43 Effect.flatMap((body) =>
44 Effect.fail(
45 new TransientUpstream({ status: response.status, body }),
46 ),
47 ),
48 )
49 : Effect.succeed(response),
50 ),
51 Effect.retry({
52 while: (e) => e._tag === "TransientUpstream",
53 schedule: Schedule.max([
54 Schedule.exponential("500 millis"),
55 Schedule.recurs(6),
56 ]),
57 }),
58 );
59
60// The fixture's role policies are freshly created on every deploy, and IAM
61// is eventually consistent — the first data-plane calls can transiently
62// return AccessDeniedException. Fetch the route's JSON outcome and repeat
63// (bounded) until propagation settles on the terminal typed tag.
64const fetchOutcome = (url: string) =>
65 send(HttpClientRequest.get(url)).pipe(
66 Effect.flatMap((response) =>
67 response.status === 200
68 ? (response.json as Effect.Effect<unknown>)
69 : Effect.fail(
70 new TransientUpstream({ status: response.status, body: "" }),
71 ),
72 ),
73 Effect.map(
74 (body) =>
75 body as {
76 ok?: boolean;
77 count?: number;
78 url?: string;
79 contentType?: string;
80 timedOut?: boolean;
81 answered?: boolean;
82 errorTag?: string;
83 errorMessage?: string;
84 },
85 ),
86 Effect.repeat({
87 schedule: Schedule.spaced("3 seconds"),
88 until: (body): boolean => body.errorTag !== "AccessDeniedException",
89 times: 20,
90 }),
91 );
92
93describe(
94 "KinesisVideo Bindings",
95 {
96 tags: [
97 "provider:aws",
98 "provider:aws:kinesisvideo",
99 "provider:aws:lambda",
100 "live",
101 ],
102 },
103 () => {
104 beforeAll(
105 Effect.gen(function* () {
106 yield* Effect.logInfo(
107 "KinesisVideo test setup: destroying previous resources",
108 );
109 yield* sharedStack.destroy();
110
111 yield* Effect.logInfo("KinesisVideo test setup: deploying fixture");
112 const { functionUrl } = yield* sharedStack.deploy(
113 Effect.gen(function* () {
114 return yield* KinesisVideoTestFunction;
115 }).pipe(Effect.provide(KinesisVideoTestFunctionLive)),
116 );
117
118 expect(functionUrl).toBeTruthy();
119 baseUrl = functionUrl!.replace(/\/+$/, "");
120
121 yield* Effect.logInfo(
122 `KinesisVideo test setup: probing readiness at ${baseUrl}/info`,
123 );
124 yield* HttpClient.get(`${baseUrl}/info`).pipe(
125 Effect.flatMap((response) =>
126 response.status === 200
127 ? Effect.void
128 : Effect.fail(
129 new Error(`Function not ready: ${response.status}`),
130 ),
131 ),
132 Effect.retry({ schedule: readinessPolicy }),
133 );
134 }),
135 { timeout: 300_000 },
136 );
137
138 afterAll.skipIf(!!process.env.NO_DESTROY)(
139 Effect.gen(function* () {
140 yield* sharedStack.destroy();
141 // assert the fixture stream + channel are gone out-of-band. Physical
142 // names are deterministic: `${stackName}-${logicalId}-${stage}-…`, so a
143 // BEGINS_WITH listing pinpoints exactly the resources this suite owns.
144 // Deletion is asynchronous — DELETING counts as gone.
145 yield* Effect.gen(function* () {
146 const streams = yield* kv.listStreams({
147 StreamNameCondition: {
148 ComparisonOperator: "BEGINS_WITH",
149 ComparisonValue: "KinesisVideoBindings-FixtureStream-test-",
150 },
151 });
152 const liveStreams = (streams.StreamInfoList ?? []).filter(
153 (s) => s.Status !== "DELETING",
154 );
155 const channels = yield* kv.listSignalingChannels({
156 ChannelNameCondition: {
157 ComparisonOperator: "BEGINS_WITH",
158 ComparisonValue: "KinesisVideoBindings-FixtureChannel-test-",
159 },
160 });
161 const liveChannels = (channels.ChannelInfoList ?? []).filter(
162 (c) => c.ChannelStatus !== "DELETING",
163 );
164 if (liveStreams.length > 0 || liveChannels.length > 0) {
165 return yield* Effect.fail(
166 new FixtureStillExists({
167 streams: liveStreams.map((s) => s.StreamName ?? "?"),
168 channels: liveChannels.map((c) => c.ChannelName ?? "?"),
169 }),
170 );
171 }
172 }).pipe(
173 Effect.retry({
174 while: (e): boolean => e._tag === "FixtureStillExists",
175 schedule: Schedule.max([
176 Schedule.spaced("3 seconds"),
177 Schedule.recurs(10),
178 ]),
179 }),
180 (effect) =>
181 Core.withProviders(effect, testOptions, "KinesisVideoBindings"),
182 );
183 }),
184 {
185 timeout: 240_000,
186 },
187 );
188
189 describe("GetHLSStreamingSessionURL", () => {
190 test.provider(
191 "lambda reaches the archived-media data plane through the binding",
192 (_stack) =>
193 Effect.gen(function* () {
194 const body = yield* fetchOutcome(`${baseUrl}/hls`);
195 // The fixture stream has never ingested media, so LIVE playback
196 // deterministically returns the archived-media data plane's typed
197 // no-fragments error. Getting THIS tag (and not NotAuthorized /
198 // an unknown error) proves GetDataEndpoint resolution, IAM, and
199 // the signed data-plane call all work.
200 expect(body.errorTag).toBe("ResourceNotFoundException");
201 }),
202 { timeout: 120_000 },
203 );
204 });
205
206 describe("GetIceServerConfig", () => {
207 test.provider(
208 "lambda fetches TURN servers through the binding",
209 (_stack) =>
210 Effect.gen(function* () {
211 const response = yield* send(
212 HttpClientRequest.get(`${baseUrl}/ice`),
213 );
214 expect(response.status).toBe(200);
215 const body = (yield* response.json) as {
216 servers?: {
217 uris: string[];
218 hasCredentials: boolean;
219 ttl?: number;
220 }[];
221 errorTag?: string;
222 };
223 expect(body.errorTag).toBeUndefined();
224 expect(body.servers!.length).toBeGreaterThan(0);
225 const turn = body.servers!.find((s) =>
226 s.uris.some((uri) => uri.startsWith("turn")),
227 );
228 expect(turn).toBeDefined();
229 expect(turn?.hasCredentials).toBe(true);
230 }),
231 { timeout: 120_000 },
232 );
233 });
234
235 describe("GetDASHStreamingSessionURL", () => {
236 test.provider(
237 "lambda reaches the archived-media data plane through the binding",
238 (_stack) =>
239 Effect.gen(function* () {
240 const body = yield* fetchOutcome(`${baseUrl}/dash`);
241 // Same rationale as /hls: LIVE playback on the never-ingested
242 // stream returns the typed no-fragments error.
243 expect(body.errorTag).toBe("ResourceNotFoundException");
244 }),
245 { timeout: 120_000 },
246 );
247 });
248
249 describe("ListFragments", () => {
250 test.provider(
251 "lambda lists fragments of the empty stream (empty page)",
252 (_stack) =>
253 Effect.gen(function* () {
254 const body = yield* fetchOutcome(`${baseUrl}/fragments`);
255 // toMatchObject so a failure diff surfaces the errorTag
256 expect(body).toMatchObject({ ok: true, count: 0 });
257 }),
258 { timeout: 120_000 },
259 );
260 });
261
262 describe("GetClip", () => {
263 test.provider(
264 "lambda requests a clip and receives the typed no-fragments error",
265 (_stack) =>
266 Effect.gen(function* () {
267 const body = yield* fetchOutcome(`${baseUrl}/clip`);
268 expect(body.ok).toBe(false);
269 expect(body.errorTag).toBe("ResourceNotFoundException");
270 }),
271 { timeout: 120_000 },
272 );
273 });
274
275 describe("GetImages", () => {
276 test.provider(
277 "lambda extracts images from the empty stream (empty page or typed error)",
278 (_stack) =>
279 Effect.gen(function* () {
280 const body = yield* fetchOutcome(`${baseUrl}/images`);
281 // The empty stream always rejects the request with a typed error;
282 // WHICH tag is a function of the seconds-old stream's age: while
283 // the trailing-60s window still predates the stream's creation the
284 // service answers InvalidArgumentException ("End timestamp … is
285 // outside of the stream retention period"), and once the window
286 // overlaps its lifetime it answers ResourceNotFoundException
287 // ("No fragments found"). Both are data-plane responses, proving
288 // endpoint discovery, IAM, and the signed call.
289 expect(body.ok).toBe(false);
290 expect([
291 "ResourceNotFoundException",
292 "InvalidArgumentException",
293 ]).toContain(body.errorTag);
294 }),
295 { timeout: 120_000 },
296 );
297 });
298
299 describe("GetMediaForFragmentList", () => {
300 test.provider(
301 "lambda requests fragment media and receives the typed no-fragment error",
302 (_stack) =>
303 Effect.gen(function* () {
304 const body = yield* fetchOutcome(`${baseUrl}/fragment-media`);
305 // The nonexistent fragment number is deterministically rejected by
306 // the media data plane: "Fragment numbers are invalid".
307 expect(body.ok).toBe(false);
308 expect(body.errorTag).toBe("InvalidArgumentException");
309 expect(body.errorMessage).toContain("Fragment numbers are invalid");
310 }),
311 { timeout: 120_000 },
312 );
313 });
314
315 describe("JoinStorageSession", () => {
316 test.provider(
317 "lambda reaches the WebRTC storage plane; unconfigured storage is a typed error",
318 (_stack) =>
319 Effect.gen(function* () {
320 const body = yield* fetchOutcome(`${baseUrl}/join-storage`);
321 // The fixture channel has no MediaStorageConfiguration, so WEBRTC
322 // endpoint discovery deterministically rejects the channel:
323 // "MediaStorageConfiguration is required for WEBRTC protocol".
324 expect(body.ok).toBe(false);
325 expect(body.errorTag).toBe("InvalidArgumentException");
326 expect(body.errorMessage).toContain("MediaStorageConfiguration");
327 }),
328 { timeout: 120_000 },
329 );
330 });
331
332 describe("JoinStorageSessionAsViewer", () => {
333 test.provider(
334 "lambda reaches the WebRTC storage plane as viewer; typed error without storage",
335 (_stack) =>
336 Effect.gen(function* () {
337 const body = yield* fetchOutcome(`${baseUrl}/join-storage-viewer`);
338 // Same deterministic rejection as /join-storage.
339 expect(body.ok).toBe(false);
340 expect(body.errorTag).toBe("InvalidArgumentException");
341 expect(body.errorMessage).toContain("MediaStorageConfiguration");
342 }),
343 { timeout: 120_000 },
344 );
345 });
346
347 describe("SendAlexaOfferToMaster", () => {
348 test.provider(
349 "lambda delivers an SDP offer through the signaling plane",
350 (_stack) =>
351 Effect.gen(function* () {
352 const body = yield* fetchOutcome(`${baseUrl}/alexa-offer`);
353 // The signaling data plane accepts the SDP stub and, with no
354 // master connected to answer, deterministically holds the offer
355 // until the fixture's bounded 5s timeout.
356 expect(body).toMatchObject({ ok: true, timedOut: true });
357 }),
358 { timeout: 120_000 },
359 );
360 });
361
362 describe("GetMedia", () => {
363 test.provider(
364 "lambda opens a media connection through the binding",
365 (_stack) =>
366 Effect.gen(function* () {
367 const body = yield* fetchOutcome(`${baseUrl}/media`);
368 // The media data plane deterministically accepts the connection to
369 // the empty stream and answers with matroska headers before any
370 // fragment exists.
371 expect(body).toMatchObject({ ok: true, contentType: "video/webm" });
372 }),
373 { timeout: 120_000 },
374 );
375 });
376 },
377);