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 logpush from "@distilled.cloud/cloudflare/logpush";
7import * as user from "@distilled.cloud/cloudflare/user";
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// Cloudflare's edge intermittently 403s ("Unable to authenticate request")
23// even for established tokens. The blip can hit engine-side calls mid-deploy
24// as well as the test's own out-of-band calls. Reconcile is idempotent
25// (observe→ensure→sync), so retrying the whole deploy is safe. The message
26// match (rather than a typed tag) is deliberate: this wraps whole deploys
27// whose error channel is the engine's aggregate, not a single distilled op.
28const retryAuthBlip = <A, E, R>(eff: Effect.Effect<A, E, R>) =>
29 eff.pipe(
30 Effect.retry({
31 while: (e) => String(e).includes("Unable to authenticate request"),
32 schedule: Schedule.exponential("1 second"),
33 times: 5,
34 }),
35 );
36
37const r2Credentials = Effect.gen(function* () {
38 const creds = yield* yield* CloudflareEnvironment;
39 if (creds.type !== "apiToken") {
40 return yield* Effect.die(
41 new Error(
42 "Logpush R2 test requires an apiToken profile (R2 S3 credentials are derived from the token)",
43 ),
44 );
45 }
46 const token = Redacted.value(creds.apiToken);
47 const verified = yield* user.verifyToken({}).pipe(
48 Effect.retry({
49 while: (e) => e._tag === "Forbidden",
50 schedule: Schedule.exponential("1 second"),
51 times: 5,
52 }),
53 );
54 const secretAccessKey = yield* Effect.sync(() =>
55 crypto.createHash("sha256").update(token).digest("hex"),
56 );
57 return {
58 accountId: creds.accountId,
59 accessKeyId: verified.id,
60 secretAccessKey,
61 };
62});
63
64interface R2Creds {
65 accountId: string;
66 accessKeyId: string;
67 secretAccessKey: string;
68}
69
70interface JobOpts {
71 dataset: Cloudflare.Logpush.Dataset;
72 enabled?: boolean;
73 maxUploadIntervalSeconds?: number;
74 outputOptions?: Cloudflare.Logpush.OutputOptions;
75}
76
77// One program deploying both the R2 destination bucket and the Logpush job
78// pushing into it. The job's destinationConf interpolates the bucket name so
79// the engine orders job-after-bucket on deploy (and the reverse on destroy).
80const program = (creds: R2Creds, opts: JobOpts) =>
81 Effect.gen(function* () {
82 const bucket = yield* Cloudflare.R2.Bucket("LogpushBucket", {
83 forceDestroy: true,
84 });
85 const job = yield* Cloudflare.Logpush.Job("Job", {
86 dataset: opts.dataset,
87 destinationConf: Output.interpolate`r2://${bucket.bucketName}/alchemy/{DATE}?account-id=${creds.accountId}&access-key-id=${creds.accessKeyId}&secret-access-key=${creds.secretAccessKey}`,
88 enabled: opts.enabled,
89 maxUploadIntervalSeconds: opts.maxUploadIntervalSeconds,
90 outputOptions: opts.outputOptions,
91 });
92 return { bucket, job };
93 });
94
95// `Forbidden` is in `getJobForAccount`'s typed error union (the same edge
96// 403 blip described above), so the retry predicate is fully inferred.
97const getJob = (accountId: string, jobId: number) =>
98 logpush.getJobForAccount({ accountId, jobId }).pipe(
99 Effect.retry({
100 while: (e) => e._tag === "Forbidden",
101 schedule: Schedule.exponential("500 millis"),
102 times: 8,
103 }),
104 );
105
106const waitForDelete = (accountId: string, jobId: number) =>
107 getJob(accountId, jobId).pipe(
108 Effect.flatMap((job) =>
109 job.id === jobId
110 ? Effect.fail({ _tag: "JobNotDeleted" } as const)
111 : Effect.void,
112 ),
113 Effect.catchTag("JobNotFound", () => Effect.void),
114 Effect.retry({
115 while: (e) => e._tag === "JobNotDeleted",
116 schedule: Schedule.max([
117 Schedule.exponential("500 millis"),
118 Schedule.recurs(10),
119 ]),
120 }),
121 );
122
123test.provider(
124 "create, update, and delete an account-scoped job pushing to R2",
125 (stack) =>
126 Effect.gen(function* () {
127 const { accountId } = yield* yield* CloudflareEnvironment;
128 const creds = yield* r2Credentials;
129
130 yield* stack.destroy();
131
132 // Create — engine-generated name, disabled to avoid pushing noise.
133 const initial = yield* retryAuthBlip(
134 stack.deploy(
135 program(creds, {
136 dataset: "workers_trace_events",
137 enabled: false,
138 }),
139 ),
140 );
141
142 expect(initial.job.jobId).toBeTypeOf("number");
143 expect(initial.job.accountId).toEqual(accountId);
144 expect(initial.job.zoneId).toBeUndefined();
145 expect(initial.job.dataset).toEqual("workers_trace_events");
146 expect(initial.job.enabled).toEqual(false);
147 expect(initial.job.destinationConf).toContain(
148 `r2://${initial.bucket.bucketName}/alchemy/{DATE}`,
149 );
150
151 const live = yield* getJob(accountId, initial.job.jobId);
152 expect(live.id).toEqual(initial.job.jobId);
153 expect(live.dataset).toEqual("workers_trace_events");
154 expect(live.enabled).toEqual(false);
155 expect(live.name).toEqual(initial.job.name);
156
157 // Update mutable props in place — same job id.
158 const updated = yield* retryAuthBlip(
159 stack.deploy(
160 program(creds, {
161 dataset: "workers_trace_events",
162 enabled: true,
163 maxUploadIntervalSeconds: 60,
164 outputOptions: {
165 fieldNames: ["EventTimestampMs", "Outcome", "ScriptName"],
166 timestampFormat: "rfc3339",
167 },
168 }),
169 ),
170 );
171
172 expect(updated.job.jobId).toEqual(initial.job.jobId);
173 expect(updated.job.enabled).toEqual(true);
174
175 const liveUpdated = yield* getJob(accountId, updated.job.jobId);
176 expect(liveUpdated.enabled).toEqual(true);
177 expect(liveUpdated.maxUploadIntervalSeconds).toEqual(60);
178 expect(liveUpdated.outputOptions?.fieldNames).toEqual([
179 "EventTimestampMs",
180 "Outcome",
181 "ScriptName",
182 ]);
183
184 yield* stack.destroy();
185
186 yield* waitForDelete(accountId, initial.job.jobId);
187
188 // Destroy again — delete must be idempotent (the job is already gone).
189 yield* stack.destroy();
190 }).pipe(logLevel),
191 {
192 tags: [
193 "provider:cloudflare",
194 "provider:cloudflare:logpush",
195 "provider:cloudflare:r2",
196 "live",
197 ],
198 timeout: 180_000,
199 },
200);
201
202test.provider(
203 "list enumerates the deployed Logpush job",
204 (stack) =>
205 Effect.gen(function* () {
206 const { accountId } = yield* yield* CloudflareEnvironment;
207 const creds = yield* r2Credentials;
208
209 yield* stack.destroy();
210
211 const deployed = yield* retryAuthBlip(
212 stack.deploy(
213 program(creds, {
214 dataset: "workers_trace_events",
215 enabled: false,
216 }),
217 ),
218 );
219
220 const provider = yield* Provider.findProvider(Cloudflare.Logpush.Job);
221 const all = yield* provider.list();
222
223 // The account-scoped job we just deployed is present in the
224 // exhaustively-paginated (account + per-zone fan-out) result.
225 const found = all.find((job) => job.jobId === deployed.job.jobId);
226 expect(found).toBeDefined();
227 expect(found?.accountId).toEqual(accountId);
228 expect(found?.zoneId).toBeUndefined();
229 expect(found?.dataset).toEqual("workers_trace_events");
230
231 yield* stack.destroy();
232
233 yield* waitForDelete(accountId, deployed.job.jobId);
234
235 yield* stack.destroy();
236 }).pipe(logLevel),
237 {
238 tags: [
239 "provider:cloudflare",
240 "provider:cloudflare:logpush",
241 "provider:cloudflare:r2",
242 "live",
243 ],
244 timeout: 180_000,
245 },
246);
247
248// Requires entitlement for a second account-scoped dataset. On the testing
249// account (Workers Paid) every dataset except `workers_trace_events` is
250// rejected with:
251// Forbidden: creating a new job (for <dataset> dataset) is not allowed:
252// exceeded max jobs allowed
253// (verified for audit_logs, audit_logs_v2, access_requests, gateway_dns,
254// casb_findings, dns_firewall_logs), so a dataset-change replacement cannot
255// be exercised here. Enable on an account with a second Logpush-entitled
256// dataset.
257test.provider.skip(
258 "changing the dataset triggers a replacement",
259 (stack) =>
260 Effect.gen(function* () {
261 const { accountId } = yield* yield* CloudflareEnvironment;
262 const creds = yield* r2Credentials;
263
264 yield* stack.destroy();
265
266 const initial = yield* retryAuthBlip(
267 stack.deploy(program(creds, { dataset: "workers_trace_events" })),
268 );
269 expect(initial.job.dataset).toEqual("workers_trace_events");
270
271 const replaced = yield* retryAuthBlip(
272 stack.deploy(program(creds, { dataset: "audit_logs_v2" })),
273 );
274
275 // The dataset is fixed at creation — a new physical job exists.
276 expect(replaced.job.jobId).not.toEqual(initial.job.jobId);
277 expect(replaced.job.dataset).toEqual("audit_logs_v2");
278
279 const liveReplaced = yield* getJob(accountId, replaced.job.jobId);
280 expect(liveReplaced.dataset).toEqual("audit_logs_v2");
281
282 // The old job was deleted as part of the replacement.
283 yield* waitForDelete(accountId, initial.job.jobId);
284
285 yield* stack.destroy();
286
287 yield* waitForDelete(accountId, replaced.job.jobId);
288 }).pipe(logLevel),
289 {
290 tags: [
291 "provider:cloudflare",
292 "provider:cloudflare:logpush",
293 "provider:cloudflare:r2",
294 "live",
295 ],
296 timeout: 180_000,
297 },
298);