1import * as ivs from "@distilled.cloud/aws/ivs";
2import * as Data from "effect/Data";
3import * as Effect from "effect/Effect";
4import * as Stream from "effect/Stream";
5import { Unowned } from "../../AdoptPolicy.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 retryWhileConflict,
13 retryWhileThrottled,
14 syncIvsTags,
15 toTagRecord,
16} from "./internal.ts";
17
18export interface ChannelProps {
19 /**
20 * Name of the channel (letters, digits, `-` and `_`; not unique across
21 * channels). If omitted, a deterministic physical name is generated.
22 * Channel names are mutable — changing the name updates the channel in
23 * place.
24 */
25 channelName?: string;
26 /**
27 * Channel latency mode. `LOW` enables low-latency live video (~3s);
28 * `NORMAL` broadcasts at higher latency for a lower cost.
29 * @default "LOW"
30 */
31 latencyMode?: "NORMAL" | "LOW";
32 /**
33 * Channel type, which determines the allowable resolution and bitrate
34 * of the delivered video (e.g. `STANDARD`, `BASIC`, `ADVANCED_SD`,
35 * `ADVANCED_HD`).
36 * @default "STANDARD"
37 */
38 type?: string;
39 /**
40 * Whether the channel is private (viewers require a playback
41 * authorization token generated with a `PlaybackKeyPair`).
42 * @default false
43 */
44 authorized?: boolean;
45 /**
46 * ARN of a recording configuration to record live broadcasts to S3.
47 * @default no recording
48 */
49 recordingConfigurationArn?: string;
50 /**
51 * Whether the channel allows insecure RTMP ingest.
52 * @default false
53 */
54 insecureIngest?: boolean;
55 /**
56 * Optional transcode preset for `ADVANCED_SD`/`ADVANCED_HD` channel
57 * types (`HIGHER_BANDWIDTH_DELIVERY` or `CONSTRAINED_BANDWIDTH_DELIVERY`).
58 */
59 preset?: string;
60 /**
61 * ARN of a playback restriction policy constraining playback by
62 * country and/or origin.
63 */
64 playbackRestrictionPolicyArn?: string;
65 /**
66 * Tags to apply to the channel. Merged with internal Alchemy tags.
67 */
68 tags?: Record<string, string>;
69}
70
71export interface Channel extends Resource<
72 "AWS.IVS.Channel",
73 ChannelProps,
74 {
75 /**
76 * The channel's physical name.
77 */
78 channelName: string;
79 /**
80 * ARN of the channel.
81 */
82 channelArn: string;
83 /**
84 * RTMPS ingest endpoint broadcast software sends video to
85 * (authenticated with a `StreamKey`).
86 */
87 ingestEndpoint: string;
88 /**
89 * Playback URL viewers use to watch the channel's live stream.
90 */
91 playbackUrl: string;
92 /**
93 * Channel type (resolution/bitrate tier) reported by IVS.
94 */
95 type: string | undefined;
96 /**
97 * Latency mode reported by IVS (`LOW` or `NORMAL`).
98 */
99 latencyMode: string | undefined;
100 /**
101 * Whether playback requires a signed authorization token.
102 */
103 authorized: boolean | undefined;
104 },
105 never,
106 Providers
107> {}
108
109/**
110 * An Amazon IVS (Interactive Video Service) channel for low-latency live
111 * video streaming.
112 *
113 * A channel stores configuration for broadcasting live streams: broadcast
114 * software sends video to the channel's `ingestEndpoint` (authenticated
115 * with a `StreamKey`) and viewers watch via the channel's `playbackUrl`.
116 * ### Creating Channels
117 * **Example:** Basic Channel
118 * ```typescript
119 * import * as IVS from "alchemy/AWS/IVS";
120 *
121 * const channel = yield* IVS.Channel("LiveChannel");
122 * ```
123 *
124 * **Example:** Basic Low-Cost Channel
125 * ```typescript
126 * const channel = yield* IVS.Channel("LiveChannel", {
127 * type: "BASIC",
128 * latencyMode: "NORMAL",
129 * });
130 * ```
131 *
132 * ### Private Channels
133 * **Example:** Channel with Playback Authorization
134 * ```typescript
135 * const channel = yield* IVS.Channel("PrivateChannel", {
136 * authorized: true,
137 * });
138 * ```
139 *
140 * ### Streaming
141 * **Example:** Channel with a Stream Key
142 * ```typescript
143 * const channel = yield* IVS.Channel("LiveChannel");
144 * const streamKey = yield* IVS.StreamKey("LiveKey", {
145 * channelArn: channel.channelArn,
146 * });
147 * // broadcast to rtmps://{channel.ingestEndpoint}:443/app/ with streamKey.value
148 * ```
149 *
150 * @resource
151 */
152export const Channel = Resource<Channel>("AWS.IVS.Channel");
153
154/**
155 * Raised when the IVS API returns a channel that is missing its ARN, name,
156 * ingest endpoint, or playback URL.
157 */
158export class IvsChannelIncomplete extends Data.TaggedError(
159 "IvsChannelIncomplete",
160)<{ message: string }> {}
161
162export const ChannelProvider = () =>
163 Provider.effect(
164 Channel,
165 Effect.gen(function* () {
166 const toName = (id: string, props: ChannelProps) =>
167 props.channelName
168 ? Effect.succeed(props.channelName)
169 : createPhysicalName({ id, maxLength: 128 });
170
171 const toAttrs = Effect.fn(function* (channel: ivs.Channel) {
172 if (
173 !channel.arn ||
174 !channel.name ||
175 channel.ingestEndpoint === undefined ||
176 channel.playbackUrl === undefined
177 ) {
178 return yield* Effect.fail(
179 new IvsChannelIncomplete({
180 message:
181 "IVS channel is missing its ARN, name, ingest endpoint, or playback URL",
182 }),
183 );
184 }
185 return {
186 channelName: channel.name,
187 channelArn: channel.arn,
188 ingestEndpoint: channel.ingestEndpoint,
189 playbackUrl: channel.playbackUrl,
190 type: channel.type,
191 latencyMode: channel.latencyMode,
192 authorized: channel.authorized,
193 };
194 });
195
196 const getByArn = Effect.fn(function* (arn: string) {
197 const response = yield* ivs.getChannel({ arn }).pipe(
198 retryWhileThrottled,
199 Effect.catchTag("ResourceNotFoundException", () =>
200 Effect.succeed(undefined),
201 ),
202 );
203 return response?.channel;
204 });
205
206 /**
207 * Channel names are not unique — `filterByName` narrows server-side
208 * and we match exactly, taking the first hit. Used only when the
209 * output ARN cache is unavailable (state-persistence failure).
210 */
211 const findByName = Effect.fn(function* (name: string) {
212 const summaries = yield* ivs.listChannels
213 .pages({ filterByName: name })
214 .pipe(
215 Stream.runCollect,
216 Effect.map((chunk) =>
217 Array.from(chunk).flatMap((page) => page.channels),
218 ),
219 retryWhileThrottled,
220 );
221 const match = summaries.find((s) => s.name === name && s.arn);
222 return match?.arn ? yield* getByArn(match.arn) : undefined;
223 });
224
225 return {
226 stables: ["channelArn"],
227
228 read: Effect.fn(function* ({ id, olds, output }) {
229 const channel = output?.channelArn
230 ? yield* getByArn(output.channelArn)
231 : yield* findByName(yield* toName(id, olds ?? {}));
232 if (channel === undefined) return undefined;
233 const attrs = yield* toAttrs(channel);
234 return (yield* hasAlchemyTags(id, toTagRecord(channel.tags)))
235 ? attrs
236 : Unowned(attrs);
237 }),
238
239 reconcile: Effect.fn(function* ({ id, news, output, session }) {
240 const name = yield* toName(id, news);
241 const internalTags = yield* createInternalTags(id);
242 const desiredTags = { ...internalTags, ...news.tags };
243
244 // 1. Observe — the live channel is authoritative; the output ARN
245 // is only an identifier cache.
246 let observed = output?.channelArn
247 ? yield* getByArn(output.channelArn)
248 : yield* findByName(name);
249
250 // 2. Ensure — create if missing. CreateChannel also provisions
251 // the channel's default stream key, which we leave in place (see
252 // the StreamKey resource).
253 if (observed === undefined) {
254 const created = yield* ivs
255 .createChannel({
256 name,
257 latencyMode: news.latencyMode,
258 type: news.type,
259 authorized: news.authorized,
260 recordingConfigurationArn: news.recordingConfigurationArn,
261 insecureIngest: news.insecureIngest,
262 preset: news.preset,
263 playbackRestrictionPolicyArn: news.playbackRestrictionPolicyArn,
264 tags: desiredTags,
265 })
266 .pipe(retryWhileThrottled);
267 observed = created.channel;
268 }
269 const arn = observed?.arn;
270 if (observed === undefined || arn === undefined) {
271 return yield* Effect.fail(
272 new IvsChannelIncomplete({
273 message: "IVS CreateChannel returned no channel ARN",
274 }),
275 );
276 }
277
278 // 3. Sync — diff observed against desired for each prop the user
279 // specified and apply only the delta. All channel settings
280 // (including the name) are mutable via UpdateChannel.
281 const patch: Partial<ivs.UpdateChannelRequest> = {};
282 if (observed.name !== name) patch.name = name;
283 if (
284 news.latencyMode !== undefined &&
285 observed.latencyMode !== news.latencyMode
286 ) {
287 patch.latencyMode = news.latencyMode;
288 }
289 if (news.type !== undefined && observed.type !== news.type) {
290 patch.type = news.type;
291 }
292 if (
293 news.authorized !== undefined &&
294 observed.authorized !== news.authorized
295 ) {
296 patch.authorized = news.authorized;
297 }
298 if (
299 news.recordingConfigurationArn !== undefined &&
300 observed.recordingConfigurationArn !==
301 news.recordingConfigurationArn
302 ) {
303 patch.recordingConfigurationArn = news.recordingConfigurationArn;
304 }
305 if (
306 news.insecureIngest !== undefined &&
307 observed.insecureIngest !== news.insecureIngest
308 ) {
309 patch.insecureIngest = news.insecureIngest;
310 }
311 if (news.preset !== undefined && observed.preset !== news.preset) {
312 patch.preset = news.preset;
313 }
314 if (
315 news.playbackRestrictionPolicyArn !== undefined &&
316 observed.playbackRestrictionPolicyArn !==
317 news.playbackRestrictionPolicyArn
318 ) {
319 patch.playbackRestrictionPolicyArn =
320 news.playbackRestrictionPolicyArn;
321 }
322 if (Object.keys(patch).length > 0) {
323 yield* ivs
324 .updateChannel({ arn, ...patch })
325 .pipe(retryWhileThrottled, retryWhileConflict);
326 }
327
328 // 3b. Sync tags — diff against OBSERVED cloud tags so adoption
329 // converges.
330 yield* syncIvsTags(arn, desiredTags);
331
332 // 4. Return fresh attributes.
333 const final = yield* getByArn(arn);
334 if (final === undefined) {
335 return yield* Effect.fail(
336 new IvsChannelIncomplete({
337 message: `IVS channel '${arn}' vanished during reconcile`,
338 }),
339 );
340 }
341 yield* session.note(arn);
342 return yield* toAttrs(final);
343 }),
344
345 delete: Effect.fn(function* ({ output }) {
346 // DeleteChannel raises ConflictException while a stream is live
347 // — retry through a bounded window, then tolerate already-gone.
348 yield* ivs.deleteChannel({ arn: output.channelArn }).pipe(
349 retryWhileThrottled,
350 retryWhileConflict,
351 Effect.catchTag("ResourceNotFoundException", () => Effect.void),
352 );
353 }),
354
355 list: () =>
356 ivs.listChannels.pages({}).pipe(
357 Stream.runCollect,
358 Effect.map((chunk) =>
359 Array.from(chunk).flatMap((page) => page.channels),
360 ),
361 Effect.flatMap(
362 Effect.forEach(
363 (summary) =>
364 summary.arn === undefined
365 ? Effect.succeed(undefined)
366 : getByArn(summary.arn).pipe(
367 Effect.flatMap((channel) =>
368 channel === undefined
369 ? Effect.succeed(undefined)
370 : toAttrs(channel),
371 ),
372 ),
373 { concurrency: 5 },
374 ),
375 ),
376 Effect.map((items) => items.filter((item) => item !== undefined)),
377 ),
378 };
379 }),
380 );