1import { CloudflareEnvironment } from "@/Cloudflare/CloudflareEnvironment";
2import * as Cloudflare from "@/Cloudflare/index.ts";
3import * as Test from "@/Test/Alchemy";
4import * as kv from "@distilled.cloud/cloudflare/kv";
5import * as workers from "@distilled.cloud/cloudflare/workers";
6import { expect } from "alchemy-test";
7import * as Effect from "effect/Effect";
8import { MinimumLogLevel } from "effect/References";
9import * as Schedule from "effect/Schedule";
10import * as Stream from "effect/Stream";
11import * as pathe from "pathe";
12import { expectUrlContains } from "../Utils/Http.ts";
13import { waitForWorkerToBeDeleted } from "../Utils/Worker.ts";
14
15const { test } = Test.make({ providers: Cloudflare.providers() });
16
17const logLevel = Effect.provideService(
18 MinimumLogLevel,
19 process.env.DEBUG ? "Debug" : "Info",
20);
21
22/**
23 * Producer: logs on every request so each invocation produces trace events
24 * for its tail consumers.
25 */
26const producerScript = `export default {
27 fetch() {
28 console.log("alchemy-tail-marker");
29 return new Response("producer-ok");
30 },
31};`;
32
33test.provider(
34 "worker with tailConsumers attaches the consumer and delivers traces",
35 (stack) =>
36 Effect.gen(function* () {
37 const { accountId } = yield* yield* CloudflareEnvironment;
38
39 yield* stack.destroy();
40
41 // The consumer is a file-based async Worker exporting a `tail()`
42 // handler that persists each trace batch into a KV namespace keyed by
43 // the producing script's name — KV makes delivery observable
44 // out-of-band, independent of which isolate received the tail event.
45 const deployStack = (tails: "attached" | "detached") =>
46 Effect.gen(function* () {
47 const events = yield* Cloudflare.KV.Namespace("TailEvents");
48 const consumer = yield* Cloudflare.Worker("TailConsumer", {
49 main: pathe.resolve(
50 import.meta.dirname,
51 "fixtures/tail/tail-consumer.ts",
52 ),
53 env: { EVENTS: events },
54 });
55 const producer = yield* Cloudflare.Worker("TailProducer", {
56 script: producerScript,
57 tailConsumers: tails === "attached" ? [consumer] : [],
58 });
59 return { events, consumer, producer };
60 });
61
62 const v1 = yield* stack.deploy(deployStack("attached"));
63
64 // The attribute records the consumer by deployed script name.
65 expect(v1.producer.tailConsumers).toEqual([
66 { service: v1.consumer.workerName },
67 ]);
68
69 // Out-of-band: the producer's script settings carry the tail consumer.
70 const settings = yield* workers.getScriptScriptAndVersionSetting({
71 accountId,
72 scriptName: v1.producer.workerName,
73 });
74 expect(settings.tailConsumers?.map((c) => c.service)).toEqual([
75 v1.consumer.workerName,
76 ]);
77
78 // Invoke the producer (retrying through workers.dev propagation),
79 // then poll KV until a tail batch keyed by the producer's script name
80 // arrives. Each poll re-invokes the producer so delivery that begins
81 // slightly after the first request still surfaces.
82 yield* expectUrlContains(v1.producer.url!, "producer-ok", {
83 label: "tail producer",
84 });
85 const eventKeys = yield* Effect.gen(function* () {
86 yield* expectUrlContains(v1.producer.url!, "producer-ok", {
87 timeout: "15 seconds",
88 label: "tail producer (poll)",
89 });
90 const keys = yield* kv.listNamespaceKeys({
91 accountId,
92 namespaceId: v1.events.namespaceId,
93 prefix: `evt:${v1.producer.workerName}:`,
94 });
95 return keys.result.map((k) => k.name);
96 }).pipe(
97 Effect.repeat({
98 schedule: Schedule.spaced("5 seconds"),
99 until: (keys): boolean => keys.length > 0,
100 times: 18,
101 }),
102 );
103 expect(eventKeys.length).toBeGreaterThan(0);
104
105 // The recorded batch references the producer script.
106 const batch = yield* kv
107 .getNamespaceValue({
108 accountId,
109 namespaceId: v1.events.namespaceId,
110 keyName: eventKeys[0],
111 })
112 .pipe(
113 Effect.flatMap((res) =>
114 Effect.tryPromise(() =>
115 new Response(
116 Stream.toReadableStream(res.body) as BodyInit,
117 ).text(),
118 ),
119 ),
120 );
121 expect(batch).toContain(v1.producer.workerName);
122
123 // Detaching every consumer is an in-place update, never a replace.
124 const v2 = yield* stack.deploy(deployStack("detached"));
125 expect(v2.producer.workerId).toEqual(v1.producer.workerId);
126 expect(v2.producer.workerName).toEqual(v1.producer.workerName);
127 expect(v2.producer.tailConsumers ?? []).toEqual([]);
128 const detached = yield* workers.getScriptScriptAndVersionSetting({
129 accountId,
130 scriptName: v2.producer.workerName,
131 });
132 expect(detached.tailConsumers ?? []).toEqual([]);
133
134 yield* stack.destroy();
135 yield* waitForWorkerToBeDeleted(v1.producer.workerName, accountId);
136 yield* waitForWorkerToBeDeleted(v1.consumer.workerName, accountId);
137 }).pipe(logLevel),
138 {
139 tags: [
140 "provider:cloudflare",
141 "provider:cloudflare:kv",
142 "provider:cloudflare:worker",
143 "live",
144 ],
145 timeout: 240_000,
146 },
147);