1import type { Pipeline, PipelineRecord } from "cloudflare:pipelines";
2import * as Data from "effect/Data";
3import type * as Effect from "effect/Effect";
4import * as Binding from "../../Binding.ts";
5import type { RuntimeContext } from "../../RuntimeContext.ts";
6import type { LegacyPipeline } from "./LegacyPipeline.ts";
7import type { Stream } from "./Stream.ts";
8
9/**
10 * Binding service that turns a Pipelines {@link Stream} (or a
11 * {@link LegacyPipeline}) into a typed {@link WriteStreamClient} you can
12 * call from a Worker's runtime Effect.
13 *
14 * The Cloudflare Worker `pipelines` binding is producer-only — `send`
15 * ingests a batch of JSON records into the stream.
16 * ### Sending Events
17 * **Example:** Producer route
18 * ```typescript
19 * const events = yield* Cloudflare.Pipelines.WriteStream(Stream);
20 *
21 * return {
22 * fetch: Effect.gen(function* () {
23 * yield* events.send([{ event: "click", at: new Date().toISOString() }]);
24 * return HttpServerResponse.empty({ status: 202 });
25 * }),
26 * };
27 * ```
28 *
29 * Provide {@link WriteStreamBinding} (native Worker binding) in the
30 * worker's runtime layer to resolve the underlying stream at request time.
31 *
32 * `WriteStream` is a single identifier that is simultaneously the binding's
33 * Context tag, its type, and the callable —
34 * `yield* Cloudflare.Pipelines.WriteStream(stream)`.
35 *
36 * @binding
37 * @product Pipelines
38 * @category Storage & Databases
39 */
40export interface WriteStream extends Binding.Service<
41 WriteStream,
42 "Cloudflare.Pipelines.WriteStream",
43 (stream: Stream | LegacyPipeline) => Effect.Effect<WriteStreamClient>
44> {}
45
46export const WriteStream = Binding.Service<WriteStream>(
47 "Cloudflare.Pipelines.WriteStream",
48);
49
50export interface WriteStreamClient {
51 raw: Effect.Effect<Pipeline, never, RuntimeContext>;
52 send(
53 records: ReadonlyArray<PipelineRecord>,
54 ): Effect.Effect<void, StreamSendError, RuntimeContext>;
55}
56
57export class StreamSendError extends Data.TaggedError("StreamSendError")<{
58 message: string;
59 cause?: unknown;
60}> {}