Add Worker Pipelines Binding to Miniflare (#7950)

* Add Worker Pipeline binding support

Adding support to Miniflare to reference bindings during `wrangler dev` or during tests via vitest.

This binding uses newer JSRPC method instead of propagating changes through `workerd.capnp`.

* Enable Workers Pipelines test for wrangler dev

* Add vitest example for Pipelines

* Wire Pipelines from Miniflare to Wrangler dev

* Update packages/wrangler/e2e/dev-with-resources.test.ts

Pipelines does not support remote bindings, only run this test for local e2e tests.

* Add id to service name

* Use correct string interpolation

Co-authored-by: emily-shen <[email protected]>

* Make pipelines not backed by DO resources

building off of Emily's feedback on PR #7950 and example commit 56c1e43

* Mark wrangler and miniflare as patch, update README and test output in e2e

* Read output until pipeline message is logged

---------

Co-authored-by: emily-shen <[email protected]>

Author: cmackenzie1

.changeset/slimy-otters-rush.md

@@ -0,0 +1,6 @@
+---
+"miniflare": patch
+"wrangler": patch
+---
+
+Add local binding support for Worker Pipelines

fixtures/vitest-pool-workers-examples/README.md

@@ -11,6 +11,7 @@ This directory contains example projects tested with `@cloudflare/vitest-pool-wo
 | [📚 d1](d1)                                                                        | Isolated tests using D1 with migrations                   |
 | [📌 durable-objects](durable-objects)                                              | Isolated tests using Durable Objects with direct access   |
 | [🚥 queues](queues)                                                                | Tests using Queue producers and consumers                 |
+| [🚰 pipelines](pipelines)                                                          | Tests using Pipelines                                     |
 | [🚀 hyperdrive](hyperdrive)                                                        | Tests using Hyperdrive with a Vitest managed TCP server   |
 | [🤹 request-mocking](request-mocking)                                              | Tests using declarative/imperative outbound request mocks |
 | [🔌 multiple-workers](multiple-workers)                                            | Tests using multiple auxiliary workers and request mocks  |

fixtures/vitest-pool-workers-examples/pipelines/README.md

@@ -0,0 +1,8 @@
+# 🚰 pipelines
+
+This Worker implements endpoint that send details of the incoming HTTP request to a Pipeline.
+
+| Test                                                                                  | Overview                                                          |
+| ------------------------------------------------------------------------------------- | ----------------------------------------------------------------- |
+| [pipeline-send-integration-self.test.ts](test/pipeline-send-integration-self.test.ts) | Integration tests for endpoints using `SELF`                      |
+| [pipeline-send-unit.test.ts](test/pipeline-send-unit.test.ts)                         | Unit tests calling `worker.fetch()` directly mocking the Pipeline |

fixtures/vitest-pool-workers-examples/pipelines/src/env.d.ts

@@ -0,0 +1,3 @@
+interface Env {
+	PIPELINE: Pipeline;
+}

fixtures/vitest-pool-workers-examples/pipelines/src/index.ts

@@ -0,0 +1,11 @@
+export default {
+	async fetch(request, env, ctx) {
+		await env.PIPELINE.send([
+			{
+				method: request.method.toUpperCase(),
+				url: request.url,
+			},
+		]);
+		return new Response("Accepted", { status: 202 });
+	},
+} satisfies ExportedHandler<Env>;

fixtures/vitest-pool-workers-examples/pipelines/src/tsconfig.json

@@ -0,0 +1,4 @@
+{
+	"extends": "../../tsconfig.workerd.json",
+	"include": ["./**/*.ts"]
+}

fixtures/vitest-pool-workers-examples/pipelines/test/env.d.ts

@@ -0,0 +1,4 @@
+declare module "cloudflare:test" {
+	// Controls the type of `import("cloudflare:test").env`
+	interface ProvidedEnv extends Env {}
+}

fixtures/vitest-pool-workers-examples/pipelines/test/pipeline-send-integration-self.test.ts

@@ -0,0 +1,12 @@
+import { SELF } from "cloudflare:test";
+import { expect, it } from "vitest";
+
+it("sends message to pipeline", async () => {
+	// Send data to the Pipeline
+	let response = await SELF.fetch("https://example.com/ingest", {
+		method: "POST",
+		body: "value",
+	});
+	expect(response.status).toBe(202);
+	expect(await response.text()).toBe("Accepted");
+});

fixtures/vitest-pool-workers-examples/pipelines/test/pipeline-send-unit.test.ts

@@ -0,0 +1,45 @@
+import {
+	createExecutionContext,
+	env,
+	waitOnExecutionContext,
+} from "cloudflare:test";
+import { afterEach, expect, it, vi } from "vitest";
+import worker from "../src/index";
+
+// This will improve in the next major version of `@cloudflare/workers-types`,
+// but for now you'll need to do something like this to get a correctly-typed
+// `Request` to pass to `worker.fetch()`.
+const IncomingRequest = Request<unknown, IncomingRequestCfProperties>;
+
+afterEach(() => {
+	vi.restoreAllMocks();
+});
+
+it("produces message to pipeline", async () => {
+	const mockPipeline = {
+		send: vi.fn().mockResolvedValue(undefined),
+	};
+
+	const testEnv = {
+		...env,
+		PIPELINE: mockPipeline,
+	};
+
+	// Send data to pipeline
+	const request = new IncomingRequest("https://example.com/ingest", {
+		method: "POST",
+		body: "value",
+	});
+	const ctx = createExecutionContext();
+	const response = await worker.fetch(request, testEnv, ctx);
+	await waitOnExecutionContext(ctx);
+
+	expect(response.status).toBe(202);
+	expect(await response.text()).toBe("Accepted");
+
+	// Check `PIPELINE.send()` was called
+	expect(mockPipeline.send).toBeCalledTimes(1);
+	expect(mockPipeline.send).toBeCalledWith([
+		{ method: "POST", url: "https://example.com/ingest" },
+	]);
+});

fixtures/vitest-pool-workers-examples/pipelines/test/tsconfig.json

@@ -0,0 +1,4 @@
+{
+	"extends": "../../tsconfig.workerd-test.json",
+	"include": ["./**/*.ts", "../src/env.d.ts"]
+}