1import { adopt } from "@/AdoptPolicy";
2import * as Cloudflare from "@/Cloudflare";
3import { CloudflareEnvironment } from "@/Cloudflare/CloudflareEnvironment";
4import * as Provider from "@/Provider";
5import * as Test from "@/Test/Alchemy";
6import * as workers from "@distilled.cloud/cloudflare/workers";
7import { expect } from "alchemy-test";
8import * as Effect from "effect/Effect";
9import { MinimumLogLevel } from "effect/References";
10import * as Schedule from "effect/Schedule";
11import * as Stream from "effect/Stream";
12import * as HttpClient from "effect/http/HttpClient";
13import * as pathe from "pathe";
14
15const { test } = Test.make({ providers: Cloudflare.providers() });
16
17const logLevel = Effect.provideService(
18 MinimumLogLevel,
19 process.env.DEBUG ? "Debug" : "Info",
20);
21
22const main = pathe.resolve(
23 import.meta.dirname,
24 "fixtures",
25 "observability-sink-worker.ts",
26);
27
28// Deterministic per-test destination names. Cloudflare enforces one
29// destination per name account-wide, so each test owns disjoint names and
30// reuses them on every run (never derive physical names from
31// Date.now()/random).
32const NAME_DEFAULT_URL = "https://example.com";
33const NAME_UPDATE = "alchemy-obsdest-update";
34const NAME_REPLACE_V1 = "alchemy-obsdest-replace-v1";
35const NAME_REPLACE_V2 = "alchemy-obsdest-replace-v2";
36const NAME_LIST = "alchemy-obsdest-list";
37
38// The scoped API token the test harness mints propagates eventually-
39// consistently across Cloudflare's edge — ride out 403 blips (`Forbidden`,
40// declared in the distilled error union) on the test's own out-of-band
41// verification calls.
42const forbiddenRetrySchedule = Schedule.exponential("500 millis");
43
44const listDestinations = (accountId: string) =>
45 workers.listObservabilityDestinations.items({ accountId }).pipe(
46 Stream.runCollect,
47 Effect.map((chunk) => Array.from(chunk)),
48 Effect.retry({
49 while: (e) => e._tag === "Forbidden",
50 schedule: forbiddenRetrySchedule,
51 times: 8,
52 }),
53 );
54
55const findByName = (accountId: string, name: string) =>
56 listDestinations(accountId).pipe(
57 Effect.map((ds) => ds.find((d) => d.name === name)),
58 );
59
60// Delete every destination with the given name — used to purge leftovers
61// from interrupted runs so tests start from a clean slate (Cloudflare
62// enforces name uniqueness, so a leaked destination would conflict).
63const purgeByName = (accountId: string, name: string) =>
64 listDestinations(accountId).pipe(
65 Effect.flatMap(
66 Effect.forEach((d) =>
67 d.name === name
68 ? workers
69 .deleteObservabilityDestination({ accountId, slug: d.slug })
70 .pipe(Effect.catch(() => Effect.void))
71 : Effect.void,
72 ),
73 ),
74 );
75
76const expectGone = (accountId: string, name: string) =>
77 findByName(accountId, name).pipe(
78 Effect.flatMap((found) =>
79 found
80 ? Effect.fail({ _tag: "DestinationNotDeleted" } as const)
81 : Effect.void,
82 ),
83 Effect.retry({
84 while: (e) => e._tag === "DestinationNotDeleted",
85 schedule: Schedule.exponential("500 millis"),
86 times: 10,
87 }),
88 );
89
90test.provider(
91 "create and delete a destination with default name",
92 (stack) =>
93 Effect.gen(function* () {
94 const { accountId } = yield* yield* CloudflareEnvironment;
95
96 yield* stack.destroy();
97
98 const dest = yield* stack.deploy(
99 Effect.gen(function* () {
100 return yield* Cloudflare.Workers.ObservabilityDestination(
101 "DefaultDest",
102 {
103 url: NAME_DEFAULT_URL,
104 logpushDataset: "opentelemetry-logs",
105 // example.com rejects POSTs, so skip the create-time probe —
106 // this test never updates, and updates are what re-preflight.
107 skipPreflightCheck: true,
108 },
109 ).pipe(adopt(true));
110 }),
111 );
112
113 expect(dest.slug).toBeDefined();
114 expect(dest.accountId).toEqual(accountId);
115 expect(dest.url).toEqual(NAME_DEFAULT_URL);
116 expect(dest.logpushDataset).toEqual("opentelemetry-logs");
117 expect(dest.enabled).toBe(true);
118 expect(dest.scripts).toEqual([]);
119
120 const live = yield* findByName(accountId, dest.name);
121 expect(live?.slug).toEqual(dest.slug);
122 expect(live?.configuration.url).toEqual(NAME_DEFAULT_URL);
123 expect(live?.configuration.logpushDataset).toEqual("opentelemetry-logs");
124
125 yield* stack.destroy();
126
127 yield* expectGone(accountId, dest.name);
128 }).pipe(logLevel),
129 {
130 tags: ["provider:cloudflare", "provider:cloudflare:worker", "live"],
131 timeout: 300_000,
132 },
133);
134
135test.provider(
136 "update url, headers, and enabled in place (same slug)",
137 (stack) =>
138 Effect.gen(function* () {
139 const { accountId } = yield* yield* CloudflareEnvironment;
140
141 yield* stack.destroy();
142 yield* purgeByName(accountId, NAME_UPDATE);
143
144 // The destination's endpoint is a deployed Worker that answers 200
145 // to everything — Cloudflare re-runs the endpoint preflight (a POST)
146 // on every in-place update, so the sink must be live.
147 const initial = yield* stack.deploy(
148 Effect.gen(function* () {
149 const worker = yield* Cloudflare.Worker("ObsSinkWorker", {
150 main,
151 compatibility: { date: "2024-01-01" },
152 });
153 const dest = yield* Cloudflare.Workers.ObservabilityDestination(
154 "Dest",
155 {
156 name: NAME_UPDATE,
157 url: worker.url.as<string>(),
158 logpushDataset: "opentelemetry-traces",
159 // Fresh workers.dev URLs take a few seconds to start serving —
160 // skip the create-time probe; the update below exercises the
161 // real preflight.
162 skipPreflightCheck: true,
163 },
164 ).pipe(adopt(true));
165 return { worker, dest };
166 }),
167 );
168
169 expect(initial.dest.name).toEqual(NAME_UPDATE);
170 expect(initial.dest.url).toEqual(initial.worker.url);
171 expect(initial.dest.enabled).toBe(true);
172
173 // Warm the sink through edge propagation so the update's preflight
174 // POST lands on a serving workers.dev URL. Fresh workers.dev hosts
175 // answer 404 until the subdomain propagates, so poll until 200.
176 const client = yield* HttpClient.HttpClient;
177 const warm = yield* client.get(initial.worker.url!).pipe(
178 Effect.retry({
179 schedule: Schedule.exponential("500 millis"),
180 times: 5,
181 }),
182 Effect.repeat({
183 schedule: Schedule.spaced("2 seconds"),
184 until: (res) => res.status === 200,
185 times: 60,
186 }),
187 );
188 expect(warm.status).toBe(200);
189
190 const updated = yield* stack.deploy(
191 Effect.gen(function* () {
192 const worker = yield* Cloudflare.Worker("ObsSinkWorker", {
193 main,
194 compatibility: { date: "2024-01-01" },
195 });
196 const dest = yield* Cloudflare.Workers.ObservabilityDestination(
197 "Dest",
198 {
199 name: NAME_UPDATE,
200 url: worker.url.as<string>(),
201 headers: { "x-alchemy-test": "1" },
202 logpushDataset: "opentelemetry-traces",
203 enabled: false,
204 },
205 ).pipe(adopt(true));
206 return { worker, dest };
207 }),
208 );
209
210 // Same destination mutated in place — not a replacement.
211 expect(updated.dest.slug).toEqual(initial.dest.slug);
212 expect(updated.dest.url).toEqual(updated.worker.url);
213 expect(updated.dest.enabled).toBe(false);
214
215 const live = yield* findByName(accountId, NAME_UPDATE);
216 expect(live?.slug).toEqual(initial.dest.slug);
217 expect(live?.enabled).toBe(false);
218 expect(live?.configuration.url).toEqual(updated.worker.url);
219 expect(live?.configuration.headers["x-alchemy-test"]).toEqual("1");
220
221 // Redeploying identical props is a no-op (and must not re-preflight).
222 const noop = yield* stack.deploy(
223 Effect.gen(function* () {
224 const worker = yield* Cloudflare.Worker("ObsSinkWorker", {
225 main,
226 compatibility: { date: "2024-01-01" },
227 });
228 const dest = yield* Cloudflare.Workers.ObservabilityDestination(
229 "Dest",
230 {
231 name: NAME_UPDATE,
232 url: worker.url.as<string>(),
233 headers: { "x-alchemy-test": "1" },
234 logpushDataset: "opentelemetry-traces",
235 enabled: false,
236 },
237 ).pipe(adopt(true));
238 return { worker, dest };
239 }),
240 );
241 expect(noop.dest.slug).toEqual(initial.dest.slug);
242
243 yield* stack.destroy();
244
245 yield* expectGone(accountId, NAME_UPDATE);
246 }).pipe(logLevel),
247 {
248 tags: ["provider:cloudflare", "provider:cloudflare:worker", "live"],
249 timeout: 300_000,
250 },
251);
252
253test.provider(
254 "replacement on name change (new slug, old destination removed)",
255 (stack) =>
256 Effect.gen(function* () {
257 const { accountId } = yield* yield* CloudflareEnvironment;
258
259 yield* stack.destroy();
260 yield* purgeByName(accountId, NAME_REPLACE_V1);
261 yield* purgeByName(accountId, NAME_REPLACE_V2);
262
263 const v1 = yield* stack.deploy(
264 Effect.gen(function* () {
265 return yield* Cloudflare.Workers.ObservabilityDestination(
266 "ReplaceDest",
267 {
268 name: NAME_REPLACE_V1,
269 url: NAME_DEFAULT_URL,
270 logpushDataset: "opentelemetry-logs",
271 skipPreflightCheck: true,
272 },
273 ).pipe(adopt(true));
274 }),
275 );
276 expect(v1.name).toEqual(NAME_REPLACE_V1);
277
278 // The slug is derived from the name and the update API cannot
279 // rename — a name change replaces the physical destination.
280 const v2 = yield* stack.deploy(
281 Effect.gen(function* () {
282 return yield* Cloudflare.Workers.ObservabilityDestination(
283 "ReplaceDest",
284 {
285 name: NAME_REPLACE_V2,
286 url: NAME_DEFAULT_URL,
287 logpushDataset: "opentelemetry-logs",
288 skipPreflightCheck: true,
289 },
290 ).pipe(adopt(true));
291 }),
292 );
293
294 expect(v2.name).toEqual(NAME_REPLACE_V2);
295 expect(v2.slug).not.toEqual(v1.slug);
296
297 const liveV2 = yield* findByName(accountId, NAME_REPLACE_V2);
298 expect(liveV2?.slug).toEqual(v2.slug);
299 // The replaced destination is cleaned up.
300 yield* expectGone(accountId, NAME_REPLACE_V1);
301
302 yield* stack.destroy();
303
304 yield* expectGone(accountId, NAME_REPLACE_V2);
305 }).pipe(logLevel),
306 {
307 tags: ["provider:cloudflare", "provider:cloudflare:worker", "live"],
308 timeout: 300_000,
309 },
310);
311
312test.provider(
313 "list enumerates the deployed destination",
314 (stack) =>
315 Effect.gen(function* () {
316 const { accountId } = yield* yield* CloudflareEnvironment;
317
318 yield* stack.destroy();
319 yield* purgeByName(accountId, NAME_LIST);
320
321 const dest = yield* stack.deploy(
322 Effect.gen(function* () {
323 return yield* Cloudflare.Workers.ObservabilityDestination(
324 "ListDest",
325 {
326 name: NAME_LIST,
327 url: NAME_DEFAULT_URL,
328 logpushDataset: "opentelemetry-logs",
329 // example.com rejects POSTs, so skip the create-time probe.
330 skipPreflightCheck: true,
331 },
332 ).pipe(adopt(true));
333 }),
334 );
335
336 // Typed provider lookup — `findProvider` infers the element type as the
337 // resource's Attributes, so `list()` is fully typed (no `any`).
338 const provider = yield* Provider.findProvider(
339 Cloudflare.Workers.ObservabilityDestination,
340 );
341 const all = yield* provider.list();
342
343 // The exhaustively-paginated result contains our deployed destination,
344 // hydrated into the exact `read` Attributes shape.
345 const found = all.find((d) => d.slug === dest.slug);
346 expect(found).toBeDefined();
347 expect(found?.accountId).toEqual(accountId);
348 expect(found?.name).toEqual(NAME_LIST);
349 expect(found?.url).toEqual(NAME_DEFAULT_URL);
350 expect(found?.logpushDataset).toEqual("opentelemetry-logs");
351
352 yield* stack.destroy();
353
354 yield* expectGone(accountId, NAME_LIST);
355 }).pipe(logLevel),
356 {
357 tags: ["provider:cloudflare", "provider:cloudflare:worker", "live"],
358 timeout: 300_000,
359 },
360);