1import * as medialive from "@distilled.cloud/aws/medialive";
2import * as Data from "effect/Data";
3import * as Effect from "effect/Effect";
4import * as Schedule from "effect/Schedule";
5import * as Stream from "effect/Stream";
6import { Unowned } from "../../AdoptPolicy.ts";
7import { isResolved } from "../../Diff.ts";
8import { createPhysicalName } from "../../PhysicalName.ts";
9import * as Provider from "../../Provider.ts";
10import { Resource } from "../../Resource.ts";
11import { createInternalTags, hasAlchemyTags } from "../../Tags.ts";
12import type { Providers } from "../Providers.ts";
13import {
14 MediaLiveResourcePending,
15 ensureIdentified,
16 ensurePresent,
17 retryWhileConflict,
18 retryWhilePending,
19 syncMlTags,
20 toTagRecord,
21} from "./internal.ts";
22
23/**
24 * The observed shape shared by `DescribeChannelResponse` and the `Channel`
25 * struct returned by create/update (and `ChannelSummary` from list).
26 */
27type ObservedChannel = Pick<
28 medialive.Channel,
29 | "Arn"
30 | "Id"
31 | "Name"
32 | "State"
33 | "ChannelClass"
34 | "EgressEndpoints"
35 | "LogLevel"
36 | "RoleArn"
37 | "Tags"
38>;
39
40type IdentifiedChannel = ObservedChannel & { Id: string; Arn: string };
41
42// Explicitly-typed pipeable retry helper (see internal.ts for why inline
43// Effect.retry is forbidden in provider ops). A freshly-created IAM role is
44// not immediately visible to MediaLive — createChannel rejects it with a
45// message-discriminated UnprocessableEntityException naming the trust
46// relationship (patched in distilled to the typed `MediaLiveRoleNotYetTrusted`
47// tag) until IAM propagation settles.
48const retryWhileRolePropagating = <A, E extends { readonly _tag: string }, R>(
49 self: Effect.Effect<A, E, R>,
50): Effect.Effect<A, E, R> =>
51 Effect.retry(self, {
52 while: (e) => e._tag === "MediaLiveRoleNotYetTrusted",
53 schedule: Schedule.max([Schedule.fixed("3 seconds"), Schedule.recurs(10)]),
54 });
55
56/**
57 * Raised when a MediaLive channel lands in `CREATE_FAILED` or `UPDATE_FAILED`
58 * instead of settling into `IDLE`.
59 */
60export class MediaLiveChannelFailed extends Data.TaggedError(
61 "MediaLiveChannelFailed",
62)<{ message: string }> {}
63
64export interface ChannelProps {
65 /**
66 * Name of the channel. If omitted, a unique name is generated from the
67 * app, stage, and logical ID. Names are mutable — changing the name
68 * updates the channel in place.
69 */
70 name?: string;
71 /**
72 * The class of the channel. `STANDARD` runs two redundant encoder
73 * pipelines; `SINGLE_PIPELINE` runs one. Changing the class replaces the
74 * channel.
75 * @default "STANDARD"
76 */
77 channelClass?: medialive.ChannelClass;
78 /**
79 * ARN of the IAM role MediaLive assumes to read inputs and write outputs
80 * (trusts `medialive.amazonaws.com`).
81 */
82 roleArn?: string;
83 /**
84 * The inputs attached to this channel.
85 */
86 inputAttachments?: medialive.InputAttachment[];
87 /**
88 * The encoder settings — audio/video descriptions, output groups, and
89 * timecode configuration.
90 */
91 encoderSettings?: medialive.EncoderSettings;
92 /**
93 * Destinations referenced by the output groups in `encoderSettings`.
94 */
95 destinations?: medialive.OutputDestination[];
96 /**
97 * Specification of the input codec/resolution/bitrate tier (drives
98 * per-hour pricing).
99 */
100 inputSpecification?: medialive.InputSpecification;
101 /**
102 * Specification of CDI inputs for this channel.
103 */
104 cdiInputSpecification?: medialive.CdiInputSpecification;
105 /**
106 * The CloudWatch log level for the channel.
107 * @default "DISABLED"
108 */
109 logLevel?: medialive.LogLevel;
110 /**
111 * Maintenance window settings (day + start hour).
112 */
113 maintenance?: medialive.MaintenanceCreateSettings;
114 /**
115 * VPC output settings. Changing the VPC settings replaces the channel.
116 */
117 vpc?: medialive.VpcOutputSettings;
118 /**
119 * User-defined tags for the channel.
120 */
121 tags?: Record<string, string>;
122}
123
124export interface Channel extends Resource<
125 "AWS.MediaLive.Channel",
126 ChannelProps,
127 {
128 /** Server-assigned unique id of the channel. */
129 channelId: string;
130 /** ARN of the channel. */
131 channelArn: string;
132 /** Name of the channel. */
133 channelName: string | undefined;
134 /** Current lifecycle state (e.g. `IDLE`, `RUNNING`). */
135 state: medialive.ChannelState | undefined;
136 /** Pipeline class (`STANDARD` or `SINGLE_PIPELINE`). */
137 channelClass: medialive.ChannelClass | undefined;
138 /** Egress endpoints the channel writes output through. */
139 egressEndpoints: medialive.ChannelEgressEndpoint[];
140 },
141 never,
142 Providers
143> {}
144
145/**
146 * An AWS Elemental MediaLive channel — the live encoder that reads from
147 * attached inputs, transcodes per its encoder settings, and writes to output
148 * destinations (HLS, RTMP, MediaPackage, ...).
149 *
150 * Channels bill per running hour; Alchemy provisions channels in the `IDLE`
151 * state and never starts them — start/stop is a runtime operation.
152 *
153 * ### Creating a Channel
154 * **Example:** Single-pipeline HLS channel
155 * ```typescript
156 * const channel = yield* MediaLive.Channel("Live", {
157 * channelClass: "SINGLE_PIPELINE",
158 * roleArn: role.roleArn,
159 * inputAttachments: [
160 * { InputId: input.inputId, InputAttachmentName: "primary" },
161 * ],
162 * inputSpecification: {
163 * Codec: "AVC",
164 * Resolution: "SD",
165 * MaximumBitrate: "MAX_10_MBPS",
166 * },
167 * destinations,
168 * encoderSettings,
169 * });
170 * ```
171 *
172 * **Example:** IAM role for MediaLive
173 * ```typescript
174 * const role = yield* IAM.Role("MediaLiveRole", {
175 * assumeRolePolicyDocument: {
176 * Version: "2012-10-17",
177 * Statement: [
178 * {
179 * Effect: "Allow",
180 * Principal: { Service: "medialive.amazonaws.com" },
181 * Action: ["sts:AssumeRole"],
182 * },
183 * ],
184 * },
185 * });
186 * ```
187 *
188 * @resource
189 */
190export const Channel = Resource<Channel>("AWS.MediaLive.Channel");
191
192export const ChannelProvider = () =>
193 Provider.effect(
194 Channel,
195 Effect.gen(function* () {
196 const createName = Effect.fn(function* (
197 id: string,
198 props: { name?: string | undefined },
199 ) {
200 return props.name ?? (yield* createPhysicalName({ id, maxLength: 64 }));
201 });
202
203 const toAttrs = (channel: IdentifiedChannel) => ({
204 channelId: channel.Id,
205 channelArn: channel.Arn,
206 channelName: channel.Name,
207 state: channel.State,
208 channelClass: channel.ChannelClass,
209 egressEndpoints: [...(channel.EgressEndpoints ?? [])],
210 });
211
212 const isGone = (state: medialive.ChannelState | undefined) =>
213 state === "DELETED" || state === "DELETING";
214
215 /** Describe by id; typed not-found (or tombstone state) → undefined. */
216 const getChannel = Effect.fn(function* (channelId: string) {
217 const channel = yield* medialive
218 .describeChannel({ ChannelId: channelId })
219 .pipe(
220 Effect.catchTag("NotFoundException", () =>
221 Effect.succeed(undefined),
222 ),
223 );
224 if (channel === undefined || isGone(channel.State)) return undefined;
225 return yield* ensureIdentified(channel, "DescribeChannel Id/Arn");
226 });
227
228 /**
229 * Channels have server-assigned ids, so a read without cached output
230 * searches the account list by the deterministic physical name.
231 */
232 const findByName = Effect.fn(function* (name: string) {
233 const channels = yield* medialive.listChannels.items({}).pipe(
234 Stream.runCollect,
235 Effect.map((chunk) => Array.from(chunk)),
236 );
237 const match = channels.find(
238 (channel) => channel.Name === name && !isGone(channel.State),
239 );
240 if (match === undefined) return undefined;
241 return yield* ensureIdentified(match, "ListChannels item Id/Arn");
242 });
243
244 /**
245 * Wait (bounded, ~90s) for a channel to settle out of the transient
246 * `CREATING`/`UPDATING` states; fail typed on `*_FAILED`.
247 */
248 const awaitSettled = Effect.fn(function* (channelId: string) {
249 return yield* retryWhilePending(
250 Effect.gen(function* () {
251 const channel = yield* medialive.describeChannel({
252 ChannelId: channelId,
253 });
254 if (
255 channel.State === "CREATE_FAILED" ||
256 channel.State === "UPDATE_FAILED"
257 ) {
258 return yield* Effect.fail(
259 new MediaLiveChannelFailed({
260 message: `channel ${channelId} is in state ${channel.State}`,
261 }),
262 );
263 }
264 if (channel.State === "CREATING" || channel.State === "UPDATING") {
265 return yield* Effect.fail(
266 new MediaLiveResourcePending({
267 message: `channel ${channelId} is still ${channel.State}`,
268 }),
269 );
270 }
271 return yield* ensureIdentified(channel, "DescribeChannel Id/Arn");
272 }),
273 );
274 });
275
276 return Channel.Provider.of({
277 stables: ["channelId", "channelArn"],
278
279 list: () =>
280 medialive.listChannels.items({}).pipe(
281 Stream.runCollect,
282 Effect.map((chunk) =>
283 Array.from(chunk)
284 .filter(
285 (
286 channel,
287 ): channel is medialive.ChannelSummary & IdentifiedChannel =>
288 channel.Id !== undefined &&
289 channel.Arn !== undefined &&
290 !isGone(channel.State),
291 )
292 .map(toAttrs),
293 ),
294 ),
295
296 read: Effect.fn(function* ({ id, olds, output }) {
297 const channel =
298 output?.channelId !== undefined
299 ? yield* getChannel(output.channelId)
300 : yield* findByName(yield* createName(id, olds ?? {}));
301 if (channel === undefined) return undefined;
302 const attrs = toAttrs(channel);
303 return (yield* hasAlchemyTags(id, toTagRecord(channel.Tags)))
304 ? attrs
305 : Unowned(attrs);
306 }),
307
308 diff: Effect.fn(function* ({ news, olds }) {
309 if (!isResolved(news)) return undefined;
310 // The channel class and VPC placement are fixed at creation.
311 if (
312 (olds.channelClass ?? "STANDARD") !==
313 (news.channelClass ?? "STANDARD")
314 ) {
315 return { action: "replace" } as const;
316 }
317 if (JSON.stringify(olds.vpc) !== JSON.stringify(news.vpc)) {
318 return { action: "replace" } as const;
319 }
320 }),
321
322 reconcile: Effect.fn(function* ({ id, news, olds, output, session }) {
323 const internalTags = yield* createInternalTags(id);
324 const desiredTags = { ...internalTags, ...news.tags };
325 const name = yield* createName(id, news);
326
327 // 1. Observe — cloud state is authoritative; output is an id cache.
328 // MediaLive channel names are not unique, so a name match is only
329 // trusted when its immutable class agrees with the desired state —
330 // during a replacement the same-named survivor is the doomed old
331 // instance, not this one.
332 let channel: IdentifiedChannel | undefined;
333 if (output?.channelId !== undefined) {
334 channel = yield* getChannel(output.channelId);
335 } else {
336 const found = yield* findByName(name);
337 channel =
338 found !== undefined &&
339 (found.ChannelClass ?? "STANDARD") ===
340 (news.channelClass ?? "STANDARD")
341 ? found
342 : undefined;
343 }
344
345 // 2. Ensure — create if missing, then wait for CREATING to settle.
346 if (channel === undefined) {
347 const created = yield* medialive
348 .createChannel({
349 Name: name,
350 ChannelClass: news.channelClass,
351 RoleArn: news.roleArn,
352 InputAttachments: news.inputAttachments,
353 EncoderSettings: news.encoderSettings,
354 Destinations: news.destinations,
355 InputSpecification: news.inputSpecification,
356 CdiInputSpecification: news.cdiInputSpecification,
357 LogLevel: news.logLevel,
358 Maintenance: news.maintenance,
359 Vpc: news.vpc,
360 Tags: desiredTags,
361 })
362 .pipe(retryWhileRolePropagating);
363 const fresh = yield* ensureIdentified(
364 created.Channel,
365 "CreateChannel Channel Id/Arn",
366 );
367 channel = yield* awaitSettled(fresh.Id);
368 } else {
369 // 3. Sync — cheap scalars are diffed against OBSERVED state; the
370 // deep encoder/attachment documents are diffed against `olds` as
371 // a no-op hint (Describe echoes them back default-expanded, so an
372 // observed-vs-desired deep compare would always report drift).
373 const nameDrift = channel.Name !== name;
374 const logDrift =
375 news.logLevel !== undefined && channel.LogLevel !== news.logLevel;
376 const roleDrift =
377 news.roleArn !== undefined && channel.RoleArn !== news.roleArn;
378 const deepDrift =
379 olds === undefined
380 ? true
381 : JSON.stringify({
382 a: olds.inputAttachments,
383 e: olds.encoderSettings,
384 d: olds.destinations,
385 i: olds.inputSpecification,
386 c: olds.cdiInputSpecification,
387 }) !==
388 JSON.stringify({
389 a: news.inputAttachments,
390 e: news.encoderSettings,
391 d: news.destinations,
392 i: news.inputSpecification,
393 c: news.cdiInputSpecification,
394 });
395 if (nameDrift || logDrift || roleDrift || deepDrift) {
396 yield* medialive
397 .updateChannel({
398 ChannelId: channel.Id,
399 Name: name,
400 RoleArn: news.roleArn,
401 InputAttachments: news.inputAttachments,
402 EncoderSettings: news.encoderSettings,
403 Destinations: news.destinations,
404 InputSpecification: news.inputSpecification,
405 CdiInputSpecification: news.cdiInputSpecification,
406 LogLevel: news.logLevel,
407 Maintenance: news.maintenance
408 ? {
409 MaintenanceDay: news.maintenance.MaintenanceDay,
410 MaintenanceStartTime:
411 news.maintenance.MaintenanceStartTime,
412 }
413 : undefined,
414 })
415 .pipe(retryWhileConflict);
416 channel = yield* awaitSettled(channel.Id);
417 }
418 }
419
420 // Both the ensure and sync branches leave `channel` assigned from a
421 // typed describe; guard with a typed failure so TS sees it defined.
422 const settled = yield* ensurePresent(
423 channel,
424 "reconciled channel state",
425 );
426
427 // 3b. Sync tags — diff against OBSERVED cloud tags.
428 yield* syncMlTags(settled.Arn, desiredTags);
429
430 yield* session.note(settled.Id);
431 return toAttrs(settled);
432 }),
433
434 delete: Effect.fn(function* ({ output }) {
435 yield* medialive.deleteChannel({ ChannelId: output.channelId }).pipe(
436 retryWhileConflict,
437 Effect.catchTag("NotFoundException", () => Effect.void),
438 Effect.asVoid,
439 );
440 // Deletion is async (DELETING → DELETED). Wait (bounded) so that
441 // dependent Inputs/InputSecurityGroups can delete without long
442 // conflict-retry loops; if the channel is still draining after the
443 // window, deletion is initiated and irreversible — let downstream
444 // retries absorb the tail.
445 yield* medialive
446 .describeChannel({ ChannelId: output.channelId })
447 .pipe(
448 Effect.flatMap((channel) =>
449 channel.State === "DELETED"
450 ? Effect.void
451 : Effect.fail(
452 new MediaLiveResourcePending({
453 message: `channel ${output.channelId} is still ${channel.State}`,
454 }),
455 ),
456 ),
457 retryWhilePending,
458 Effect.catchTag("NotFoundException", () => Effect.void),
459 Effect.catchTag("MediaLiveResourcePending", () => Effect.void),
460 );
461 }),
462 });
463 }),
464 );