1import * as Cloudflare from "@/Cloudflare";
2import { CloudflareEnvironment } from "@/Cloudflare/CloudflareEnvironment";
3import * as Provider from "@/Provider";
4import * as Test from "@/Test/Alchemy";
5import * as user from "@distilled.cloud/cloudflare/user";
6import { expect } from "alchemy-test";
7import * as Effect from "effect/Effect";
8import * as Redacted from "effect/Redacted";
9import { MinimumLogLevel } from "effect/References";
10import * as Schedule from "effect/Schedule";
11import crypto from "node:crypto";
12
13const { test } = Test.make({ providers: Cloudflare.providers() });
14
15const logLevel = Effect.provideService(
16 MinimumLogLevel,
17 process.env.DEBUG ? "Debug" : "Info",
18);
19
20// The scoped API token the harness mints propagates eventually-
21// consistently across Cloudflare's edge — ride out auth blips.
22const retryAuthBlip = <A, E, R>(eff: Effect.Effect<A, E, R>) =>
23 eff.pipe(
24 Effect.retry({
25 while: (e) => String(e).includes("Unable to authenticate request"),
26 schedule: Schedule.exponential("1 second"),
27 times: 5,
28 }),
29 );
30
31// R2 S3-compatible credentials are derived from the active API token:
32// the access key id is the token id and the secret is the SHA-256 hex
33// digest of the token value.
34const r2Credentials = Effect.gen(function* () {
35 const creds = yield* yield* CloudflareEnvironment;
36 if (creds.type !== "apiToken") {
37 return yield* Effect.die(
38 new Error(
39 "Pipelines sink test requires an apiToken profile (R2 S3 credentials are derived from the token)",
40 ),
41 );
42 }
43 const token = Redacted.value(creds.apiToken);
44 const verified = yield* retryAuthBlip(
45 user.verifyToken({}).pipe(
46 Effect.retry({
47 while: (e) => e._tag === "Forbidden",
48 schedule: Schedule.exponential("500 millis"),
49 times: 8,
50 }),
51 ),
52 );
53 const secretAccessKey = yield* Effect.sync(() =>
54 crypto.createHash("sha256").update(token).digest("hex"),
55 );
56 return {
57 accessKeyId: Redacted.make(verified.id),
58 secretAccessKey: Redacted.make(secretAccessKey),
59 };
60});
61
62test.provider(
63 "list enumerates the deployed sink",
64 (stack) =>
65 Effect.gen(function* () {
66 yield* stack.destroy();
67
68 const creds = yield* r2Credentials;
69
70 const deployed = yield* retryAuthBlip(
71 stack.deploy(
72 Effect.gen(function* () {
73 const bucket = yield* Cloudflare.R2.Bucket("SinkBucket", {
74 forceDestroy: true,
75 });
76 return yield* Cloudflare.Pipelines.Sink("ListSink", {
77 type: "r2",
78 config: {
79 bucket: bucket.bucketName,
80 credentials: creds,
81 rollingPolicy: { intervalSeconds: 10 },
82 },
83 });
84 }),
85 ),
86 );
87
88 expect(deployed.sinkId).toBeTruthy();
89
90 // Account collection: list() exhaustively paginates every sink in
91 // the account and hydrates each into the read Attributes shape.
92 const provider = yield* Provider.findProvider(Cloudflare.Pipelines.Sink);
93 const all = yield* provider.list();
94
95 const match = all.find((s) => s.sinkId === deployed.sinkId);
96 expect(match).toBeDefined();
97 expect(match?.name).toEqual(deployed.name);
98 expect(match?.type).toEqual("r2");
99 expect(match?.accountId).toEqual(deployed.accountId);
100
101 yield* stack.destroy();
102 }).pipe(logLevel),
103 {
104 tags: [
105 "provider:cloudflare",
106 "provider:cloudflare:pipelines",
107 "provider:cloudflare:r2",
108 "live",
109 ],
110 timeout: 300_000,
111 },
112);