1import * as AWS from "@/AWS";
2import * as Core from "@/Test/Core";
3import * as Test from "@/Test/Alchemy";
4import * as eventbridge from "@distilled.cloud/aws/eventbridge";
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 IvsTestFunctionLive, { IvsTestFunction } from "./fixtures/handler";
12
13const testOptions = { providers: AWS.providers() };
14const { test, beforeAll, afterAll } = Test.make(testOptions);
15const sharedStack = Core.scratchStack(testOptions, "IVSBindings");
16
17// Lambda function URL cold-start (DNS, IAM propagation, init) can take well
18// over 60s on a fresh deploy.
19const readinessPolicy = Schedule.max([
20 Schedule.fixed("2 seconds"),
21 Schedule.recurs(75),
22]);
23
24let baseUrl: string;
25let functionArn: string;
26
27class TransientUpstream extends Data.TaggedError("TransientUpstream")<{
28 readonly status: number;
29 readonly body: string;
30}> {}
31
32// The shared Lambda fixture occasionally answers a transient 5xx under load
33// (cold re-init, IAM propagation on the freshly attached policy that the
34// handler's `Effect.orDie` surfaces as a 500). Retry only 5xx; a genuine
35// 4xx/assertion failure surfaces immediately.
36const send = (request: HttpClientRequest.HttpClientRequest) =>
37 HttpClient.execute(request).pipe(
38 Effect.flatMap((response) =>
39 response.status >= 500
40 ? response.text.pipe(
41 Effect.flatMap((body) =>
42 Effect.fail(
43 new TransientUpstream({ status: response.status, body }),
44 ),
45 ),
46 )
47 : Effect.succeed(response),
48 ),
49 Effect.retry({
50 while: (e) => e._tag === "TransientUpstream",
51 schedule: Schedule.max([
52 Schedule.exponential("500 millis"),
53 Schedule.recurs(6),
54 ]),
55 }),
56 );
57
58const getJson = (path: string) =>
59 send(HttpClientRequest.get(`${baseUrl}${path}`)).pipe(
60 Effect.flatMap((r) => r.json),
61 );
62
63const postJson = (path: string) =>
64 send(HttpClientRequest.post(`${baseUrl}${path}`)).pipe(
65 Effect.flatMap((r) => r.json),
66 );
67
68describe.sequential(
69 "IVS Bindings",
70 {
71 tags: [
72 "provider:aws",
73 "provider:aws:batch",
74 "provider:aws:ivs",
75 "provider:aws:lambda",
76 "live",
77 ],
78 },
79 () => {
80 beforeAll(
81 Effect.gen(function* () {
82 yield* Effect.logInfo("IVS test setup: destroying previous resources");
83 yield* sharedStack.destroy();
84
85 yield* Effect.logInfo("IVS test setup: deploying fixture");
86 const attrs = yield* sharedStack.deploy(
87 Effect.gen(function* () {
88 return yield* IvsTestFunction;
89 }).pipe(Effect.provide(IvsTestFunctionLive)),
90 );
91
92 expect(attrs.functionUrl).toBeTruthy();
93 baseUrl = attrs.functionUrl!.replace(/\/+$/, "");
94 functionArn = attrs.functionArn;
95
96 const readinessUrl = `${baseUrl}/bindings`;
97 yield* Effect.logInfo(
98 `IVS test setup: probing readiness at ${readinessUrl}`,
99 );
100 yield* HttpClient.get(readinessUrl).pipe(
101 Effect.flatMap((response) =>
102 response.status === 200
103 ? Effect.succeed(response)
104 : Effect.fail(
105 new Error(`Function not ready: ${response.status}`),
106 ),
107 ),
108 Effect.tapError((error) =>
109 Effect.logWarning(
110 `IVS test setup: fixture not ready yet (${String(error)})`,
111 ),
112 ),
113 Effect.retry({ schedule: readinessPolicy }),
114 );
115 }),
116 { timeout: 240_000 },
117 );
118
119 afterAll(sharedStack.destroy(), { timeout: 120_000 });
120
121 describe("binding registration", () => {
122 test.provider("all nine capabilities initialize in the runtime", () =>
123 Effect.gen(function* () {
124 const response = (yield* getJson("/bindings")) as { bound: string[] };
125 expect(response.bound).toHaveLength(9);
126 }),
127 );
128 });
129
130 describe("GetStream", () => {
131 test.provider(
132 "an idle channel answers with the typed ChannelNotBroadcasting",
133 () =>
134 Effect.gen(function* () {
135 const response = (yield* getJson("/stream")) as {
136 live: boolean;
137 viewers: number;
138 };
139 expect(response.live).toBe(false);
140 expect(response.viewers).toBe(0);
141 }),
142 );
143 });
144
145 describe("GetStreamSession", () => {
146 test.provider("a never-broadcast channel has no latest session", () =>
147 Effect.gen(function* () {
148 const response = (yield* getJson("/session")) as { found: boolean };
149 expect(response.found).toBe(false);
150 }),
151 );
152 });
153
154 describe("ListStreamSessions", () => {
155 test.provider("the idle channel's session history is empty", () =>
156 Effect.gen(function* () {
157 const response = (yield* getJson("/sessions")) as { count: number };
158 expect(response.count).toBe(0);
159 }),
160 );
161 });
162
163 describe("ListStreams", () => {
164 test.provider("account-level live-stream enumeration succeeds", () =>
165 Effect.gen(function* () {
166 const response = (yield* getJson("/streams")) as { count: number };
167 expect(response.count).toBeGreaterThanOrEqual(0);
168 }),
169 );
170 });
171
172 describe("PutMetadata", () => {
173 test.provider(
174 "inserting metadata into an idle stream is the typed ChannelNotBroadcasting",
175 () =>
176 Effect.gen(function* () {
177 const response = (yield* postJson("/metadata")) as {
178 inserted: boolean;
179 tag?: string;
180 };
181 expect(response.inserted).toBe(false);
182 expect(response.tag).toBe("ChannelNotBroadcasting");
183 }),
184 );
185 });
186
187 describe("StopStream", () => {
188 test.provider(
189 "stopping an idle channel is the typed ChannelNotBroadcasting",
190 () =>
191 Effect.gen(function* () {
192 const response = (yield* postJson("/stop")) as {
193 stopped: boolean;
194 tag?: string;
195 };
196 expect(response.stopped).toBe(false);
197 expect(response.tag).toBe("ChannelNotBroadcasting");
198 }),
199 );
200 });
201
202 describe("StartViewerSessionRevocation", () => {
203 test.provider("revoking a viewer session succeeds", () =>
204 Effect.gen(function* () {
205 const response = (yield* postJson("/revoke")) as {
206 ok: boolean;
207 tag?: string;
208 };
209 expect(response.ok).toBe(true);
210 }),
211 );
212 });
213
214 describe("InsertAdBreak", () => {
215 test.provider(
216 "inserting an ad break into an idle stream is a typed failure",
217 () =>
218 Effect.gen(function* () {
219 const response = (yield* postJson("/adbreak")) as {
220 inserted: boolean;
221 tag?: string;
222 };
223 expect(response.inserted).toBe(false);
224 expect([
225 "ChannelNotBroadcasting",
226 "ValidationException",
227 "ConflictException",
228 ]).toContain(response.tag);
229 }),
230 );
231 });
232
233 describe("BatchStartViewerSessionRevocation", () => {
234 test.provider(
235 "batch-revoking a viewer session reports no per-pair errors",
236 () =>
237 Effect.gen(function* () {
238 const response = (yield* postJson("/revoke-batch")) as {
239 errorCount: number;
240 };
241 expect(response.errorCount).toBe(0);
242 }),
243 );
244 });
245
246 describe(
247 "consumeStreamEvents",
248 { tags: ["provider:aws:eventbridge"] },
249 () => {
250 test.provider(
251 "the deploy created an EventBridge rule targeting the function",
252 () =>
253 Effect.gen(function* () {
254 // Out-of-band via distilled: the fixture's consumeStreamEvents
255 // must have materialized as a rule on the default bus with the
256 // Lambda as target.
257 const { RuleNames } = yield* eventbridge.listRuleNamesByTarget({
258 TargetArn: functionArn,
259 });
260 expect((RuleNames ?? []).length).toBeGreaterThanOrEqual(1);
261 }),
262 );
263 },
264 );
265 },
266);