1import * as IVSRealtime from "@/AWS/IVSRealtime";
2import * as Lambda from "@/AWS/Lambda";
3import * as Duration from "effect/Duration";
4import * as Effect from "effect/Effect";
5import * as Layer from "effect/Layer";
6import * as Redacted from "effect/Redacted";
7import * as Stream from "effect/Stream";
8import { HttpServerRequest } from "effect/http/HttpServerRequest";
9import * as HttpServerResponse from "effect/http/HttpServerResponse";
10import path from "pathe";
11
12const main = path.resolve(import.meta.dirname, "handler.ts");
13
14/** The user ID the fixture mints participant tokens for. */
15export const TEST_USER_ID = "alchemy-test-user";
16
17/**
18 * A well-formed participant id that does not exist — participant/session
19 * routes prove end-to-end wiring by observing the API's typed response.
20 */
21const MISSING_PARTICIPANT_ID = "abcDEF123456";
22/** A well-formed stage session id that does not exist. */
23const MISSING_SESSION_ID = "st-0000AbCd0000";
24
25export class IVSRealtimeTestFunction extends Lambda.Function<Lambda.Function>()(
26 "IVSRealtimeTestFunction",
27) {}
28
29export default IVSRealtimeTestFunction.make(
30 {
31 main,
32 functionUrl: true,
33 timeout: Duration.seconds(30),
34 },
35 Effect.gen(function* () {
36 const stage = yield* IVSRealtime.Stage("BindingsStage", {
37 stageName: "alchemy-test-ivsrealtime-bindings",
38 tags: { fixture: "ivsrealtime-bindings" },
39 });
40 const overflowStage = yield* IVSRealtime.Stage("BindingsOverflowStage", {
41 stageName: "alchemy-test-ivsrealtime-bindings-b",
42 tags: { fixture: "ivsrealtime-bindings" },
43 });
44
45 // Event source: subscribe the host to IVS Real-Time stage updates and
46 // composition state changes. The deploy proves the EventBridge rule +
47 // invoke permission wiring.
48 yield* IVSRealtime.consumeStageEvents(
49 { kinds: ["stage-update", "composition-state-change"] },
50 (events) =>
51 Stream.runForEach(events, (event) =>
52 Effect.log(
53 `ivs stage event: ${event.detail.event_name} (${event.detail.participant_id})`,
54 ),
55 ),
56 );
57
58 const createParticipantToken =
59 yield* IVSRealtime.CreateParticipantToken(stage);
60 const disconnectParticipant =
61 yield* IVSRealtime.DisconnectParticipant(stage);
62 const getParticipant = yield* IVSRealtime.GetParticipant(stage);
63 const listParticipants = yield* IVSRealtime.ListParticipants(stage);
64 const listParticipantEvents =
65 yield* IVSRealtime.ListParticipantEvents(stage);
66 const listParticipantReplicas =
67 yield* IVSRealtime.ListParticipantReplicas(stage);
68 const getStageSession = yield* IVSRealtime.GetStageSession(stage);
69 const listStageSessions = yield* IVSRealtime.ListStageSessions(stage);
70 const startParticipantReplication =
71 yield* IVSRealtime.StartParticipantReplication(stage, overflowStage);
72 const stopParticipantReplication =
73 yield* IVSRealtime.StopParticipantReplication(stage, overflowStage);
74 const startComposition = yield* IVSRealtime.StartComposition(stage);
75 const stopComposition = yield* IVSRealtime.StopComposition();
76 const getComposition = yield* IVSRealtime.GetComposition();
77 const listCompositions = yield* IVSRealtime.ListCompositions();
78
79 const bound = {
80 createParticipantToken,
81 disconnectParticipant,
82 getParticipant,
83 listParticipants,
84 listParticipantEvents,
85 listParticipantReplicas,
86 getStageSession,
87 listStageSessions,
88 startParticipantReplication,
89 stopParticipantReplication,
90 startComposition,
91 stopComposition,
92 getComposition,
93 listCompositions,
94 };
95
96 return {
97 fetch: Effect.gen(function* () {
98 const request = yield* HttpServerRequest;
99 const url = new URL(request.originalUrl);
100 const pathname = url.pathname;
101
102 // Cheap readiness route — no AWS call.
103 if (request.method === "GET" && pathname === "/ping") {
104 return yield* HttpServerResponse.json({ ok: true });
105 }
106
107 if (request.method === "GET" && pathname === "/bindings") {
108 return yield* HttpServerResponse.json({ bound: Object.keys(bound) });
109 }
110
111 if (request.method === "POST" && pathname === "/token") {
112 const { participantToken } = yield* createParticipantToken({
113 userId: TEST_USER_ID,
114 capabilities: ["PUBLISH", "SUBSCRIBE"],
115 // Exercises the Duration.Input -> wire minutes conversion.
116 duration: "30 minutes",
117 attributes: { displayName: "Alchemy" },
118 });
119 // The token is sensitive — assert on its shape, never echo it.
120 const token = participantToken?.token;
121 const tokenLength =
122 token === undefined
123 ? 0
124 : Redacted.isRedacted(token)
125 ? Redacted.value(token).length
126 : token.length;
127 return yield* HttpServerResponse.json({
128 tokenLength,
129 tokenIsRedacted: token !== undefined && Redacted.isRedacted(token),
130 participantId: participantToken?.participantId,
131 duration: participantToken?.duration,
132 expirationTime: participantToken?.expirationTime,
133 });
134 }
135
136 if (request.method === "POST" && pathname === "/disconnect") {
137 const result = yield* disconnectParticipant({
138 participantId: MISSING_PARTICIPANT_ID,
139 reason: "alchemy test",
140 }).pipe(
141 Effect.map(() => ({ ok: true })),
142 Effect.catchTag(
143 ["ResourceNotFoundException", "ValidationException"],
144 (e) => Effect.succeed({ tag: e._tag }),
145 ),
146 );
147 return yield* HttpServerResponse.json(result);
148 }
149
150 if (request.method === "GET" && pathname === "/sessions") {
151 const { stageSessions } = yield* listStageSessions();
152 return yield* HttpServerResponse.json({
153 count: stageSessions.length,
154 });
155 }
156
157 if (request.method === "GET" && pathname === "/session-missing") {
158 const result = yield* getStageSession({
159 sessionId: MISSING_SESSION_ID,
160 }).pipe(
161 Effect.map(() => ({ found: true })),
162 Effect.catchTag(
163 ["ResourceNotFoundException", "ValidationException"],
164 (e) => Effect.succeed({ tag: e._tag }),
165 ),
166 );
167 return yield* HttpServerResponse.json(result);
168 }
169
170 if (request.method === "GET" && pathname === "/participants") {
171 const result = yield* listParticipants({
172 sessionId: MISSING_SESSION_ID,
173 }).pipe(
174 Effect.map((r) => ({ count: r.participants.length })),
175 Effect.catchTag("ValidationException", (e) =>
176 Effect.succeed({ tag: e._tag }),
177 ),
178 );
179 return yield* HttpServerResponse.json(result);
180 }
181
182 if (request.method === "GET" && pathname === "/participant-missing") {
183 const result = yield* getParticipant({
184 sessionId: MISSING_SESSION_ID,
185 participantId: MISSING_PARTICIPANT_ID,
186 }).pipe(
187 Effect.map(() => ({ found: true })),
188 Effect.catchTag(
189 ["ResourceNotFoundException", "ValidationException"],
190 (e) => Effect.succeed({ tag: e._tag }),
191 ),
192 );
193 return yield* HttpServerResponse.json(result);
194 }
195
196 if (request.method === "GET" && pathname === "/participant-events") {
197 const result = yield* listParticipantEvents({
198 sessionId: MISSING_SESSION_ID,
199 participantId: MISSING_PARTICIPANT_ID,
200 }).pipe(
201 Effect.map((r) => ({ count: r.events.length })),
202 Effect.catchTag("ValidationException", (e) =>
203 Effect.succeed({ tag: e._tag }),
204 ),
205 );
206 return yield* HttpServerResponse.json(result);
207 }
208
209 if (request.method === "GET" && pathname === "/replicas") {
210 const result = yield* listParticipantReplicas({
211 participantId: MISSING_PARTICIPANT_ID,
212 }).pipe(
213 Effect.map((r) => ({ count: r.replicas.length })),
214 Effect.catchTag("ValidationException", (e) =>
215 Effect.succeed({ tag: e._tag }),
216 ),
217 );
218 return yield* HttpServerResponse.json(result);
219 }
220
221 if (request.method === "POST" && pathname === "/replication-start") {
222 const result = yield* startParticipantReplication({
223 participantId: MISSING_PARTICIPANT_ID,
224 reconnectWindow: "30 seconds",
225 }).pipe(
226 Effect.map(() => ({ ok: true })),
227 Effect.catchTag(
228 [
229 "ResourceNotFoundException",
230 "ValidationException",
231 "ConflictException",
232 "PendingVerification",
233 "ServiceQuotaExceededException",
234 ],
235 (e) => Effect.succeed({ tag: e._tag }),
236 ),
237 );
238 return yield* HttpServerResponse.json(result);
239 }
240
241 if (request.method === "POST" && pathname === "/replication-stop") {
242 const result = yield* stopParticipantReplication({
243 participantId: MISSING_PARTICIPANT_ID,
244 }).pipe(
245 Effect.map(() => ({ ok: true })),
246 Effect.catchTag(
247 ["ResourceNotFoundException", "ValidationException"],
248 (e) => Effect.succeed({ tag: e._tag }),
249 ),
250 );
251 return yield* HttpServerResponse.json(result);
252 }
253
254 if (request.method === "GET" && pathname === "/compositions") {
255 const { compositions } = yield* listCompositions();
256 return yield* HttpServerResponse.json({
257 count: compositions.length,
258 });
259 }
260
261 // Composition routes take a well-formed-but-nonexistent ARN from the
262 // test (the test knows the region/account) and prove wiring by
263 // observing the API's typed response.
264 if (request.method === "GET" && pathname === "/composition-missing") {
265 const arn = url.searchParams.get("arn")!;
266 const result = yield* getComposition({ arn }).pipe(
267 Effect.map(() => ({ found: true })),
268 Effect.catchTag(
269 ["ResourceNotFoundException", "ValidationException"],
270 (e) => Effect.succeed({ tag: e._tag }),
271 ),
272 );
273 return yield* HttpServerResponse.json(result);
274 }
275
276 if (request.method === "POST" && pathname === "/composition-stop") {
277 const arn = url.searchParams.get("arn")!;
278 const result = yield* stopComposition({ arn }).pipe(
279 Effect.map(() => ({ ok: true })),
280 Effect.catchTag(
281 ["ResourceNotFoundException", "ValidationException"],
282 (e) => Effect.succeed({ tag: e._tag }),
283 ),
284 );
285 return yield* HttpServerResponse.json(result);
286 }
287
288 if (request.method === "POST" && pathname === "/composition-start") {
289 const storageConfigurationArn = url.searchParams.get("arn")!;
290 const result = yield* startComposition({
291 destinations: [
292 {
293 s3: {
294 storageConfigurationArn,
295 encoderConfigurationArns: [],
296 },
297 },
298 ],
299 }).pipe(
300 Effect.map((r) => ({ ok: true, arn: r.composition?.arn })),
301 Effect.catchTag(
302 [
303 "ResourceNotFoundException",
304 "ValidationException",
305 "ConflictException",
306 "PendingVerification",
307 "ServiceQuotaExceededException",
308 ],
309 (e) => Effect.succeed({ tag: e._tag }),
310 ),
311 );
312 return yield* HttpServerResponse.json(result);
313 }
314
315 return yield* HttpServerResponse.json(
316 { error: "Not found", method: request.method, pathname },
317 { status: 404 },
318 );
319 }).pipe(Effect.orDie),
320 };
321 }).pipe(
322 Effect.provide(
323 Layer.mergeAll(
324 Lambda.EventSource,
325 IVSRealtime.CreateParticipantTokenHttp,
326 IVSRealtime.DisconnectParticipantHttp,
327 IVSRealtime.GetParticipantHttp,
328 IVSRealtime.ListParticipantsHttp,
329 IVSRealtime.ListParticipantEventsHttp,
330 IVSRealtime.ListParticipantReplicasHttp,
331 IVSRealtime.GetStageSessionHttp,
332 IVSRealtime.ListStageSessionsHttp,
333 IVSRealtime.StartParticipantReplicationHttp,
334 IVSRealtime.StopParticipantReplicationHttp,
335 IVSRealtime.StartCompositionHttp,
336 IVSRealtime.StopCompositionHttp,
337 IVSRealtime.GetCompositionHttp,
338 IVSRealtime.ListCompositionsHttp,
339 ),
340 ),
341 ),
342);