1import * as logpush from "@distilled.cloud/cloudflare/logpush";
2import * as Effect from "effect/Effect";
3import * as Predicate from "effect/Predicate";
4import * as Stream from "effect/Stream";
5
6import { Unowned } from "../../AdoptPolicy.ts";
7import { createPhysicalName } from "../../PhysicalName.ts";
8import * as Provider from "../../Provider.ts";
9import { Resource } from "../../Resource.ts";
10import { CloudflareEnvironment } from "../CloudflareEnvironment.ts";
11import type { Providers } from "../Providers.ts";
12import { listAllZones } from "../Zone/lookup.ts";
13
14const TypeId = "Cloudflare.Logpush.Job" as const;
15type TypeId = typeof TypeId;
16
17/**
18 * Name of the dataset a Logpush job pushes. The available datasets depend
19 * on the job's scope (account vs zone) and the account's plan — e.g.
20 * `workers_trace_events` and `audit_logs` are account-scoped, while
21 * `http_requests` and `firewall_events` are zone-scoped (Enterprise).
22 */
23export type Dataset = Exclude<
24 NonNullable<logpush.CreateJobForAccountRequest["dataset"]>,
25 null
26>;
27
28/**
29 * Structured output configuration for a Logpush job — the replacement for
30 * the deprecated `logpull_options` string.
31 */
32export interface OutputOptions {
33 /**
34 * String prepended to each batch (e.g. `[` to build a JSON array).
35 */
36 batchPrefix?: string;
37 /**
38 * String appended to each batch (e.g. `]`).
39 */
40 batchSuffix?: string;
41 /**
42 * Mitigation flag for CVE-2021-44228 — when `true`, verbatim
43 * `${` sequences are replaced with `x{`.
44 */
45 "cve-2021-44228"?: boolean;
46 /**
47 * Delimiter between fields in `csv` output.
48 */
49 fieldDelimiter?: string;
50 /**
51 * Log fields to include in the output. Available fields per dataset are
52 * listed on the Cloudflare developer docs.
53 */
54 fieldNames?: string[];
55 /**
56 * Whether to merge sub-request data into the parent request record
57 * (only meaningful for some datasets).
58 */
59 mergeSubrequests?: boolean;
60 /**
61 * Output format.
62 * @default "ndjson"
63 */
64 outputType?: "ndjson" | "csv";
65 /**
66 * Delimiter inserted between records.
67 */
68 recordDelimiter?: string;
69 /**
70 * String prepended to each record.
71 */
72 recordPrefix?: string;
73 /**
74 * String appended to each record.
75 */
76 recordSuffix?: string;
77 /**
78 * Go-template string used to render each record (mutually exclusive
79 * with prefix/suffix/delimiter options).
80 */
81 recordTemplate?: string;
82 /**
83 * Floating point fraction (0.0–1.0) of records to include.
84 * @default 1
85 */
86 sampleRate?: number;
87 /**
88 * Timestamp rendering format.
89 * @default "unixnano"
90 */
91 timestampFormat?: "unixnano" | "unix" | "rfc3339" | "rfc3339ms" | "rfc3339ns";
92}
93
94export interface JobProps {
95 /**
96 * Zone the job is scoped to. When omitted, the job is account-scoped
97 * (using the account from the active Cloudflare credentials).
98 *
99 * Stable — moving a job between scopes triggers a replacement.
100 */
101 zoneId?: string;
102 /**
103 * Name of the dataset to push (e.g. `workers_trace_events`,
104 * `audit_logs`, `http_requests`).
105 *
106 * Stable — the dataset is fixed at creation, so changing it triggers
107 * a replacement.
108 */
109 dataset: Dataset;
110 /**
111 * Destination URI, including any credentials the destination needs —
112 * e.g. `r2://bucket/{DATE}?account-id=…&access-key-id=…&secret-access-key=…`
113 * for R2, or an `s3://`, `gs://`, `https://` endpoint.
114 *
115 * Cloudflare validates the destination synchronously on create/update by
116 * writing a test object, so the destination (e.g. the R2 bucket) must
117 * already exist.
118 *
119 * Mutable — but Cloudflare may reject changing the destination
120 * *provider/domain* of an existing job; in that case change a
121 * replacement-triggering prop instead.
122 */
123 destinationConf: string;
124 /**
125 * Optional human readable job name (not unique). If omitted, a unique
126 * name will be generated.
127 * @default ${app}-${stage}-${id}
128 */
129 name?: string;
130 /**
131 * Whether the job actively pushes logs.
132 * @default false
133 */
134 enabled?: boolean;
135 /**
136 * Filter expression (JSON-encoded) selecting which events to push.
137 * See https://developers.cloudflare.com/logs/reference/filters/.
138 */
139 filter?: string;
140 /**
141 * Differentiates Logpush (`""`) from Edge Log Delivery (`"edge"`) jobs.
142 *
143 * Stable — changing the kind triggers a replacement.
144 * @default ""
145 */
146 kind?: "" | "edge";
147 /**
148 * Structured output configuration (fields, format, delimiters, sampling).
149 */
150 outputOptions?: OutputOptions;
151 /**
152 * Maximum uncompressed file size of a batch, in bytes. Between 5 MB and
153 * 1 GB, or `0` to disable the limit.
154 * @default 0
155 */
156 maxUploadBytes?: number;
157 /**
158 * Maximum interval in seconds between log batches. Between 30 and 300,
159 * or `0` to disable the limit.
160 * @default 30
161 */
162 maxUploadIntervalSeconds?: number;
163 /**
164 * Maximum number of log lines per batch. Between 1,000 and 1,000,000,
165 * or `0` to disable the limit.
166 * @default 100000
167 */
168 maxUploadRecords?: number;
169 /**
170 * Ownership challenge token proving destination ownership — required
171 * for destinations like S3/GCS/Azure. R2 destinations authenticated via
172 * credentials in `destinationConf` do not need it.
173 *
174 * Write-only: Cloudflare never echoes it back, so it does not
175 * participate in drift detection.
176 */
177 ownershipChallenge?: string;
178}
179
180export interface JobAttributes {
181 /** Cloudflare-assigned numeric job id. */
182 jobId: number;
183 /** Account that owns the job. */
184 accountId: string;
185 /** Zone the job is scoped to, or `undefined` for account-scoped jobs. */
186 zoneId: string | undefined;
187 /** Job name. */
188 name: string;
189 /** Dataset the job pushes. */
190 dataset: Dataset;
191 /**
192 * Destination URI. Note Cloudflare redacts embedded secrets (e.g. the
193 * R2 `secret-access-key`) when echoing this back.
194 */
195 destinationConf: string;
196 /** Whether the job is enabled. */
197 enabled: boolean;
198 /** Job kind (`""` for Logpush, `"edge"` for Edge Log Delivery). */
199 kind: string;
200 /** Last failure message, if the job is currently failing. */
201 errorMessage: string | undefined;
202 /** End of the last successfully-pushed log range. */
203 lastComplete: string | undefined;
204 /** Time of the last push failure. */
205 lastError: string | undefined;
206}
207
208export type Job = Resource<TypeId, JobProps, JobAttributes, never, Providers>;
209
210/**
211 * A Cloudflare Logpush job that pushes batches of logs (HTTP requests,
212 * Workers trace events, audit logs, Zero Trust datasets, …) to a
213 * destination such as R2, S3, GCS, or an HTTP endpoint.
214 *
215 * Jobs can be account-scoped (default) or zone-scoped (pass `zoneId`).
216 * The dataset, kind, and scope are fixed at creation — changing any of
217 * them triggers a replacement; everything else updates in place.
218 * ### Pushing Workers trace events to R2
219 * **Example:** Account-scoped job writing to an R2 bucket
220 * The R2 destination authenticates with S3-compatible credentials embedded
221 * in the destination URI, so no ownership challenge is required.
222 * ```typescript
223 * const bucket = yield* Cloudflare.R2.Bucket("logs", {});
224 *
225 * const job = yield* Cloudflare.Logpush.Job("worker-logs", {
226 * dataset: "workers_trace_events",
227 * destinationConf: Output.interpolate`r2://${bucket.bucketName}/{DATE}?account-id=${accountId}&access-key-id=${r2AccessKeyId}&secret-access-key=${r2SecretAccessKey}`,
228 * enabled: true,
229 * });
230 * ```
231 *
232 * ### Zone-scoped jobs
233 * **Example:** HTTP requests dataset on a zone (Enterprise)
234 * ```typescript
235 * const job = yield* Cloudflare.Logpush.Job("http-logs", {
236 * zoneId: zone.zoneId,
237 * dataset: "http_requests",
238 * destinationConf: "s3://my-bucket/logs?region=us-east-1",
239 * ownershipChallenge: "00000000000000000000",
240 * });
241 * ```
242 *
243 * ### Output configuration
244 * **Example:** Selecting fields and batching limits
245 * ```typescript
246 * const job = yield* Cloudflare.Logpush.Job("worker-logs", {
247 * dataset: "workers_trace_events",
248 * destinationConf,
249 * outputOptions: {
250 * fieldNames: ["EventTimestampMs", "Outcome", "ScriptName"],
251 * outputType: "ndjson",
252 * timestampFormat: "rfc3339",
253 * },
254 * maxUploadIntervalSeconds: 60,
255 * });
256 * ```
257 *
258 * @see https://developers.cloudflare.com/logs/logpush/
259 *
260 * @resource
261 * @product Logpush
262 * @category Observability & Analytics
263 */
264export const Job = Resource<Job>(TypeId);
265
266/**
267 * Returns true if the given value is a Job resource.
268 */
269export const isJob = (value: unknown): value is Job =>
270 Predicate.hasProperty(value, "Type") && value.Type === TypeId;
271
272export const JobProvider = () =>
273 Provider.succeed(Job, {
274 stables: ["jobId", "accountId", "zoneId", "dataset", "kind"],
275
276 diff: Effect.fn(function* ({ olds = {}, news, output }) {
277 const { accountId } = yield* yield* CloudflareEnvironment;
278 const o = olds as JobProps;
279 const n = news as JobProps;
280 if ((output?.accountId ?? accountId) !== accountId) {
281 return { action: "replace" } as const;
282 }
283 if (o.dataset !== undefined && o.dataset !== n.dataset) {
284 return { action: "replace" } as const;
285 }
286 if (o.dataset !== undefined && (o.kind ?? "") !== (n.kind ?? "")) {
287 return { action: "replace" } as const;
288 }
289 // zoneId is string; only comparable when both sides are
290 // concrete strings (or one side is absent => scope change).
291 const oZone = typeof o.zoneId === "string" ? o.zoneId : undefined;
292 const nZone = typeof n.zoneId === "string" ? n.zoneId : undefined;
293 if (
294 o.dataset !== undefined &&
295 (typeof o.zoneId === "string" || o.zoneId === undefined) &&
296 (typeof n.zoneId === "string" || n.zoneId === undefined) &&
297 oZone !== nZone
298 ) {
299 return { action: "replace" } as const;
300 }
301 }),
302
303 read: Effect.fn(function* ({ id, output, olds }) {
304 const { accountId } = yield* yield* CloudflareEnvironment;
305 const scope: Scope = {
306 accountId: output?.accountId ?? accountId,
307 zoneId: output?.zoneId ?? (olds?.zoneId as string | undefined),
308 };
309 // Owned path: refresh by the persisted job id.
310 if (output?.jobId !== undefined) {
311 const observed = yield* observeById(scope, output.jobId);
312 const attrs = toAttributes(observed, scope);
313 if (attrs) return attrs;
314 }
315 // Cold/adoption path: no persisted id — find the job by name. The
316 // engine-generated physical name embeds the instance id, so a match
317 // on a generated name is proof we created it. A match on a
318 // user-provided name is not — report it `Unowned` so the engine
319 // gates takeover behind the adopt policy.
320 const name = yield* createJobName(id, olds?.name);
321 const match = yield* findByName(scope, name, olds?.dataset);
322 const attrs = toAttributes(match, scope);
323 if (attrs) {
324 return olds?.name !== undefined ? Unowned(attrs) : attrs;
325 }
326 return undefined;
327 }),
328
329 // Logpush jobs are hybrid-scoped: account-scoped jobs live under the
330 // account, zone-scoped jobs under each zone. Enumerate both — the account
331 // collection plus a fan-out over every zone — and hydrate each into the
332 // same Attributes shape `read` returns.
333 list: Effect.fn(function* () {
334 const { accountId } = yield* yield* CloudflareEnvironment;
335
336 const accountRows = yield* logpush.listJobsForAccount
337 .pages({ accountId })
338 .pipe(
339 Stream.runCollect,
340 Effect.map((chunk) =>
341 Array.from(chunk).flatMap((page) =>
342 (page.result ?? [])
343 .filter((job): job is NonNullable<typeof job> => job != null)
344 .map((job) =>
345 toAttributes(job, { accountId, zoneId: undefined }),
346 )
347 .filter((attrs): attrs is JobAttributes => attrs !== undefined),
348 ),
349 ),
350 );
351
352 const zones = yield* listAllZones(accountId);
353 const zoneRows = yield* Effect.forEach(
354 zones,
355 (zone) =>
356 logpush.listJobsForZone.pages({ zoneId: zone.id }).pipe(
357 Stream.runCollect,
358 Effect.map((chunk) =>
359 Array.from(chunk).flatMap((page) =>
360 (page.result ?? [])
361 .filter((job): job is NonNullable<typeof job> => job != null)
362 .map((job) =>
363 toAttributes(job, { accountId, zoneId: zone.id }),
364 )
365 .filter(
366 (attrs): attrs is JobAttributes => attrs !== undefined,
367 ),
368 ),
369 ),
370 // Best-effort account-wide fan-out: a zone the token can't read
371 // for Logpush (plan-gated route, missing permission, or a code-
372 // 10000 auth blip) must be skipped, not fail the whole
373 // enumeration. Drop only that zone and keep the rest.
374 Effect.catchTag(
375 ["InvalidRoute", "Unauthorized", "Forbidden", "NotFound"],
376 () => Effect.succeed([]),
377 ),
378 ),
379 { concurrency: 10 },
380 );
381
382 return [...accountRows, ...zoneRows.flat()];
383 }),
384
385 reconcile: Effect.fn(function* ({ id, news, olds, output }) {
386 const { accountId } = yield* yield* CloudflareEnvironment;
387 const scope: Scope = {
388 accountId: output?.accountId ?? accountId,
389 zoneId: (news.zoneId as string | undefined) ?? undefined,
390 };
391 const name = yield* createJobName(id, news.name);
392 const destinationConf = news.destinationConf as string;
393 const body = buildMutableBody(news, name, destinationConf);
394
395 // 1. Observe — by cached id first, falling back to a list scan by
396 // name so we recover from lost state or out-of-band deletes.
397 let observed: logpush.GetJobResponse | undefined;
398 if (output?.jobId !== undefined) {
399 observed = yield* observeById(scope, output.jobId);
400 }
401 let justCreated = false;
402 if (!observed) {
403 observed = yield* findByName(scope, name, news.dataset);
404 }
405
406 // 2. Ensure — create when missing. Logpush job names are not
407 // unique, so there is no AlreadyExists race to tolerate; create
408 // failures are real (e.g. invalid destination).
409 if (!observed) {
410 observed = yield* createJob(scope, news.dataset, body);
411 justCreated = true;
412 }
413
414 const jobId = observed.id;
415 if (jobId === undefined || jobId === null) {
416 return yield* Effect.fail(
417 new Error("Cloudflare did not return an id for the Logpush job"),
418 );
419 }
420
421 // 3. Sync — Cloudflare's update endpoint is PUT-style; resend the
422 // full desired body when anything differs. `destination_conf`
423 // and `filter` are compared against `olds` because Cloudflare
424 // redacts embedded secrets (destination) or omits the field
425 // entirely (filter) in get/list responses.
426 if (!justCreated && needsUpdate(body, news, olds, observed)) {
427 observed = yield* updateJob(scope, jobId, body);
428 }
429
430 const attrs = toAttributes({ ...observed, id: jobId }, scope);
431 if (!attrs) {
432 return yield* Effect.fail(
433 new Error(
434 "Cloudflare returned a Logpush job without id/dataset/destination",
435 ),
436 );
437 }
438 // Prefer the desired destination — the observed echo redacts secrets.
439 return { ...attrs, destinationConf, name: attrs.name || name };
440 }),
441
442 delete: Effect.fn(function* ({ output }) {
443 const scope: Scope = {
444 accountId: output.accountId,
445 zoneId: output.zoneId,
446 };
447 yield* (
448 scope.zoneId !== undefined
449 ? logpush.deleteJobForZone({
450 zoneId: scope.zoneId,
451 jobId: output.jobId,
452 })
453 : logpush.deleteJobForAccount({
454 accountId: scope.accountId,
455 jobId: output.jobId,
456 })
457 ).pipe(Effect.catchTag("JobNotFound", () => Effect.void));
458 }),
459 });
460
461interface Scope {
462 accountId: string;
463 zoneId: string | undefined;
464}
465
466const createJobName = (id: string, name: string | undefined) =>
467 Effect.gen(function* () {
468 if (name) return name;
469 return yield* createPhysicalName({ id, lowercase: true });
470 });
471
472const observeById = (scope: Scope, jobId: number) =>
473 (scope.zoneId !== undefined
474 ? logpush.getJobForZone({ zoneId: scope.zoneId, jobId })
475 : logpush.getJobForAccount({ accountId: scope.accountId, jobId })
476 ).pipe(
477 Effect.map((job): logpush.GetJobResponse | undefined => job),
478 Effect.catchTag("JobNotFound", () => Effect.succeed(undefined)),
479 );
480
481const findByName = (scope: Scope, name: string, dataset: Dataset | undefined) =>
482 (scope.zoneId !== undefined
483 ? logpush.listJobsForZone.items({ zoneId: scope.zoneId })
484 : logpush.listJobsForAccount.items({ accountId: scope.accountId })
485 ).pipe(
486 Stream.runCollect,
487 Effect.map((chunk): logpush.GetJobResponse | undefined => {
488 const match = Array.from(chunk)
489 .filter((job): job is NonNullable<typeof job> => job != null)
490 .find(
491 (job) =>
492 job.name === name &&
493 (dataset === undefined || job.dataset === dataset),
494 );
495 return match ?? undefined;
496 }),
497 );
498
499/**
500 * The mutable body shared by create and update (PUT) requests.
501 */
502interface JobMutableBody {
503 destinationConf: string;
504 enabled: boolean;
505 filter: string | undefined;
506 kind: "" | "edge" | undefined;
507 maxUploadBytes: number | undefined;
508 maxUploadIntervalSeconds: number | undefined;
509 maxUploadRecords: number | undefined;
510 name: string;
511 outputOptions: OutputOptions | undefined;
512 ownershipChallenge: string | undefined;
513}
514
515const buildMutableBody = (
516 news: JobProps,
517 name: string,
518 destinationConf: string,
519): JobMutableBody => ({
520 destinationConf,
521 enabled: news.enabled ?? false,
522 filter: news.filter,
523 kind: news.kind,
524 maxUploadBytes: news.maxUploadBytes,
525 maxUploadIntervalSeconds: news.maxUploadIntervalSeconds,
526 maxUploadRecords: news.maxUploadRecords,
527 name,
528 outputOptions: news.outputOptions,
529 ownershipChallenge: news.ownershipChallenge,
530});
531
532const createJob = (scope: Scope, dataset: Dataset, body: JobMutableBody) =>
533 (scope.zoneId !== undefined
534 ? logpush.createJobForZone({ zoneId: scope.zoneId, dataset, ...body })
535 : logpush.createJobForAccount({
536 accountId: scope.accountId,
537 dataset,
538 ...body,
539 })
540 ).pipe(Effect.map((job): logpush.GetJobResponse => job));
541
542const updateJob = (scope: Scope, jobId: number, body: JobMutableBody) =>
543 (scope.zoneId !== undefined
544 ? logpush.updateJobForZone({ zoneId: scope.zoneId, jobId, ...body })
545 : logpush.updateJobForAccount({
546 accountId: scope.accountId,
547 jobId,
548 ...body,
549 })
550 ).pipe(Effect.map((job): logpush.GetJobResponse => job));
551
552const asNumber = (v: "0" | number | null | undefined): number | undefined =>
553 v == null ? undefined : typeof v === "string" ? Number(v) : v;
554
555/**
556 * Decide whether a PUT is needed. Observable fields are diffed against the
557 * live job; write-only/redacted fields (`destination_conf`, `filter`,
558 * `ownership_challenge`) fall back to a `news` vs `olds` comparison —
559 * `olds === undefined` (adoption) forces a PUT so the job converges to the
560 * declared destination/filter.
561 */
562const needsUpdate = (
563 desired: JobMutableBody,
564 news: JobProps,
565 olds: JobProps | undefined,
566 observed: logpush.GetJobResponse,
567): boolean => {
568 if (olds === undefined) return true;
569 if ((news.destinationConf as string) !== (olds.destinationConf as string)) {
570 return true;
571 }
572 if (news.filter !== olds.filter) return true;
573 if (desired.enabled !== (observed.enabled ?? false)) return true;
574 if (desired.name !== (observed.name ?? undefined)) return true;
575 if (
576 desired.maxUploadBytes !== undefined &&
577 desired.maxUploadBytes !== asNumber(observed.maxUploadBytes)
578 ) {
579 return true;
580 }
581 if (
582 desired.maxUploadIntervalSeconds !== undefined &&
583 desired.maxUploadIntervalSeconds !==
584 asNumber(observed.maxUploadIntervalSeconds)
585 ) {
586 return true;
587 }
588 if (
589 desired.maxUploadRecords !== undefined &&
590 desired.maxUploadRecords !== asNumber(observed.maxUploadRecords)
591 ) {
592 return true;
593 }
594 if (
595 desired.outputOptions !== undefined &&
596 !outputOptionsEqual(desired.outputOptions, observed.outputOptions)
597 ) {
598 return true;
599 }
600 return false;
601};
602
603/**
604 * Compare only the keys the user declared — Cloudflare fills the rest with
605 * defaults we must not fight.
606 */
607const outputOptionsEqual = (
608 desired: OutputOptions,
609 observed: logpush.GetJobResponse["outputOptions"],
610): boolean => {
611 const o = observed ?? {};
612 for (const key of Object.keys(desired) as (keyof OutputOptions)[]) {
613 const want = desired[key];
614 if (want === undefined) continue;
615 const have = o[key] ?? undefined;
616 if (Array.isArray(want)) {
617 if (
618 !Array.isArray(have) ||
619 want.length !== have.length ||
620 want.some((v, i) => v !== have[i])
621 ) {
622 return false;
623 }
624 } else if (want !== have) {
625 return false;
626 }
627 }
628 return true;
629};
630
631const undef = <T>(v: T | null | undefined): T | undefined =>
632 v == null ? undefined : v;
633
634const toAttributes = (
635 observed: logpush.GetJobResponse | undefined,
636 scope: Scope,
637): JobAttributes | undefined => {
638 if (
639 observed?.id == null ||
640 observed.dataset == null ||
641 observed.destinationConf == null
642 ) {
643 return undefined;
644 }
645 return {
646 jobId: observed.id,
647 accountId: scope.accountId,
648 zoneId: scope.zoneId,
649 name: observed.name ?? "",
650 dataset: observed.dataset,
651 destinationConf: observed.destinationConf,
652 enabled: observed.enabled ?? false,
653 kind: observed.kind ?? "",
654 errorMessage: undef(observed.errorMessage),
655 lastComplete: undef(observed.lastComplete),
656 lastError: undef(observed.lastError),
657 };
658};