1import * as workers from "@distilled.cloud/cloudflare/workers";
2import * as Effect from "effect/Effect";
3import * as Predicate from "effect/Predicate";
4import * as Schedule from "effect/Schedule";
5import * as Stream from "effect/Stream";
6
7import { Unowned } from "../../AdoptPolicy.ts";
8import { isResolved } from "../../Diff.ts";
9import { createPhysicalName } from "../../PhysicalName.ts";
10import * as Provider from "../../Provider.ts";
11import { Resource } from "../../Resource.ts";
12import { CloudflareEnvironment } from "../CloudflareEnvironment.ts";
13import type { Providers } from "../Providers.ts";
14
15const TypeId = "Cloudflare.Workers.ObservabilityDestination" as const;
16type TypeId = typeof TypeId;
17
18/**
19 * The Workers Logs dataset exported by an observability destination.
20 */
21export type ObservabilityDataset =
22 | "opentelemetry-traces"
23 | "opentelemetry-logs"
24 | "opentelemetry-metrics";
25
26export interface ObservabilityDestinationProps {
27 /**
28 * Human readable destination name. Cloudflare derives the destination's
29 * stable `slug` from it and the name cannot be changed afterwards —
30 * updating this property triggers a replacement. If omitted, a unique
31 * name is generated from the app, stage, and logical ID.
32 * @default ${app}-${stage}-${id}
33 */
34 name?: string;
35 /**
36 * HTTPS endpoint the OTLP payloads are pushed to (e.g. an OTLP/HTTP
37 * collector). Mutable — updated in place.
38 *
39 * Cloudflare verifies the endpoint with a preflight request on create
40 * (unless {@link skipPreflightCheck} is set) and on **every** update, so
41 * the endpoint must accept a `POST` with a `2xx` response for updates to
42 * succeed.
43 */
44 url: string;
45 /**
46 * Extra HTTP headers sent with each push (e.g. authentication tokens).
47 * Cloudflare always adds a `content-type: application/json` header of
48 * its own. Mutable — updated in place.
49 * @default {}
50 */
51 headers?: Record<string, string>;
52 /**
53 * Which Workers Logs dataset to export. Cannot be changed after
54 * creation — updating this property triggers a replacement.
55 */
56 logpushDataset: ObservabilityDataset;
57 /**
58 * Whether the destination actively exports data. Mutable — updated in
59 * place.
60 * @default true
61 */
62 enabled?: boolean;
63 /**
64 * Skip the create-time preflight request against {@link url}. Useful
65 * when the collector rejects empty probe payloads. Create-only — the
66 * update API always performs the preflight check.
67 * @default false
68 */
69 skipPreflightCheck?: boolean;
70}
71
72export interface ObservabilityDestinationAttributes {
73 /**
74 * Cloudflare-assigned stable identifier, derived from the name at
75 * creation time.
76 */
77 slug: string;
78 /**
79 * The Cloudflare account the destination belongs to.
80 */
81 accountId: string;
82 /**
83 * Human readable destination name.
84 */
85 name: string;
86 /**
87 * Whether the destination actively exports data.
88 */
89 enabled: boolean;
90 /**
91 * HTTPS endpoint the OTLP payloads are pushed to.
92 */
93 url: string;
94 /**
95 * The Workers Logs dataset this destination exports.
96 */
97 logpushDataset: ObservabilityDataset;
98 /**
99 * The underlying Logpush destination string (the URL with the
100 * configured headers encoded as query parameters).
101 */
102 destinationConf: string;
103 /**
104 * Names of the Worker scripts currently opted in to this destination.
105 */
106 scripts: string[];
107}
108
109export type ObservabilityDestination = Resource<
110 TypeId,
111 ObservabilityDestinationProps,
112 ObservabilityDestinationAttributes,
113 never,
114 Providers
115>;
116
117/**
118 * A Workers observability destination — an account-level OTLP export of
119 * Workers Logs telemetry (traces, logs, or metrics) pushed to an external
120 * HTTPS collector via Logpush.
121 *
122 * A destination is identified by its Cloudflare-derived `slug` (stable,
123 * computed from the name at creation). The endpoint URL, headers, and
124 * enabled flag are mutable in place; `name` and `logpushDataset` force a
125 * replacement.
126 *
127 * Cloudflare preflights the endpoint with a `POST` on create (skippable
128 * via `skipPreflightCheck`) and on every in-place update (not skippable),
129 * so the collector must answer `2xx` for updates to converge.
130 *
131 * Safety: destinations carry no ownership markers and Cloudflare enforces
132 * one destination per name. When there is no prior state, `read` scans the
133 * account for a destination with the same name and reports it as
134 * `Unowned`, so the engine refuses to take it over unless `--adopt` (or
135 * `adopt(true)`) is set.
136 * ### Exporting Workers traces
137 * **Example:** Push traces to an OTLP collector
138 * ```typescript
139 * const traces = yield* Cloudflare.Workers.ObservabilityDestination("Traces", {
140 * url: "https://otel.example.com/v1/traces",
141 * headers: { authorization: secret },
142 * logpushDataset: "opentelemetry-traces",
143 * });
144 * ```
145 *
146 * ### Exporting Workers logs
147 * **Example:** Push logs, skipping the create-time preflight
148 * ```typescript
149 * const logs = yield* Cloudflare.Workers.ObservabilityDestination("Logs", {
150 * name: "my-app-logs",
151 * url: "https://collector.example.com/v1/logs",
152 * logpushDataset: "opentelemetry-logs",
153 * skipPreflightCheck: true,
154 * });
155 * ```
156 *
157 * ### Pausing an export
158 * **Example:** Disable the destination without deleting it
159 * ```typescript
160 * yield* Cloudflare.Workers.ObservabilityDestination("Logs", {
161 * name: "my-app-logs",
162 * url: "https://collector.example.com/v1/logs",
163 * logpushDataset: "opentelemetry-logs",
164 * enabled: false,
165 * });
166 * ```
167 *
168 * @resource
169 * @product Workers
170 * @category Workers & Compute
171 */
172export const ObservabilityDestination =
173 Resource<ObservabilityDestination>(TypeId);
174
175/**
176 * Returns true if the given value is an ObservabilityDestination resource.
177 */
178export const isObservabilityDestination = (
179 value: unknown,
180): value is ObservabilityDestination =>
181 Predicate.hasProperty(value, "Type") && value.Type === TypeId;
182
183export const ObservabilityDestinationProvider = () =>
184 Provider.succeed(ObservabilityDestination, {
185 stables: ["slug", "accountId", "name", "logpushDataset"],
186
187 diff: Effect.fn(function* ({ id, olds, news, output }) {
188 if (!isResolved(news)) return undefined;
189 const { accountId } = yield* yield* CloudflareEnvironment;
190 if ((output?.accountId ?? accountId) !== accountId) {
191 return { action: "replace" } as const;
192 }
193 // The slug is derived from the name at creation and the update API
194 // cannot rename — a name change is a replacement.
195 const oldName =
196 output?.name ?? olds?.name ?? (yield* createDestinationName(id));
197 // Auto-generated names are engine-owned: the deployed name stays
198 // authoritative even if the generator would name this id differently
199 // today. Only an explicit user-provided name can force a replace.
200 const newName = news.name ?? oldName;
201 if (oldName !== newName) {
202 return { action: "replace" } as const;
203 }
204 // The update API cannot change the dataset.
205 const oldDataset = output?.logpushDataset ?? olds?.logpushDataset;
206 if (oldDataset !== undefined && oldDataset !== news.logpushDataset) {
207 return { action: "replace" } as const;
208 }
209 return undefined;
210 }),
211
212 read: Effect.fn(function* ({ id, output, olds }) {
213 const { accountId } = yield* yield* CloudflareEnvironment;
214 const acct = output?.accountId ?? accountId;
215
216 // Owned path: refresh by our persisted slug.
217 if (output?.slug) {
218 const observed = yield* findBySlug(acct, output.slug);
219 return observed ? toAttributes(observed, acct) : undefined;
220 }
221 // Adoption path: no state of our own, but Cloudflare enforces one
222 // destination per name, so an exact name match is the same logical
223 // destination. Destinations carry no ownership markers, so brand it
224 // `Unowned` and let the engine gate takeover behind the adopt
225 // policy.
226 const name = olds?.name ?? (yield* createDestinationName(id));
227 const observed = yield* findByName(acct, name);
228 return observed ? Unowned(toAttributes(observed, acct)) : undefined;
229 }),
230
231 reconcile: Effect.fn(function* ({ id, news, output }) {
232 const { accountId } = yield* yield* CloudflareEnvironment;
233 // Prefer the deployed name: regenerating would target a different
234 // resource if the generator's output for this id ever drifts.
235 const name =
236 news.name ?? output?.name ?? (yield* createDestinationName(id));
237 // Inputs have been resolved to concrete strings by Plan.
238 const url = news.url as string;
239 const headers = (news.headers ?? {}) as Record<string, string>;
240 const enabled = news.enabled ?? true;
241
242 // 1. Observe — the slug cached on `output` is a hint, not a
243 // guarantee; fall back to the name (Cloudflare enforces name
244 // uniqueness, and ownership of a name match has already been
245 // verified upstream by `read` + the adopt policy).
246 let observed = output?.slug
247 ? yield* findBySlug(accountId, output.slug)
248 : undefined;
249 if (!observed) {
250 observed = yield* findByName(accountId, name);
251 }
252
253 // 2. Ensure. The API reports both a duplicate name and a failed
254 // endpoint preflight as the same opaque 400 — re-observe by name
255 // to tell a creation race (converge via sync below) from a
256 // genuine failure (re-fail).
257 if (!observed) {
258 const created = yield* workers
259 .createObservabilityDestination({
260 accountId,
261 name,
262 enabled,
263 skipPreflightCheck: news.skipPreflightCheck,
264 configuration: {
265 type: "logpush",
266 url,
267 headers,
268 logpushDataset: news.logpushDataset,
269 },
270 })
271 .pipe(
272 Effect.catchTag(
273 "ObservabilityDestinationCreateFailed",
274 (originalError) =>
275 Effect.gen(function* () {
276 const match = yield* findByName(accountId, name);
277 if (!match) {
278 return yield* Effect.fail(originalError);
279 }
280 return match;
281 }),
282 ),
283 );
284 observed = isObservedDestination(created)
285 ? created
286 : yield* freshObserved(accountId, created.slug);
287 }
288
289 // 3. Sync — diff observed cloud state against desired and PATCH only
290 // on drift. Skipping the no-op matters doubly here: every PATCH
291 // re-runs the endpoint preflight.
292 if (
293 observed.enabled !== enabled ||
294 observed.configuration.url !== url ||
295 !sameHeaders(observed.configuration.headers, headers)
296 ) {
297 const patched = yield* workers
298 .patchObservabilityDestination({
299 accountId,
300 slug: observed.slug,
301 enabled,
302 configuration: { type: "logpush", url, headers },
303 })
304 .pipe(
305 // Every PATCH re-runs the endpoint preflight POST from
306 // Cloudflare's edge. A freshly deployed endpoint (e.g. a
307 // workers.dev sink) can transiently fail that probe while the
308 // route propagates, so ride out preflight failures briefly.
309 Effect.retry({
310 while: (e) =>
311 e._tag === "ObservabilityDestinationPreflightFailed",
312 schedule: Schedule.min([
313 Schedule.exponential("1 second"),
314 Schedule.spaced("5 seconds"),
315 ]),
316 times: 8,
317 }),
318 );
319 observed = yield* freshObserved(accountId, patched.slug);
320 }
321
322 // 4. Return.
323 return toAttributes(observed, accountId);
324 }),
325
326 delete: Effect.fn(function* ({ output }) {
327 yield* workers
328 .deleteObservabilityDestination({
329 accountId: output.accountId,
330 slug: output.slug,
331 })
332 .pipe(
333 Effect.catchTag(
334 "ObservabilityDestinationNotFound",
335 () => Effect.void,
336 ),
337 );
338 }),
339
340 // Account collection. Observability destinations are account-scoped and
341 // the distilled list op paginates on `result`; exhaustively collect every
342 // page and hydrate each item into the exact `read` Attributes shape via
343 // `toAttributes`. The destination's stored credentials/headers token are
344 // write-only on the API, so — like `read` — they never appear here.
345 list: Effect.fn(function* () {
346 const { accountId } = yield* yield* CloudflareEnvironment;
347 return yield* workers.listObservabilityDestinations
348 .items({ accountId })
349 .pipe(
350 Stream.runCollect,
351 Effect.map((chunk) =>
352 Array.from(chunk).map((observed) =>
353 toAttributes(observed, accountId),
354 ),
355 ),
356 );
357 }),
358 });
359
360type ObservedDestination =
361 workers.ListObservabilityDestinationsResponse["result"][number];
362
363const isObservedDestination = (
364 value: ObservedDestination | workers.CreateObservabilityDestinationResponse,
365): value is ObservedDestination =>
366 Predicate.hasProperty(value.configuration, "headers");
367
368const createDestinationName = (id: string) =>
369 createPhysicalName({ id, lowercase: true });
370
371/**
372 * Locate a destination by its stable slug. There is no get-by-slug API,
373 * so observation goes through the list endpoint.
374 */
375const findBySlug = (accountId: string, slug: string) =>
376 workers.listObservabilityDestinations.items({ accountId }).pipe(
377 Stream.filter((d) => d.slug === slug),
378 Stream.runCollect,
379 Effect.map((chunk) => Array.from(chunk).at(0)),
380 );
381
382/**
383 * Locate a destination by exact name. Cloudflare enforces name
384 * uniqueness, so a match identifies the destination.
385 */
386const findByName = (accountId: string, name: string) =>
387 workers.listObservabilityDestinations.items({ accountId }).pipe(
388 Stream.filter((d) => d.name === name),
389 Stream.runCollect,
390 Effect.map((chunk) => Array.from(chunk).at(0)),
391 );
392
393/**
394 * Re-read a destination through the list endpoint after a write. The
395 * create/patch responses omit the stored headers, so the list item is the
396 * canonical observed shape.
397 */
398const freshObserved = (accountId: string, slug: string) =>
399 findBySlug(accountId, slug).pipe(
400 Effect.flatMap((observed) =>
401 observed
402 ? Effect.succeed(observed)
403 : // The destination cannot vanish between the write that just
404 // succeeded and this read except via an out-of-band delete racing
405 // us — surface that as the typed not-found the API would raise.
406 Effect.fail(
407 new workers.ObservabilityDestinationNotFound({
408 code: 0,
409 message: `observability destination "${slug}" disappeared during reconcile`,
410 }),
411 ),
412 ),
413 );
414
415/**
416 * Compare observed headers against the desired set. Cloudflare always
417 * injects a `content-type: application/json` header of its own — ignore
418 * it unless the user explicitly pinned one, so an empty desired set is
419 * not perpetually "dirty".
420 */
421const sameHeaders = (
422 observed: Record<string, unknown>,
423 desired: Record<string, string>,
424) => {
425 const normalized = Object.fromEntries(
426 Object.entries(observed)
427 .filter(([key]) => key !== "content-type" || "content-type" in desired)
428 .map(([key, value]) => [key, String(value)]),
429 );
430 const keys = Object.keys(normalized);
431 return (
432 keys.length === Object.keys(desired).length &&
433 keys.every((key) => normalized[key] === desired[key])
434 );
435};
436
437const toAttributes = (
438 observed: ObservedDestination,
439 accountId: string,
440): ObservabilityDestinationAttributes => ({
441 slug: observed.slug,
442 accountId,
443 name: observed.name,
444 enabled: observed.enabled,
445 url: observed.configuration.url,
446 // Distilled widens generated string enums to open unions (`string & {}`).
447 logpushDataset: observed.configuration.logpushDataset as ObservabilityDataset,
448 destinationConf: observed.configuration.destinationConf,
449 scripts: [...observed.scripts],
450});