1import * as stream from "@distilled.cloud/cloudflare/stream";
2import * as Effect from "effect/Effect";
3import * as Predicate from "effect/Predicate";
4import * as Stream from "effect/Stream";
5
6import { isResolved } from "../../Diff.ts";
7import * as Provider from "../../Provider.ts";
8import { Resource } from "../../Resource.ts";
9import { CloudflareEnvironment } from "../CloudflareEnvironment.ts";
10import type { Providers } from "../Providers.ts";
11
12const TypeId = "Cloudflare.Stream.LiveInputOutput" as const;
13type TypeId = typeof TypeId;
14
15export type LiveInputOutputProps = {
16 /**
17 * The unique identifier (`uid`) of the live input the output restreams
18 * from. Usually a reference to a `LiveInput`'s `liveInputId`
19 * attribute.
20 *
21 * Immutable — an output belongs to exactly one live input, so changing
22 * it triggers a replacement.
23 */
24 liveInputId: string;
25 /**
26 * The URL the output restreams to, e.g.
27 * `rtmps://a.rtmps.youtube.com/live2`.
28 *
29 * Immutable — Cloudflare's update endpoint only toggles `enabled`, so
30 * changing the URL triggers a replacement.
31 */
32 url: string;
33 /**
34 * The streamKey used to authenticate against the output's target.
35 *
36 * Immutable — Cloudflare's update endpoint only toggles `enabled`, so
37 * changing the streamKey triggers a replacement.
38 */
39 streamKey: string;
40 /**
41 * When enabled, live video streamed to the associated live input is
42 * sent to the output URL. When disabled, live video is not sent to the
43 * output URL, even while streaming to the associated live input.
44 * Mutable.
45 * @default true
46 */
47 enabled?: boolean;
48};
49
50export type LiveInputOutputAttributes = {
51 /**
52 * The unique identifier for the output (Cloudflare `uid`).
53 */
54 outputId: string;
55 /**
56 * The unique identifier of the live input the output belongs to.
57 */
58 liveInputId: string;
59 /**
60 * The Cloudflare account the output belongs to.
61 */
62 accountId: string;
63 /**
64 * The URL the output restreams to.
65 */
66 url: string;
67 /**
68 * The streamKey used to authenticate against the output's target.
69 */
70 streamKey: string;
71 /**
72 * Whether live video is sent to the output URL while streaming to the
73 * associated live input.
74 */
75 enabled: boolean;
76};
77
78export type LiveInputOutput = Resource<
79 TypeId,
80 LiveInputOutputProps,
81 LiveInputOutputAttributes,
82 never,
83 Providers
84>;
85
86/**
87 * A Cloudflare Stream live input output — restreams (simulcasts) live
88 * video received by a `LiveInput` to another RTMP(S) destination
89 * such as YouTube Live or Twitch.
90 *
91 * The destination (`url` + `streamKey`) is immutable: Cloudflare's update
92 * endpoint only toggles `enabled`, so changing the destination replaces
93 * the output. Toggling `enabled` updates the output in place.
94 *
95 * Requires the Stream subscription to be enabled on the account.
96 * ### Creating an output
97 * **Example:** Restream a live input to YouTube
98 * ```typescript
99 * const input = yield* Cloudflare.Stream.LiveInput("Broadcast", {});
100 *
101 * const youtube = yield* Cloudflare.Stream.LiveInputOutput("YouTube", {
102 * liveInputId: input.liveInputId,
103 * url: "rtmps://a.rtmps.youtube.com/live2",
104 * streamKey: youtubeStreamKey,
105 * });
106 * ```
107 *
108 * ### Managing an output
109 * **Example:** Pause restreaming without deleting the output
110 * ```typescript
111 * const youtube = yield* Cloudflare.Stream.LiveInputOutput("YouTube", {
112 * liveInputId: input.liveInputId,
113 * url: "rtmps://a.rtmps.youtube.com/live2",
114 * streamKey: youtubeStreamKey,
115 * enabled: false,
116 * });
117 * ```
118 *
119 * @see https://developers.cloudflare.com/stream/stream-live/simulcasting/
120 *
121 * @resource
122 * @product Stream
123 * @category Media
124 */
125export const LiveInputOutput = Resource<LiveInputOutput>(TypeId);
126
127/**
128 * Returns true if the given value is a LiveInputOutput resource.
129 */
130export const isLiveInputOutput = (value: unknown): value is LiveInputOutput =>
131 Predicate.hasProperty(value, "Type") && value.Type === TypeId;
132
133export const LiveInputOutputProvider = () =>
134 Provider.succeed(LiveInputOutput, {
135 stables: ["outputId", "liveInputId", "accountId", "url", "streamKey"],
136
137 diff: Effect.fn(function* ({ olds, news, output }) {
138 const { accountId } = yield* yield* CloudflareEnvironment;
139 if (!isResolved(news)) return undefined;
140 if (output !== undefined && output.accountId !== accountId) {
141 return { action: "replace" } as const;
142 }
143 // An output belongs to exactly one live input.
144 const oldLiveInput = output?.liveInputId ?? olds?.liveInputId;
145 if (
146 typeof oldLiveInput === "string" &&
147 typeof news.liveInputId === "string" &&
148 oldLiveInput !== news.liveInputId
149 ) {
150 return { action: "replace" } as const;
151 }
152 // The destination is immutable — only `enabled` can be updated.
153 const oldUrl = output?.url ?? olds?.url;
154 if (oldUrl !== undefined && oldUrl !== news.url) {
155 return { action: "replace" } as const;
156 }
157 const oldStreamKey = output?.streamKey ?? olds?.streamKey;
158 if (oldStreamKey !== undefined && oldStreamKey !== news.streamKey) {
159 return { action: "replace" } as const;
160 }
161 // `enabled` is mutable via PUT — default engine update logic.
162 return undefined;
163 }),
164
165 read: Effect.fn(function* ({ output }) {
166 // Outputs have no deterministic name or tags, so a cold read (state
167 // lost before the uid was persisted) cannot recover the resource —
168 // rely on the cached uid.
169 if (output?.outputId === undefined) return undefined;
170 const observed = yield* findOutput(
171 output.accountId,
172 output.liveInputId,
173 output.outputId,
174 );
175 return observed
176 ? toAttributes(observed, output.accountId, output.liveInputId)
177 : undefined;
178 }),
179
180 reconcile: Effect.fn(function* ({ news, output }) {
181 const { accountId } = yield* yield* CloudflareEnvironment;
182 // Inputs have been resolved to concrete strings by Plan.
183 const liveInputId = news.liveInputId as string;
184
185 // Observe — the uid cached on `output` is a hint, not a guarantee:
186 // a missing output (or live input) falls through to "missing" and
187 // we recreate.
188 const observed = output?.outputId
189 ? yield* findOutput(
190 output.accountId ?? accountId,
191 liveInputId,
192 output.outputId,
193 )
194 : undefined;
195
196 if (!observed) {
197 // Ensure — greenfield (or out-of-band delete): create with the
198 // full desired body. uids are auto-assigned so there is no
199 // AlreadyExists race to tolerate.
200 const created = yield* stream.createLiveInputOutput({
201 accountId,
202 liveInputIdentifier: liveInputId,
203 url: news.url,
204 streamKey: news.streamKey,
205 enabled: news.enabled,
206 });
207 return toAttributes(created, accountId, liveInputId);
208 }
209
210 // Sync — `enabled` is the only mutable aspect; diff observed cloud
211 // state against desired and skip the PUT on a no-op.
212 const observedAccount = output?.accountId ?? accountId;
213 const desiredEnabled = news.enabled ?? true;
214 if (desiredEnabled === (observed.enabled ?? true)) {
215 return toAttributes(observed, observedAccount, liveInputId);
216 }
217
218 const updated = yield* stream.updateLiveInputOutput({
219 accountId: observedAccount,
220 liveInputIdentifier: liveInputId,
221 outputIdentifier: observed.uid ?? output!.outputId,
222 enabled: desiredEnabled,
223 });
224 return toAttributes(updated, observedAccount, liveInputId);
225 }),
226
227 list: Effect.fn(function* () {
228 const { accountId } = yield* yield* CloudflareEnvironment;
229 // Parent fan-out: outputs are sub-resources of a live input and
230 // have no account-wide enumeration endpoint. Enumerate every live
231 // input on the account, then list each input's outputs.
232 // Cloudflare returns this list either wrapped (`{ liveInputs: [...] }`)
233 // or as a bare `result` array depending on the account — handle both.
234 const inputs = yield* stream.listLiveInputs({ accountId });
235 const liveInputIds = (
236 Array.isArray(inputs) ? inputs : (inputs.liveInputs ?? [])
237 )
238 .map((input) => input.uid)
239 .filter((uid): uid is string => typeof uid === "string");
240
241 const rows = yield* Effect.forEach(
242 liveInputIds,
243 (liveInputId) =>
244 stream.listLiveInputOutputs
245 .pages({
246 accountId,
247 liveInputIdentifier: liveInputId,
248 })
249 .pipe(
250 Stream.runCollect,
251 Effect.map((chunk) =>
252 Array.from(chunk).flatMap((page) =>
253 page.result.map((observed) =>
254 toAttributes(observed, accountId, liveInputId),
255 ),
256 ),
257 ),
258 // A live input deleted between enumeration and listing its
259 // outputs counts as having no outputs.
260 Effect.catchTag("LiveInputNotFound", () => Effect.succeed([])),
261 ),
262 { concurrency: 10 },
263 );
264 return rows.flat();
265 }),
266
267 delete: Effect.fn(function* ({ output }) {
268 // Idempotent — an output (or its parent live input) already deleted
269 // out-of-band surfaces as `OutputNotFound` (Cloudflare error code
270 // 10003).
271 yield* stream
272 .deleteLiveInputOutput({
273 accountId: output.accountId,
274 liveInputIdentifier: output.liveInputId,
275 outputIdentifier: output.outputId,
276 })
277 .pipe(Effect.catchTag("OutputNotFound", () => Effect.void));
278 }),
279 });
280
281type ObservedOutput = {
282 enabled?: boolean | null;
283 streamKey?: string | null;
284 uid?: string | null;
285 url?: string | null;
286};
287
288/**
289 * Find an output by uid under a live input, mapping "gone" — either the
290 * output missing from the list or the parent live input deleted
291 * (`LiveInputNotFound`, Cloudflare error code 10003) — to `undefined`.
292 * There is no get-by-uid endpoint, so observation goes through the list.
293 */
294const findOutput = (accountId: string, liveInputId: string, outputId: string) =>
295 stream
296 .listLiveInputOutputs({ accountId, liveInputIdentifier: liveInputId })
297 .pipe(
298 Effect.map((page) =>
299 page.result.find((candidate) => candidate.uid === outputId),
300 ),
301 Effect.catchTag("LiveInputNotFound", () => Effect.succeed(undefined)),
302 );
303
304const toAttributes = (
305 output:
306 | stream.CreateLiveInputOutputResponse
307 | stream.UpdateLiveInputOutputResponse
308 | ObservedOutput,
309 accountId: string,
310 liveInputId: string,
311): LiveInputOutputAttributes => ({
312 outputId: output.uid ?? "",
313 liveInputId,
314 accountId,
315 url: output.url ?? "",
316 streamKey: output.streamKey ?? "",
317 enabled: output.enabled ?? true,
318});