ch23-pipelines

取得並執行

這個範例可以獨立 clone 執行，不依賴其他章節。

npx degit mrsuner/cloudflare-serverless-platform-manuals/examples/ch23-pipelines
cd ch23-pipelines
npm install

可用指令

npm run dev	# wrangler dev
npm run deploy	# wrangler deploy
npm run types	# wrangler types --env-interface CloudflareBindings

說明

ch23 — Pipelines / R2 SQL

Probe project for chapter 23. Pipelines, R2 SQL and R2 Data Catalog are all in open beta and the docs contradict themselves in several places; this project pins down what the tooling actually does.

Run

npm install
npx wrangler types --env-interface CloudflareBindings
npx wrangler dev --port 9025

Routes

Route	What it shows
`/shape`	Walks the binding’s whole prototype chain. It is a `Fetcher`, and `send` is a `[object JsRpcProperty]` — Pipelines is built on Workers RPC (chapter 18).
`/send`	Seven probes: normal, 50 records, empty array, non-array, no arguments, deeply nested, non-serializable. All resolve to `undefined` locally.
`/rpc`	Shows that any property name is `typeof "function"`; a typo only fails at call time with `The RPC receiver does not implement the method "sendd"`.
`/others`	The same `send()` through a binding declared with the deprecated `pipeline` key.

The binding key was renamed

config-schema.json is explicit:

"pipeline": { "type": "string", "deprecated": "Use `stream` instead." }

Using it produces, on every wrangler invocation:

▲ WARNING Processing wrangler.jsonc configuration:
    - The "pipeline" field in "pipelines[1]" bindings is deprecated. Use "stream" instead.

The canonical docs page for this — /pipelines/streams/writing-to-streams/ — still shows the old key with no deprecation note. Use stream.

Note the outer array is still called pipelines; only the inner key changed.

Schema is looser than the validator

The JSON schema requires only binding, so an editor will accept { "binding": "X" }. Wrangler will not:

✘ ERROR  - "pipelines[2]" bindings must have a string "stream" field but got {"binding":"NO_TARGET"}.

Trust wrangler deploy --dry-run, not editor autocomplete.

`remote` works, and is undocumented

Pipelines has no row in the bindings-per-env matrix and no doc page mentions wrangler dev. But the schema has:

"remote": { "type": "boolean",
            "description": "Whether the pipeline should be remote or not in local development" }

and "remote": true passes deploy --dry-run without a warning. Since the local binding stores nothing (below), remote: true is the only way to develop against Pipelines with any confidence.

The local binding is a no-op

curl -s localhost:9025/send | jq 'to_entries|map(.value.ok)|unique'   # ["undefined"]
ls .wrangler/state/v3/                                               # no pipelines dir

No validation, no persistence, no query path.

Reproducing the RPC finding

curl -s localhost:9025/shape | jq '{sendSource, protoChain: [.protoChain[].ctor]}'
# sendSource: "[object JsRpcProperty]"
# protoChain: ["Fetcher","Fetcher","Object"]

curl -s localhost:9025/rpc | jq
# typeofTypo: "function"  -- feature detection is useless on this binding
# callTypo:   TypeError: The RPC receiver does not implement the method "sendd".
# callFetch:  Error: Handler does not export a fetch() function.

A fossil in the runtime types

cloudflare:pipelines still exports PipelineTransformationEntrypoint, the pre-2025-09 Worker-based transform model. The current architecture uses SQL transforms and this class appears nowhere in the docs. Do not use it.

CLI surface (wrangler 4.118.0)

wrangler pipelines setup | create | list | get | delete
wrangler pipelines update <p>          # legacy pipelines only
wrangler pipelines streams create|list|get|delete
wrangler pipelines sinks   create|list|get|delete
wrangler r2 sql query <warehouse> <query>
wrangler r2 bucket catalog enable|disable|get|compaction|snapshot-expiration

There is no streams update and no sinks update, and pipeline SQL cannot be modified after creation. Everything is immutable — change means delete and recreate.

Defaults the docs never state: --http-enabled true, --http-auth true, --format parquet, --compression zstd, --roll-interval 300.

