1import * as Lambda from "@/AWS/Lambda";
2import * as MediaConnect from "@/AWS/MediaConnect";
3import * as Duration from "effect/Duration";
4import * as Effect from "effect/Effect";
5import * as Layer from "effect/Layer";
6import * as Schedule from "effect/Schedule";
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
14export class MediaConnectTestFunction extends Lambda.Function<Lambda.Function>()(
15 "MediaConnectTestFunction",
16) {}
17
18export default MediaConnectTestFunction.make(
19 {
20 main,
21 functionUrl: true,
22 // The gated /start-stop roundtrip waits for the flow to accept StopFlow.
23 timeout: Duration.seconds(120),
24 },
25 Effect.gen(function* () {
26 // The flow the flow-scoped bindings are bound to. It is created in
27 // STANDBY (which does not bill for transport) and nothing ever sends
28 // media to it, so the observability ops see the typed
29 // not-running/no-content behavior.
30 const flow = yield* MediaConnect.Flow("BindingFlow", {
31 source: {
32 Name: "primary",
33 Protocol: "rtp",
34 WhitelistCidr: "10.24.34.0/23",
35 IngestPort: 5000,
36 },
37 tags: { fixture: "mediaconnect-bindings" },
38 });
39
40 // Event source: subscribe the host to MediaConnect alert/status events.
41 // The deploy proves the EventBridge rule + invoke permission wiring.
42 yield* MediaConnect.consumeFlowEvents(
43 { kinds: ["alert", "flow-status-change"] },
44 (events) =>
45 Stream.runForEach(events, (event) =>
46 Effect.log(
47 `mediaconnect event: ${event["detail-type"]} ${event.detail.message ?? event.detail.status ?? ""}`,
48 ),
49 ),
50 );
51
52 // Accessor for the flow's ARN, resolvable inside the runtime fetch
53 // handler (the entitlement roundtrip derives the account id from it).
54 const FlowArn = yield* flow.flowArn;
55
56 const describeFlow = yield* MediaConnect.DescribeFlow(flow);
57 const sourceMetadata = yield* MediaConnect.DescribeFlowSourceMetadata(flow);
58 const sourceThumbnail =
59 yield* MediaConnect.DescribeFlowSourceThumbnail(flow);
60 const startFlow = yield* MediaConnect.StartFlow(flow);
61 const stopFlow = yield* MediaConnect.StopFlow(flow);
62 const grantEntitlements = yield* MediaConnect.GrantFlowEntitlements(flow);
63 const revokeEntitlement = yield* MediaConnect.RevokeFlowEntitlement(flow);
64 const listFlows = yield* MediaConnect.ListFlows();
65 const listEntitlements = yield* MediaConnect.ListEntitlements();
66
67 const bound = {
68 describeFlow,
69 sourceMetadata,
70 sourceThumbnail,
71 startFlow,
72 stopFlow,
73 grantEntitlements,
74 revokeEntitlement,
75 listFlows,
76 listEntitlements,
77 };
78
79 return {
80 fetch: Effect.gen(function* () {
81 const request = yield* HttpServerRequest;
82 const url = new URL(request.originalUrl);
83 const pathname = url.pathname;
84
85 if (request.method === "GET" && pathname === "/bindings") {
86 return yield* HttpServerResponse.json({
87 bound: Object.keys(bound),
88 });
89 }
90
91 // Read the flow's live state — the fixture flow idles in STANDBY.
92 if (request.method === "GET" && pathname === "/flow") {
93 const { Flow: current } = yield* describeFlow();
94 return yield* HttpServerResponse.json({
95 status: current?.Status,
96 sourceName: current?.Source?.Name,
97 });
98 }
99
100 // Transport-stream metadata on a flow that is not receiving
101 // content: MediaConnect answers with status messages (or a typed
102 // BadRequestException while the flow is not running).
103 if (request.method === "GET" && pathname === "/source-metadata") {
104 const result = yield* sourceMetadata().pipe(
105 Effect.map(({ TransportMediaInfo, Messages }) => ({
106 programs: TransportMediaInfo?.Programs?.length ?? 0,
107 messages: (Messages ?? []).length,
108 tag: undefined as string | undefined,
109 })),
110 Effect.catchTag("BadRequestException", (e) =>
111 Effect.succeed({ programs: 0, messages: 0, tag: e._tag }),
112 ),
113 );
114 return yield* HttpServerResponse.json(result);
115 }
116
117 // Source thumbnail on an idle flow: no image, only messages (or a
118 // typed BadRequestException while the flow is not running).
119 if (request.method === "GET" && pathname === "/thumbnail") {
120 const result = yield* sourceThumbnail().pipe(
121 Effect.map(({ ThumbnailDetails }) => ({
122 hasImage: ThumbnailDetails?.Thumbnail !== undefined,
123 messages: ThumbnailDetails?.ThumbnailMessages?.length ?? 0,
124 tag: undefined as string | undefined,
125 })),
126 Effect.catchTag("BadRequestException", (e) =>
127 Effect.succeed({ hasImage: false, messages: 0, tag: e._tag }),
128 ),
129 );
130 return yield* HttpServerResponse.json(result);
131 }
132
133 // Account-level enumerations.
134 if (request.method === "GET" && pathname === "/flows") {
135 const { Flows } = yield* listFlows({ MaxResults: 20 });
136 return yield* HttpServerResponse.json({
137 count: (Flows ?? []).length,
138 });
139 }
140 if (request.method === "GET" && pathname === "/entitlements") {
141 const { Entitlements } = yield* listEntitlements();
142 return yield* HttpServerResponse.json({
143 count: (Entitlements ?? []).length,
144 });
145 }
146
147 // Stopping a flow that is already in STANDBY: the typed
148 // BadRequestException tag is the real observable behavior.
149 if (request.method === "POST" && pathname === "/stop") {
150 const result = yield* stopFlow().pipe(
151 Effect.map(({ Status }) => ({
152 status: Status,
153 tag: undefined as string | undefined,
154 })),
155 Effect.catchTag("BadRequestException", (e) =>
156 Effect.succeed({
157 status: undefined as string | undefined,
158 tag: e._tag,
159 }),
160 ),
161 );
162 return yield* HttpServerResponse.json(result);
163 }
164
165 // Grant an entitlement to this very account, then revoke it —
166 // a full grant/revoke roundtrip with no cross-account dependency.
167 // Failures surface as their typed tag so the test can diagnose
168 // (e.g. an IAM gap is a visible ForbiddenException, not a 500).
169 if (request.method === "POST" && pathname === "/entitlement-cycle") {
170 const flowArn = yield* FlowArn;
171 const accountId = flowArn.split(":")[4]!;
172 const result = yield* grantEntitlements({
173 Entitlements: [
174 {
175 Name: "alchemy-binding-test",
176 Subscribers: [accountId],
177 },
178 ],
179 }).pipe(
180 Effect.flatMap(({ Entitlements }) => {
181 const entitlementArn = Entitlements?.[0]?.EntitlementArn;
182 if (entitlementArn === undefined) {
183 return Effect.succeed({
184 granted: false,
185 revoked: false,
186 tag: undefined as string | undefined,
187 });
188 }
189 return revokeEntitlement({
190 EntitlementArn: entitlementArn,
191 }).pipe(
192 Effect.map(() => ({
193 granted: true,
194 revoked: true,
195 tag: undefined as string | undefined,
196 })),
197 );
198 }),
199 Effect.catchTag(
200 [
201 "BadRequestException",
202 "GrantFlowEntitlements420Exception",
203 "ForbiddenException",
204 "NotFoundException",
205 ],
206 (e) =>
207 Effect.succeed({
208 granted: false,
209 revoked: false,
210 tag: e._tag as string | undefined,
211 }),
212 ),
213 );
214 return yield* HttpServerResponse.json(result);
215 }
216
217 // Start the flow, then stop it as soon as MediaConnect accepts a
218 // StopFlow (a STARTING flow rejects it with BadRequestException).
219 // Only driven by the AWS_TEST_MEDIACONNECT-gated test: an ACTIVE
220 // flow bills for transport by the hour.
221 if (request.method === "POST" && pathname === "/start-stop") {
222 const { Status: startStatus } = yield* startFlow();
223 const { Status: stopStatus } = yield* stopFlow().pipe(
224 Effect.retry({
225 while: (e): boolean => e._tag === "BadRequestException",
226 schedule: Schedule.max([
227 Schedule.fixed("5 seconds"),
228 Schedule.recurs(18),
229 ]),
230 }),
231 );
232 return yield* HttpServerResponse.json({
233 startStatus,
234 stopStatus,
235 });
236 }
237
238 return yield* HttpServerResponse.json(
239 { error: "Not found", method: request.method, pathname },
240 { status: 404 },
241 );
242 }).pipe(Effect.orDie),
243 };
244 }).pipe(
245 Effect.provide(
246 Layer.mergeAll(
247 Lambda.EventSource,
248 MediaConnect.DescribeFlowHttp,
249 MediaConnect.DescribeFlowSourceMetadataHttp,
250 MediaConnect.DescribeFlowSourceThumbnailHttp,
251 MediaConnect.StartFlowHttp,
252 MediaConnect.StopFlowHttp,
253 MediaConnect.GrantFlowEntitlementsHttp,
254 MediaConnect.RevokeFlowEntitlementHttp,
255 MediaConnect.ListFlowsHttp,
256 MediaConnect.ListEntitlementsHttp,
257 ),
258 ),
259 ),
260);