1import * as stream from "@distilled.cloud/cloudflare/stream";
2import * as Effect from "effect/Effect";
3import * as Predicate from "effect/Predicate";
4
5import { createPhysicalName } from "../../PhysicalName.ts";
6import * as Provider from "../../Provider.ts";
7import { Resource } from "../../Resource.ts";
8import { CloudflareEnvironment } from "../CloudflareEnvironment.ts";
9import type { Providers } from "../Providers.ts";
10
11const TypeId = "Cloudflare.Stream.LiveInput" as const;
12type TypeId = typeof TypeId;
13
14/**
15 * Recording behavior of a live input.
16 */
17export type LiveInputRecording = {
18 /**
19 * Origins allowed to display videos created from this live input.
20 * Enter allowed origin domains in an array, e.g. `["example.com"]`.
21 */
22 allowedOrigins?: string[];
23 /**
24 * Disables reporting the number of live viewers when this is set to
25 * `true`.
26 * @default false
27 */
28 hideLiveViewerCount?: boolean;
29 /**
30 * Specifies the recording behavior for the live input. `off` prevents
31 * the input from being recorded; `automatic` records the input as a
32 * Cloudflare Stream video whenever a stream is live.
33 * @default "off"
34 */
35 mode?: "off" | "automatic";
36 /**
37 * Indicates if signed URL tokens are required to view the recorded
38 * video.
39 * @default false
40 */
41 requireSignedURLs?: boolean;
42 /**
43 * Number of seconds the live input is considered live after the
44 * broadcast stops, before the recording transitions to on-demand.
45 * @default 0
46 */
47 timeoutSeconds?: number;
48};
49
50export type LiveInputProps = {
51 /**
52 * Sets the creator ID associated with this live input. Mutable.
53 */
54 defaultCreator?: string;
55 /**
56 * Number of days after which the live input's recordings are deleted.
57 * The minimum accepted value is 30. Omit to retain recordings
58 * indefinitely. Mutable.
59 */
60 deleteRecordingAfterDays?: number;
61 /**
62 * Whether the live input is enabled and can accept streams. Mutable.
63 * @default true
64 */
65 enabled?: boolean;
66 /**
67 * A user modifiable key-value store used to reference other systems of
68 * record for managing live inputs. By convention, set a `name` field
69 * for display in the Cloudflare dashboard. If omitted, a `name`
70 * derived from the app, stage, and logical ID is used. Mutable.
71 * @default { name: ${app}-${stage}-${id} }
72 */
73 meta?: Record<string, unknown>;
74 /**
75 * Records the input to a Cloudflare Stream video when set to
76 * `automatic` mode. Mutable.
77 * @default { mode: "off" }
78 */
79 recording?: LiveInputRecording;
80};
81
82export type LiveInputAttributes = {
83 /**
84 * The unique identifier for the live input (Cloudflare `uid`).
85 */
86 liveInputId: string;
87 /**
88 * The Cloudflare account the live input belongs to.
89 */
90 accountId: string;
91 /**
92 * The date and time the live input was created.
93 */
94 created: string | undefined;
95 /**
96 * The date and time the live input was last modified.
97 */
98 modified: string | undefined;
99 /**
100 * Whether the live input is enabled and can accept streams.
101 */
102 enabled: boolean;
103 /**
104 * Number of days after which the live input's recordings are deleted,
105 * if configured.
106 */
107 deleteRecordingAfterDays: number | undefined;
108 /**
109 * The user-modifiable key-value store associated with the live input.
110 */
111 meta: Record<string, unknown>;
112};
113
114export type LiveInput = Resource<
115 TypeId,
116 LiveInputProps,
117 LiveInputAttributes,
118 never,
119 Providers
120>;
121
122/**
123 * A Cloudflare Stream live input — an ingest endpoint (RTMPS/SRT/WebRTC)
124 * that accepts live video and optionally records it as a Stream video.
125 *
126 * Live inputs are identified by an auto-assigned `uid`; every prop is
127 * mutable in place via Cloudflare's PUT endpoint, so the resource is
128 * never replaced. Deleting a live input does not delete videos already
129 * recorded from it.
130 *
131 * Requires the Stream subscription to be enabled on the account.
132 * ### Creating a live input
133 * **Example:** Basic live input
134 * ```typescript
135 * const input = yield* Cloudflare.Stream.LiveInput("Broadcast", {});
136 * ```
137 *
138 * **Example:** Live input with automatic recording
139 * ```typescript
140 * const input = yield* Cloudflare.Stream.LiveInput("Broadcast", {
141 * meta: { name: "town-hall" },
142 * recording: {
143 * mode: "automatic",
144 * timeoutSeconds: 10,
145 * },
146 * deleteRecordingAfterDays: 30,
147 * });
148 * ```
149 *
150 * ### Managing a live input
151 * **Example:** Disable ingest without deleting the input
152 * ```typescript
153 * const input = yield* Cloudflare.Stream.LiveInput("Broadcast", {
154 * enabled: false,
155 * });
156 * ```
157 *
158 * @see https://developers.cloudflare.com/stream/stream-live/
159 *
160 * @resource
161 * @product Stream
162 * @category Media
163 */
164export const LiveInput = Resource<LiveInput>(TypeId);
165
166/**
167 * Returns true if the given value is a LiveInput resource.
168 */
169export const isLiveInput = (value: unknown): value is LiveInput =>
170 Predicate.hasProperty(value, "Type") && value.Type === TypeId;
171
172export const LiveInputProvider = () =>
173 Provider.succeed(LiveInput, {
174 stables: ["liveInputId", "accountId", "created"],
175
176 diff: Effect.fn(function* ({ output }) {
177 const { accountId } = yield* yield* CloudflareEnvironment;
178 if (output !== undefined && output.accountId !== accountId) {
179 return { action: "replace" } as const;
180 }
181 // Everything else is mutable via PUT.
182 return undefined;
183 }),
184
185 read: Effect.fn(function* ({ output }) {
186 // Live inputs have no deterministic name or tags, so a cold read
187 // (state lost before the uid was persisted) cannot recover the
188 // resource — rely on the cached uid.
189 if (output?.liveInputId === undefined) return undefined;
190 const accountId = output.accountId;
191 const observed = yield* getLiveInput(accountId, output.liveInputId);
192 return observed ? toAttributes(observed, accountId) : undefined;
193 }),
194
195 // Account collection: enumerate every live input in the account. The
196 // Cloudflare list endpoint returns all inputs in a single response
197 // (no cursor), so there is no extra pagination to drive. Hydrate each
198 // item into the same Attributes shape `read` produces.
199 list: Effect.fn(function* () {
200 const { accountId } = yield* yield* CloudflareEnvironment;
201 // Cloudflare returns this list either wrapped (`{ liveInputs: [...] }`)
202 // or as a bare `result` array depending on the account — handle both.
203 const response = yield* stream.listLiveInputs({ accountId });
204 const inputs = Array.isArray(response)
205 ? response
206 : (response.liveInputs ?? []);
207 return inputs
208 .filter(
209 (li): li is typeof li & { uid: string } =>
210 li.uid !== null && li.uid !== undefined,
211 )
212 .map((li) => toAttributes(li, accountId));
213 }),
214
215 reconcile: Effect.fn(function* ({ id, news, olds, output }) {
216 const { accountId } = yield* yield* CloudflareEnvironment;
217 const meta = news.meta ?? {
218 name: yield* createPhysicalName({ id, lowercase: true }),
219 };
220
221 // Observe — the uid cached on `output` is a hint, not a guarantee:
222 // a not-found falls through to "missing" and we recreate.
223 const observed = output?.liveInputId
224 ? yield* getLiveInput(output.accountId ?? accountId, output.liveInputId)
225 : undefined;
226
227 if (!observed) {
228 // Ensure — greenfield (or out-of-band delete): create with the
229 // full desired body. uids are auto-assigned so there is no
230 // AlreadyExists race to tolerate.
231 const created = yield* stream.createLiveInput({
232 accountId,
233 defaultCreator: news.defaultCreator,
234 deleteRecordingAfterDays: news.deleteRecordingAfterDays,
235 enabled: news.enabled,
236 meta,
237 recording: news.recording,
238 });
239 return toAttributes(created, accountId);
240 }
241
242 // Sync — diff observed cloud state against desired. The update API
243 // is a PUT, so send the full desired body, but skip the call
244 // entirely on a no-op. `defaultCreator` and `recording` are not
245 // echoed by the API, so `olds` is the best available baseline for
246 // them.
247 const observedAccount = output?.accountId ?? accountId;
248 const dirty =
249 (news.enabled ?? true) !== (observed.enabled ?? true) ||
250 (news.deleteRecordingAfterDays ?? undefined) !==
251 (observed.deleteRecordingAfterDays ?? undefined) ||
252 !deepValueEquals(meta, observed.meta ?? {}) ||
253 news.defaultCreator !== olds?.defaultCreator ||
254 !deepValueEquals(news.recording ?? {}, olds?.recording ?? {});
255
256 if (!dirty) {
257 return toAttributes(observed, observedAccount);
258 }
259
260 const updated = yield* stream.updateLiveInput({
261 accountId: observedAccount,
262 liveInputIdentifier: observed.uid ?? output!.liveInputId,
263 defaultCreator: news.defaultCreator,
264 deleteRecordingAfterDays: news.deleteRecordingAfterDays,
265 enabled: news.enabled,
266 meta,
267 recording: news.recording,
268 });
269 return toAttributes(updated, observedAccount);
270 }),
271
272 delete: Effect.fn(function* ({ output }) {
273 // Idempotent — a live input already deleted out-of-band surfaces
274 // as `LiveInputNotFound` (Cloudflare error code 10003).
275 yield* stream
276 .deleteLiveInput({
277 accountId: output.accountId,
278 liveInputIdentifier: output.liveInputId,
279 })
280 .pipe(Effect.catchTag("LiveInputNotFound", () => Effect.void));
281 }),
282 });
283
284/**
285 * Read a live input by uid, mapping "gone" (`LiveInputNotFound`,
286 * Cloudflare error code 10003) to `undefined`.
287 */
288const getLiveInput = (accountId: string, liveInputId: string) =>
289 stream
290 .getLiveInput({ accountId, liveInputIdentifier: liveInputId })
291 .pipe(
292 Effect.catchTag("LiveInputNotFound", () => Effect.succeed(undefined)),
293 );
294
295const toAttributes = (
296 input:
297 | stream.GetLiveInputResponse
298 | stream.CreateLiveInputResponse
299 | stream.UpdateLiveInputResponse,
300 accountId: string,
301): LiveInputAttributes => ({
302 liveInputId: input.uid ?? "",
303 accountId,
304 created: input.created ?? undefined,
305 modified: input.modified ?? undefined,
306 enabled: input.enabled ?? true,
307 deleteRecordingAfterDays: input.deleteRecordingAfterDays ?? undefined,
308 meta: (input.meta ?? {}) as Record<string, unknown>,
309});
310
311/**
312 * Structural equality for plain-JSON values (`meta`, `recording`) —
313 * primitives, arrays, and plain objects, `null`-tolerant and key-order
314 * insensitive.
315 */
316const deepValueEquals = (a: unknown, b: unknown): boolean => {
317 if (a === b) return true;
318 if (a === null || b === null || a === undefined || b === undefined) {
319 return false;
320 }
321 if (Array.isArray(a) && Array.isArray(b)) {
322 return a.length === b.length && a.every((v, i) => deepValueEquals(v, b[i]));
323 }
324 if (typeof a === "object" && typeof b === "object") {
325 const ka = Object.keys(a as Record<string, unknown>);
326 const kb = Object.keys(b as Record<string, unknown>);
327 return (
328 ka.length === kb.length &&
329 ka.every((k) =>
330 deepValueEquals(
331 (a as Record<string, unknown>)[k],
332 (b as Record<string, unknown>)[k],
333 ),
334 )
335 );
336 }
337 return false;
338};