1import * as pipelines from "@distilled.cloud/cloudflare/pipelines";
2import * as Effect from "effect/Effect";
3import * as Predicate from "effect/Predicate";
4import * as Schedule from "effect/Schedule";
5import * as EffectStream 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 StreamTypeId = "Cloudflare.Pipelines.Stream" as const;
16type StreamTypeId = typeof StreamTypeId;
17
18/**
19 * Scalar field types accepted by a structured stream schema.
20 */
21export type StreamFieldType =
22 | "int32"
23 | "int64"
24 | "float32"
25 | "float64"
26 | "bool"
27 | "string"
28 | "binary"
29 | "json";
30
31/**
32 * A single field of a structured stream schema. `timestamp` fields
33 * additionally accept a `unit`.
34 */
35export type StreamField =
36 | {
37 /** Field type. */
38 type: StreamFieldType;
39 /** Field name as it appears in ingested events. */
40 name: string;
41 /** Whether the field must be present in every event. */
42 required?: boolean;
43 /** Name to expose to SQL when it differs from `name`. */
44 sqlName?: string;
45 /** Metadata key the field is populated from instead of the event body. */
46 metadataKey?: string;
47 }
48 | {
49 /** Timestamp field type. */
50 type: "timestamp";
51 /** Field name as it appears in ingested events. */
52 name: string;
53 /** Whether the field must be present in every event. */
54 required?: boolean;
55 /** Name to expose to SQL when it differs from `name`. */
56 sqlName?: string;
57 /** Metadata key the field is populated from instead of the event body. */
58 metadataKey?: string;
59 /**
60 * Precision of the timestamp.
61 * @default "millisecond"
62 */
63 unit?: "second" | "millisecond" | "microsecond" | "nanosecond";
64 };
65
66/**
67 * Input format of events ingested by the stream.
68 */
69export interface StreamFormat {
70 /** Only JSON ingestion is supported. */
71 type: "json";
72 /**
73 * Accept events of any shape (no schema enforcement).
74 */
75 unstructured?: boolean;
76 /**
77 * How timestamps are rendered in ingested JSON.
78 * @default "rfc3339"
79 */
80 timestampFormat?: "rfc3339" | "unix_millis";
81 /** How decimals are encoded in ingested JSON. */
82 decimalEncoding?: "number" | "string" | "bytes";
83}
84
85/**
86 * HTTP ingest endpoint configuration. Mutable in place.
87 */
88export interface StreamHttp {
89 /**
90 * Whether the stream exposes an HTTP ingest endpoint.
91 * @default true
92 */
93 enabled?: boolean;
94 /**
95 * Whether requests to the HTTP endpoint must carry a Cloudflare API
96 * token.
97 * @default false
98 */
99 authentication?: boolean;
100 /** CORS configuration for browser-originated ingestion. */
101 cors?: {
102 /** Allowed origins, e.g. `["https://app.example.com"]` or `["*"]`. */
103 origins?: string[];
104 };
105}
106
107export interface StreamProps {
108 /**
109 * Name of the stream. Unique per account; must be alphanumeric and
110 * underscores only (it is referenced as a SQL table name). If omitted,
111 * a unique name is generated from the app, stage, and logical ID.
112 * Changing the name triggers a replacement.
113 * @default ${app}_${id}_${stage}_${suffix}
114 */
115 name?: string;
116 /**
117 * Structured schema of ingested events. Immutable — changing the schema
118 * triggers a replacement. When omitted, the stream accepts unstructured
119 * JSON events.
120 */
121 schema?: {
122 /** Fields of the structured schema. */
123 fields: StreamField[];
124 };
125 /**
126 * Input format configuration. Immutable — changing it triggers a
127 * replacement.
128 * @default { type: "json" }
129 */
130 format?: StreamFormat;
131 /**
132 * HTTP ingest endpoint configuration. Mutable in place.
133 * @default { enabled: true, authentication: false }
134 */
135 http?: StreamHttp;
136 /**
137 * Whether Workers can send events to this stream via a `pipelines`
138 * binding. Mutable in place.
139 * @default { enabled: true }
140 */
141 workerBinding?: {
142 /** Whether the Worker binding is enabled. */
143 enabled: boolean;
144 };
145}
146
147export interface StreamAttributes {
148 /** Cloudflare-assigned stream identifier. */
149 streamId: string;
150 /** Account that owns the stream. */
151 accountId: string;
152 /** Stream name (unique per account). */
153 name: string;
154 /** HTTP ingest endpoint URL, when HTTP ingestion is enabled. */
155 endpoint: string | undefined;
156 /** Whether the HTTP ingest endpoint is enabled. */
157 httpEnabled: boolean;
158 /** Whether the HTTP ingest endpoint requires authentication. */
159 httpAuthentication: boolean;
160 /** Allowed CORS origins of the HTTP ingest endpoint. */
161 corsOrigins: string[] | undefined;
162 /** Whether Workers can send events via a `pipelines` binding. */
163 workerBindingEnabled: boolean;
164 /** Current version of the stream. */
165 version: number;
166 /** When the stream was created. */
167 createdAt: string;
168 /** When the stream was last modified. */
169 modifiedAt: string;
170}
171
172export type Stream = Resource<
173 StreamTypeId,
174 StreamProps,
175 StreamAttributes,
176 never,
177 Providers
178>;
179
180/**
181 * A Cloudflare Pipelines stream — the ingestion endpoint of the Pipelines
182 * product. Events are sent to a stream over HTTP (and/or from Workers via
183 * a binding), transformed by a SQL {@link Pipeline}, and written to a
184 * {@link Sink}.
185 *
186 * The stream's `schema` and `format` are fixed at creation (changing them
187 * triggers a replacement); the HTTP endpoint and Worker-binding toggles
188 * are mutable in place.
189 * ### Creating a Stream
190 * **Example:** Unstructured stream with default settings
191 * ```typescript
192 * const stream = yield* Cloudflare.Pipelines.Stream("events", {});
193 * ```
194 *
195 * **Example:** Structured stream with a typed schema
196 * ```typescript
197 * const stream = yield* Cloudflare.Pipelines.Stream("clicks", {
198 * schema: {
199 * fields: [
200 * { type: "string", name: "url", required: true },
201 * { type: "timestamp", name: "ts", unit: "millisecond" },
202 * ],
203 * },
204 * });
205 * ```
206 *
207 * ### HTTP ingestion
208 * **Example:** Authenticated endpoint with CORS
209 * ```typescript
210 * const stream = yield* Cloudflare.Pipelines.Stream("events", {
211 * http: {
212 * enabled: true,
213 * authentication: true,
214 * cors: { origins: ["https://app.example.com"] },
215 * },
216 * });
217 * // POST events to stream.endpoint with an API token
218 * ```
219 *
220 * ### Wiring into a Pipeline
221 * **Example:** Stream → SQL Pipeline → R2 Sink
222 * ```typescript
223 * const pipeline = yield* Cloudflare.Pipelines.Pipeline("etl", {
224 * sql: Output.interpolate`INSERT INTO ${sink.name} SELECT * FROM ${stream.name}`,
225 * });
226 * ```
227 *
228 * @see https://developers.cloudflare.com/pipelines/
229 *
230 * @resource
231 * @product Pipelines
232 * @category Storage & Databases
233 */
234export const Stream = Resource<Stream>(StreamTypeId);
235
236/**
237 * Returns true if the given value is a Stream resource.
238 */
239export const isStream = (value: unknown): value is Stream =>
240 Predicate.hasProperty(value, "Type") && value.Type === StreamTypeId;
241
242export const StreamProvider = () =>
243 Provider.succeed(Stream, {
244 stables: ["streamId", "accountId", "name", "createdAt"],
245
246 diff: Effect.fn(function* ({ id, olds, news, output }) {
247 const { accountId } = yield* yield* CloudflareEnvironment;
248 if (!isResolved(news)) return undefined;
249 if ((output?.accountId ?? accountId) !== accountId) {
250 return { action: "replace" } as const;
251 }
252 const o = olds as StreamProps | undefined;
253 const newName = yield* streamName(id, news.name);
254 const oldName = output?.name ?? (yield* streamName(id, o?.name));
255 if (newName !== oldName) {
256 return { action: "replace" } as const;
257 }
258 // Schema and format are immutable — only http/workerBinding are
259 // patchable.
260 if (o !== undefined && !stableEquals(news.schema, o.schema)) {
261 return { action: "replace" } as const;
262 }
263 if (o !== undefined && !stableEquals(news.format, o.format)) {
264 return { action: "replace" } as const;
265 }
266 return undefined;
267 }),
268
269 read: Effect.fn(function* ({ id, output, olds }) {
270 const { accountId } = yield* yield* CloudflareEnvironment;
271 const acct = output?.accountId ?? accountId;
272 if (output?.streamId) {
273 const observed = yield* getStream(acct, output.streamId);
274 if (observed) return toAttributes(observed, acct);
275 }
276 // Cold read — stream names are unique per account, so an exact name
277 // match identifies the resource. A generated name embeds the
278 // instance id (proof of ownership); a user-provided name does not,
279 // so gate takeover behind the adopt policy.
280 const name = yield* streamName(id, olds?.name);
281 const match = yield* findStreamByName(acct, name);
282 if (match) {
283 const attrs = toAttributes(match, acct);
284 return olds?.name !== undefined ? Unowned(attrs) : attrs;
285 }
286 return undefined;
287 }),
288
289 reconcile: Effect.fn(function* ({ id, news, output }) {
290 const { accountId } = yield* yield* CloudflareEnvironment;
291 const name = yield* streamName(id, news.name);
292
293 // 1. Observe — the cached streamId is a hint, not a guarantee; a
294 // missing stream falls through to a name lookup and then create.
295 let observed = output?.streamId
296 ? yield* getStream(output.accountId ?? accountId, output.streamId)
297 : undefined;
298 if (!observed) {
299 observed = yield* findStreamByName(accountId, name);
300 }
301
302 // 2. Ensure — create when missing. Stream names are unique, so an
303 // AlreadyExists is a race (or recovery from a lost state write):
304 // fall back to the name lookup.
305 if (!observed) {
306 observed = yield* pipelines
307 .createStream({
308 accountId,
309 name,
310 format: news.format,
311 schema: news.schema,
312 http: news.http ? desiredHttp(news.http) : undefined,
313 workerBinding: news.workerBinding,
314 })
315 .pipe(
316 Effect.catchTag("StreamAlreadyExists", (error) =>
317 findStreamByName(accountId, name).pipe(
318 Effect.flatMap((match) =>
319 match ? Effect.succeed(match) : Effect.fail(error),
320 ),
321 ),
322 ),
323 );
324 }
325
326 // 3. Sync — http and workerBinding are the only mutable aspects.
327 // Diff observed cloud state against the declared desired state
328 // and PATCH only the delta; skip the API entirely on a no-op.
329 const patch: pipelines.PatchStreamRequest = {
330 accountId,
331 streamId: observed.id,
332 };
333 let dirty = false;
334 if (news.http !== undefined) {
335 const desired = desiredHttp(news.http);
336 if (
337 desired.enabled !== observed.http.enabled ||
338 desired.authentication !== observed.http.authentication ||
339 !sameOrigins(desired.cors?.origins, observed.http.cors?.origins)
340 ) {
341 patch.http = desired;
342 dirty = true;
343 }
344 }
345 if (
346 news.workerBinding !== undefined &&
347 news.workerBinding.enabled !== observed.workerBinding.enabled
348 ) {
349 patch.workerBinding = { enabled: news.workerBinding.enabled };
350 dirty = true;
351 }
352 if (dirty) {
353 observed = yield* pipelines.patchStream(patch);
354 }
355
356 return toAttributes(observed, accountId);
357 }),
358
359 delete: Effect.fn(function* ({ output }) {
360 // A dependent pipeline's own deletion may still be propagating —
361 // ride out `StreamInUse` (HTTP 422 "still in use") with a bounded
362 // retry. A stream that is already gone is success.
363 yield* pipelines
364 .deleteStream({
365 accountId: output.accountId,
366 streamId: output.streamId,
367 })
368 .pipe(
369 Effect.retry({
370 while: (e) => e._tag === "StreamInUse",
371 schedule: Schedule.max([
372 Schedule.exponential("500 millis"),
373 Schedule.recurs(8),
374 ]),
375 }),
376 Effect.catchTag("StreamNotFound", () => Effect.void),
377 Effect.catchTag("InvalidStreamId", () => Effect.void),
378 );
379 }),
380
381 // Account-scoped collection: exhaustively paginate every stream in the
382 // account and hydrate each into the same Attributes shape `read`
383 // returns. The list item shape is structurally identical to the
384 // get/create response consumed by `toAttributes`.
385 list: Effect.fn(function* () {
386 const { accountId } = yield* yield* CloudflareEnvironment;
387 return yield* pipelines.listStreams.pages({ accountId }).pipe(
388 EffectStream.runCollect,
389 Effect.map((chunk) =>
390 Array.from(chunk).flatMap((page) =>
391 (page.result ?? []).map((s) => toAttributes(s, accountId)),
392 ),
393 ),
394 );
395 }),
396 });
397
398/**
399 * The subset of stream state shared by get/list/create/patch responses.
400 */
401interface ObservedStream {
402 id: string;
403 name: string;
404 endpoint?: string | null;
405 http: {
406 enabled: boolean;
407 authentication: boolean;
408 cors?: { origins?: readonly string[] | null } | null;
409 };
410 workerBinding: { enabled: boolean };
411 version: number;
412 createdAt: string;
413 modifiedAt: string;
414}
415
416// Pipelines entity names must be alphanumeric/underscore only (they are
417// referenced as SQL table names), so swap the default hyphen delimiter
418// for underscores.
419const streamName = (id: string, name: string | undefined) =>
420 Effect.gen(function* () {
421 if (name) return name;
422 const generated = yield* createPhysicalName({
423 id,
424 lowercase: true,
425 delimiter: "_",
426 });
427 return generated.replaceAll(/[^a-zA-Z0-9_]/g, "_");
428 });
429
430/**
431 * Read a stream by id, mapping "gone" (`StreamNotFound`, Cloudflare error
432 * code 1016, or `InvalidStreamId` for a malformed/foreign id) to
433 * `undefined`.
434 */
435const getStream = (accountId: string, streamId: string) =>
436 pipelines.getStream({ accountId, streamId }).pipe(
437 Effect.map((s): ObservedStream | undefined => s),
438 Effect.catchTag("StreamNotFound", () => Effect.succeed(undefined)),
439 Effect.catchTag("InvalidStreamId", () => Effect.succeed(undefined)),
440 );
441
442const findStreamByName = (accountId: string, name: string) =>
443 pipelines.listStreams.items({ accountId }).pipe(
444 EffectStream.runCollect,
445 Effect.map((chunk): ObservedStream | undefined =>
446 Array.from(chunk).find((s) => s.name === name),
447 ),
448 );
449
450const desiredHttp = (http: StreamHttp) => ({
451 enabled: http.enabled ?? true,
452 authentication: http.authentication ?? false,
453 cors: http.cors?.origins ? { origins: http.cors.origins } : undefined,
454});
455
456const sameOrigins = (
457 desired: readonly string[] | undefined,
458 observed: readonly string[] | null | undefined,
459) => {
460 // Only diff origins when the user declared them; Cloudflare's default
461 // is not ours to fight.
462 if (desired === undefined) return true;
463 const have = observed ?? [];
464 return (
465 desired.length === have.length &&
466 [...desired].sort().join(",") === [...have].sort().join(",")
467 );
468};
469
470/**
471 * Key-order-insensitive structural equality for plain JSON-ish prop
472 * values (schemas, formats).
473 */
474const stableEquals = (a: unknown, b: unknown): boolean =>
475 stableStringify(a) === stableStringify(b);
476
477const stableStringify = (value: unknown): string =>
478 JSON.stringify(value, (_key, v) =>
479 v !== null && typeof v === "object" && !Array.isArray(v)
480 ? Object.fromEntries(
481 Object.entries(v as Record<string, unknown>).sort(([x], [y]) =>
482 x.localeCompare(y),
483 ),
484 )
485 : v,
486 ) ?? "undefined";
487
488const toAttributes = (
489 observed: ObservedStream,
490 accountId: string,
491): StreamAttributes => ({
492 streamId: observed.id,
493 accountId,
494 name: observed.name,
495 endpoint: observed.endpoint ?? undefined,
496 httpEnabled: observed.http.enabled,
497 httpAuthentication: observed.http.authentication,
498 corsOrigins: observed.http.cors?.origins
499 ? [...observed.http.cors.origins]
500 : undefined,
501 workerBindingEnabled: observed.workerBinding.enabled,
502 version: observed.version,
503 createdAt: observed.createdAt,
504 modifiedAt: observed.modifiedAt,
505});