1import * as medialive from "@distilled.cloud/aws/medialive";
2import * as Effect from "effect/Effect";
3import * as Stream from "effect/Stream";
4import { Unowned } from "../../AdoptPolicy.ts";
5import { isResolved } from "../../Diff.ts";
6import { createPhysicalName } from "../../PhysicalName.ts";
7import * as Provider from "../../Provider.ts";
8import { Resource } from "../../Resource.ts";
9import { createInternalTags, hasAlchemyTags } from "../../Tags.ts";
10import type { Providers } from "../Providers.ts";
11import {
12 MediaLiveResourcePending,
13 ensureIdentified,
14 retryWhileConflict,
15 retryWhilePending,
16 syncMlTags,
17 toTagRecord,
18} from "./internal.ts";
19
20/**
21 * The observed shape shared by `DescribeInputResponse` and the `Input`
22 * struct returned by create/update/list.
23 */
24type ObservedInput = Pick<
25 medialive.Input,
26 | "Arn"
27 | "Id"
28 | "Name"
29 | "State"
30 | "Type"
31 | "InputClass"
32 | "Destinations"
33 | "SecurityGroups"
34 | "Sources"
35 | "RoleArn"
36 | "MediaConnectFlows"
37 | "Tags"
38>;
39
40type IdentifiedInput = ObservedInput & { Id: string; Arn: string };
41
42export interface InputProps {
43 /**
44 * Name of the input. If omitted, a unique name is generated from the app,
45 * stage, and logical ID. Names are mutable — changing the name updates the
46 * input in place.
47 */
48 name?: string;
49 /**
50 * The input type (e.g. `RTMP_PUSH`, `URL_PULL`, `MEDIACONNECT`,
51 * `MP4_FILE`). Changing the type replaces the input.
52 */
53 type: medialive.InputType;
54 /**
55 * Destination settings for PUSH-type inputs — e.g. the application/stream
56 * name pair for `RTMP_PUSH` (`{ StreamName: "live/stream" }`).
57 */
58 destinations?: medialive.InputDestinationRequest[];
59 /**
60 * Source URLs for PULL-type inputs (`URL_PULL`, `RTMP_PULL`, `MP4_FILE`,
61 * `TS_FILE`), with optional basic-auth credentials.
62 */
63 sources?: medialive.InputSourceRequest[];
64 /**
65 * IDs of input security groups to attach. Required for PUSH-type inputs
66 * that ingest over the public internet.
67 */
68 inputSecurityGroups?: string[];
69 /**
70 * MediaConnect flows to use as the input source (for `MEDIACONNECT`
71 * inputs).
72 */
73 mediaConnectFlows?: medialive.MediaConnectFlowRequest[];
74 /**
75 * ARN of the IAM role MediaLive assumes to access the input (required for
76 * VPC and MediaConnect inputs).
77 */
78 roleArn?: string;
79 /**
80 * VPC settings for a VPC push input. Changing the VPC settings replaces
81 * the input.
82 */
83 vpc?: medialive.InputVpcRequest;
84 /**
85 * User-defined tags for the input.
86 */
87 tags?: Record<string, string>;
88}
89
90export interface Input extends Resource<
91 "AWS.MediaLive.Input",
92 InputProps,
93 {
94 /** Server-assigned unique id of the input. */
95 inputId: string;
96 /** ARN of the input. */
97 inputArn: string;
98 /** Name of the input. */
99 inputName: string | undefined;
100 /** Current lifecycle state (e.g. `DETACHED`, `ATTACHED`). */
101 state: medialive.InputState | undefined;
102 /** The input type (e.g. `RTMP_PUSH`, `URL_PULL`). */
103 type: medialive.InputType | undefined;
104 /** `STANDARD` (two ingest endpoints) or `SINGLE_PIPELINE`. */
105 inputClass: medialive.InputClass | undefined;
106 /** Resolved ingest destinations (push URLs) for the input. */
107 destinations: medialive.InputDestination[];
108 /** IDs of the attached input security groups. */
109 securityGroups: string[];
110 },
111 never,
112 Providers
113> {}
114
115/**
116 * An AWS Elemental MediaLive input — the ingest endpoint or source locator a
117 * channel reads live content from (RTMP/RTP push, HLS/MP4 pull,
118 * MediaConnect flow, ...).
119 *
120 * ### Creating Inputs
121 * **Example:** RTMP push input behind an allowlist
122 * ```typescript
123 * const isg = yield* MediaLive.InputSecurityGroup("Allowlist", {
124 * whitelistRules: ["0.0.0.0/0"],
125 * });
126 * const input = yield* MediaLive.Input("Stream", {
127 * type: "RTMP_PUSH",
128 * inputSecurityGroups: [isg.inputSecurityGroupId],
129 * destinations: [{ StreamName: "live/stream" }],
130 * });
131 * ```
132 *
133 * **Example:** HLS pull input
134 * ```typescript
135 * const input = yield* MediaLive.Input("Vod", {
136 * type: "URL_PULL",
137 * sources: [{ Url: "https://example.com/stream/index.m3u8" }],
138 * });
139 * ```
140 *
141 * ### Attaching to a Channel
142 * **Example:** Feed a channel
143 * ```typescript
144 * const channel = yield* MediaLive.Channel("Live", {
145 * roleArn: role.roleArn,
146 * inputAttachments: [
147 * { InputId: input.inputId, InputAttachmentName: "primary" },
148 * ],
149 * encoderSettings,
150 * destinations,
151 * });
152 * ```
153 *
154 * @resource
155 */
156export const Input = Resource<Input>("AWS.MediaLive.Input");
157
158export const InputProvider = () =>
159 Provider.effect(
160 Input,
161 Effect.gen(function* () {
162 const createName = Effect.fn(function* (
163 id: string,
164 props: { name?: string | undefined },
165 ) {
166 return props.name ?? (yield* createPhysicalName({ id, maxLength: 64 }));
167 });
168
169 const toAttrs = (input: IdentifiedInput) => ({
170 inputId: input.Id,
171 inputArn: input.Arn,
172 inputName: input.Name,
173 state: input.State,
174 type: input.Type,
175 inputClass: input.InputClass,
176 destinations: [...(input.Destinations ?? [])],
177 securityGroups: [...(input.SecurityGroups ?? [])],
178 });
179
180 const isGone = (state: medialive.InputState | undefined) =>
181 state === "DELETED" || state === "DELETING";
182
183 /** Describe by id; typed not-found (or tombstone state) → undefined. */
184 const getInput = Effect.fn(function* (inputId: string) {
185 const input = yield* medialive
186 .describeInput({ InputId: inputId })
187 .pipe(
188 Effect.catchTag("NotFoundException", () =>
189 Effect.succeed(undefined),
190 ),
191 );
192 if (input === undefined || isGone(input.State)) return undefined;
193 return yield* ensureIdentified(input, "DescribeInput Id/Arn");
194 });
195
196 /**
197 * Inputs have server-assigned ids, so a read without cached output
198 * searches the account list by the deterministic physical name.
199 */
200 const findByName = Effect.fn(function* (name: string) {
201 const inputs = yield* medialive.listInputs.items({}).pipe(
202 Stream.runCollect,
203 Effect.map((chunk) => Array.from(chunk)),
204 );
205 const match = inputs.find(
206 (input) => input.Name === name && !isGone(input.State),
207 );
208 if (match === undefined) return undefined;
209 return yield* ensureIdentified(match, "ListInputs item Id/Arn");
210 });
211
212 /** Wait (bounded) for a fresh input to leave `CREATING`. */
213 const awaitActive = Effect.fn(function* (inputId: string) {
214 return yield* medialive.describeInput({ InputId: inputId }).pipe(
215 Effect.flatMap((input) =>
216 input.State === "CREATING"
217 ? Effect.fail(
218 new MediaLiveResourcePending({
219 message: `input ${inputId} is still CREATING`,
220 }),
221 )
222 : Effect.succeed(input),
223 ),
224 retryWhilePending,
225 Effect.flatMap((input) =>
226 ensureIdentified(input, "DescribeInput Id/Arn"),
227 ),
228 );
229 });
230
231 return Input.Provider.of({
232 stables: ["inputId", "inputArn"],
233
234 list: () =>
235 medialive.listInputs.items({}).pipe(
236 Stream.runCollect,
237 Effect.map((chunk) =>
238 Array.from(chunk)
239 .filter(
240 (input): input is medialive.Input & IdentifiedInput =>
241 input.Id !== undefined &&
242 input.Arn !== undefined &&
243 !isGone(input.State),
244 )
245 .map(toAttrs),
246 ),
247 ),
248
249 read: Effect.fn(function* ({ id, olds, output }) {
250 const input =
251 output?.inputId !== undefined
252 ? yield* getInput(output.inputId)
253 : yield* findByName(yield* createName(id, olds ?? {}));
254 if (input === undefined) return undefined;
255 const attrs = toAttrs(input);
256 return (yield* hasAlchemyTags(id, toTagRecord(input.Tags)))
257 ? attrs
258 : Unowned(attrs);
259 }),
260
261 diff: Effect.fn(function* ({ news, olds }) {
262 if (!isResolved(news)) return undefined;
263 // The input type is immutable.
264 if (olds.type !== news.type) return { action: "replace" } as const;
265 // VPC placement is fixed at creation.
266 if (JSON.stringify(olds.vpc) !== JSON.stringify(news.vpc)) {
267 return { action: "replace" } as const;
268 }
269 }),
270
271 reconcile: Effect.fn(function* ({ id, news, olds, output, session }) {
272 const internalTags = yield* createInternalTags(id);
273 const desiredTags = { ...internalTags, ...news.tags };
274 const name = yield* createName(id, news);
275
276 // 1. Observe — cloud state is authoritative; output is an id cache.
277 // MediaLive input names are not unique, so a name match is only
278 // trusted when its immutable type agrees with the desired state —
279 // during a replacement the same-named survivor is the doomed old
280 // instance, not this one.
281 let input: IdentifiedInput | undefined;
282 if (output?.inputId !== undefined) {
283 input = yield* getInput(output.inputId);
284 } else {
285 const found = yield* findByName(name);
286 input =
287 found !== undefined && found.Type === news.type
288 ? found
289 : undefined;
290 }
291
292 // 2. Ensure — create if missing, then wait for CREATING to settle.
293 if (input === undefined) {
294 const created = yield* medialive.createInput({
295 Name: name,
296 Type: news.type,
297 Destinations: news.destinations,
298 Sources: news.sources,
299 InputSecurityGroups: news.inputSecurityGroups,
300 MediaConnectFlows: news.mediaConnectFlows,
301 RoleArn: news.roleArn,
302 Vpc: news.vpc,
303 Tags: desiredTags,
304 });
305 const fresh = yield* ensureIdentified(
306 created.Input,
307 "CreateInput Input Id/Arn",
308 );
309 input = yield* awaitActive(fresh.Id);
310 } else {
311 // 3. Sync — apply the in-place-updatable aspects only on drift.
312 // Destination stream names are not echoed back by Describe (the
313 // API returns resolved URLs), so `olds` is the no-op hint there.
314 const sameStringSet = (
315 a: readonly (string | undefined)[],
316 b: readonly (string | undefined)[],
317 ) =>
318 JSON.stringify([...a].sort()) === JSON.stringify([...b].sort());
319 const nameDrift = input.Name !== name;
320 const roleDrift =
321 news.roleArn !== undefined && input.RoleArn !== news.roleArn;
322 const sgDrift = !sameStringSet(
323 news.inputSecurityGroups ?? [],
324 input.SecurityGroups ?? [],
325 );
326 const sourceDrift = !sameStringSet(
327 (news.sources ?? []).map((s) => s.Url),
328 (input.Sources ?? []).map((s) => s.Url),
329 );
330 const flowDrift = !sameStringSet(
331 (news.mediaConnectFlows ?? []).map((f) => f.FlowArn),
332 (input.MediaConnectFlows ?? []).map((f) => f.FlowArn),
333 );
334 const destinationDrift =
335 olds === undefined
336 ? news.destinations !== undefined
337 : JSON.stringify(olds.destinations) !==
338 JSON.stringify(news.destinations);
339 if (
340 nameDrift ||
341 roleDrift ||
342 sgDrift ||
343 sourceDrift ||
344 flowDrift ||
345 destinationDrift
346 ) {
347 yield* medialive
348 .updateInput({
349 InputId: input.Id,
350 Name: name,
351 Destinations: news.destinations,
352 Sources: news.sources,
353 InputSecurityGroups: news.inputSecurityGroups,
354 MediaConnectFlows: news.mediaConnectFlows,
355 RoleArn: news.roleArn,
356 })
357 .pipe(retryWhileConflict);
358 input = yield* awaitActive(input.Id);
359 }
360 }
361
362 // 3b. Sync tags — diff against OBSERVED cloud tags.
363 yield* syncMlTags(input.Arn, desiredTags);
364
365 yield* session.note(input.Id);
366 return toAttrs(input);
367 }),
368
369 delete: Effect.fn(function* ({ output }) {
370 // An input attached to a still-deleting channel transiently
371 // rejects deletion; deletion itself is async (DELETING → DELETED),
372 // which `read` treats as gone.
373 yield* medialive.deleteInput({ InputId: output.inputId }).pipe(
374 retryWhileConflict,
375 Effect.catchTag("NotFoundException", () => Effect.void),
376 );
377 }),
378 });
379 }),
380 );