1import type * as runtime from "@cloudflare/workers-types";
2import * as Effect from "effect/Effect";
3import * as Stream from "effect/Stream";
4import * as Output from "../../Output.ts";
5import { Worker, WorkerEnvironment } from "../Workers/Worker.ts";
6import type { Bucket } from "./Bucket.ts";
7import type { R2Object, ObjectBody } from "./BucketTypes.ts";
8import { R2Error } from "./BucketTypes.ts";
9
10/**
11 * Shared scaffolding for the Worker-binding implementations of the R2 services.
12 *
13 * Resolves the {@link WorkerEnvironment} and host {@link Worker}, registers the
14 * `r2_bucket` binding at deploy time, then delegates to `makeClient` with the
15 * shared {@link Helpers} to build the read/write/read-write client.
16 */
17export const makeBucketBinding = <Client>(options: {
18 makeClient: (helpers: ReturnType<typeof makeHelpers>) => Client;
19}) =>
20 Effect.gen(function* () {
21 const env = yield* WorkerEnvironment;
22 const host = yield* Worker;
23
24 return Effect.fn(function* (bucket: Bucket) {
25 if (!globalThis.__ALCHEMY_RUNTIME__) {
26 yield* host.bind`${bucket}`({
27 bindings: [
28 {
29 type: "r2_bucket",
30 name: bucket.LogicalId,
31 bucketName: bucket.bucketName,
32 jurisdiction: bucket.jurisdiction.pipe(
33 Output.map((jurisdiction) =>
34 jurisdiction === "default" ? undefined : jurisdiction,
35 ),
36 ),
37 },
38 ],
39 });
40 }
41
42 return options.makeClient(makeHelpers(env, bucket));
43 });
44 });
45
46/**
47 * Helpers shared by both the read and write halves of the binding client.
48 *
49 * Read-only (`wrapR2Objects`) and write-only (`wrapR2MultipartUpload`) wrappers
50 * live in {@link makeRead}/{@link makeWrite} respectively — only the primitives
51 * used by both sides are exposed here.
52 */
53export const makeHelpers = (env: Record<string, any>, bucket: Bucket) => {
54 const raw = Effect.sync(
55 () => (env as Record<string, runtime.R2Bucket>)[bucket.LogicalId]!,
56 );
57 const tryPromise = <T>(fn: () => Promise<T>): Effect.Effect<T, R2Error> =>
58 Effect.tryPromise({
59 try: fn,
60 catch: (error: any) =>
61 new R2Error({
62 message: error.message ?? "Unknown error",
63 cause: error,
64 }),
65 });
66
67 const use = <T>(
68 fn: (raw: runtime.R2Bucket) => Promise<T>,
69 ): Effect.Effect<T, R2Error> =>
70 raw.pipe(Effect.flatMap((raw) => tryPromise(() => fn(raw))));
71
72 return {
73 raw,
74 use,
75 tryPromise,
76 ...makeR2ObjectWrappers(tryPromise),
77 };
78};
79
80/**
81 * The `R2Object`/`R2ObjectBody` → Effect-client wrappers, shared by the
82 * Worker-binding helpers above and the local platform-proxy helpers
83 * (`LocalR2Gateway.ts`), which differ only in how `raw`/`use` obtain the
84 * native bucket.
85 */
86export const makeR2ObjectWrappers = (
87 tryPromise: <T>(fn: () => Promise<T>) => Effect.Effect<T, R2Error>,
88) => {
89 const wrapR2Object = (object: runtime.R2Object): R2Object => ({
90 ...object,
91 writeHttpMetadata: (headers: Headers) =>
92 Effect.sync(() => object.writeHttpMetadata(headers)),
93 });
94 const wrapR2ObjectBody = (object: runtime.R2ObjectBody): ObjectBody => ({
95 ...wrapR2Object(object),
96 body: Stream.fromReadableStream({
97 evaluate: () =>
98 object.body as any as ReadableStream<Uint8Array<ArrayBufferLike>>,
99 onError: (error: any) =>
100 new R2Error({
101 message: error.message ?? "Unknown error",
102 cause: error,
103 }),
104 }),
105 readable: object.body as any as ReadableStream<Uint8Array>,
106 bodyUsed: object.bodyUsed,
107 arrayBuffer: () => tryPromise(() => object.arrayBuffer()),
108 bytes: () => tryPromise(() => object.bytes()),
109 text: () => tryPromise(() => object.text()),
110 json: <T>() => tryPromise(() => object.json<T>()),
111 blob: () => tryPromise(() => object.blob()),
112 });
113
114 const isR2ObjectBody = (object: any): object is runtime.R2ObjectBody =>
115 object !== null && typeof object === "object" && "body" in object;
116
117 const wrapR2ObjectOrBody = (
118 object: runtime.R2Object | runtime.R2ObjectBody | null,
119 ): R2Object | ObjectBody | null =>
120 object === null
121 ? object
122 : isR2ObjectBody(object)
123 ? wrapR2ObjectBody(object)
124 : wrapR2Object(object);
125
126 return {
127 wrapR2Object,
128 wrapR2ObjectOrBody,
129 };
130};