1import * as pipelines from "@distilled.cloud/cloudflare/pipelines";
2import * as Effect from "effect/Effect";
3import * as Predicate from "effect/Predicate";
4import * as Redacted from "effect/Redacted";
5import * as Schedule from "effect/Schedule";
6import * as Stream from "effect/Stream";
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.Sink" as const;
17type TypeId = typeof TypeId;
18
19/**
20 * Batching policy controlling when the sink rolls (closes and uploads)
21 * the current output file.
22 */
23export interface SinkRollingPolicy {
24 /**
25 * Roll the file once it reaches this size in bytes.
26 */
27 fileSizeBytes?: number;
28 /**
29 * Roll the file after this many seconds without new events.
30 */
31 inactivitySeconds?: number;
32 /**
33 * Roll the file at most this many seconds after it was opened.
34 * @default 300
35 */
36 intervalSeconds?: number;
37}
38
39/**
40 * Configuration of an `r2` sink writing raw files to an R2 bucket.
41 */
42export interface SinkR2Config {
43 /**
44 * Name of the destination R2 bucket. The bucket must already exist.
45 */
46 bucket: string;
47 /**
48 * R2 S3-compatible credentials the sink uses to write objects.
49 * Write-only — Cloudflare never echoes them back.
50 */
51 credentials: {
52 /** R2 access key id (the API token id). */
53 accessKeyId: Redacted.Redacted<string>;
54 /** R2 secret access key (SHA-256 hex of the API token value). */
55 secretAccessKey: Redacted.Redacted<string>;
56 };
57 /**
58 * Key prefix under which output objects are written.
59 */
60 path?: string;
61 /**
62 * Time-based partitioning of output object keys.
63 */
64 partitioning?: {
65 /**
66 * strftime-style pattern, e.g. `year=%Y/month=%m/day=%d`.
67 */
68 timePattern?: string;
69 };
70 /**
71 * Naming of output files within a partition.
72 */
73 fileNaming?: {
74 /** Prefix prepended to each file name. */
75 prefix?: string;
76 /** Suffix appended to each file name. */
77 suffix?: string;
78 /**
79 * Strategy generating the unique part of each file name.
80 * @default "uuid_v7"
81 */
82 strategy?: "serial" | "uuid" | "uuid_v7" | "ulid";
83 };
84 /**
85 * When the sink rolls output files.
86 */
87 rollingPolicy?: SinkRollingPolicy;
88 /**
89 * Jurisdiction the bucket was created in (`eu`, `fedramp`), when not
90 * the default.
91 */
92 jurisdiction?: string;
93}
94
95/**
96 * Configuration of an `r2_data_catalog` sink writing Iceberg tables via
97 * the R2 Data Catalog.
98 */
99export interface SinkR2DataCatalogConfig {
100 /**
101 * Name of the R2 bucket backing the catalog. The bucket must already
102 * exist and have the Data Catalog enabled.
103 */
104 bucket: string;
105 /**
106 * Name of the Iceberg table to write.
107 */
108 tableName: string;
109 /**
110 * Catalog namespace the table lives in.
111 * @default "default"
112 */
113 namespace?: string;
114 /**
115 * Cloudflare API token with R2 Data Catalog permissions. Write-only —
116 * Cloudflare never echoes it back.
117 */
118 token: Redacted.Redacted<string>;
119 /**
120 * When the sink rolls output files.
121 */
122 rollingPolicy?: SinkRollingPolicy;
123}
124
125/**
126 * Output file format written by the sink.
127 */
128export type SinkFormat =
129 | {
130 /** Newline-delimited JSON output. */
131 type: "json";
132 }
133 | {
134 /** Parquet output. */
135 type: "parquet";
136 /**
137 * Compression codec.
138 * @default "zstd"
139 */
140 compression?: "uncompressed" | "snappy" | "gzip" | "zstd" | "lz4";
141 /** Target row-group size in bytes. */
142 rowGroupBytes?: number;
143 };
144
145interface SinkBaseProps {
146 /**
147 * Name of the sink. Unique per account; must be alphanumeric and
148 * underscores only (it is referenced as a SQL table name). If omitted,
149 * a unique name is generated from the app, stage, and logical ID.
150 *
151 * Sinks have no update API, so changing this (or any other) property
152 * triggers a replacement.
153 * @default ${app}_${id}_${stage}_${suffix}
154 */
155 name?: string;
156 /**
157 * Output file format.
158 * @default { type: "json" }
159 */
160 format?: SinkFormat;
161}
162
163export type SinkProps =
164 | (SinkBaseProps & {
165 /**
166 * Sink type — `r2` writes raw files to an R2 bucket.
167 */
168 type: "r2";
169 /**
170 * R2 destination configuration.
171 */
172 config: SinkR2Config;
173 })
174 | (SinkBaseProps & {
175 /**
176 * Sink type — `r2_data_catalog` writes Iceberg tables via the R2
177 * Data Catalog.
178 */
179 type: "r2_data_catalog";
180 /**
181 * R2 Data Catalog destination configuration.
182 */
183 config: SinkR2DataCatalogConfig;
184 });
185
186export interface SinkAttributes {
187 /** Cloudflare-assigned sink identifier. */
188 sinkId: string;
189 /** Account that owns the sink. */
190 accountId: string;
191 /** Sink name (unique per account). */
192 name: string;
193 /** Sink type. */
194 type: "r2" | "r2_data_catalog";
195 /** Destination R2 bucket name. */
196 bucket: string;
197 /** Key prefix output objects are written under (r2 sinks). */
198 path: string | undefined;
199 /** When the sink was created. */
200 createdAt: string;
201 /** When the sink was last modified. */
202 modifiedAt: string;
203}
204
205export type Sink = Resource<
206 TypeId,
207 SinkProps,
208 SinkAttributes,
209 never,
210 Providers
211>;
212
213/**
214 * A Cloudflare Pipelines sink — the destination of the Pipelines product.
215 * A SQL {@link Pipeline} reads events from a {@link Stream} and
216 * writes them to a sink, which stores them in R2 either as raw files
217 * (`r2`) or as Iceberg tables via the R2 Data Catalog
218 * (`r2_data_catalog`).
219 *
220 * Sinks have no update API: every property change triggers a
221 * replacement. With engine-generated names this is seamless (the new
222 * sink gets a fresh name before the old one is deleted); with an
223 * explicit `name` the create-before-delete replacement collides, so
224 * prefer generated names.
225 * ### Creating a Sink
226 * **Example:** R2 sink with JSON output
227 * The S3-compatible credentials are derived from a Cloudflare API token:
228 * the access key id is the token id and the secret is the SHA-256 hex
229 * digest of the token value.
230 * ```typescript
231 * const bucket = yield* Cloudflare.R2.Bucket("events", {});
232 *
233 * const sink = yield* Cloudflare.Pipelines.Sink("events-sink", {
234 * type: "r2",
235 * config: {
236 * bucket: bucket.bucketName,
237 * credentials: {
238 * accessKeyId: yield* Config.Redacted("R2_ACCESS_KEY_ID"),
239 * secretAccessKey: yield* Config.Redacted("R2_SECRET_ACCESS_KEY"),
240 * },
241 * path: "ingest",
242 * rollingPolicy: { intervalSeconds: 30 },
243 * },
244 * });
245 * ```
246 *
247 * **Example:** Parquet output
248 * ```typescript
249 * const sink = yield* Cloudflare.Pipelines.Sink("parquet-sink", {
250 * type: "r2",
251 * config: { bucket: bucket.bucketName, credentials },
252 * format: { type: "parquet", compression: "zstd" },
253 * });
254 * ```
255 *
256 * ### R2 Data Catalog
257 * **Example:** Iceberg table sink
258 * ```typescript
259 * const sink = yield* Cloudflare.Pipelines.Sink("iceberg-sink", {
260 * type: "r2_data_catalog",
261 * config: {
262 * bucket: bucket.bucketName,
263 * tableName: "events",
264 * namespace: "default",
265 * token: yield* Config.Redacted("CATALOG_TOKEN"),
266 * },
267 * });
268 * ```
269 *
270 * @see https://developers.cloudflare.com/pipelines/
271 *
272 * @resource
273 * @product Pipelines
274 * @category Storage & Databases
275 */
276export const Sink = Resource<Sink>(TypeId);
277
278/**
279 * Returns true if the given value is a Sink resource.
280 */
281export const isSink = (value: unknown): value is Sink =>
282 Predicate.hasProperty(value, "Type") && value.Type === TypeId;
283
284export const SinkProvider = () =>
285 Provider.succeed(Sink, {
286 stables: ["sinkId", "accountId", "name", "type", "createdAt"],
287
288 diff: Effect.fn(function* ({ id, olds, news, output }) {
289 const { accountId } = yield* yield* CloudflareEnvironment;
290 if (!isResolved(news)) return undefined;
291 if ((output?.accountId ?? accountId) !== accountId) {
292 return { action: "replace" } as const;
293 }
294 const o = olds as SinkProps | undefined;
295 if (o === undefined) return undefined;
296 // Sinks have no update API — any change is a replacement.
297 const newName = yield* sinkName(id, news.name);
298 const oldName = output?.name ?? (yield* sinkName(id, o.name));
299 if (
300 newName !== oldName ||
301 news.type !== o.type ||
302 !stableEquals(normalizeProps(news), normalizeProps(o))
303 ) {
304 return { action: "replace" } as const;
305 }
306 return undefined;
307 }),
308
309 read: Effect.fn(function* ({ id, output, olds }) {
310 const { accountId } = yield* yield* CloudflareEnvironment;
311 const acct = output?.accountId ?? accountId;
312 if (output?.sinkId) {
313 const observed = yield* getSink(acct, output.sinkId);
314 if (observed) return toAttributes(observed, acct);
315 }
316 // Cold read — sink names are unique per account; a generated-name
317 // match is proof of ownership, an explicit name is not.
318 const name = yield* sinkName(id, olds?.name);
319 const match = yield* findSinkByName(acct, name);
320 if (match) {
321 const attrs = toAttributes(match, acct);
322 return olds?.name !== undefined ? Unowned(attrs) : attrs;
323 }
324 return undefined;
325 }),
326
327 reconcile: Effect.fn(function* ({ id, news, output }) {
328 const { accountId } = yield* yield* CloudflareEnvironment;
329 const name = yield* sinkName(id, news.name);
330
331 // 1. Observe — by cached id first, then by (unique) name so we
332 // recover from lost state writes.
333 let observed = output?.sinkId
334 ? yield* getSink(output.accountId ?? accountId, output.sinkId)
335 : undefined;
336 if (!observed) {
337 observed = yield* findSinkByName(accountId, name);
338 }
339
340 // Converge drift — there is no update API, so when observable
341 // echoed config (bucket/path/table) differs from the desired state
342 // (e.g. the bucket reference was unresolved at diff time), delete
343 // the stale sink and fall through to recreate it under the same
344 // name. The delete rides out `SinkInUse` while a pipeline that is
345 // being repointed still references it.
346 if (observed && sinkDrifted(observed, news)) {
347 yield* deleteSink(accountId, observed.id);
348 // Wait until the delete is visible so the recreate below does not
349 // race a `SinkAlreadyExists` against the dying sink.
350 yield* getSink(accountId, observed.id).pipe(
351 Effect.repeat({
352 schedule: Schedule.max([
353 Schedule.exponential("250 millis"),
354 Schedule.recurs(8),
355 ]),
356 until: (s) => s === undefined,
357 }),
358 );
359 observed = undefined;
360 }
361
362 // 2. Ensure — create when missing. There is no sync step: sinks
363 // have no update API, so prop changes arrive as replacements
364 // (diff) rather than in-place updates. An AlreadyExists on
365 // create is a race or recovery — resolve it via the name lookup.
366 if (!observed) {
367 observed = yield* pipelines
368 .createSink({
369 accountId,
370 name,
371 type: news.type,
372 config: toRequestConfig(accountId, news),
373 format: news.format,
374 })
375 .pipe(
376 Effect.catchTag("SinkAlreadyExists", (error) =>
377 findSinkByName(accountId, name).pipe(
378 Effect.flatMap((match) =>
379 match ? Effect.succeed(match) : Effect.fail(error),
380 ),
381 ),
382 ),
383 );
384 }
385
386 return toAttributes(observed, accountId);
387 }),
388
389 delete: Effect.fn(function* ({ output }) {
390 yield* deleteSink(output.accountId, output.sinkId);
391 }),
392
393 // Account collection: sinks are account-scoped and enumerable via
394 // `listSinks` (paginated, items in `result`). Hydrate each page item
395 // into the exact `read` Attributes shape. Credentials/token are
396 // write-only and never echoed, matching `read`.
397 list: Effect.fn(function* () {
398 const { accountId } = yield* yield* CloudflareEnvironment;
399 return yield* pipelines.listSinks.pages({ accountId }).pipe(
400 Stream.runCollect,
401 Effect.map((chunk) =>
402 Array.from(chunk).flatMap((page) =>
403 (page.result ?? []).map((sink) => toAttributes(sink, accountId)),
404 ),
405 ),
406 );
407 }),
408 });
409
410/**
411 * The subset of sink state shared by get/list/create responses.
412 */
413interface ObservedSink {
414 id: string;
415 name: string;
416 type: string;
417 config?:
418 | {
419 bucket: string;
420 path?: string | null;
421 }
422 | { bucket: string; tableName: string; namespace?: string | null }
423 | null;
424 createdAt: string;
425 modifiedAt: string;
426}
427
428// Pipelines entity names must be alphanumeric/underscore only (they are
429// referenced as SQL table names), so swap the default hyphen delimiter
430// for underscores.
431const sinkName = (id: string, name: string | undefined) =>
432 Effect.gen(function* () {
433 if (name) return name;
434 const generated = yield* createPhysicalName({
435 id,
436 lowercase: true,
437 delimiter: "_",
438 });
439 return generated.replaceAll(/[^a-zA-Z0-9_]/g, "_");
440 });
441
442/**
443 * Read a sink by id, mapping "gone" (`SinkNotFound`, Cloudflare error
444 * code 1015, or `InvalidSinkId` for a malformed/foreign id) to
445 * `undefined`.
446 */
447const getSink = (accountId: string, sinkId: string) =>
448 pipelines.getSink({ accountId, sinkId }).pipe(
449 Effect.map((s): ObservedSink | undefined => s),
450 Effect.catchTag("SinkNotFound", () => Effect.succeed(undefined)),
451 Effect.catchTag("InvalidSinkId", () => Effect.succeed(undefined)),
452 );
453
454/**
455 * Idempotent delete — a sink that is already gone is success. Rides out
456 * `SinkInUse` (HTTP 422 "Sink still in use") with a bounded retry while
457 * a dependent pipeline's own deletion propagates.
458 */
459const deleteSink = (accountId: string, sinkId: string) =>
460 pipelines.deleteSink({ accountId, sinkId }).pipe(
461 Effect.retry({
462 while: (e) => e._tag === "SinkInUse",
463 schedule: Schedule.max([
464 Schedule.exponential("500 millis"),
465 Schedule.recurs(8),
466 ]),
467 }),
468 Effect.catchTag("SinkNotFound", () => Effect.void),
469 Effect.catchTag("InvalidSinkId", () => Effect.void),
470 );
471
472/**
473 * Detect drift between the echoed cloud config and the desired props.
474 * Only fields Cloudflare echoes back are comparable (credentials and the
475 * catalog token are write-only); only user-declared optional fields are
476 * diffed so we don't fight server-side defaults.
477 */
478const sinkDrifted = (observed: ObservedSink, news: SinkProps): boolean => {
479 if (observed.type !== news.type) return true;
480 const cfg = observed.config;
481 if (!cfg) return false;
482 if (cfg.bucket !== (news.config.bucket as string)) return true;
483 if (news.type === "r2") {
484 const observedPath = "path" in cfg ? (cfg.path ?? undefined) : undefined;
485 if (news.config.path !== undefined && news.config.path !== observedPath) {
486 return true;
487 }
488 } else if ("tableName" in cfg) {
489 if (cfg.tableName !== news.config.tableName) return true;
490 if (
491 news.config.namespace !== undefined &&
492 news.config.namespace !== (cfg.namespace ?? undefined)
493 ) {
494 return true;
495 }
496 }
497 return false;
498};
499
500const findSinkByName = (accountId: string, name: string) =>
501 pipelines.listSinks.items({ accountId }).pipe(
502 Stream.runCollect,
503 Effect.map((chunk): ObservedSink | undefined =>
504 Array.from(chunk).find((s) => s.name === name),
505 ),
506 );
507
508/**
509 * Build the distilled create request `config` body from props,
510 * unwrapping write-only secrets.
511 */
512const toRequestConfig = (accountId: string, news: SinkProps) => {
513 if (news.type === "r2") {
514 const c = news.config;
515 return {
516 accountId,
517 bucket: c.bucket as string,
518 credentials: {
519 accessKeyId: Redacted.value(c.credentials.accessKeyId),
520 secretAccessKey: Redacted.value(c.credentials.secretAccessKey),
521 },
522 path: c.path,
523 partitioning: c.partitioning,
524 fileNaming: c.fileNaming,
525 rollingPolicy: c.rollingPolicy,
526 jurisdiction: c.jurisdiction,
527 };
528 }
529 const c = news.config;
530 return {
531 accountId,
532 bucket: c.bucket as string,
533 tableName: c.tableName,
534 namespace: c.namespace,
535 token: Redacted.value(c.token),
536 rollingPolicy: c.rollingPolicy,
537 };
538};
539
540/**
541 * Normalize props for change detection: unwrap redacted secrets so two
542 * `Redacted` wrappers holding the same value compare equal.
543 */
544const normalizeProps = (props: SinkProps): unknown => {
545 if (props.type === "r2") {
546 return {
547 type: props.type,
548 format: props.format,
549 config: {
550 ...props.config,
551 credentials: {
552 accessKeyId: Redacted.value(props.config.credentials.accessKeyId),
553 secretAccessKey: Redacted.value(
554 props.config.credentials.secretAccessKey,
555 ),
556 },
557 },
558 };
559 }
560 return {
561 type: props.type,
562 format: props.format,
563 config: {
564 ...props.config,
565 token: Redacted.value(props.config.token),
566 },
567 };
568};
569
570/**
571 * Key-order-insensitive structural equality for plain JSON-ish prop
572 * values.
573 */
574const stableEquals = (a: unknown, b: unknown): boolean =>
575 stableStringify(a) === stableStringify(b);
576
577const stableStringify = (value: unknown): string =>
578 JSON.stringify(value, (_key, v) =>
579 v !== null && typeof v === "object" && !Array.isArray(v)
580 ? Object.fromEntries(
581 Object.entries(v as Record<string, unknown>).sort(([x], [y]) =>
582 x.localeCompare(y),
583 ),
584 )
585 : v,
586 ) ?? "undefined";
587
588const toAttributes = (
589 observed: ObservedSink,
590 accountId: string,
591): SinkAttributes => ({
592 sinkId: observed.id,
593 accountId,
594 name: observed.name,
595 type: observed.type as "r2" | "r2_data_catalog",
596 bucket: observed.config?.bucket ?? "",
597 path:
598 observed.config && "path" in observed.config
599 ? (observed.config.path ?? undefined)
600 : undefined,
601 createdAt: observed.createdAt,
602 modifiedAt: observed.modifiedAt,
603});