1import * as Cloudflare from "@/Cloudflare";
2import * as Effect from "effect/Effect";
3import { HttpServerRequest } from "effect/http/HttpServerRequest";
4import * as HttpServerResponse from "effect/http/HttpServerResponse";
5import { EventsStream } from "./stream.ts";
6
7/**
8 * Effect-native Worker fixture for the Pipelines stream binding, going
9 * through the `WriteStream` capability: the binding is registered by
10 * `yield* Cloudflare.Pipelines.WriteStream(EventsStream)` and the client
11 * exposes `send` as an Effect — no `env` declaration, no `Effect.promise`.
12 */
13export default class PipelinesEffectWorker extends Cloudflare.Worker<PipelinesEffectWorker>()(
14 "PipelinesEffectWorker",
15 {
16 main: import.meta.url,
17 },
18 Effect.gen(function* () {
19 const events = yield* Cloudflare.Pipelines.WriteStream(EventsStream);
20
21 return {
22 fetch: Effect.gen(function* () {
23 const request = yield* HttpServerRequest;
24 const url = new URL(request.originalUrl, "http://x");
25 // `send` only exists on a real `pipelines` binding — a `json`
26 // binding would deliver the stream's attributes instead.
27 const kind = typeof (yield* events.raw)?.send;
28 if (url.pathname === "/send") {
29 yield* events
30 .send([
31 {
32 source: "pipelines-binding-test",
33 nonce: url.searchParams.get("nonce") ?? "none",
34 },
35 ])
36 .pipe(Effect.orDie);
37 return yield* HttpServerResponse.json({
38 mode: "effect",
39 sent: true,
40 kind,
41 });
42 }
43 return yield* HttpServerResponse.json({
44 mode: "effect",
45 sent: false,
46 kind,
47 });
48 }),
49 };
50 }).pipe(Effect.provide(Cloudflare.Pipelines.WriteStreamBinding)),
51) {}