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 getLiveInput = (accountId: string, liveInputId: string) =>
23 stream.getLiveInput({ accountId, liveInputIdentifier: liveInputId }).pipe(
24 Effect.retry({
25 while: (e) => e._tag === "Forbidden",
26 schedule: Schedule.exponential("500 millis"),
27 times: 8,
28 }),
29 );
30
31const expectGone = (accountId: string, liveInputId: string) =>
32 getLiveInput(accountId, liveInputId).pipe(
33 Effect.flatMap(() => Effect.fail({ _tag: "LiveInputNotDeleted" } as const)),
34 // A missing live input surfaces as `LiveInputNotFound` (Cloudflare
35 // error code 10003) — that's the success condition here.
36 Effect.catchTag("LiveInputNotFound", () => Effect.void),
37 Effect.retry({
38 while: (e) => e._tag === "LiveInputNotDeleted",
39 schedule: Schedule.max([
40 Schedule.exponential("500 millis"),
41 Schedule.recurs(10),
42 ]),
43 }),
44 );
45
46test.provider(
47 "create, update in place, and delete a live input",
48 (stack) =>
49 Effect.gen(function* () {
50 const { accountId } = yield* yield* CloudflareEnvironment;
51
52 yield* stack.destroy();
53
54 const input = yield* stack.deploy(
55 Cloudflare.Stream.LiveInput("BroadcastInput", {
56 meta: { name: "alchemy-stream-live-input" },
57 recording: { mode: "automatic", timeoutSeconds: 10 },
58 }),
59 );
60
61 expect(input.liveInputId).toBeTruthy();
62 expect(input.accountId).toEqual(accountId);
63 expect(input.enabled).toBe(true);
64 expect(input.meta).toMatchObject({ name: "alchemy-stream-live-input" });
65
66 const live = yield* getLiveInput(accountId, input.liveInputId);
67 expect(live.uid).toEqual(input.liveInputId);
68 expect(live.enabled).toBe(true);
69
70 // Update mutable props in place — same uid, no replacement.
71 const updated = yield* stack.deploy(
72 Cloudflare.Stream.LiveInput("BroadcastInput", {
73 enabled: false,
74 meta: { name: "alchemy-stream-live-input-v2" },
75 recording: { mode: "automatic", timeoutSeconds: 10 },
76 }),
77 );
78
79 expect(updated.liveInputId).toEqual(input.liveInputId);
80 expect(updated.enabled).toBe(false);
81 expect(updated.meta).toMatchObject({
82 name: "alchemy-stream-live-input-v2",
83 });
84
85 const observed = yield* getLiveInput(accountId, updated.liveInputId);
86 expect(observed.enabled).toBe(false);
87
88 // Redeploying identical props is a no-op (still the same input).
89 const noop = yield* stack.deploy(
90 Cloudflare.Stream.LiveInput("BroadcastInput", {
91 enabled: false,
92 meta: { name: "alchemy-stream-live-input-v2" },
93 recording: { mode: "automatic", timeoutSeconds: 10 },
94 }),
95 );
96 expect(noop.liveInputId).toEqual(input.liveInputId);
97
98 yield* stack.destroy();
99
100 yield* expectGone(accountId, input.liveInputId);
101 }).pipe(logLevel),
102 {
103 tags: ["provider:cloudflare", "provider:cloudflare:stream", "live"],
104 timeout: 120_000,
105 },
106);
107
108test.provider(
109 "recreates after out-of-band delete",
110 (stack) =>
111 Effect.gen(function* () {
112 const { accountId } = yield* yield* CloudflareEnvironment;
113
114 yield* stack.destroy();
115
116 const input = yield* stack.deploy(
117 Cloudflare.Stream.LiveInput("HealInput", {
118 meta: { name: "alchemy-stream-heal-input" },
119 }),
120 );
121
122 // Delete the live input out-of-band. A redeploy with identical props
123 // is a planner no-op, so change a prop to force reconcile — it must
124 // observe the input as missing and recreate it instead of failing.
125 yield* stream
126 .deleteLiveInput({ accountId, liveInputIdentifier: input.liveInputId })
127 .pipe(
128 Effect.retry({
129 while: (e) => e._tag === "Forbidden",
130 schedule: Schedule.exponential("500 millis"),
131 times: 8,
132 }),
133 );
134
135 const healed = yield* stack.deploy(
136 Cloudflare.Stream.LiveInput("HealInput", {
137 enabled: false,
138 meta: { name: "alchemy-stream-heal-input" },
139 }),
140 );
141
142 expect(healed.liveInputId).not.toEqual(input.liveInputId);
143 expect(healed.enabled).toBe(false);
144
145 yield* stack.destroy();
146
147 yield* expectGone(accountId, healed.liveInputId);
148 }).pipe(logLevel),
149 {
150 tags: ["provider:cloudflare", "provider:cloudflare:stream", "live"],
151 timeout: 120_000,
152 },
153);
154
155// Canonical `list()` test (account collection): deploy a live input, then
156// resolve the typed provider and assert the deployed uid appears in the
157// exhaustively-enumerated result.
158//
159// GATED — blocked on a distilled schema mismatch, not an entitlement: the
160// account *can* create/list live inputs (the CRUD cases above pass), but
161// `stream.listLiveInputs` fails to decode the live response. The Cloudflare
162// REST API returns `result` as a bare array of live-input objects, while the
163// upstream cloudflare-typescript SDK (and therefore distilled) types the
164// `result` payload as the object `{ liveInputs, range, total }`. The decode
165// throws, verbatim:
166//
167// CloudflareHttpError { status: 200, statusText: "Schema decode failed" }
168// GET /accounts/{account_id}/stream/live_inputs
169// body: {"result":[{"uid":"…","created":"…",…}],"success":true,…}
170//
171// Needed distilled patch: model `ListLiveInputsResponse` as
172// `Array<LiveInput>` decoded at ResponsePath("result") (not the wrapper
173// object). The current patch DSL can't express this — `op.responsePath`
174// ("result", from the SDK's `_thenUnwrap`) wins over `patch.responsePath`,
175// and `responseType: "array"` wraps the wrapper object rather than the item
176// — so it requires a generator/spec-level fix. Once landed, drop this gate
177// and map `response` (the array) directly in LiveInput.ts's `list()`.
178test.provider.skipIf(!process.env.CLOUDFLARE_TEST_STREAM_LIST)(
179 "list enumerates the deployed live input",
180 (stack) =>
181 Effect.gen(function* () {
182 yield* stack.destroy();
183
184 const input = yield* stack.deploy(
185 Cloudflare.Stream.LiveInput("ListInput", {
186 meta: { name: "alchemy-stream-list-input" },
187 }),
188 );
189
190 const provider = yield* Provider.findProvider(
191 Cloudflare.Stream.LiveInput,
192 );
193 const all = yield* provider.list();
194
195 expect(all.some((x) => x.liveInputId === input.liveInputId)).toBe(true);
196
197 yield* stack.destroy();
198 }).pipe(logLevel),
199 {
200 tags: ["provider:cloudflare", "provider:cloudflare:stream", "live"],
201 timeout: 120_000,
202 },
203);