1import { Credentials } from "@distilled.cloud/cloudflare/Credentials";
2import * as pipelines from "@distilled.cloud/cloudflare/pipelines";
3import * as Effect from "effect/Effect";
4import * as Predicate from "effect/Predicate";
5import * as Redacted from "effect/Redacted";
6import type * as HttpClient from "effect/http/HttpClient";
7
8import { Unowned } from "../../AdoptPolicy.ts";
9import { isResolved } from "../../Diff.ts";
10import { createPhysicalName } from "../../PhysicalName.ts";
11import * as Provider from "../../Provider.ts";
12import { Resource } from "../../Resource.ts";
13import { CloudflareEnvironment } from "../CloudflareEnvironment.ts";
14import type { Providers } from "../Providers.ts";
15
16const TypeId = "Cloudflare.Pipelines.LegacyPipeline" as const;
17type TypeId = typeof TypeId;
18
19/**
20 * HTTP ingest source — events are POSTed as JSON to the pipeline's
21 * `endpoint` URL.
22 */
23export interface LegacyPipelineHttpSource {
24 /** Accept events over HTTP at the pipeline endpoint. */
25 type: "http";
26 /**
27 * Require Cloudflare API-token authentication on the ingest endpoint.
28 * @default false
29 */
30 authentication?: boolean;
31 /**
32 * CORS configuration for browser-originated ingestion.
33 */
34 cors?: {
35 /** Allowed origins, e.g. `["https://example.com"]` or `["*"]`. */
36 origins?: string[];
37 };
38}
39
40/**
41 * Worker binding ingest source — events are sent from a Worker via a
42 * `pipelines` binding.
43 */
44export interface LegacyPipelineBindingSource {
45 /** Accept events from a Worker `pipelines` binding. */
46 type: "binding";
47}
48
49/**
50 * An ingest source of a legacy pipeline.
51 */
52export type LegacyPipelineSource =
53 | LegacyPipelineHttpSource
54 | LegacyPipelineBindingSource;
55
56/**
57 * R2 destination configuration of a legacy pipeline.
58 */
59export interface LegacyPipelineDestination {
60 /**
61 * Name of the destination R2 bucket. The bucket must already exist.
62 */
63 bucket: string;
64 /**
65 * R2 S3-compatible credentials the pipeline uses to write objects.
66 * Write-only — Cloudflare never echoes them back.
67 */
68 credentials: {
69 /** R2 access key id (the API token id). */
70 accessKeyId: Redacted.Redacted<string>;
71 /** R2 secret access key (SHA-256 hex of the API token value). */
72 secretAccessKey: Redacted.Redacted<string>;
73 /**
74 * S3-compatible endpoint of the R2 bucket.
75 * @default https://{accountId}.r2.cloudflarestorage.com
76 */
77 endpoint?: string;
78 };
79 /**
80 * Batching policy controlling when the pipeline flushes a batch of
81 * events to R2.
82 */
83 batch?: {
84 /** Flush once the batch reaches this size in bytes. */
85 maxBytes?: number;
86 /** Flush at most this many seconds after the batch was opened. */
87 maxDurationS?: number;
88 /** Flush once the batch reaches this many events. */
89 maxRows?: number;
90 };
91 /**
92 * Compression applied to output files.
93 * @default "gzip"
94 */
95 compression?: "none" | "gzip" | "deflate";
96 /**
97 * Key prefix under which output objects are written.
98 */
99 prefix?: string;
100 /**
101 * Time-partitioned directory layout of output object keys
102 * (strftime-style, e.g. `event_date=%F/hr=%H`).
103 */
104 filepath?: string;
105 /**
106 * Name template of output files within a partition.
107 */
108 filename?: string;
109}
110
111export interface LegacyPipelineProps {
112 /**
113 * Name of the pipeline. Unique per account; the legacy API addresses
114 * pipelines by name, so changing it triggers a replacement. If
115 * omitted, a unique name is generated from the app, stage, and
116 * logical ID.
117 * @default ${app}-${id}-${stage}-${suffix}
118 */
119 name?: string;
120 /**
121 * Ingest sources accepted by the pipeline.
122 * @default [{ type: "http" }, { type: "binding" }]
123 */
124 source?: LegacyPipelineSource[];
125 /**
126 * R2 destination the pipeline batches events into.
127 */
128 destination: LegacyPipelineDestination;
129}
130
131export interface LegacyPipelineAttributes {
132 /** Cloudflare-assigned pipeline identifier. */
133 pipelineId: string;
134 /** Account that owns the pipeline. */
135 accountId: string;
136 /** Pipeline name (unique per account; the API identifier). */
137 name: string;
138 /** HTTP endpoint URL events can be POSTed to. */
139 endpoint: string;
140 /** Destination R2 bucket name. */
141 bucket: string;
142 /** Version number of the last saved configuration. */
143 version: number;
144}
145
146export type LegacyPipeline = Resource<
147 TypeId,
148 LegacyPipelineProps,
149 LegacyPipelineAttributes,
150 never,
151 Providers
152>;
153
154/**
155 * A **legacy** Cloudflare Pipeline — the original HTTP-ingest → R2 batch
156 * product (`/accounts/{account}/pipelines`).
157 *
158 * :::caution
159 * This is the **deprecated, legacy** Pipelines API. Cloudflare has
160 * superseded it with the SQL-based product — prefer
161 * {@link Stream}, {@link Sink}, and {@link Pipeline}
162 * for new infrastructure. This resource exists only to manage
163 * pre-existing legacy pipelines.
164 * :::
165 *
166 * A legacy pipeline accepts JSON events over HTTP (and/or a Worker
167 * `pipelines` binding) and batches them into an R2 bucket using
168 * S3-compatible credentials.
169 * ### Creating a Legacy Pipeline
170 * **Example:** HTTP ingest into R2
171 * The S3-compatible credentials are derived from a Cloudflare API token:
172 * the access key id is the token id and the secret is the SHA-256 hex
173 * digest of the token value.
174 * ```typescript
175 * const bucket = yield* Cloudflare.R2.Bucket("events", {});
176 *
177 * const pipeline = yield* Cloudflare.Pipelines.LegacyPipeline("ingest", {
178 * destination: {
179 * bucket: bucket.bucketName,
180 * credentials: {
181 * accessKeyId: yield* Config.Redacted("R2_ACCESS_KEY_ID"),
182 * secretAccessKey: yield* Config.Redacted("R2_SECRET_ACCESS_KEY"),
183 * },
184 * },
185 * });
186 * // POST events to pipeline.endpoint
187 * ```
188 *
189 * **Example:** Tuned batching and CORS
190 * ```typescript
191 * const pipeline = yield* Cloudflare.Pipelines.LegacyPipeline("ingest", {
192 * source: [
193 * { type: "http", cors: { origins: ["https://example.com"] } },
194 * ],
195 * destination: {
196 * bucket: bucket.bucketName,
197 * credentials,
198 * batch: { maxDurationS: 10, maxRows: 1000 },
199 * compression: "gzip",
200 * prefix: "ingest",
201 * },
202 * });
203 * ```
204 *
205 * @see https://developers.cloudflare.com/pipelines/
206 *
207 * @resource
208 * @product Pipelines
209 * @category Storage & Databases
210 */
211export const LegacyPipeline = Resource<LegacyPipeline>(TypeId);
212
213/**
214 * Returns true if the given value is a LegacyPipeline resource.
215 */
216export const isLegacyPipeline = (value: unknown): value is LegacyPipeline =>
217 Predicate.hasProperty(value, "Type") && value.Type === TypeId;
218
219export const LegacyPipelineProvider = () =>
220 Provider.succeed(LegacyPipeline, {
221 stables: ["pipelineId", "accountId", "name", "endpoint"],
222
223 diff: Effect.fn(function* ({ id, olds, news, output }) {
224 const { accountId } = yield* yield* CloudflareEnvironment;
225 if (!isResolved(news)) return undefined;
226 if ((output?.accountId ?? accountId) !== accountId) {
227 return { action: "replace" } as const;
228 }
229 const o = olds as LegacyPipelineProps | undefined;
230 if (o === undefined) return undefined;
231 // The legacy API addresses pipelines by name — a name change is a
232 // replacement. Everything else updates in place via PUT.
233 const newName = yield* legacyPipelineName(id, news.name);
234 const oldName = output?.name ?? (yield* legacyPipelineName(id, o.name));
235 if (newName !== oldName) {
236 return { action: "replace" } as const;
237 }
238 return undefined;
239 }),
240
241 list: Effect.fn(function* () {
242 const { accountId } = yield* yield* CloudflareEnvironment;
243 // The list endpoint returns summary items (id/name/endpoint only),
244 // so hydrate each by name into the exact `read` Attributes shape
245 // with bounded concurrency and a typed per-item not-found skip
246 // (a pipeline can vanish between the list and the get).
247 const summaries = yield* listLegacyPipelineSummaries(accountId);
248 const rows = yield* Effect.forEach(
249 summaries,
250 (summary) =>
251 getLegacyPipeline(accountId, summary.name).pipe(
252 Effect.map((observed) =>
253 observed ? toAttributes(observed, accountId) : undefined,
254 ),
255 ),
256 { concurrency: 10 },
257 );
258 return rows.filter(
259 (row): row is LegacyPipelineAttributes => row !== undefined,
260 );
261 }),
262
263 read: Effect.fn(function* ({ id, output, olds }) {
264 const { accountId } = yield* yield* CloudflareEnvironment;
265 const acct = output?.accountId ?? accountId;
266 if (output?.name) {
267 const observed = yield* getLegacyPipeline(acct, output.name);
268 if (observed) return toAttributes(observed, acct);
269 }
270 // Cold read — names are unique per account; a generated-name match
271 // is proof of ownership, an explicit name is not.
272 const name = yield* legacyPipelineName(id, olds?.name);
273 const observed = yield* getLegacyPipeline(acct, name);
274 if (observed) {
275 const attrs = toAttributes(observed, acct);
276 return olds?.name !== undefined ? Unowned(attrs) : attrs;
277 }
278 return undefined;
279 }),
280
281 reconcile: Effect.fn(function* ({ id, news, olds, output }) {
282 const { accountId } = yield* yield* CloudflareEnvironment;
283 const name = yield* legacyPipelineName(id, news.name);
284
285 // 1. Observe — the name is the API identifier, so a single get
286 // covers both the cached-output and cold-recovery cases.
287 let observed = yield* getLegacyPipeline(
288 output?.accountId ?? accountId,
289 output?.name ?? name,
290 );
291
292 // 2. Ensure — create when missing.
293 if (!observed) {
294 observed = yield* pipelines.createPipeline({
295 accountId,
296 name,
297 source: toRequestSource(news.source),
298 destination: toRequestDestination(accountId, news.destination),
299 });
300 return toAttributes(observed, accountId);
301 }
302
303 // 3. Sync — diff observed config against desired and PUT the full
304 // desired state only when something changed. Credentials are
305 // write-only, so `olds` is the only (best-effort) baseline for
306 // them: on adoption (`olds` undefined) or a credential change we
307 // push an update unconditionally.
308 const credsRotated =
309 olds === undefined ||
310 Redacted.value(news.destination.credentials.accessKeyId) !==
311 Redacted.value(olds.destination.credentials.accessKeyId) ||
312 Redacted.value(news.destination.credentials.secretAccessKey) !==
313 Redacted.value(olds.destination.credentials.secretAccessKey);
314 if (credsRotated || drifted(observed, news)) {
315 observed = yield* pipelines.updatePipeline({
316 accountId,
317 pipelineName: observed.name,
318 name,
319 source: toRequestSource(news.source),
320 destination: toRequestDestination(accountId, news.destination),
321 });
322 }
323
324 return toAttributes(observed, accountId);
325 }),
326
327 delete: Effect.fn(function* ({ output }) {
328 yield* pipelines
329 .deletePipeline({
330 accountId: output.accountId,
331 pipelineName: output.name,
332 })
333 .pipe(Effect.catchTag("PipelineNotExists", () => Effect.void));
334 }),
335 });
336
337/**
338 * The pipeline state shared by get/create/update responses.
339 */
340interface ObservedLegacyPipeline {
341 id: string;
342 name: string;
343 endpoint: string;
344 version: number;
345 destination: {
346 batch: { maxBytes: number; maxDurationS: number; maxRows: number };
347 compression: { type: string };
348 path: {
349 bucket: string;
350 filename?: string | null;
351 filepath?: string | null;
352 prefix?: string | null;
353 };
354 };
355 source: readonly {
356 type: string;
357 authentication?: boolean | null;
358 cors?: { origins?: readonly string[] | null } | null;
359 }[];
360}
361
362const legacyPipelineName = (id: string, name: string | undefined) =>
363 Effect.gen(function* () {
364 if (name) return name;
365 return yield* createPhysicalName({ id, lowercase: true, delimiter: "-" });
366 });
367
368/**
369 * Read a legacy pipeline by name, mapping "gone" (`PipelineNotExists`,
370 * Cloudflare error code 1000) to `undefined`.
371 */
372const getLegacyPipeline = (accountId: string, pipelineName: string) =>
373 pipelines.getPipeline({ accountId, pipelineName }).pipe(
374 Effect.map((p): ObservedLegacyPipeline | undefined => p),
375 Effect.catchTag("PipelineNotExists", () => Effect.succeed(undefined)),
376 );
377
378interface LegacyPipelineSummary {
379 id: string;
380 name: string;
381 endpoint: string;
382}
383
384/**
385 * Enumerate every legacy pipeline summary in an account. The distilled
386 * `listPipelines` op is not stream-paginated, so walk the `page`/
387 * `per_page` query params using `result_info.total_count` (falling back
388 * to a short-page sentinel) until every page is collected. The list
389 * endpoint caps `per_page` at 50 and returns only summary fields.
390 */
391const listLegacyPipelineSummaries = (accountId: string) => {
392 const perPage = 50;
393 const collect = (
394 page: number,
395 acc: LegacyPipelineSummary[],
396 ): Effect.Effect<
397 LegacyPipelineSummary[],
398 pipelines.ListPipelinesError,
399 Credentials | HttpClient.HttpClient
400 > =>
401 Effect.gen(function* () {
402 const response = yield* pipelines.listPipelines({
403 accountId,
404 page: String(page),
405 perPage: String(perPage),
406 });
407 const results = (response.results ?? []).map(
408 (p): LegacyPipelineSummary => ({
409 id: p.id,
410 name: p.name ?? "",
411 endpoint: p.endpoint ?? "",
412 }),
413 );
414 const next = [...acc, ...results];
415 const total = response.resultInfo?.totalCount;
416 const done =
417 results.length < perPage || (total != null && next.length >= total);
418 return done ? next : yield* collect(page + 1, next);
419 });
420 return collect(1, []);
421};
422
423const defaultSources: LegacyPipelineSource[] = [
424 { type: "http" },
425 { type: "binding" },
426];
427
428const toRequestSource = (source: LegacyPipelineSource[] | undefined) =>
429 (source ?? defaultSources).map((s) =>
430 s.type === "http"
431 ? {
432 format: "json" as const,
433 type: s.type,
434 authentication: s.authentication,
435 cors: s.cors,
436 }
437 : { format: "json" as const, type: s.type },
438 );
439
440const toRequestDestination = (
441 accountId: string,
442 destination: LegacyPipelineDestination,
443) => ({
444 type: "r2" as const,
445 format: "json" as const,
446 batch: destination.batch ?? {},
447 compression: { type: destination.compression },
448 credentials: {
449 accessKeyId: Redacted.value(destination.credentials.accessKeyId),
450 secretAccessKey: Redacted.value(destination.credentials.secretAccessKey),
451 endpoint:
452 destination.credentials.endpoint ??
453 `https://${accountId}.r2.cloudflarestorage.com`,
454 },
455 path: {
456 bucket: destination.bucket as string,
457 prefix: destination.prefix,
458 filepath: destination.filepath,
459 filename: destination.filename,
460 },
461});
462
463/**
464 * Detect drift between echoed cloud config and desired props. Only
465 * user-declared optional fields are diffed so we don't fight server-side
466 * defaults; credentials are write-only and handled separately.
467 */
468const drifted = (
469 observed: ObservedLegacyPipeline,
470 news: LegacyPipelineProps,
471): boolean => {
472 const d = news.destination;
473 const path = observed.destination.path;
474 if (path.bucket !== (d.bucket as string)) return true;
475 if (d.prefix !== undefined && d.prefix !== (path.prefix ?? undefined)) {
476 return true;
477 }
478 if (d.filepath !== undefined && d.filepath !== (path.filepath ?? undefined)) {
479 return true;
480 }
481 if (d.filename !== undefined && d.filename !== (path.filename ?? undefined)) {
482 return true;
483 }
484 if (
485 d.compression !== undefined &&
486 d.compression !== observed.destination.compression.type
487 ) {
488 return true;
489 }
490 const batch = observed.destination.batch;
491 if (d.batch?.maxBytes !== undefined && d.batch.maxBytes !== batch.maxBytes) {
492 return true;
493 }
494 if (
495 d.batch?.maxDurationS !== undefined &&
496 d.batch.maxDurationS !== batch.maxDurationS
497 ) {
498 return true;
499 }
500 if (d.batch?.maxRows !== undefined && d.batch.maxRows !== batch.maxRows) {
501 return true;
502 }
503 // Sources — compare the declared set of source types and the declared
504 // http options against what Cloudflare echoes.
505 const desired = news.source ?? defaultSources;
506 const observedTypes = observed.source.map((s) => s.type).sort();
507 const desiredTypes = desired.map((s) => s.type).sort();
508 if (observedTypes.join(",") !== desiredTypes.join(",")) return true;
509 const desiredHttp = desired.find(
510 (s): s is LegacyPipelineHttpSource => s.type === "http",
511 );
512 const observedHttp = observed.source.find((s) => s.type === "http");
513 if (desiredHttp && observedHttp) {
514 if (
515 desiredHttp.authentication !== undefined &&
516 desiredHttp.authentication !== (observedHttp.authentication ?? false)
517 ) {
518 return true;
519 }
520 if (desiredHttp.cors?.origins !== undefined) {
521 const observedOrigins = observedHttp.cors?.origins ?? [];
522 if (
523 desiredHttp.cors.origins.length !== observedOrigins.length ||
524 desiredHttp.cors.origins.some((o, i) => o !== observedOrigins[i])
525 ) {
526 return true;
527 }
528 }
529 }
530 return false;
531};
532
533const toAttributes = (
534 observed: ObservedLegacyPipeline,
535 accountId: string,
536): LegacyPipelineAttributes => ({
537 pipelineId: observed.id,
538 accountId,
539 name: observed.name,
540 endpoint: observed.endpoint,
541 bucket: observed.destination.path.bucket,
542 version: observed.version,
543});