1import * as IVS from "@/AWS/IVS";
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 Stream from "effect/Stream";
7import { HttpServerRequest } from "effect/http/HttpServerRequest";
8import * as HttpServerResponse from "effect/http/HttpServerResponse";
9import path from "pathe";
10
11const main = path.resolve(import.meta.dirname, "handler.ts");
12
13export class IvsTestFunction extends Lambda.Function<Lambda.Function>()(
14 "IvsTestFunction",
15) {}
16
17export default IvsTestFunction.make(
18 {
19 main,
20 functionUrl: true,
21 timeout: Duration.seconds(30),
22 },
23 Effect.gen(function* () {
24 // The channel the channel-scoped bindings are bound to. Nothing ever
25 // broadcasts to it, so the data-plane ops observe the typed
26 // ChannelNotBroadcasting behavior.
27 const channel = yield* IVS.Channel("BindingChannel", {
28 type: "BASIC",
29 latencyMode: "NORMAL",
30 tags: { fixture: "ivs-bindings" },
31 });
32
33 // Event source: subscribe the host to IVS stream state-change events.
34 // The deploy proves the EventBridge rule + invoke permission wiring.
35 yield* IVS.consumeStreamEvents(
36 { kinds: ["stream-state-change", "recording-state-change"] },
37 (events) =>
38 Stream.runForEach(events, (event) =>
39 Effect.log(
40 `ivs event: ${event.detail.channel_name} ${event.detail.event_name}`,
41 ),
42 ),
43 );
44
45 // Accessor for the channel's ARN, resolvable inside the runtime fetch
46 // handler (the batch revocation request shape carries explicit
47 // channel ARNs rather than binding to a single channel).
48 const ChannelArn = yield* channel.channelArn;
49
50 const getStream = yield* IVS.GetStream(channel);
51 const getStreamSession = yield* IVS.GetStreamSession(channel);
52 const listStreamSessions = yield* IVS.ListStreamSessions(channel);
53 const putMetadata = yield* IVS.PutMetadata(channel);
54 const stopStream = yield* IVS.StopStream(channel);
55 const revokeViewerSession =
56 yield* IVS.StartViewerSessionRevocation(channel);
57 const insertAdBreak = yield* IVS.InsertAdBreak(channel);
58 const listStreams = yield* IVS.ListStreams();
59 const batchRevokeViewerSessions =
60 yield* IVS.BatchStartViewerSessionRevocation();
61
62 const bound = {
63 getStream,
64 getStreamSession,
65 listStreamSessions,
66 putMetadata,
67 stopStream,
68 revokeViewerSession,
69 insertAdBreak,
70 listStreams,
71 batchRevokeViewerSessions,
72 };
73
74 return {
75 fetch: Effect.gen(function* () {
76 const request = yield* HttpServerRequest;
77 const url = new URL(request.originalUrl);
78 const pathname = url.pathname;
79
80 if (request.method === "GET" && pathname === "/bindings") {
81 return yield* HttpServerResponse.json({
82 bound: Object.keys(bound),
83 });
84 }
85
86 // Live-stream read on an idle channel: the typed
87 // ChannelNotBroadcasting tag is the real observable behavior.
88 if (request.method === "GET" && pathname === "/stream") {
89 const result = yield* getStream().pipe(
90 Effect.map(({ stream }) => ({
91 live: true,
92 viewers: stream?.viewerCount ?? 0,
93 })),
94 Effect.catchTag("ChannelNotBroadcasting", () =>
95 Effect.succeed({ live: false, viewers: 0 }),
96 ),
97 );
98 return yield* HttpServerResponse.json(result);
99 }
100
101 // Latest session on a channel that never broadcast: typed
102 // not-found.
103 if (request.method === "GET" && pathname === "/session") {
104 const result = yield* getStreamSession({}).pipe(
105 Effect.map(({ streamSession }) => ({
106 found: true,
107 streamId: streamSession?.streamId,
108 })),
109 Effect.catchTag("ResourceNotFoundException", () =>
110 Effect.succeed({ found: false, streamId: undefined }),
111 ),
112 );
113 return yield* HttpServerResponse.json(result);
114 }
115
116 // Channel-scoped session history (empty for the idle fixture).
117 if (request.method === "GET" && pathname === "/sessions") {
118 const { streamSessions } = yield* listStreamSessions({
119 maxResults: 10,
120 });
121 return yield* HttpServerResponse.json({
122 count: streamSessions.length,
123 });
124 }
125
126 // Account-level live-stream enumeration.
127 if (request.method === "GET" && pathname === "/streams") {
128 const { streams } = yield* listStreams();
129 return yield* HttpServerResponse.json({
130 count: streams.length,
131 });
132 }
133
134 // Timed-metadata insert requires a live stream — the idle channel
135 // answers with the typed ChannelNotBroadcasting tag.
136 if (request.method === "POST" && pathname === "/metadata") {
137 const result = yield* putMetadata({
138 metadata: JSON.stringify({ hello: "viewers" }),
139 }).pipe(
140 Effect.map(() => ({ inserted: true, tag: undefined })),
141 Effect.catchTag("ChannelNotBroadcasting", (e) =>
142 Effect.succeed({ inserted: false, tag: e._tag }),
143 ),
144 );
145 return yield* HttpServerResponse.json(result);
146 }
147
148 // Stopping an idle channel: typed ChannelNotBroadcasting.
149 if (request.method === "POST" && pathname === "/stop") {
150 const result = yield* stopStream().pipe(
151 Effect.map(() => ({ stopped: true, tag: undefined })),
152 Effect.catchTag(
153 ["ChannelNotBroadcasting", "StreamUnavailable"],
154 (e) => Effect.succeed({ stopped: false, tag: e._tag }),
155 ),
156 );
157 return yield* HttpServerResponse.json(result);
158 }
159
160 // Viewer session revocation is fire-and-forget; report the typed
161 // outcome either way.
162 if (request.method === "POST" && pathname === "/revoke") {
163 const result = yield* revokeViewerSession({
164 viewerId: "alchemy-test-viewer",
165 }).pipe(
166 Effect.map(() => ({
167 ok: true,
168 tag: undefined as string | undefined,
169 })),
170 Effect.catchTag(
171 ["ValidationException", "PendingVerification"],
172 (e) =>
173 Effect.succeed({
174 ok: false,
175 tag: e._tag as string | undefined,
176 }),
177 ),
178 );
179 return yield* HttpServerResponse.json(result);
180 }
181
182 // Server-side ad insertion requires a live stream (and an ad
183 // configuration); report the typed outcome.
184 if (request.method === "POST" && pathname === "/adbreak") {
185 const result = yield* insertAdBreak({ durationSeconds: 30 }).pipe(
186 Effect.map(({ adBreakId }) => ({
187 inserted: true,
188 adBreakId,
189 tag: undefined as string | undefined,
190 })),
191 Effect.catchTag(
192 [
193 "ChannelNotBroadcasting",
194 "ValidationException",
195 "ConflictException",
196 ],
197 (e) =>
198 Effect.succeed({
199 inserted: false,
200 adBreakId: undefined,
201 tag: e._tag as string | undefined,
202 }),
203 ),
204 );
205 return yield* HttpServerResponse.json(result);
206 }
207
208 // Batch revocation across channels — per-pair failures surface in
209 // the response's `errors` array rather than failing the call.
210 if (request.method === "POST" && pathname === "/revoke-batch") {
211 const channelArn = yield* ChannelArn;
212 const { errors } = yield* batchRevokeViewerSessions({
213 viewerSessions: [{ channelArn, viewerId: "alchemy-test-viewer" }],
214 });
215 return yield* HttpServerResponse.json({
216 errorCount: (errors ?? []).length,
217 });
218 }
219
220 return yield* HttpServerResponse.json(
221 { error: "Not found", method: request.method, pathname },
222 { status: 404 },
223 );
224 }).pipe(Effect.orDie),
225 };
226 }).pipe(
227 Effect.provide(
228 Layer.mergeAll(
229 Lambda.EventSource,
230 IVS.GetStreamHttp,
231 IVS.GetStreamSessionHttp,
232 IVS.ListStreamSessionsHttp,
233 IVS.PutMetadataHttp,
234 IVS.StopStreamHttp,
235 IVS.StartViewerSessionRevocationHttp,
236 IVS.InsertAdBreakHttp,
237 IVS.ListStreamsHttp,
238 IVS.BatchStartViewerSessionRevocationHttp,
239 ),
240 ),
241 ),
242);