1import * as workers from "@distilled.cloud/cloudflare/workers";
2import type * as Cause from "effect/Cause";
3import * as Effect from "effect/Effect";
4import * as Queue from "effect/Queue";
5import * as Schedule from "effect/Schedule";
6import * as Stream from "effect/Stream";
7import * as Socket from "effect/socket/Socket";
8import type { LogLine, LogsInput } from "../Provider.ts";
9import { profileCommandHint } from "../Util/interactive.ts";
10
11const DEFAULT_LOOKBACK_MS = 1 * 60 * 60 * 1000;
12
13export interface TelemetryFilter {
14 key: string;
15 operation:
16 | "eq"
17 | "neq"
18 | "includes"
19 | "not_includes"
20 | "starts_with"
21 | "gt"
22 | "gte"
23 | "lt"
24 | "lte"
25 | "in"
26 | "not_in";
27 type: "string" | "number" | "boolean";
28 value?: string | number | boolean;
29}
30
31interface TailEventMessage {
32 eventTimestamp?: number;
33 wallTime: number;
34 cpuTime: number;
35 truncated: boolean;
36 outcome: string;
37 scriptName: string;
38 exceptions: {
39 name: string;
40 message: string;
41 stack: string;
42 timestamp: string;
43 }[];
44 logs: {
45 message: string[];
46 level: string;
47 timestamp: string;
48 }[];
49 event:
50 | {
51 request: { method: string; url: string };
52 response?: { status: number };
53 }
54 | null
55 | undefined;
56}
57
58const parseEvents = (
59 response: workers.QueryObservabilityTelemetryResponse,
60): LogLine[] => {
61 const lines: LogLine[] = [];
62 if (response.events?.events) {
63 for (const event of response.events.events) {
64 const ts = new Date(event.timestamp);
65 const meta = event.metadata;
66 const msg =
67 meta.message ??
68 (meta.level === "error"
69 ? `error: ${meta.error ?? "unknown"}`
70 : `${meta.level ?? "log"}`);
71 lines.push({ timestamp: ts, message: msg });
72 }
73 }
74 return lines;
75};
76
77export const CloudflareLogs = Effect.gen(function* () {
78 const queryTelemetry = yield* workers.queryObservabilityTelemetry;
79 const createScriptTail = yield* workers.createScriptTail;
80 const deleteScriptTail = yield* workers.deleteScriptTail;
81
82 /**
83 * The telemetry query needs the `workers_observability:read` OAuth scope.
84 * A token minted by an older login (before the scope joined the
85 * defaults) keeps its original grants forever, so the query fails with a
86 * bare `Unauthorized`/`Forbidden` even though a fresh login would work —
87 * explain the fix instead of surfacing the raw tag.
88 */
89 const explainMissingObservabilityScope = <A, R>(
90 effect: Effect.Effect<A, workers.QueryObservabilityTelemetryError, R>,
91 ): Effect.Effect<A, workers.QueryObservabilityTelemetryError, R> =>
92 effect.pipe(
93 Effect.catchTag("Unauthorized", () =>
94 Effect.gen(function* () {
95 const command = yield* profileCommandHint(
96 "alchemy profile edit --reconfigure Cloudflare",
97 );
98 return yield* Effect.die(
99 new Error(
100 "Cloudflare rejected the observability telemetry query (Unauthorized). " +
101 'Your stored credentials are likely missing the "workers_observability:read" scope — ' +
102 "OAuth tokens keep the scopes they were minted with, so tokens minted by " +
103 `an older login won't have it. Run \`${command}\` to mint a token with ` +
104 "the current default scopes, or use an API token that grants Workers Observability read access.",
105 ),
106 );
107 }),
108 ),
109 );
110
111 const queryLogs = (opts: {
112 accountId: string;
113 filters: TelemetryFilter[];
114 options: LogsInput;
115 }) =>
116 Effect.gen(function* () {
117 const now = Date.now();
118 const limit = opts.options.limit ?? 100;
119
120 if (opts.options.since) {
121 const response = yield* explainMissingObservabilityScope(
122 queryTelemetry({
123 accountId: opts.accountId,
124 queryId: "events",
125 view: "events",
126 timeframe: { from: opts.options.since.getTime(), to: now },
127 limit,
128 parameters: {
129 filters: opts.filters,
130 // orderBy: { value: "timestamp", order: "desc" },
131 },
132 }),
133 );
134 return parseEvents(response);
135 }
136
137 const response = yield* explainMissingObservabilityScope(
138 queryTelemetry({
139 accountId: opts.accountId,
140 queryId: "events",
141 view: "events",
142 timeframe: { from: now - DEFAULT_LOOKBACK_MS, to: now },
143 limit,
144 parameters: {
145 filters: opts.filters,
146 },
147 }),
148 );
149
150 return parseEvents(response);
151 });
152
153 /**
154 * Open a real-time Workers tail session against `scriptName` and
155 * surface its messages as a {@link LogLine} stream. The websocket is
156 * automatically reconnected (with `Stream.repeat`) when Cloudflare
157 * closes it after its idle window.
158 */
159 const tailScript = (opts: { accountId: string; scriptName: string }) => {
160 const runTailSession = Effect.gen(function* () {
161 const { id: tailId, url } = yield* createScriptTail({
162 scriptName: opts.scriptName,
163 accountId: opts.accountId,
164 filters: [],
165 });
166
167 const socket = yield* Socket.makeWebSocket(url, {
168 protocols: ["trace-v1"],
169 });
170
171 const queue = yield* Queue.make<LogLine, Cause.Done>();
172
173 const decoder = new TextDecoder();
174 const offerTailMessage = (raw: Uint8Array) => {
175 const data: TailEventMessage = JSON.parse(decoder.decode(raw));
176 const eventTs = new Date(data.eventTimestamp ?? Date.now());
177
178 if (data.event && "request" in data.event) {
179 const reqEvent = data.event;
180 const pathname = (() => {
181 try {
182 return new URL(reqEvent.request.url).pathname;
183 } catch {
184 return reqEvent.request.url;
185 }
186 })();
187 const status = reqEvent.response?.status ?? 500;
188 Queue.offerUnsafe(queue, {
189 timestamp: eventTs,
190 message: `${reqEvent.request.method} ${pathname} > ${status} (cpu: ${Math.round(data.cpuTime)}ms, wall: ${Math.round(data.wallTime)}ms)`,
191 });
192 }
193
194 for (const log of data.logs) {
195 const msg = log.message.join(" ");
196 Queue.offerUnsafe(queue, {
197 timestamp: new Date(log.timestamp),
198 message: log.level === "log" ? msg : `${log.level}: ${msg}`,
199 });
200 }
201
202 for (const exception of data.exceptions) {
203 Queue.offerUnsafe(queue, {
204 timestamp: new Date(exception.timestamp),
205 message: `${exception.name} ${exception.message}\n${exception.stack}`,
206 });
207 }
208 };
209
210 yield* Socket.toStream(socket).pipe(
211 Stream.runForEach((raw) => Effect.sync(() => offerTailMessage(raw))),
212 Effect.catchIf(
213 (error) =>
214 Socket.isSocketError(error) &&
215 error.reason._tag === "SocketCloseError" &&
216 (error.reason.code === 1000 || error.reason.code === 1006),
217 () => Effect.void,
218 ),
219 Effect.ensuring(
220 Effect.all([
221 deleteScriptTail({
222 scriptName: opts.scriptName,
223 id: tailId,
224 accountId: opts.accountId,
225 }).pipe(Effect.ignore),
226 Queue.end(queue),
227 ]),
228 ),
229 Effect.scoped,
230 Effect.ignore,
231 Effect.forkChild(),
232 );
233
234 return Stream.fromQueue(queue);
235 });
236
237 return Stream.unwrap(runTailSession).pipe(
238 Stream.repeat(Schedule.spaced("1 second")),
239 );
240 };
241
242 const tailStream = (opts: {
243 accountId: string;
244 filters: TelemetryFilter[];
245 }) => {
246 const poll = (since: number): Stream.Stream<LogLine, any> =>
247 Stream.unwrap(
248 Effect.gen(function* () {
249 yield* Effect.sleep("2 seconds");
250 const now = Date.now();
251
252 const response = yield* explainMissingObservabilityScope(
253 queryTelemetry({
254 accountId: opts.accountId,
255 queryId: "events",
256 view: "events",
257 timeframe: { from: since, to: now },
258 limit: 100,
259 parameters: {
260 filters: opts.filters,
261 orderBy: { value: "timestamp", order: "asc" },
262 },
263 }),
264 );
265
266 const lines = parseEvents(response);
267 const nextSince =
268 lines.length > 0
269 ? Math.max(...lines.map((l) => l.timestamp.getTime())) + 1
270 : since;
271
272 return Stream.concat(Stream.fromIterable(lines), poll(nextSince));
273 }),
274 );
275
276 return poll(Date.now());
277 };
278
279 return { queryLogs, tailScript, tailStream };
280});