1import type * as cf from "@cloudflare/workers-types";
2import * as Effect from "effect/Effect";
3import type { RuntimeContext } from "../../RuntimeContext.ts";
4import type * as Binding from "../Workers/Binding.ts";
5import { makeBindingLayer } from "../Workers/BindingLayer.ts";
6import {
7 Stream,
8 StreamError,
9 type StreamClient,
10 type StreamVideoClient,
11} from "./Stream.ts";
12
13/** The binding value produced by calling {@link Stream} (declared on `env` or `yield*`-ed). */
14export type StreamBinding = Binding.Binding<
15 Stream["key"],
16 StreamClient,
17 Stream
18>;
19
20/**
21 * The layer that provides the Effect-native interface for the Cloudflare
22 * Stream binding.
23 *
24 * Provide it on the Worker effect (`Effect.provide(Cloudflare.Stream.StreamBinding)`)
25 * so that yielding a {@link Stream} binding attaches the native `stream`
26 * binding to the surrounding Worker at deploy time and, at runtime, resolves to
27 * the Effect-native {@link StreamClient} (wrapping the raw `cf.StreamBinding`
28 * handle so every operation returns an `Effect`).
29 */
30export const StreamBinding = makeBindingLayer<
31 Stream,
32 cf.StreamBinding,
33 StreamClient
34>(Stream, (raw): StreamClient => {
35 const call = <T>(
36 fn: (binding: cf.StreamBinding) => Promise<T>,
37 ): Effect.Effect<T, StreamError, RuntimeContext> =>
38 raw.pipe(Effect.flatMap((binding) => tryPromise(() => fn(binding))));
39
40 const video = (id: string): StreamVideoClient => ({
41 id,
42 details: () => call((binding) => binding.video(id).details()),
43 update: (params) => call((binding) => binding.video(id).update(params)),
44 delete: () => call((binding) => binding.video(id).delete()),
45 generateToken: () => call((binding) => binding.video(id).generateToken()),
46 downloads: {
47 generate: (downloadType) =>
48 call((binding) => binding.video(id).downloads.generate(downloadType)),
49 get: () => call((binding) => binding.video(id).downloads.get()),
50 delete: (downloadType) =>
51 call((binding) => binding.video(id).downloads.delete(downloadType)),
52 },
53 captions: {
54 upload: (language, input) =>
55 call((binding) =>
56 binding
57 .video(id)
58 .captions.upload(language, input as unknown as cf.ReadableStream),
59 ),
60 generate: (language) =>
61 call((binding) => binding.video(id).captions.generate(language)),
62 list: (language) =>
63 call((binding) => binding.video(id).captions.list(language)),
64 delete: (language) =>
65 call((binding) => binding.video(id).captions.delete(language)),
66 },
67 });
68
69 return {
70 raw,
71 upload: (url, params) => call((binding) => binding.upload(url, params)),
72 createDirectUpload: (params) =>
73 call((binding) => binding.createDirectUpload(params)),
74 video,
75 videos: {
76 list: (params) => call((binding) => binding.videos.list(params)),
77 },
78 watermarks: {
79 generate: (input, params) =>
80 call((binding) =>
81 // The runtime accepts a stream or a URL string for the first
82 // argument; the published types declare the two as overloads.
83 binding.watermarks.generate(
84 input as unknown as cf.ReadableStream,
85 params,
86 ),
87 ),
88 list: () => call((binding) => binding.watermarks.list()),
89 get: (watermarkId) =>
90 call((binding) => binding.watermarks.get(watermarkId)),
91 delete: (watermarkId) =>
92 call((binding) => binding.watermarks.delete(watermarkId)),
93 },
94 } satisfies StreamClient;
95});
96
97const tryPromise = <T>(fn: () => Promise<T>): Effect.Effect<T, StreamError> =>
98 Effect.tryPromise({
99 try: fn,
100 catch: (error: any) =>
101 new StreamError({
102 message: error?.message ?? "Unknown Cloudflare Stream error",
103 code: typeof error?.code === "number" ? error.code : undefined,
104 statusCode:
105 typeof error?.statusCode === "number" ? error.statusCode : undefined,
106 cause: error,
107 }),
108 });