1import type { Pipeline } from "cloudflare:pipelines";
2import * as Effect from "effect/Effect";
3import * as Layer from "effect/Layer";
4import { Worker, WorkerEnvironment } from "../Workers/Worker.ts";
5import { isLegacyPipeline, type LegacyPipeline } from "./LegacyPipeline.ts";
6import type { Stream } from "./Stream.ts";
7import {
8 StreamSendError,
9 WriteStream,
10 type WriteStreamClient,
11} from "./WriteStream.ts";
12
13/**
14 * Implementation of the {@link WriteStream} service that uses a native
15 * Worker `pipelines` binding.
16 */
17export const WriteStreamBinding = Layer.effect(
18 WriteStream,
19 Effect.gen(function* () {
20 const env = yield* WorkerEnvironment;
21 const host = yield* Worker;
22
23 return Effect.fn(function* (stream: Stream | LegacyPipeline) {
24 if (!globalThis.__ALCHEMY_RUNTIME__) {
25 yield* host.bind`${stream}`({
26 bindings: [
27 {
28 type: "pipelines",
29 name: stream.LogicalId,
30 // A stream is bound by id; a legacy pipeline by name (the
31 // API identifier of the legacy generation).
32 pipeline: isLegacyPipeline(stream)
33 ? stream.name
34 : stream.streamId,
35 },
36 ],
37 });
38 }
39
40 return makeWriteStreamClient(env, stream);
41 });
42 }),
43);
44
45/** Build the producer client over a native Worker `pipelines` binding. */
46export const makeWriteStreamClient = (
47 env: Record<string, any>,
48 stream: Stream | LegacyPipeline,
49): WriteStreamClient => {
50 const raw = Effect.sync(
51 () => (env as Record<string, Pipeline>)[stream.LogicalId]!,
52 );
53 return {
54 raw,
55 send: (records) =>
56 raw.pipe(
57 Effect.flatMap((pipeline) =>
58 Effect.tryPromise({
59 try: () => pipeline.send([...records]),
60 catch: (error: any) =>
61 new StreamSendError({
62 message: error?.message ?? "Unknown pipeline error",
63 cause: error,
64 }),
65 }),
66 ),
67 ),
68 };
69};