1import * as Cloudflare from "@/Cloudflare";
2import { CloudflareEnvironment } from "@/Cloudflare/CloudflareEnvironment";
3import * as Provider from "@/Provider";
4import * as Test from "@/Test/Alchemy";
5import * as stream from "@distilled.cloud/cloudflare/stream";
6import { expect } from "alchemy-test";
7import * as Effect from "effect/Effect";
8import { MinimumLogLevel } from "effect/References";
9import * as Schedule from "effect/Schedule";
10
11const { test } = Test.make({ providers: Cloudflare.providers() });
12
13const logLevel = Effect.provideService(
14 MinimumLogLevel,
15 process.env.DEBUG ? "Debug" : "Info",
16);
17
18// The scoped API token the test harness mints propagates eventually-
19// consistently across Cloudflare's edge — ride out 403 blips
20// (`Forbidden`, declared in the distilled error union) on the test's
21// own out-of-band verification calls.
22const listOutputs = (accountId: string, liveInputId: string) =>
23 stream
24 .listLiveInputOutputs({ accountId, liveInputIdentifier: liveInputId })
25 .pipe(
26 Effect.retry({
27 while: (e) => e._tag === "Forbidden",
28 schedule: Schedule.exponential("500 millis"),
29 times: 8,
30 }),
31 );
32
33const findOutput = (accountId: string, liveInputId: string, outputId: string) =>
34 listOutputs(accountId, liveInputId).pipe(
35 Effect.map((page) => page.result.find((o) => o.uid === outputId)),
36 );
37
38const expectGone = (accountId: string, liveInputId: string, outputId: string) =>
39 findOutput(accountId, liveInputId, outputId).pipe(
40 Effect.flatMap((found) =>
41 found === undefined
42 ? Effect.void
43 : Effect.fail({ _tag: "OutputNotDeleted" } as const),
44 ),
45 // The parent live input being gone counts as the output being gone.
46 Effect.catchTag("LiveInputNotFound", () => Effect.void),
47 Effect.retry({
48 while: (e) => e._tag === "OutputNotDeleted",
49 schedule: Schedule.max([
50 Schedule.exponential("500 millis"),
51 Schedule.recurs(10),
52 ]),
53 }),
54 );
55
56test.provider(
57 "create, toggle enabled in place, replace destination, and delete",
58 (stack) =>
59 Effect.gen(function* () {
60 const { accountId } = yield* yield* CloudflareEnvironment;
61
62 yield* stack.destroy();
63
64 const deployOutput = (props: {
65 url: string;
66 streamKey: string;
67 enabled?: boolean;
68 }) =>
69 Effect.gen(function* () {
70 const input = yield* Cloudflare.Stream.LiveInput("RestreamInput", {
71 meta: { name: "alchemy-stream-output-input" },
72 });
73 const output = yield* Cloudflare.Stream.LiveInputOutput("Restream", {
74 liveInputId: input.liveInputId,
75 ...props,
76 });
77 return { input, output };
78 });
79
80 const created = yield* stack.deploy(
81 deployOutput({
82 url: "rtmps://a.rtmps.youtube.com/live2",
83 streamKey: "alchemy-test-stream-key",
84 }),
85 );
86
87 expect(created.output.outputId).toBeTruthy();
88 expect(created.output.liveInputId).toEqual(created.input.liveInputId);
89 expect(created.output.accountId).toEqual(accountId);
90 expect(created.output.url).toEqual("rtmps://a.rtmps.youtube.com/live2");
91 expect(created.output.streamKey).toEqual("alchemy-test-stream-key");
92 expect(created.output.enabled).toBe(true);
93
94 // Out-of-band verify via distilled.
95 const observed = yield* findOutput(
96 accountId,
97 created.input.liveInputId,
98 created.output.outputId,
99 );
100 expect(observed?.uid).toEqual(created.output.outputId);
101 expect(observed?.enabled).toBe(true);
102
103 // Toggle `enabled` — updates in place, same uid.
104 const disabled = yield* stack.deploy(
105 deployOutput({
106 url: "rtmps://a.rtmps.youtube.com/live2",
107 streamKey: "alchemy-test-stream-key",
108 enabled: false,
109 }),
110 );
111 expect(disabled.output.outputId).toEqual(created.output.outputId);
112 expect(disabled.output.enabled).toBe(false);
113
114 const observedDisabled = yield* findOutput(
115 accountId,
116 created.input.liveInputId,
117 created.output.outputId,
118 );
119 expect(observedDisabled?.enabled).toBe(false);
120
121 // Redeploying identical props is a no-op (still the same output).
122 const noop = yield* stack.deploy(
123 deployOutput({
124 url: "rtmps://a.rtmps.youtube.com/live2",
125 streamKey: "alchemy-test-stream-key",
126 enabled: false,
127 }),
128 );
129 expect(noop.output.outputId).toEqual(created.output.outputId);
130
131 // Changing the destination (url/streamKey) replaces the output.
132 const replaced = yield* stack.deploy(
133 deployOutput({
134 url: "rtmps://b.rtmps.youtube.com/live2?backup=1",
135 streamKey: "alchemy-test-stream-key-v2",
136 }),
137 );
138 expect(replaced.output.outputId).not.toEqual(created.output.outputId);
139 expect(replaced.output.url).toEqual(
140 "rtmps://b.rtmps.youtube.com/live2?backup=1",
141 );
142 expect(replaced.output.enabled).toBe(true);
143
144 // The replaced (old) output is gone; the new one exists.
145 yield* expectGone(
146 accountId,
147 created.input.liveInputId,
148 created.output.outputId,
149 );
150 const observedReplaced = yield* findOutput(
151 accountId,
152 replaced.input.liveInputId,
153 replaced.output.outputId,
154 );
155 expect(observedReplaced?.uid).toEqual(replaced.output.outputId);
156
157 yield* stack.destroy();
158
159 yield* expectGone(
160 accountId,
161 replaced.input.liveInputId,
162 replaced.output.outputId,
163 );
164 }).pipe(logLevel),
165 {
166 tags: ["provider:cloudflare", "provider:cloudflare:stream", "live"],
167 timeout: 120_000,
168 },
169);
170
171test.provider(
172 "recreates after out-of-band delete",
173 (stack) =>
174 Effect.gen(function* () {
175 const { accountId } = yield* yield* CloudflareEnvironment;
176
177 yield* stack.destroy();
178
179 const deployOutput = (enabled?: boolean) =>
180 Effect.gen(function* () {
181 const input = yield* Cloudflare.Stream.LiveInput("HealOutputInput", {
182 meta: { name: "alchemy-stream-output-heal-input" },
183 });
184 const output = yield* Cloudflare.Stream.LiveInputOutput(
185 "HealOutput",
186 {
187 liveInputId: input.liveInputId,
188 url: "rtmps://a.rtmps.youtube.com/live2",
189 streamKey: "alchemy-heal-stream-key",
190 enabled,
191 },
192 );
193 return { input, output };
194 });
195
196 const created = yield* stack.deploy(deployOutput());
197
198 // Delete the output out-of-band. A redeploy with identical props is
199 // a planner no-op, so toggle `enabled` to force reconcile — it must
200 // observe the output as missing and recreate it instead of failing.
201 yield* stream
202 .deleteLiveInputOutput({
203 accountId,
204 liveInputIdentifier: created.input.liveInputId,
205 outputIdentifier: created.output.outputId,
206 })
207 .pipe(
208 Effect.retry({
209 while: (e) => e._tag === "Forbidden",
210 schedule: Schedule.exponential("500 millis"),
211 times: 8,
212 }),
213 );
214
215 const healed = yield* stack.deploy(deployOutput(false));
216
217 expect(healed.output.outputId).not.toEqual(created.output.outputId);
218 expect(healed.output.enabled).toBe(false);
219
220 yield* stack.destroy();
221
222 yield* expectGone(
223 accountId,
224 healed.input.liveInputId,
225 healed.output.outputId,
226 );
227 }).pipe(logLevel),
228 {
229 tags: ["provider:cloudflare", "provider:cloudflare:stream", "live"],
230 timeout: 120_000,
231 },
232);
233
234// `list()` enumerates every live input on the account (via
235// `stream.listLiveInputs`) and then lists each input's outputs. The
236// distilled `listLiveInputs` response schema is currently wrong: it
237// decodes `result` as an object `{ liveInputs?, range?, total? }`, but
238// Cloudflare returns `result` as a *bare array* of live-input items.
239// This surfaces as an untyped `CloudflareHttpError` (status 200,
240// statusText "Schema decode failed") thrown by the schema decoder before
241// `list()`'s own code runs. It is a deterministic schema mismatch (not an
242// entitlement issue), so the live list test is gated until distilled is
243// patched. See the JSON report for the exact needed distilled patch.
244// Run with CLOUDFLARE_TEST_STREAM_LIST=1 once distilled is fixed.
245test.provider.skipIf(!process.env.CLOUDFLARE_TEST_STREAM_LIST)(
246 "list enumerates outputs across all live inputs",
247 (stack) =>
248 Effect.gen(function* () {
249 const { accountId } = yield* yield* CloudflareEnvironment;
250
251 yield* stack.destroy();
252
253 const deployed = yield* stack.deploy(
254 Effect.gen(function* () {
255 const input = yield* Cloudflare.Stream.LiveInput("ListOutputInput", {
256 meta: { name: "alchemy-stream-output-list-input" },
257 });
258 const output = yield* Cloudflare.Stream.LiveInputOutput(
259 "ListOutput",
260 {
261 liveInputId: input.liveInputId,
262 url: "rtmps://a.rtmps.youtube.com/live2",
263 streamKey: "alchemy-list-stream-key",
264 },
265 );
266 return { input, output };
267 }),
268 );
269
270 const provider = yield* Provider.findProvider(
271 Cloudflare.Stream.LiveInputOutput,
272 );
273
274 // Edge propagation: the freshly-created output (and its parent live
275 // input enumeration) is eventually consistent — retry until present.
276 const all = yield* provider.list().pipe(
277 Effect.flatMap((rows) =>
278 rows.some((o) => o.outputId === deployed.output.outputId)
279 ? Effect.succeed(rows)
280 : Effect.fail({ _tag: "OutputNotListed" } as const),
281 ),
282 Effect.retry({
283 while: (e) => e._tag === "OutputNotListed",
284 schedule: Schedule.max([
285 Schedule.exponential("500 millis"),
286 Schedule.recurs(10),
287 ]),
288 }),
289 );
290
291 const found = all.find((o) => o.outputId === deployed.output.outputId);
292 expect(found).toBeDefined();
293 expect(found?.liveInputId).toEqual(deployed.input.liveInputId);
294 expect(found?.accountId).toEqual(accountId);
295 expect(found?.url).toEqual("rtmps://a.rtmps.youtube.com/live2");
296 expect(found?.enabled).toBe(true);
297
298 yield* stack.destroy();
299 }).pipe(logLevel),
300 {
301 tags: ["provider:cloudflare", "provider:cloudflare:stream", "live"],
302 timeout: 120_000,
303 },
304);