1import * as Cloudflare from "@/Cloudflare";
2import { CloudflareEnvironment } from "@/Cloudflare/CloudflareEnvironment";
3import * as Output from "@/Output";
4import * as Test from "@/Test/Alchemy";
5import * as pipelines from "@distilled.cloud/cloudflare/pipelines";
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
21// The scoped API token the test harness mints propagates eventually-
22// consistently across Cloudflare's edge — ride out 403 blips
23// (`Forbidden`, declared in the distilled error unions) on the test's
24// own out-of-band verification calls.
25const forbiddenBlips = {
26 while: (e: { _tag: string }) => e._tag === "Forbidden",
27 schedule: Schedule.exponential("500 millis"),
28 times: 8,
29} as const;
30
31// The same blip can hit engine-side calls mid-deploy. Reconcile is
32// idempotent (observe→ensure→sync), so retrying the whole deploy is safe.
33const retryAuthBlip = <A, E, R>(eff: Effect.Effect<A, E, R>) =>
34 eff.pipe(
35 Effect.retry({
36 while: (e) => String(e).includes("Unable to authenticate request"),
37 schedule: Schedule.exponential("1 second"),
38 times: 5,
39 }),
40 );
41
42// R2 S3-compatible credentials are derived from the active API token:
43// the access key id is the token id and the secret is the SHA-256 hex
44// digest of the token value.
45const r2Credentials = Effect.gen(function* () {
46 const creds = yield* yield* CloudflareEnvironment;
47 if (creds.type !== "apiToken") {
48 return yield* Effect.die(
49 new Error(
50 "Pipelines sink test requires an apiToken profile (R2 S3 credentials are derived from the token)",
51 ),
52 );
53 }
54 const token = Redacted.value(creds.apiToken);
55 const verified = yield* retryAuthBlip(user.verifyToken({}));
56 const secretAccessKey = yield* Effect.sync(() =>
57 crypto.createHash("sha256").update(token).digest("hex"),
58 );
59 return {
60 accessKeyId: Redacted.make(verified.id),
61 secretAccessKey: Redacted.make(secretAccessKey),
62 };
63});
64
65const getStream = (accountId: string, streamId: string) =>
66 pipelines
67 .getStream({ accountId, streamId })
68 .pipe(Effect.retry(forbiddenBlips));
69
70const getSink = (accountId: string, sinkId: string) =>
71 pipelines.getSink({ accountId, sinkId }).pipe(Effect.retry(forbiddenBlips));
72
73const getPipeline = (accountId: string, pipelineId: string) =>
74 pipelines
75 .getV1Pipeline({ accountId, pipelineId })
76 .pipe(Effect.retry(forbiddenBlips));
77
78const expectStreamGone = (accountId: string, streamId: string) =>
79 getStream(accountId, streamId).pipe(
80 Effect.flatMap(() => Effect.fail({ _tag: "StreamNotDeleted" } as const)),
81 Effect.catchTag("StreamNotFound", () => Effect.void),
82 Effect.retry({
83 while: (e) => e._tag === "StreamNotDeleted",
84 schedule: Schedule.max([
85 Schedule.exponential("500 millis"),
86 Schedule.recurs(10),
87 ]),
88 }),
89 );
90
91const expectSinkGone = (accountId: string, sinkId: string) =>
92 getSink(accountId, sinkId).pipe(
93 Effect.flatMap(() => Effect.fail({ _tag: "SinkNotDeleted" } as const)),
94 Effect.catchTag("SinkNotFound", () => Effect.void),
95 Effect.retry({
96 while: (e) => e._tag === "SinkNotDeleted",
97 schedule: Schedule.max([
98 Schedule.exponential("500 millis"),
99 Schedule.recurs(10),
100 ]),
101 }),
102 );
103
104const expectPipelineGone = (accountId: string, pipelineId: string) =>
105 getPipeline(accountId, pipelineId).pipe(
106 Effect.flatMap(() => Effect.fail({ _tag: "PipelineNotDeleted" } as const)),
107 Effect.catchTag("PipelineNotExists", () => Effect.void),
108 Effect.retry({
109 while: (e) => e._tag === "PipelineNotDeleted",
110 schedule: Schedule.max([
111 Schedule.exponential("500 millis"),
112 Schedule.recurs(10),
113 ]),
114 }),
115 );
116
117test.provider(
118 "stream: create with defaults, patch http in place, replace on schema change",
119 (stack) =>
120 Effect.gen(function* () {
121 const { accountId } = yield* yield* CloudflareEnvironment;
122
123 yield* stack.destroy();
124
125 // Create — engine-generated name, default http/workerBinding.
126 const initial = yield* retryAuthBlip(
127 stack.deploy(Cloudflare.Pipelines.Stream("Stream", {})),
128 );
129
130 expect(initial.streamId).toBeTruthy();
131 expect(initial.accountId).toEqual(accountId);
132 expect(initial.httpEnabled).toEqual(true);
133 expect(initial.httpAuthentication).toEqual(false);
134 expect(initial.workerBindingEnabled).toEqual(true);
135 expect(initial.endpoint).toBeTruthy();
136
137 const live = yield* getStream(accountId, initial.streamId);
138 expect(live.id).toEqual(initial.streamId);
139 expect(live.name).toEqual(initial.name);
140
141 // Patch http in place — same streamId.
142 const updated = yield* retryAuthBlip(
143 stack.deploy(
144 Cloudflare.Pipelines.Stream("Stream", {
145 http: {
146 enabled: true,
147 authentication: true,
148 cors: { origins: ["https://example.com"] },
149 },
150 }),
151 ),
152 );
153
154 expect(updated.streamId).toEqual(initial.streamId);
155 expect(updated.httpAuthentication).toEqual(true);
156 expect(updated.corsOrigins).toEqual(["https://example.com"]);
157
158 const liveUpdated = yield* getStream(accountId, updated.streamId);
159 expect(liveUpdated.http.authentication).toEqual(true);
160 expect(liveUpdated.http.cors?.origins).toEqual(["https://example.com"]);
161
162 // The schema is immutable — declaring one triggers a replacement.
163 const replaced = yield* retryAuthBlip(
164 stack.deploy(
165 Cloudflare.Pipelines.Stream("Stream", {
166 schema: {
167 fields: [
168 { type: "string", name: "url", required: true },
169 { type: "timestamp", name: "ts", unit: "millisecond" },
170 ],
171 },
172 }),
173 ),
174 );
175
176 expect(replaced.streamId).not.toEqual(initial.streamId);
177 yield* expectStreamGone(accountId, initial.streamId);
178
179 yield* stack.destroy();
180 yield* expectStreamGone(accountId, replaced.streamId);
181
182 // Destroy again — delete must be idempotent.
183 yield* stack.destroy();
184 }).pipe(logLevel),
185 {
186 tags: ["provider:cloudflare", "provider:cloudflare:pipelines", "live"],
187 timeout: 300_000,
188 },
189);
190
191interface EtlOpts {
192 path?: string;
193 where?: string;
194}
195
196// One program deploying the whole Pipelines topology: an R2 destination
197// bucket, the sink writing to it, an ingest stream, and the SQL pipeline
198// connecting them. The pipeline SQL interpolates the stream/sink names so
199// the engine orders pipeline-after-stream/sink on deploy (and the reverse
200// on destroy).
201const etl = (
202 creds: {
203 accessKeyId: Redacted.Redacted<string>;
204 secretAccessKey: Redacted.Redacted<string>;
205 },
206 opts: EtlOpts = {},
207) =>
208 Effect.gen(function* () {
209 const bucket = yield* Cloudflare.R2.Bucket("SinkBucket", {
210 forceDestroy: true,
211 });
212 const stream = yield* Cloudflare.Pipelines.Stream("Stream", {});
213 const sink = yield* Cloudflare.Pipelines.Sink("Sink", {
214 type: "r2",
215 config: {
216 bucket: bucket.bucketName,
217 credentials: creds,
218 path: opts.path,
219 rollingPolicy: { intervalSeconds: 10 },
220 },
221 });
222 const pipeline = yield* Cloudflare.Pipelines.Pipeline("Etl", {
223 sql: Output.interpolate`INSERT INTO ${sink.name} SELECT * FROM ${stream.name}${opts.where ? ` WHERE ${opts.where}` : ""}`,
224 });
225 return { bucket, stream, sink, pipeline };
226 });
227
228test.provider(
229 "end-to-end: stream → sql pipeline → r2 sink, replacement on sink change",
230 (stack) =>
231 Effect.gen(function* () {
232 const { accountId } = yield* yield* CloudflareEnvironment;
233 const creds = yield* r2Credentials;
234
235 yield* stack.destroy();
236
237 const initial = yield* retryAuthBlip(stack.deploy(etl(creds)));
238
239 expect(initial.sink.sinkId).toBeTruthy();
240 expect(initial.sink.type).toEqual("r2");
241 expect(initial.sink.bucket).toEqual(initial.bucket.bucketName);
242 expect(initial.pipeline.pipelineId).toBeTruthy();
243 expect(initial.pipeline.sql).toContain(
244 `INSERT INTO ${initial.sink.name} SELECT * FROM ${initial.stream.name}`,
245 );
246
247 const liveSink = yield* getSink(accountId, initial.sink.sinkId);
248 expect(liveSink.name).toEqual(initial.sink.name);
249 expect(liveSink.config?.bucket).toEqual(initial.bucket.bucketName);
250
251 const livePipeline = yield* getPipeline(
252 accountId,
253 initial.pipeline.pipelineId,
254 );
255 expect(livePipeline.name).toEqual(initial.pipeline.name);
256 expect(livePipeline.sql).toEqual(initial.pipeline.sql);
257 expect(livePipeline.status).toBeTruthy();
258
259 // Redeploying identical props is a no-op.
260 const noop = yield* retryAuthBlip(stack.deploy(etl(creds)));
261 expect(noop.sink.sinkId).toEqual(initial.sink.sinkId);
262 expect(noop.pipeline.pipelineId).toEqual(initial.pipeline.pipelineId);
263 expect(noop.stream.streamId).toEqual(initial.stream.streamId);
264
265 // Sinks have no update API — changing the path replaces the sink,
266 // and the pipeline's SQL change replaces the pipeline with it. The
267 // stream is untouched.
268 const replaced = yield* retryAuthBlip(
269 stack.deploy(etl(creds, { path: "ingest", where: "1 = 1" })),
270 );
271
272 expect(replaced.sink.sinkId).not.toEqual(initial.sink.sinkId);
273 expect(replaced.sink.path).toEqual("ingest");
274 expect(replaced.pipeline.pipelineId).not.toEqual(
275 initial.pipeline.pipelineId,
276 );
277 expect(replaced.pipeline.sql).toContain("WHERE 1 = 1");
278 expect(replaced.stream.streamId).toEqual(initial.stream.streamId);
279
280 // The old sink and pipeline were deleted as part of the replacement.
281 yield* expectPipelineGone(accountId, initial.pipeline.pipelineId);
282 yield* expectSinkGone(accountId, initial.sink.sinkId);
283
284 // Destroy tears down pipeline → sink/stream → bucket in dependency
285 // order.
286 yield* stack.destroy();
287
288 yield* expectPipelineGone(accountId, replaced.pipeline.pipelineId);
289 yield* expectSinkGone(accountId, replaced.sink.sinkId);
290 yield* expectStreamGone(accountId, replaced.stream.streamId);
291
292 // Destroy again — deletes must be idempotent.
293 yield* stack.destroy();
294 }).pipe(logLevel),
295 {
296 tags: [
297 "provider:cloudflare",
298 "provider:cloudflare:pipelines",
299 "provider:cloudflare:r2",
300 "live",
301 ],
302 timeout: 600_000,
303 },
304);