1import * as Cloudflare from "@/Cloudflare/index.ts";
2import * as Test from "@/Test/Alchemy";
3import { expect } from "alchemy-test";
4import * as Data from "effect/Data";
5import * as Effect from "effect/Effect";
6import { MinimumLogLevel } from "effect/References";
7import * as Schedule from "effect/Schedule";
8import * as HttpClient from "effect/http/HttpClient";
9import * as pathe from "pathe";
10
11// Mirrors the marker logged by `fixtures/tail/streaming-tail-producer.ts` on
12// every request. Duplicated (not imported) because the fixture's default
13// export must be its only export — extra named exports become workerd
14// top-level exports and fail startup validation.
15const STREAM_TAIL_MARKER = "alchemy-local-streaming-tail-marker";
16
17// `dev: true` runs local providers behind the RPC sidecar proxy by default,
18// matching the process topology of the real `alchemy dev` command.
19const { test } = Test.make({
20 providers: Cloudflare.providers(),
21 dev: true,
22});
23
24const logLevel = Effect.provideService(
25 MinimumLogLevel,
26 process.env.DEBUG ? "Debug" : "Info",
27);
28
29class WorkerNotReady extends Data.TaggedError("WorkerNotReady")<{
30 status: number;
31}> {}
32
33/** GET a route, retrying until the freshly started workerd serves a 200. */
34const getTextReady = (url: string) =>
35 Effect.gen(function* () {
36 const client = yield* HttpClient.HttpClient;
37 const res = yield* client.get(url).pipe(
38 Effect.flatMap((res) =>
39 res.status === 200
40 ? Effect.succeed(res)
41 : Effect.fail(new WorkerNotReady({ status: res.status })),
42 ),
43 Effect.retry({
44 while: (e): e is WorkerNotReady => e instanceof WorkerNotReady,
45 schedule: Schedule.max([
46 Schedule.min([
47 Schedule.exponential("500 millis"),
48 Schedule.spaced("2 seconds"),
49 ]),
50 Schedule.recurs(10),
51 ]),
52 }),
53 );
54 return yield* res.text;
55 }).pipe(Effect.orDie);
56
57/** The recorded shape of one streamed `TailEvent` in a consumer session. */
58interface RecordedTailEvent {
59 invocationId?: string;
60 event?: { type?: string; outcome?: string; message?: unknown };
61}
62
63/**
64 * Under `alchemy dev` a Worker's `streamingTailConsumers` lower into
65 * workerd's `streamingTails` service designators: each consumer script name
66 * resolves through the dev registry proxy — the same path plain tails and
67 * cross-worker service bindings take — and the consumer's `tailStream()`
68 * handler receives the invocation's `onset` while the producer is still
69 * executing, then every subsequent event ending with the terminal `outcome`,
70 * exactly like the deployed `streaming_tail_consumers` setting. The consumer
71 * records each completed session into a locally-emulated KV namespace and
72 * serves it back on `/events`, so this exercises the full local roundtrip
73 * with zero cloud calls.
74 */
75test.provider(
76 "local producer streams events to its local streaming tail consumer",
77 (stack) =>
78 Effect.gen(function* () {
79 yield* stack.destroy();
80
81 const deployed = yield* stack.deploy(
82 Effect.gen(function* () {
83 const events = yield* Cloudflare.KV.Namespace("StreamTailEvents");
84 const consumer = yield* Cloudflare.Worker("StreamingTailConsumer", {
85 main: pathe.resolve(
86 import.meta.dirname,
87 "fixtures/tail/streaming-tail-consumer.ts",
88 ),
89 env: { EVENTS: events },
90 });
91 const producer = yield* Cloudflare.Worker("StreamingTailProducer", {
92 main: pathe.resolve(
93 import.meta.dirname,
94 "fixtures/tail/streaming-tail-producer.ts",
95 ),
96 streamingTailConsumers: [consumer],
97 });
98 return { events, consumer, producer };
99 }),
100 );
101
102 // Dev markers: the KV namespace is locally emulated (no cloud call
103 // ran), and both workers serve from the local dev proxy.
104 expect(deployed.events.namespaceId).toMatch(/^dev:/);
105 expect(deployed.consumer.url).toMatch(/^http:\/\/localhost:\d+$/);
106 expect(deployed.producer.url).toMatch(/^http:\/\/localhost:\d+$/);
107
108 // Attribute parity with the live provider: the consumer is recorded
109 // by script name.
110 expect(deployed.producer.streamingTailConsumers).toEqual([
111 { service: deployed.consumer.workerName },
112 ]);
113 expect(deployed.producer.tailConsumers).toBeUndefined();
114
115 // Readiness probes for both freshly started workerds.
116 expect(yield* getTextReady(deployed.consumer.url!)).toBe(
117 "streaming-tail-consumer-ok",
118 );
119 expect(yield* getTextReady(deployed.producer.url!)).toBe(
120 "streaming-producer-ok",
121 );
122
123 // Drive the producer and poll the consumer's read route until a full
124 // session carrying the producer's log marker arrives. The producer's
125 // registry proxy may pick the consumer up asynchronously — each poll
126 // re-invokes the producer so early dropped sessions don't strand the
127 // test.
128 const client = yield* HttpClient.HttpClient;
129 const sessions = yield* Effect.gen(function* () {
130 yield* client.get(deployed.producer.url!).pipe(
131 Effect.flatMap((res) => res.text),
132 Effect.orDie,
133 );
134 const res = yield* client
135 .get(`${deployed.consumer.url}/events`)
136 .pipe(Effect.orDie);
137 const body = (yield* res.json.pipe(Effect.orDie)) as {
138 sessions?: unknown;
139 };
140 return Array.isArray(body.sessions) ? (body.sessions as string[]) : [];
141 }).pipe(
142 Effect.repeat({
143 schedule: Schedule.spaced("1 second"),
144 until: (sessions): boolean =>
145 sessions.some((session) => session.includes(STREAM_TAIL_MARKER)),
146 times: 20,
147 }),
148 );
149
150 const session = sessions.find((s) => s.includes(STREAM_TAIL_MARKER));
151 expect(session).toBeDefined();
152 // The recorded session is the JSON-serialized event stream: the onset
153 // handed to `tailStream()`, the console.log marker, and the terminal
154 // outcome — all from one producer invocation.
155 const events = JSON.parse(session!) as RecordedTailEvent[];
156 const types = events.map((tailEvent) => tailEvent.event?.type);
157 expect(types).toContain("onset");
158 expect(types).toContain("outcome");
159 const log = events.find((tailEvent) => tailEvent.event?.type === "log");
160 expect(log).toBeDefined();
161 expect(log?.event?.message).toEqual([STREAM_TAIL_MARKER]);
162 const onset = events.find(
163 (tailEvent) => tailEvent.event?.type === "onset",
164 );
165 // Every streamed event belongs to the onset's invocation.
166 for (const tailEvent of events) {
167 expect(tailEvent.invocationId).toBe(onset?.invocationId);
168 }
169 const outcome = events.find(
170 (tailEvent) => tailEvent.event?.type === "outcome",
171 );
172 expect(outcome?.event?.outcome).toBe("ok");
173
174 yield* stack.destroy();
175 }).pipe(logLevel),
176 {
177 tags: [
178 "provider:cloudflare",
179 "provider:cloudflare:kv",
180 "provider:cloudflare:worker",
181 "local",
182 ],
183 timeout: 180_000,
184 },
185);