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 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.Pipelines.Pipeline" as const;
16type TypeId = typeof TypeId;
17
18export interface PipelineProps {
19 /**
20 * Name of the pipeline. Unique per account; must be alphanumeric and
21 * underscores only. If omitted, a unique name is generated from the
22 * app, stage, and logical ID.
23 *
24 * Pipelines have no update API, so changing this (or the `sql`)
25 * triggers a replacement.
26 * @default ${app}_${id}_${stage}_${suffix}
27 */
28 name?: string;
29 /**
30 * SQL statement describing the processing flow, e.g.
31 * `INSERT INTO my_sink SELECT * FROM my_stream`. Streams and sinks are
32 * referenced by name — interpolate `stream.name` / `sink.name` outputs
33 * so the engine orders the pipeline after them on deploy (and before
34 * them on destroy).
35 *
36 * Immutable — changing the SQL triggers a replacement.
37 */
38 sql: string;
39}
40
41export interface PipelineAttributes {
42 /** Cloudflare-assigned pipeline identifier. */
43 pipelineId: string;
44 /** Account that owns the pipeline. */
45 accountId: string;
46 /** Pipeline name (unique per account). */
47 name: string;
48 /** SQL statement of the processing flow. */
49 sql: string;
50 /** Current status of the pipeline. */
51 status: string;
52 /** When the pipeline was created. */
53 createdAt: string;
54 /** When the pipeline was last modified. */
55 modifiedAt: string;
56}
57
58export type Pipeline = Resource<
59 TypeId,
60 PipelineProps,
61 PipelineAttributes,
62 never,
63 Providers
64>;
65
66/**
67 * A Cloudflare SQL Pipeline — the transform of the Pipelines product. A
68 * pipeline is a single SQL statement that reads events from a
69 * {@link Stream} and writes them to a {@link Sink}, both
70 * referenced by name.
71 *
72 * The SQL is fixed at creation: changing it (or the name) triggers a
73 * replacement. Nothing references a pipeline downstream, so replacements
74 * are cheap.
75 * ### Creating a Pipeline
76 * **Example:** Stream → Sink passthrough
77 * ```typescript
78 * const stream = yield* Cloudflare.Pipelines.Stream("events", {});
79 * const sink = yield* Cloudflare.Pipelines.Sink("events-sink", {
80 * type: "r2",
81 * config: { bucket: bucket.bucketName, credentials },
82 * });
83 *
84 * const pipeline = yield* Cloudflare.Pipelines.Pipeline("etl", {
85 * sql: Output.interpolate`INSERT INTO ${sink.name} SELECT * FROM ${stream.name}`,
86 * });
87 * ```
88 *
89 * **Example:** Filtering transform
90 * ```typescript
91 * const pipeline = yield* Cloudflare.Pipelines.Pipeline("errors-only", {
92 * sql: Output.interpolate`INSERT INTO ${sink.name} SELECT * FROM ${stream.name} WHERE level = 'error'`,
93 * });
94 * ```
95 *
96 * @see https://developers.cloudflare.com/pipelines/
97 *
98 * @resource
99 * @product Pipelines
100 * @category Storage & Databases
101 */
102export const Pipeline = Resource<Pipeline>(TypeId);
103
104/**
105 * Returns true if the given value is a Pipeline resource.
106 */
107export const isPipeline = (value: unknown): value is Pipeline =>
108 Predicate.hasProperty(value, "Type") && value.Type === TypeId;
109
110export const PipelineProvider = () =>
111 Provider.succeed(Pipeline, {
112 stables: ["pipelineId", "accountId", "name", "createdAt"],
113
114 diff: Effect.fn(function* ({ id, olds, news, output }) {
115 const { accountId } = yield* yield* CloudflareEnvironment;
116 if (!isResolved(news)) return undefined;
117 if ((output?.accountId ?? accountId) !== accountId) {
118 return { action: "replace" } as const;
119 }
120 const o = olds as PipelineProps | undefined;
121 if (o === undefined) return undefined;
122 // Pipelines have no update API — name and sql changes replace.
123 const newName = yield* pipelineName(id, news.name);
124 const oldName = output?.name ?? (yield* pipelineName(id, o.name));
125 if (newName !== oldName) {
126 return { action: "replace" } as const;
127 }
128 const oldSql = typeof o.sql === "string" ? o.sql : output?.sql;
129 if (oldSql !== undefined && news.sql !== oldSql) {
130 return { action: "replace" } as const;
131 }
132 return undefined;
133 }),
134
135 read: Effect.fn(function* ({ id, output, olds }) {
136 const { accountId } = yield* yield* CloudflareEnvironment;
137 const acct = output?.accountId ?? accountId;
138 if (output?.pipelineId) {
139 const observed = yield* getPipeline(acct, output.pipelineId);
140 if (observed) return toAttributes(observed, acct);
141 }
142 // Cold read — pipeline names are unique per account; a
143 // generated-name match is proof of ownership, an explicit name is
144 // not.
145 const name = yield* pipelineName(id, olds?.name);
146 const match = yield* findPipelineByName(acct, name);
147 if (match) {
148 const attrs = toAttributes(match, acct);
149 return olds?.name !== undefined ? Unowned(attrs) : attrs;
150 }
151 return undefined;
152 }),
153
154 reconcile: Effect.fn(function* ({ id, news, output }) {
155 const { accountId } = yield* yield* CloudflareEnvironment;
156 const name = yield* pipelineName(id, news.name);
157 const sql = news.sql as string;
158
159 // 1. Observe — by cached id first, then by (unique) name so we
160 // recover from lost state writes.
161 let observed = output?.pipelineId
162 ? yield* getPipeline(output.accountId ?? accountId, output.pipelineId)
163 : undefined;
164 if (!observed) {
165 observed = yield* findPipelineByName(accountId, name);
166 }
167
168 // Converge drift — there is no update API, so when the observed SQL
169 // differs from the desired SQL (e.g. a referenced stream/sink was
170 // replaced and `sql` was unresolved at diff time), delete the stale
171 // pipeline and fall through to recreate it under the same name.
172 // Nothing references a pipeline downstream, so this is safe.
173 if (observed && (observed.sql !== sql || observed.name !== name)) {
174 yield* deletePipeline(accountId, observed.id);
175 // Wait until the delete is visible so the recreate below does not
176 // race an `PipelineAlreadyExists` against the dying pipeline.
177 yield* getPipeline(accountId, observed.id).pipe(
178 Effect.repeat({
179 schedule: Schedule.max([
180 Schedule.exponential("250 millis"),
181 Schedule.recurs(8),
182 ]),
183 until: (p) => p === undefined,
184 }),
185 );
186 observed = undefined;
187 }
188
189 // 2. Ensure — create when missing. The referenced stream/sink may
190 // have been created moments ago, so ride out eventual-consistency
191 // `TableNotFound` blips with a bounded retry. An AlreadyExists is
192 // a race or recovery — resolve it via the name lookup. There is
193 // no sync step: the SQL is immutable, so changes arrive as
194 // replacements (diff).
195 if (!observed) {
196 observed = yield* pipelines
197 .createV1Pipeline({ accountId, name, sql })
198 .pipe(
199 Effect.retry({
200 while: (e) => e._tag === "TableNotFound",
201 schedule: Schedule.max([
202 Schedule.exponential("500 millis"),
203 Schedule.recurs(6),
204 ]),
205 }),
206 Effect.catchTag("PipelineAlreadyExists", (error) =>
207 findPipelineByName(accountId, name).pipe(
208 Effect.flatMap((match) =>
209 match ? Effect.succeed(match) : Effect.fail(error),
210 ),
211 ),
212 ),
213 );
214 }
215
216 return toAttributes(observed, accountId);
217 }),
218
219 delete: Effect.fn(function* ({ output }) {
220 yield* deletePipeline(output.accountId, output.pipelineId);
221 }),
222
223 // Account collection — pipelines are account-scoped. Exhaustively
224 // paginate the account-wide list and hydrate each row into the same
225 // Attributes shape `read` returns.
226 list: Effect.fn(function* () {
227 const { accountId } = yield* yield* CloudflareEnvironment;
228 return yield* pipelines.listV1Pipeline.pages({ accountId }).pipe(
229 Stream.runCollect,
230 Effect.map((chunk) =>
231 Array.from(chunk).flatMap((page) =>
232 (page.result ?? []).map((p) => toAttributes(p, accountId)),
233 ),
234 ),
235 );
236 }),
237 });
238
239/**
240 * The subset of pipeline state shared by get/list/create responses.
241 */
242interface ObservedPipeline {
243 id: string;
244 name: string;
245 sql: string;
246 status: string;
247 createdAt: string;
248 modifiedAt: string;
249}
250
251// Pipelines entity names must be alphanumeric/underscore only, so swap
252// the default hyphen delimiter for underscores.
253const pipelineName = (id: string, name: string | undefined) =>
254 Effect.gen(function* () {
255 if (name) return name;
256 const generated = yield* createPhysicalName({
257 id,
258 lowercase: true,
259 delimiter: "_",
260 });
261 return generated.replaceAll(/[^a-zA-Z0-9_]/g, "_");
262 });
263
264/**
265 * Read a pipeline by id, mapping "gone" (`PipelineNotExists`, Cloudflare
266 * error code 1000) to `undefined`.
267 */
268const getPipeline = (accountId: string, pipelineId: string) =>
269 pipelines.getV1Pipeline({ accountId, pipelineId }).pipe(
270 Effect.map((p): ObservedPipeline | undefined => p),
271 Effect.catchTag("PipelineNotExists", () => Effect.succeed(undefined)),
272 );
273
274/**
275 * Idempotent delete — a pipeline that is already gone is success.
276 */
277const deletePipeline = (accountId: string, pipelineId: string) =>
278 pipelines
279 .deleteV1Pipeline({ accountId, pipelineId })
280 .pipe(Effect.catchTag("PipelineNotExists", () => Effect.void));
281
282const findPipelineByName = (accountId: string, name: string) =>
283 pipelines.listV1Pipeline.items({ accountId }).pipe(
284 Stream.runCollect,
285 Effect.map((chunk): ObservedPipeline | undefined =>
286 Array.from(chunk).find((p) => p.name === name),
287 ),
288 );
289
290const toAttributes = (
291 observed: ObservedPipeline,
292 accountId: string,
293): PipelineAttributes => ({
294 pipelineId: observed.id,
295 accountId,
296 name: observed.name,
297 sql: observed.sql,
298 status: observed.status,
299 createdAt: observed.createdAt,
300 modifiedAt: observed.modifiedAt,
301});