1import * as Cloudflare from "@/Cloudflare";
2import { CloudflareEnvironment } from "@/Cloudflare/CloudflareEnvironment";
3import * as Provider from "@/Provider";
4import * as Test from "@/Test/Alchemy";
5import * as pipelines from "@distilled.cloud/cloudflare/pipelines";
6import * as user from "@distilled.cloud/cloudflare/user";
7import * as workers from "@distilled.cloud/cloudflare/workers";
8import { expect } from "alchemy-test";
9import * as Effect from "effect/Effect";
10import * as Redacted from "effect/Redacted";
11import { MinimumLogLevel } from "effect/References";
12import * as Schedule from "effect/Schedule";
13import crypto from "node:crypto";
14
15const { test } = Test.make({ providers: Cloudflare.providers() });
16
17const logLevel = Effect.provideService(
18 MinimumLogLevel,
19 process.env.DEBUG ? "Debug" : "Info",
20);
21
22// The scoped API token the test harness mints propagates eventually-
23// consistently across Cloudflare's edge — ride out 403 blips
24// (`Forbidden`, declared in the distilled error unions) on the test's
25// own out-of-band verification calls.
26const forbiddenBlips = {
27 while: (e: { _tag: string }) => e._tag === "Forbidden",
28 schedule: Schedule.exponential("500 millis"),
29 times: 8,
30} as const;
31
32// The same blip can hit engine-side calls mid-deploy. Reconcile is
33// idempotent (observe→ensure→sync), so retrying the whole deploy is safe.
34const retryAuthBlip = <A, E, R>(eff: Effect.Effect<A, E, R>) =>
35 eff.pipe(
36 Effect.retry({
37 while: (e) => String(e).includes("Unable to authenticate request"),
38 schedule: Schedule.exponential("1 second"),
39 times: 5,
40 }),
41 );
42
43// R2 S3-compatible credentials are derived from the active API token:
44// the access key id is the token id and the secret is the SHA-256 hex
45// digest of the token value.
46const r2Credentials = Effect.gen(function* () {
47 const creds = yield* yield* CloudflareEnvironment;
48 if (creds.type !== "apiToken") {
49 return yield* Effect.die(
50 new Error(
51 "LegacyPipeline test requires an apiToken profile (R2 S3 credentials are derived from the token)",
52 ),
53 );
54 }
55 const token = Redacted.value(creds.apiToken);
56 const verified = yield* retryAuthBlip(user.verifyToken({}));
57 const secretAccessKey = yield* Effect.sync(() =>
58 crypto.createHash("sha256").update(token).digest("hex"),
59 );
60 return {
61 accessKeyId: Redacted.make(verified.id),
62 secretAccessKey: Redacted.make(secretAccessKey),
63 };
64});
65
66const getLegacyPipeline = (accountId: string, pipelineName: string) =>
67 pipelines
68 .getPipeline({ accountId, pipelineName })
69 .pipe(Effect.retry(forbiddenBlips));
70
71const expectLegacyPipelineGone = (accountId: string, pipelineName: string) =>
72 getLegacyPipeline(accountId, pipelineName).pipe(
73 Effect.flatMap(() => Effect.fail({ _tag: "PipelineNotDeleted" } as const)),
74 Effect.catchTag("PipelineNotExists", () => Effect.void),
75 Effect.retry({
76 while: (e) => e._tag === "PipelineNotDeleted",
77 schedule: Schedule.max([
78 Schedule.exponential("500 millis"),
79 Schedule.recurs(10),
80 ]),
81 }),
82 );
83
84interface LegacyOpts {
85 prefix?: string;
86 corsOrigins?: string[];
87 name?: string;
88}
89
90const legacy = (
91 creds: {
92 accessKeyId: Redacted.Redacted<string>;
93 secretAccessKey: Redacted.Redacted<string>;
94 },
95 opts: LegacyOpts = {},
96) =>
97 Effect.gen(function* () {
98 const bucket = yield* Cloudflare.R2.Bucket("LegacyBucket", {
99 forceDestroy: true,
100 });
101 const pipeline = yield* Cloudflare.Pipelines.LegacyPipeline("Legacy", {
102 name: opts.name,
103 source: [
104 {
105 type: "http",
106 ...(opts.corsOrigins ? { cors: { origins: opts.corsOrigins } } : {}),
107 },
108 { type: "binding" },
109 ],
110 destination: {
111 bucket: bucket.bucketName,
112 credentials: creds,
113 batch: { maxDurationS: 5 },
114 ...(opts.prefix ? { prefix: opts.prefix } : {}),
115 },
116 });
117 return { bucket, pipeline };
118 });
119
120const asyncWorkerScript = `export default {
121 async fetch() {
122 return new Response("ok");
123 },
124};
125`;
126
127test.provider(
128 "binds a legacy pipeline to a worker by name",
129 (stack) =>
130 Effect.gen(function* () {
131 const { accountId } = yield* yield* CloudflareEnvironment;
132 const creds = yield* r2Credentials;
133
134 yield* stack.destroy();
135
136 const { pipeline, worker } = yield* retryAuthBlip(
137 stack.deploy(
138 Effect.gen(function* () {
139 const { pipeline } = yield* legacy(creds);
140 const worker = yield* Cloudflare.Worker(
141 "legacy-pipeline-binding-worker",
142 {
143 script: asyncWorkerScript,
144 env: {
145 LEGACY: pipeline,
146 },
147 },
148 );
149 return { pipeline, worker };
150 }),
151 ),
152 );
153
154 const settings = yield* workers
155 .getScriptScriptAndVersionSetting({
156 accountId,
157 scriptName: worker.workerName,
158 })
159 .pipe(Effect.retry(forbiddenBlips));
160
161 const binding = (settings.bindings ?? []).find(
162 (b) => b.name === "LEGACY",
163 );
164
165 expect(binding).toMatchObject({
166 type: "pipelines",
167 name: "LEGACY",
168 pipeline: pipeline.pipelineId,
169 });
170
171 yield* stack.destroy();
172 yield* expectLegacyPipelineGone(accountId, pipeline.name);
173 }).pipe(logLevel),
174 {
175 tags: [
176 "provider:cloudflare",
177 "provider:cloudflare:pipelines",
178 "provider:cloudflare:r2",
179 "provider:cloudflare:worker",
180 "live",
181 ],
182 timeout: 420_000,
183 },
184);
185
186test.provider(
187 "legacy pipeline: create, in-place update, replace on name change",
188 (stack) =>
189 Effect.gen(function* () {
190 const { accountId } = yield* yield* CloudflareEnvironment;
191 const creds = yield* r2Credentials;
192
193 yield* stack.destroy();
194
195 // Create — engine-generated name, http + binding sources.
196 const initial = yield* retryAuthBlip(stack.deploy(legacy(creds)));
197
198 expect(initial.pipeline.pipelineId).toBeTruthy();
199 expect(initial.pipeline.accountId).toEqual(accountId);
200 expect(initial.pipeline.endpoint).toBeTruthy();
201 expect(initial.pipeline.bucket).toEqual(initial.bucket.bucketName);
202
203 const live = yield* getLegacyPipeline(accountId, initial.pipeline.name);
204 expect(live.id).toEqual(initial.pipeline.pipelineId);
205 expect(live.destination.path.bucket).toEqual(initial.bucket.bucketName);
206 expect(live.source.map((s) => s.type).sort()).toEqual([
207 "binding",
208 "http",
209 ]);
210
211 // Redeploying identical props is a no-op.
212 const noop = yield* retryAuthBlip(stack.deploy(legacy(creds)));
213 expect(noop.pipeline.pipelineId).toEqual(initial.pipeline.pipelineId);
214 expect(noop.pipeline.version).toEqual(initial.pipeline.version);
215
216 // In-place update — prefix and CORS change via PUT, same id/name.
217 const updated = yield* retryAuthBlip(
218 stack.deploy(
219 legacy(creds, {
220 prefix: "ingest",
221 corsOrigins: ["https://example.com"],
222 }),
223 ),
224 );
225
226 expect(updated.pipeline.pipelineId).toEqual(initial.pipeline.pipelineId);
227 expect(updated.pipeline.name).toEqual(initial.pipeline.name);
228
229 const liveUpdated = yield* getLegacyPipeline(
230 accountId,
231 updated.pipeline.name,
232 );
233 expect(liveUpdated.destination.path.prefix).toEqual("ingest");
234 const httpSource = liveUpdated.source.find((s) => s.type === "http");
235 expect(
236 httpSource && "cors" in httpSource
237 ? httpSource.cors?.origins
238 : undefined,
239 ).toEqual(["https://example.com"]);
240
241 // Name change — the legacy API addresses pipelines by name, so this
242 // is a replacement.
243 const replacementName = `${initial.pipeline.name}-r`;
244 const replaced = yield* retryAuthBlip(
245 stack.deploy(
246 legacy(creds, {
247 name: replacementName,
248 prefix: "ingest",
249 corsOrigins: ["https://example.com"],
250 }),
251 ),
252 );
253
254 expect(replaced.pipeline.name).toEqual(replacementName);
255 expect(replaced.pipeline.pipelineId).not.toEqual(
256 initial.pipeline.pipelineId,
257 );
258 yield* expectLegacyPipelineGone(accountId, initial.pipeline.name);
259
260 yield* stack.destroy();
261 yield* expectLegacyPipelineGone(accountId, replaced.pipeline.name);
262
263 // Destroy again — delete must be idempotent.
264 yield* stack.destroy();
265 }).pipe(logLevel),
266 {
267 tags: [
268 "provider:cloudflare",
269 "provider:cloudflare:pipelines",
270 "provider:cloudflare:r2",
271 "live",
272 ],
273 timeout: 420_000,
274 },
275);
276
277// The list endpoint returns truncated summary items
278// (`{id, name, endpoint}` only), but distilled's `ListPipelinesResponse`
279// schema marks the per-item `destination`, `source`, and `version`
280// fields as required, so the valid response is rejected as a catch-all
281// `CloudflareHttpError`:
282// CloudflareHttpError: {"success":true,...,"result":[{"id":"...",
283// "name":"...","endpoint":"..."}],"result_info":{...}}
284// NEEDED DISTILLED PATCH (pipelines/listPipelines): make the per-item
285// `destination`, `source`, and `version` fields optional in
286// `ListPipelinesResponse.results` (the list endpoint is summary-only;
287// the provider hydrates each item via `getPipeline`). Until then this is
288// gated — set CLOUDFLARE_TEST_LEGACY_PIPELINE_LIST=1 to run it.
289test.provider.skipIf(!process.env.CLOUDFLARE_TEST_LEGACY_PIPELINE_LIST)(
290 "list enumerates the deployed legacy pipeline",
291 (stack) =>
292 Effect.gen(function* () {
293 const { accountId } = yield* yield* CloudflareEnvironment;
294 const creds = yield* r2Credentials;
295
296 yield* stack.destroy();
297
298 const deployed = yield* retryAuthBlip(stack.deploy(legacy(creds)));
299 expect(deployed.pipeline.pipelineId).toBeTruthy();
300
301 // Account collection: list() exhaustively paginates every legacy
302 // pipeline in the account and hydrates each into the read
303 // Attributes shape.
304 const provider = yield* Provider.findProvider(
305 Cloudflare.Pipelines.LegacyPipeline,
306 );
307 const all = yield* provider.list();
308
309 const match = all.find(
310 (p) => p.pipelineId === deployed.pipeline.pipelineId,
311 );
312 expect(match).toBeDefined();
313 expect(match?.name).toEqual(deployed.pipeline.name);
314 expect(match?.accountId).toEqual(accountId);
315 expect(match?.bucket).toEqual(deployed.bucket.bucketName);
316 expect(match?.endpoint).toEqual(deployed.pipeline.endpoint);
317
318 yield* stack.destroy();
319 }).pipe(logLevel),
320 {
321 tags: [
322 "provider:cloudflare",
323 "provider:cloudflare:pipelines",
324 "provider:cloudflare:r2",
325 "live",
326 ],
327 timeout: 300_000,
328 },
329);