`wrangler.jsonc`

{
  "$schema": "./node_modules/wrangler/config-schema.json",
  "name": "ch23-pipelines",
  "main": "src/index.ts",
  "compatibility_date": "2026-07-24",
  "observability": { "enabled": true },
  "pipelines": [
    { "binding": "EVENTS", "stream": "ch23-clicks" },
    // The pre-2026-05 key. Schema marks it deprecated: "Use `stream` instead."
    { "binding": "OLD_KEY", "pipeline": "ch23-legacy" }
  ]
}

`package.json`

{
  "name": "ch23-pipelines",
  "private": true,
  "scripts": { "dev": "wrangler dev", "deploy": "wrangler deploy", "types": "wrangler types --env-interface CloudflareBindings" },
  "devDependencies": { "wrangler": "^4.118.0", "typescript": "^5.9.2" }
}

`src/index.ts`

const t = async (fn: () => unknown): Promise<unknown> => {
  try { const v = await fn(); return { ok: v === undefined ? "undefined" : v }; }
  catch (e) { return { threwName: (e as Error).name, threw: String(e).slice(0, 300) }; }
};

export default {
  async fetch(request: Request, env: CloudflareBindings): Promise<Response> {
    const url = new URL(request.url);
    const p = env.EVENTS as unknown as CloudflareBindings["EVENTS"] & { fetch(u: string): Promise<Response> };
    switch (url.pathname) {
      case "/shape": {
        const chain: unknown[] = [];
        let o: object | null = p as unknown as object;
        while (o) {
          chain.push({ ctor: (o as { constructor?: { name?: string } })?.constructor?.name ?? null,
                       keys: Object.getOwnPropertyNames(o) });
          o = Object.getPrototypeOf(o);
        }
        return Response.json({
          typeofBinding: typeof p,
          hasSend: "send" in (p as object),
          typeofSend: typeof (p as { send?: unknown }).send,
          sendName: ((p as { send?: { name?: string } }).send)?.name ?? null,
          sendLength: ((p as { send?: { length?: number } }).send)?.length ?? null,
          sendSource: String((p as { send?: unknown }).send).slice(0, 200),
          protoChain: chain,
        });
      }
      case "/send":
        return Response.json({
          one: await t(() => p.send([{ slug: "abc", country: "TW", ts: Date.now() }])),
          many: await t(() => p.send(Array.from({ length: 50 }, (_, i) => ({ i })))),
          empty: await t(() => p.send([])),
          notArray: await t(() => (p as unknown as { send(x: unknown): Promise<void> }).send({ a: 1 })),
          noArgs: await t(() => (p as unknown as { send(): Promise<void> }).send()),
          nested: await t(() => p.send([{ a: { b: [1, 2, { c: "d" }] } }])),
          nonSerializable: await t(() => p.send([{ fn: (() => 1) as unknown as string }])),
        });
      case "/rpc": {
        const anyP = p as unknown as Record<string, (...a: unknown[]) => Promise<unknown>>;
        return Response.json({
          // The binding is a Fetcher; `send` is a JsRpcProperty, not a real method.
          // So does a name that does not exist also look like a function?
          typeofTypo: typeof anyP.sendd,
          typoSource: String(anyP.sendd).slice(0, 60),
          callTypo: await t(() => anyP.sendd([{ a: 1 }])),
          callFetch: await t(() => p.fetch("http://pipeline/")),
        });
      }
      case "/others":
        return Response.json({
          oldKey: await t(() => env.OLD_KEY.send([{ via: "deprecated pipeline key" }])),
        });
      default:
        return new Response("/shape /send /others\n", { status: 404 });
    }
  },
} satisfies ExportedHandler<CloudflareBindings>;

`tsconfig.json`

{
  "compilerOptions": {
    "target": "esnext",
    "lib": ["esnext"],
    "module": "esnext",
    "moduleResolution": "bundler",
    "types": ["./worker-configuration.d.ts"],
    "strict": true,
    "skipLibCheck": true,
    "noEmit": true,
    "isolatedModules": true
  },
  "include": ["src/**/*.ts", "worker-configuration.d.ts"]
}