1import * as Cloudflare from "@/Cloudflare";
2import { CloudflareEnvironment } from "@/Cloudflare/CloudflareEnvironment";
3import * as Output from "@/Output";
4import * as Provider from "@/Provider";
5import * as Test from "@/Test/Alchemy";
6import * as user from "@distilled.cloud/cloudflare/user";
7import { expect } from "alchemy-test";
8import * as Effect from "effect/Effect";
9import * as Redacted from "effect/Redacted";
10import { MinimumLogLevel } from "effect/References";
11import * as Schedule from "effect/Schedule";
12import crypto from "node:crypto";
13
14const { test } = Test.make({ providers: Cloudflare.providers() });
15
16const logLevel = Effect.provideService(
17 MinimumLogLevel,
18 process.env.DEBUG ? "Debug" : "Info",
19);
20
21const retryAuthBlip = <A, E, R>(eff: Effect.Effect<A, E, R>) =>
22 eff.pipe(
23 Effect.retry({
24 while: (e) => String(e).includes("Unable to authenticate request"),
25 schedule: Schedule.exponential("1 second"),
26 times: 5,
27 }),
28 );
29
30// R2 S3-compatible credentials are derived from the active API token:
31// the access key id is the token id and the secret is the SHA-256 hex
32// digest of the token value.
33const r2Credentials = Effect.gen(function* () {
34 const creds = yield* yield* CloudflareEnvironment;
35 if (creds.type !== "apiToken") {
36 return yield* Effect.die(
37 new Error(
38 "Pipelines list test requires an apiToken profile (R2 S3 credentials are derived from the token)",
39 ),
40 );
41 }
42 const token = Redacted.value(creds.apiToken);
43 const verified = yield* retryAuthBlip(user.verifyToken({}));
44 const secretAccessKey = yield* Effect.sync(() =>
45 crypto.createHash("sha256").update(token).digest("hex"),
46 );
47 return {
48 accessKeyId: Redacted.make(verified.id),
49 secretAccessKey: Redacted.make(secretAccessKey),
50 };
51});
52
53// A pipeline references a stream + sink by name in its SQL, so the
54// smallest deployable unit is the full R2 topology.
55const etl = (creds: {
56 accessKeyId: Redacted.Redacted<string>;
57 secretAccessKey: Redacted.Redacted<string>;
58}) =>
59 Effect.gen(function* () {
60 const bucket = yield* Cloudflare.R2.Bucket("SinkBucket", {
61 forceDestroy: true,
62 });
63 const stream = yield* Cloudflare.Pipelines.Stream("Stream", {});
64 const sink = yield* Cloudflare.Pipelines.Sink("Sink", {
65 type: "r2",
66 config: {
67 bucket: bucket.bucketName,
68 credentials: creds,
69 rollingPolicy: { intervalSeconds: 10 },
70 },
71 });
72 const pipeline = yield* Cloudflare.Pipelines.Pipeline("Etl", {
73 sql: Output.interpolate`INSERT INTO ${sink.name} SELECT * FROM ${stream.name}`,
74 });
75 return { bucket, stream, sink, pipeline };
76 });
77
78test.provider(
79 "list enumerates the deployed pipeline",
80 (stack) =>
81 Effect.gen(function* () {
82 yield* stack.destroy();
83
84 const creds = yield* r2Credentials;
85 const deployed = yield* retryAuthBlip(stack.deploy(etl(creds)));
86
87 const provider = yield* Provider.findProvider(
88 Cloudflare.Pipelines.Pipeline,
89 );
90 const all = yield* provider.list();
91
92 // Each element is the full `read` Attributes shape, usable by delete.
93 expect(
94 all.some((p) => p.pipelineId === deployed.pipeline.pipelineId),
95 ).toBe(true);
96 const found = all.find(
97 (p) => p.pipelineId === deployed.pipeline.pipelineId,
98 )!;
99 expect(found.name).toEqual(deployed.pipeline.name);
100 expect(found.sql).toEqual(deployed.pipeline.sql);
101 expect(found.accountId).toEqual(deployed.pipeline.accountId);
102
103 yield* stack.destroy();
104 }).pipe(logLevel),
105 {
106 tags: [
107 "provider:cloudflare",
108 "provider:cloudflare:pipelines",
109 "provider:cloudflare:r2",
110 "live",
111 ],
112 timeout: 600_000,
113 },
114);