1/**
2 * Async (non-Effect) streaming Tail Worker fixture for
3 * `StreamingTailConsumers.test.ts` and `StreamingTailConsumers.local.test.ts`.
4 *
5 * Contract delta from the plain `tail-consumer.ts` fixture: Cloudflare
6 * invokes the exported `tailStream()` handler with the invocation's `onset`
7 * event while the producer is still executing, and the returned function
8 * receives every subsequent event of the session (`log`, `spanOpen`, ...)
9 * ending with the terminal `outcome`. On `outcome`, the whole recorded
10 * session (onset included) is persisted to the bound KV namespace so both
11 * the live test (out-of-band distilled KV reads) and the local test (the
12 * `/events` fetch route below) can assert delivery.
13 *
14 * NOTE: the default export must be this module's ONLY export — extra named
15 * exports become workerd top-level exports and fail startup validation.
16 */
17
18interface TailEventsKV {
19 put(key: string, value: string): Promise<void>;
20 get(key: string): Promise<string | null>;
21 list(options?: { prefix?: string }): Promise<{ keys: { name: string }[] }>;
22}
23
24interface StreamTailEvent {
25 invocationId?: string;
26 timestamp?: string;
27 sequence?: number;
28 event?: {
29 type?: string;
30 outcome?: string;
31 level?: string;
32 message?: unknown;
33 info?: { type?: string; url?: string; method?: string };
34 };
35}
36
37export default {
38 async fetch(
39 request: Request,
40 env: { EVENTS: TailEventsKV },
41 ): Promise<Response> {
42 const url = new URL(request.url);
43 if (url.pathname === "/events") {
44 const list = await env.EVENTS.list({ prefix: "evt:" });
45 const sessions = await Promise.all(
46 list.keys.map((key) => env.EVENTS.get(key.name)),
47 );
48 return Response.json({
49 keys: list.keys.map((key) => key.name),
50 sessions: sessions.filter(
51 (session): session is string => session !== null,
52 ),
53 });
54 }
55 return new Response("streaming-tail-consumer-ok");
56 },
57 tailStream(
58 onset: StreamTailEvent,
59 env: { EVENTS: TailEventsKV },
60 ): (tailEvent: StreamTailEvent) => Promise<void> {
61 const events: StreamTailEvent[] = [onset];
62 return async (tailEvent: StreamTailEvent): Promise<void> => {
63 events.push(tailEvent);
64 // `outcome` terminates the session — flush the full recording. The
65 // key suffix is payload data (not a resource name), so wall-clock
66 // uniqueness is fine here.
67 if (tailEvent.event?.type === "outcome") {
68 const id = onset.invocationId ?? `${Date.now()}`;
69 await env.EVENTS.put(`evt:${id}:${Date.now()}`, JSON.stringify(events));
70 }
71 };
72 },
73};