Cloudflare.Pipelines reference

LegacyPipeline

Source: src/Cloudflare/Pipelines/LegacyPipeline.ts

A legacy Cloudflare Pipeline — the original HTTP-ingest → R2 batch product (/accounts/{account}/pipelines).

A legacy pipeline accepts JSON events over HTTP (and/or a Worker pipelines binding) and batches them into an R2 bucket using S3-compatible credentials.

LegacyPipeline: Creating a Legacy Pipeline

HTTP ingest into R2

The S3-compatible credentials are derived from a Cloudflare API token: the access key id is the token id and the secret is the SHA-256 hex digest of the token value.

const bucket = yield* Cloudflare.R2.Bucket("events", {});

const pipeline = yield* Cloudflare.Pipelines.LegacyPipeline("ingest", {
  destination: {
    bucket: bucket.bucketName,
    credentials: {
      accessKeyId: yield* Config.Redacted("R2_ACCESS_KEY_ID"),
      secretAccessKey: yield* Config.Redacted("R2_SECRET_ACCESS_KEY"),
    },
  },
});
// POST events to pipeline.endpoint

Tuned batching and CORS

const pipeline = yield* Cloudflare.Pipelines.LegacyPipeline("ingest", {
  source: [
    { type: "http", cors: { origins: ["https://example.com"] } },
  ],
  destination: {
    bucket: bucket.bucketName,
    credentials,
    batch: { maxDurationS: 10, maxRows: 1000 },
    compression: "gzip",
    prefix: "ingest",
  },
});

Pipeline

Source: src/Cloudflare/Pipelines/Pipeline.ts

A Cloudflare SQL Pipeline — the transform of the Pipelines product. A pipeline is a single SQL statement that reads events from a Stream and writes them to a Sink, both referenced by name.

The SQL is fixed at creation: changing it (or the name) triggers a replacement. Nothing references a pipeline downstream, so replacements are cheap.

Pipeline: Creating a Pipeline

Stream → Sink passthrough

const stream = yield* Cloudflare.Pipelines.Stream("events", {});
const sink = yield* Cloudflare.Pipelines.Sink("events-sink", {
  type: "r2",
  config: { bucket: bucket.bucketName, credentials },
});

const pipeline = yield* Cloudflare.Pipelines.Pipeline("etl", {
  sql: Output.interpolate`INSERT INTO ${sink.name} SELECT * FROM ${stream.name}`,
});

Filtering transform

const pipeline = yield* Cloudflare.Pipelines.Pipeline("errors-only", {
  sql: Output.interpolate`INSERT INTO ${sink.name} SELECT * FROM ${stream.name} WHERE level = 'error'`,
});

Sink

Source: src/Cloudflare/Pipelines/Sink.ts

A Cloudflare Pipelines sink — the destination of the Pipelines product. A SQL Pipeline reads events from a Stream and writes them to a sink, which stores them in R2 either as raw files (r2) or as Iceberg tables via the R2 Data Catalog (r2_data_catalog).

Sinks have no update API: every property change triggers a replacement. With engine-generated names this is seamless (the new sink gets a fresh name before the old one is deleted); with an explicit name the create-before-delete replacement collides, so prefer generated names.

Sink: Creating a Sink

R2 sink with JSON output

The S3-compatible credentials are derived from a Cloudflare API token: the access key id is the token id and the secret is the SHA-256 hex digest of the token value.

const bucket = yield* Cloudflare.R2.Bucket("events", {});

const sink = yield* Cloudflare.Pipelines.Sink("events-sink", {
  type: "r2",
  config: {
    bucket: bucket.bucketName,
    credentials: {
      accessKeyId: yield* Config.Redacted("R2_ACCESS_KEY_ID"),
      secretAccessKey: yield* Config.Redacted("R2_SECRET_ACCESS_KEY"),
    },
    path: "ingest",
    rollingPolicy: { intervalSeconds: 30 },
  },
});

Parquet output

const sink = yield* Cloudflare.Pipelines.Sink("parquet-sink", {
  type: "r2",
  config: { bucket: bucket.bucketName, credentials },
  format: { type: "parquet", compression: "zstd" },
});

Sink: R2 Data Catalog

const sink = yield* Cloudflare.Pipelines.Sink("iceberg-sink", {
  type: "r2_data_catalog",
  config: {
    bucket: bucket.bucketName,
    tableName: "events",
    namespace: "default",
    token: yield* Config.Redacted("CATALOG_TOKEN"),
  },
});

Stream

Source: src/Cloudflare/Pipelines/Stream.ts

A Cloudflare Pipelines stream — the ingestion endpoint of the Pipelines product. Events are sent to a stream over HTTP (and/or from Workers via a binding), transformed by a SQL Pipeline, and written to a Sink.

The stream’s schema and format are fixed at creation (changing them triggers a replacement); the HTTP endpoint and Worker-binding toggles are mutable in place.

Stream: Creating a Stream

Unstructured stream with default settings

const stream = yield* Cloudflare.Pipelines.Stream("events", {});

Structured stream with a typed schema

const stream = yield* Cloudflare.Pipelines.Stream("clicks", {
  schema: {
    fields: [
      { type: "string", name: "url", required: true },
      { type: "timestamp", name: "ts", unit: "millisecond" },
    ],
  },
});

Stream: HTTP ingestion

const stream = yield* Cloudflare.Pipelines.Stream("events", {
  http: {
    enabled: true,
    authentication: true,
    cors: { origins: ["https://app.example.com"] },
  },
});
// POST events to stream.endpoint with an API token

Stream: Wiring into a Pipeline

const pipeline = yield* Cloudflare.Pipelines.Pipeline("etl", {
  sql: Output.interpolate`INSERT INTO ${sink.name} SELECT * FROM ${stream.name}`,
});

WriteStream

Source: src/Cloudflare/Pipelines/WriteStream.ts

Binding service that turns a Pipelines Stream (or a LegacyPipeline) into a typed WriteStreamClient you can call from a Worker’s runtime Effect.

The Cloudflare Worker pipelines binding is producer-only — send ingests a batch of JSON records into the stream.

WriteStream: Sending Events

const events = yield* Cloudflare.Pipelines.WriteStream(Stream);

return {
  fetch: Effect.gen(function* () {
    yield* events.send([{ event: "click", at: new Date().toISOString() }]);
    return HttpServerResponse.empty({ status: 202 });
  }),
};

Provide WriteStreamBinding (native Worker binding) in the worker’s runtime layer to resolve the underlying stream at request time.

WriteStream is a single identifier that is simultaneously the binding’s Context tag, its type, and the callable — yield* Cloudflare.Pipelines.WriteStream(stream).