1import * as mediaconnect from "@distilled.cloud/aws/mediaconnect";
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, diffTags, hasAlchemyTags } from "../../Tags.ts";
12import type { Providers } from "../Providers.ts";
13
14/**
15 * Every flow output managed by alchemy must carry a `Name` so the reconciler
16 * can converge outputs by identity across updates.
17 */
18export class FlowOutputNameMissing extends Data.TaggedError(
19 "FlowOutputNameMissing",
20)<{ message: string }> {}
21
22export interface FlowProps {
23 /**
24 * Name of the flow. If omitted, a deterministic physical name is generated
25 * from the app, stage and logical ID. Changing the name replaces the flow
26 * (MediaConnect has no rename operation).
27 */
28 flowName?: string;
29 /**
30 * Availability Zone the flow is created in. Must be within the current
31 * region. Changing the Availability Zone replaces the flow.
32 * @default MediaConnect picks an Availability Zone
33 */
34 availabilityZone?: string;
35 /**
36 * The source the flow ingests. Raw distilled `SetSourceRequest` shape —
37 * e.g. `{ Protocol: "rtp", WhitelistCidr: "10.0.0.0/8", IngestPort: 5000 }`.
38 * Mutable source settings (protocol, whitelist CIDR, ingest port, bitrate,
39 * latency, stream id, sender address/port, description) are updated in
40 * place via UpdateFlowSource.
41 */
42 source: mediaconnect.SetSourceRequest;
43 /**
44 * Outputs of the flow (up to 50). Each output MUST have a `Name` — the
45 * reconciler converges outputs by name: missing outputs are added, extra
46 * non-entitlement outputs are removed, and changed settings (destination,
47 * port, protocol, CIDR allow list, latency, description) are updated in
48 * place.
49 * @default no outputs
50 */
51 outputs?: mediaconnect.AddOutputRequest[];
52 /**
53 * User-defined tags for the flow.
54 */
55 tags?: Record<string, string>;
56}
57
58export interface Flow extends Resource<
59 "AWS.MediaConnect.Flow",
60 FlowProps,
61 {
62 /** Name of the flow. */
63 flowName: string;
64 /** ARN of the flow. */
65 flowArn: string;
66 /** Current status — a freshly created flow is `STANDBY` (not billing for transport). */
67 status: string;
68 /** Availability Zone the flow runs in. */
69 availabilityZone: string;
70 /** Flow description, if any. */
71 description: string | undefined;
72 /** IP address the flow egresses media from. */
73 egressIp: string | undefined;
74 /** ARN of the flow's source. */
75 sourceArn: string | undefined;
76 /** IP address the flow listens on for incoming content. */
77 sourceIngestIp: string | undefined;
78 /** Port the flow listens on for incoming content. */
79 sourceIngestPort: number | undefined;
80 /** The flow's outputs (name, ARN, destination endpoint). */
81 outputs: {
82 name: string;
83 outputArn: string;
84 destination: string | undefined;
85 port: number | undefined;
86 listenerAddress: string | undefined;
87 }[];
88 /** Observed tags on the flow. */
89 tags: Record<string, string>;
90 },
91 never,
92 Providers
93> {}
94
95/**
96 * An AWS Elemental MediaConnect flow — a reliable live-video transport
97 * between a source and one or more outputs.
98 *
99 * Creating a flow leaves it in `STANDBY`; a flow only ingests/egresses media
100 * (and bills for transport) once started with the StartFlow API. Flows bill
101 * hourly while ACTIVE, so alchemy never starts a flow implicitly.
102 * ### Creating a Flow
103 * **Example:** RTP Flow with a CIDR-Whitelisted Source
104 * ```typescript
105 * const flow = yield* Flow("Broadcast", {
106 * source: {
107 * Name: "primary",
108 * Protocol: "rtp",
109 * WhitelistCidr: "10.24.34.0/23",
110 * IngestPort: 5000,
111 * },
112 * });
113 * ```
114 *
115 * ### Outputs
116 * **Example:** Flow with an RTP Output
117 * ```typescript
118 * const flow = yield* Flow("Distribution", {
119 * source: {
120 * Name: "primary",
121 * Protocol: "rtp",
122 * WhitelistCidr: "10.24.34.0/23",
123 * IngestPort: 5000,
124 * },
125 * outputs: [
126 * {
127 * Name: "affiliate-east",
128 * Protocol: "rtp",
129 * Destination: "198.51.100.11",
130 * Port: 5010,
131 * },
132 * ],
133 * });
134 * ```
135 *
136 * ### Tags
137 * **Example:** Tagged Flow
138 * ```typescript
139 * const flow = yield* Flow("Broadcast", {
140 * source: { Protocol: "rtp", WhitelistCidr: "10.0.0.0/8", IngestPort: 5000 },
141 * tags: { team: "live-video" },
142 * });
143 * ```
144 *
145 * @resource
146 */
147export const Flow = Resource<Flow>("AWS.MediaConnect.Flow");
148
149type DescribedFlow = NonNullable<mediaconnect.DescribeFlowResponse["Flow"]>;
150// The element type of a described flow's Outputs, INCLUDING the response
151// refinements (required Name/OutputArn). Note: DescribedFlow["Outputs"] is an
152// intersection of array types, so `.map` resolves its callback against the
153// base `Output[]` signature and drops these refinements — iterate with
154// `for..of` (which uses the number-index element type) instead of `.map`.
155type ObservedOutput = DescribedFlow["Outputs"][number];
156
157const sameStringList = (a: string[] | undefined, b: string[] | undefined) => {
158 const l = [...(a ?? [])].sort();
159 const r = [...(b ?? [])].sort();
160 return l.length === r.length && l.every((v, i) => v === r[i]);
161};
162
163export const FlowProvider = () =>
164 Provider.effect(
165 Flow,
166 Effect.gen(function* () {
167 const toName = (id: string, props: { flowName?: string }) =>
168 props.flowName
169 ? Effect.succeed(props.flowName)
170 : createPhysicalName({ id, maxLength: 60 });
171
172 const readFlow = Effect.fn(function* (flowArn: string) {
173 const response = yield* mediaconnect
174 .describeFlow({ FlowArn: flowArn })
175 .pipe(
176 Effect.catchTag("NotFoundException", () =>
177 Effect.succeed(undefined),
178 ),
179 );
180 return response?.Flow;
181 });
182
183 // Flow names are not unique in MediaConnect — identity is the ARN.
184 // When the ARN cache is lost (state persistence failure) fall back to
185 // the first flow bearing our deterministic name.
186 const findFlowArnByName = Effect.fn(function* (name: string) {
187 const flows = yield* mediaconnect.listFlows
188 .items({})
189 .pipe(Stream.runCollect);
190 for (const flow of flows) {
191 if (flow.Name === name && flow.FlowArn !== undefined) {
192 return flow.FlowArn;
193 }
194 }
195 return undefined;
196 });
197
198 const readFlowTags = Effect.fn(function* (flowArn: string) {
199 const response = yield* mediaconnect
200 .listTagsForResource({ ResourceArn: flowArn })
201 .pipe(
202 Effect.catchTag("NotFoundException", () =>
203 Effect.succeed(undefined),
204 ),
205 );
206 const tags: Record<string, string> = {};
207 for (const [key, value] of Object.entries(response?.Tags ?? {})) {
208 if (value !== undefined) tags[key] = value;
209 }
210 return tags;
211 });
212
213 // Flow creation and in-place updates are async; wait (bounded, ~3 min)
214 // for the flow to settle into a steady state before mutating further.
215 const waitUntilSettled = Effect.fn(function* (flowArn: string) {
216 const policy = Schedule.max([
217 Schedule.fixed("5 seconds"),
218 Schedule.recurs(36),
219 ]);
220 return yield* readFlow(flowArn).pipe(
221 Effect.flatMap((flow) => {
222 if (flow === undefined) {
223 return Effect.fail(
224 new Error(`MediaConnect flow '${flowArn}' not found`),
225 );
226 }
227 if (
228 flow.Status !== "STANDBY" &&
229 flow.Status !== "ACTIVE" &&
230 flow.Status !== "ERROR"
231 ) {
232 return Effect.fail(
233 new Error(
234 `MediaConnect flow '${flowArn}' still settling (status: ${flow.Status})`,
235 ),
236 );
237 }
238 return Effect.succeed(flow);
239 }),
240 Effect.retry({ schedule: policy }),
241 );
242 });
243
244 // Deletion is verified as fully gone (bounded, ~3 min) so dependents
245 // (and re-creates of the same name) never race a half-deleted flow.
246 const waitUntilGone = Effect.fn(function* (flowArn: string) {
247 const policy = Schedule.max([
248 Schedule.fixed("5 seconds"),
249 Schedule.recurs(36),
250 ]);
251 yield* readFlow(flowArn).pipe(
252 Effect.flatMap((flow) =>
253 flow === undefined
254 ? Effect.void
255 : Effect.fail(
256 new Error(
257 `MediaConnect flow '${flowArn}' still exists (status: ${flow.Status})`,
258 ),
259 ),
260 ),
261 Effect.retry({ schedule: policy }),
262 );
263 });
264
265 const toAttrs = Effect.fn(function* (flow: DescribedFlow) {
266 const outputs: {
267 name: string;
268 outputArn: string;
269 destination: string | undefined;
270 port: number | undefined;
271 listenerAddress: string | undefined;
272 }[] = [];
273 for (const output of flow.Outputs ?? []) {
274 outputs.push({
275 name: output.Name,
276 outputArn: output.OutputArn,
277 destination: output.Destination,
278 port: output.Port,
279 listenerAddress: output.ListenerAddress,
280 });
281 }
282 return {
283 flowName: flow.Name,
284 flowArn: flow.FlowArn,
285 status: flow.Status,
286 availabilityZone: flow.AvailabilityZone,
287 description: flow.Description,
288 egressIp: flow.EgressIp,
289 sourceArn: flow.Source?.SourceArn,
290 sourceIngestIp: flow.Source?.IngestIp,
291 sourceIngestPort: flow.Source?.IngestPort,
292 outputs,
293 tags: yield* readFlowTags(flow.FlowArn),
294 };
295 });
296
297 return {
298 stables: ["flowName", "flowArn", "availabilityZone"],
299
300 diff: Effect.fn(function* ({ id, olds, news }) {
301 if (!isResolved(news)) return undefined;
302 const oldName = yield* toName(id, { flowName: olds?.flowName });
303 const newName = yield* toName(id, { flowName: news?.flowName });
304 if (oldName !== newName) {
305 return { action: "replace" } as const;
306 }
307 // The Availability Zone is create-only.
308 if (
309 news?.availabilityZone !== undefined &&
310 olds?.availabilityZone !== undefined &&
311 news.availabilityZone !== olds.availabilityZone
312 ) {
313 return { action: "replace" } as const;
314 }
315 }),
316
317 read: Effect.fn(function* ({ id, olds, output }) {
318 const name = output?.flowName ?? (yield* toName(id, olds ?? {}));
319 const arn = output?.flowArn ?? (yield* findFlowArnByName(name));
320 if (arn === undefined) return undefined;
321 const flow = yield* readFlow(arn);
322 if (flow === undefined) return undefined;
323 const attrs = yield* toAttrs(flow);
324 return (yield* hasAlchemyTags(id, attrs.tags))
325 ? attrs
326 : Unowned(attrs);
327 }),
328
329 reconcile: Effect.fn(function* ({ id, news, output, session }) {
330 const props = news!;
331 const name = output?.flowName ?? (yield* toName(id, props));
332 const internalTags = yield* createInternalTags(id);
333 const desiredTags = { ...internalTags, ...props.tags };
334
335 // Validate up front (before any AWS call): outputs are converged
336 // by name, so every desired output must carry one.
337 const desiredOutputs: (mediaconnect.AddOutputRequest & {
338 Name: string;
339 })[] = [];
340 for (const desired of props.outputs ?? []) {
341 if (desired.Name === undefined) {
342 return yield* Effect.fail(
343 new FlowOutputNameMissing({
344 message: `every output of MediaConnect flow '${name}' must set a Name — outputs are converged by name`,
345 }),
346 );
347 }
348 desiredOutputs.push({ ...desired, Name: desired.Name });
349 }
350
351 // 1. Observe — cloud state is authoritative; output is only an
352 // ARN cache (flow names are not unique, so fall back to search).
353 let arn = output?.flowArn ?? (yield* findFlowArnByName(name));
354 let observed = arn !== undefined ? yield* readFlow(arn) : undefined;
355
356 // 2. Ensure — create if missing. Creation leaves the flow in
357 // STANDBY; alchemy never starts a flow (ACTIVE bills hourly).
358 if (observed === undefined) {
359 const created = yield* mediaconnect.createFlow({
360 Name: name,
361 AvailabilityZone: props.availabilityZone,
362 Source: props.source,
363 Outputs: desiredOutputs.length > 0 ? desiredOutputs : undefined,
364 });
365 arn = created.Flow?.FlowArn;
366 }
367 if (arn === undefined) {
368 return yield* Effect.fail(
369 new Error(`MediaConnect flow '${name}' is missing its ARN`),
370 );
371 }
372 observed = yield* waitUntilSettled(arn);
373
374 // 3. Sync source — diff OBSERVED source settings against desired
375 // and apply only the delta via UpdateFlowSource.
376 const source = observed.Source;
377 if (source?.SourceArn !== undefined) {
378 const desired = props.source;
379 const transport = source.Transport;
380 const update: Omit<
381 mediaconnect.UpdateFlowSourceRequest,
382 "FlowArn" | "SourceArn"
383 > = {};
384 let mutated = false;
385 if (
386 desired.Description !== undefined &&
387 desired.Description !== source.Description
388 ) {
389 update.Description = desired.Description;
390 mutated = true;
391 }
392 if (
393 desired.Protocol !== undefined &&
394 desired.Protocol !== transport?.Protocol
395 ) {
396 update.Protocol = desired.Protocol;
397 mutated = true;
398 }
399 if (
400 desired.WhitelistCidr !== undefined &&
401 desired.WhitelistCidr !== source.WhitelistCidr
402 ) {
403 update.WhitelistCidr = desired.WhitelistCidr;
404 mutated = true;
405 }
406 if (
407 desired.IngestPort !== undefined &&
408 desired.IngestPort !== source.IngestPort
409 ) {
410 update.IngestPort = desired.IngestPort;
411 mutated = true;
412 }
413 if (
414 desired.MaxBitrate !== undefined &&
415 desired.MaxBitrate !== transport?.MaxBitrate
416 ) {
417 update.MaxBitrate = desired.MaxBitrate;
418 mutated = true;
419 }
420 if (
421 desired.MaxLatency !== undefined &&
422 desired.MaxLatency !== transport?.MaxLatency
423 ) {
424 update.MaxLatency = desired.MaxLatency;
425 mutated = true;
426 }
427 if (
428 desired.MinLatency !== undefined &&
429 desired.MinLatency !== transport?.MinLatency
430 ) {
431 update.MinLatency = desired.MinLatency;
432 mutated = true;
433 }
434 if (
435 desired.StreamId !== undefined &&
436 desired.StreamId !== transport?.StreamId
437 ) {
438 update.StreamId = desired.StreamId;
439 mutated = true;
440 }
441 if (
442 desired.SenderIpAddress !== undefined &&
443 desired.SenderIpAddress !== source.SenderIpAddress
444 ) {
445 update.SenderIpAddress = desired.SenderIpAddress;
446 mutated = true;
447 }
448 if (
449 desired.SenderControlPort !== undefined &&
450 desired.SenderControlPort !== source.SenderControlPort
451 ) {
452 update.SenderControlPort = desired.SenderControlPort;
453 mutated = true;
454 }
455 if (mutated) {
456 yield* mediaconnect.updateFlowSource({
457 ...update,
458 FlowArn: arn,
459 SourceArn: source.SourceArn,
460 });
461 observed = yield* waitUntilSettled(arn);
462 }
463 }
464
465 // 3b. Sync outputs by name — add missing, update drifted, remove
466 // extras (entitlement-generated outputs are not ours to remove).
467 const observedOutputs = observed.Outputs ?? [];
468 const observedByName = new Map<string, ObservedOutput>();
469 for (const observedOutput of observedOutputs) {
470 observedByName.set(observedOutput.Name, observedOutput);
471 }
472 const desiredNames = new Set(desiredOutputs.map((o) => o.Name));
473 let outputsMutated = false;
474
475 const toAdd = desiredOutputs.filter(
476 (o) => !observedByName.has(o.Name),
477 );
478 if (toAdd.length > 0) {
479 yield* mediaconnect.addFlowOutputs({
480 FlowArn: arn,
481 Outputs: toAdd,
482 });
483 outputsMutated = true;
484 }
485
486 for (const existing of observedOutputs) {
487 if (
488 !desiredNames.has(existing.Name) &&
489 existing.EntitlementArn === undefined
490 ) {
491 yield* mediaconnect
492 .removeFlowOutput({
493 FlowArn: arn,
494 OutputArn: existing.OutputArn,
495 })
496 .pipe(Effect.catchTag("NotFoundException", () => Effect.void));
497 outputsMutated = true;
498 }
499 }
500
501 for (const desired of desiredOutputs) {
502 const existing = observedByName.get(desired.Name);
503 if (existing === undefined) continue;
504 const transport = existing.Transport;
505 const update: Omit<
506 mediaconnect.UpdateFlowOutputRequest,
507 "FlowArn" | "OutputArn"
508 > = {};
509 let mutated = false;
510 if (
511 desired.Description !== undefined &&
512 desired.Description !== existing.Description
513 ) {
514 update.Description = desired.Description;
515 mutated = true;
516 }
517 if (
518 desired.Destination !== undefined &&
519 desired.Destination !== existing.Destination
520 ) {
521 update.Destination = desired.Destination;
522 mutated = true;
523 }
524 if (desired.Port !== undefined && desired.Port !== existing.Port) {
525 update.Port = desired.Port;
526 mutated = true;
527 }
528 if (
529 desired.Protocol !== undefined &&
530 desired.Protocol !== transport?.Protocol
531 ) {
532 update.Protocol = desired.Protocol;
533 mutated = true;
534 }
535 if (
536 desired.CidrAllowList !== undefined &&
537 !sameStringList(desired.CidrAllowList, transport?.CidrAllowList)
538 ) {
539 update.CidrAllowList = desired.CidrAllowList;
540 mutated = true;
541 }
542 if (
543 desired.MaxLatency !== undefined &&
544 desired.MaxLatency !== transport?.MaxLatency
545 ) {
546 update.MaxLatency = desired.MaxLatency;
547 mutated = true;
548 }
549 if (
550 desired.MinLatency !== undefined &&
551 desired.MinLatency !== transport?.MinLatency
552 ) {
553 update.MinLatency = desired.MinLatency;
554 mutated = true;
555 }
556 if (
557 desired.SmoothingLatency !== undefined &&
558 desired.SmoothingLatency !== transport?.SmoothingLatency
559 ) {
560 update.SmoothingLatency = desired.SmoothingLatency;
561 mutated = true;
562 }
563 if (
564 desired.StreamId !== undefined &&
565 desired.StreamId !== transport?.StreamId
566 ) {
567 update.StreamId = desired.StreamId;
568 mutated = true;
569 }
570 if (
571 desired.RemoteId !== undefined &&
572 desired.RemoteId !== transport?.RemoteId
573 ) {
574 update.RemoteId = desired.RemoteId;
575 mutated = true;
576 }
577 if (mutated) {
578 yield* mediaconnect.updateFlowOutput({
579 ...update,
580 FlowArn: arn,
581 OutputArn: existing.OutputArn,
582 });
583 outputsMutated = true;
584 }
585 }
586 if (outputsMutated) {
587 observed = yield* waitUntilSettled(arn);
588 }
589
590 // 3c. Sync tags — diff against OBSERVED cloud tags.
591 const observedTags = yield* readFlowTags(arn);
592 const { removed, upsert } = diffTags(observedTags, desiredTags);
593 if (upsert.length > 0) {
594 yield* mediaconnect.tagResource({
595 ResourceArn: arn,
596 Tags: Object.fromEntries(upsert.map((t) => [t.Key, t.Value])),
597 });
598 }
599 if (removed.length > 0) {
600 yield* mediaconnect.untagResource({
601 ResourceArn: arn,
602 TagKeys: removed,
603 });
604 }
605
606 yield* session.note(name);
607 return yield* toAttrs(observed);
608 }),
609
610 delete: Effect.fn(function* ({ output }) {
611 const arn = output.flowArn;
612 const observed = yield* readFlow(arn);
613 if (observed === undefined) return;
614 // A flow must be stopped before it can be deleted. Alchemy never
615 // starts flows, but the flow may have been started out-of-band.
616 if (observed.Status === "ACTIVE" || observed.Status === "STARTING") {
617 yield* mediaconnect.stopFlow({ FlowArn: arn }).pipe(
618 // A STARTING flow rejects StopFlow until the start completes.
619 Effect.retry({
620 while: (e) => e._tag === "BadRequestException",
621 schedule: Schedule.max([
622 Schedule.fixed("5 seconds"),
623 Schedule.recurs(12),
624 ]),
625 }),
626 Effect.catchTag("NotFoundException", () => Effect.void),
627 );
628 }
629 // Wait for any in-flight transition (STOPPING/UPDATING) to settle,
630 // then delete. A flow already DELETING (or gone) is success.
631 const settled = yield* waitUntilSettled(arn).pipe(
632 Effect.catch(() => Effect.succeed(undefined)),
633 );
634 if (settled !== undefined) {
635 yield* mediaconnect.deleteFlow({ FlowArn: arn }).pipe(
636 // Status races (e.g. a transition that began after our read)
637 // surface as BadRequestException — bounded retry through it.
638 Effect.retry({
639 while: (e) => e._tag === "BadRequestException",
640 schedule: Schedule.max([
641 Schedule.fixed("5 seconds"),
642 Schedule.recurs(12),
643 ]),
644 }),
645 Effect.catchTag("NotFoundException", () => Effect.void),
646 );
647 }
648 yield* waitUntilGone(arn);
649 }),
650
651 list: () =>
652 Effect.gen(function* () {
653 const listed = yield* mediaconnect.listFlows
654 .items({})
655 .pipe(Stream.runCollect);
656 const arns: string[] = [];
657 for (const flow of listed) {
658 if (flow.FlowArn !== undefined) arns.push(flow.FlowArn);
659 }
660 const flows = yield* Effect.forEach(arns, (arn) => readFlow(arn), {
661 concurrency: 4,
662 });
663 return yield* Effect.forEach(
664 flows.filter((flow): flow is DescribedFlow => flow !== undefined),
665 (flow) => toAttrs(flow),
666 { concurrency: 4 },
667 );
668 }),
669 };
670 }),
671 );