1import type * as runtime from "@cloudflare/workers-types";
2import * as Effect from "effect/Effect";
3import * as Layer from "effect/Layer";
4import * as Stream from "effect/Stream";
5import { getRawStream } from "../../Util/Stream.ts";
6import { makeBucketBinding, makeHelpers } from "./BucketBinding.ts";
7import type {
8 Conditional,
9 MultipartOptions,
10 MultipartUpload,
11 PutOptions,
12 UploadPartOptions,
13 UploadedPart,
14} from "./BucketTypes.ts";
15import { WriteBucket, type WriteBucketClient } from "./WriteBucket.ts";
16
17/**
18 * Implementation of the {@link WriteBucket} binding that uses a Worker binding.
19 */
20export const WriteBucketBinding = Layer.effect(
21 WriteBucket,
22 Effect.suspend(() => makeBucketBinding({ makeClient: makeWrite })),
23);
24
25/** Build the write half of the binding client. */
26export const makeWrite = ({
27 raw,
28 use,
29 tryPromise,
30 wrapR2Object,
31 wrapR2ObjectOrBody,
32}: ReturnType<typeof makeHelpers>): WriteBucketClient => {
33 const wrapR2MultipartUpload = (
34 upload: runtime.R2MultipartUpload,
35 ): MultipartUpload => ({
36 ...upload,
37 raw: upload,
38 uploadId: upload.uploadId,
39 abort: () => tryPromise(() => upload.abort()),
40 complete: (uploadedParts: UploadedPart[]) =>
41 tryPromise(() => upload.complete(uploadedParts)).pipe(
42 Effect.map(wrapR2Object),
43 ),
44 uploadPart: (
45 partNumber: number,
46 value:
47 | ReadableStream
48 | ArrayBuffer
49 | ArrayBufferView
50 | string
51 | Blob
52 | Stream.Stream<Uint8Array>,
53 options?: UploadPartOptions,
54 ) =>
55 tryPromise(() =>
56 upload.uploadPart(
57 partNumber,
58 Stream.isStream(value)
59 ? value.pipe(Stream.toReadableStream())
60 : (value as any),
61 options,
62 ),
63 ),
64 });
65
66 return {
67 // @ts-expect-error
68 put: (
69 key: string,
70 value:
71 | ReadableStream
72 | ArrayBuffer
73 | ArrayBufferView
74 | string
75 | null
76 | Blob
77 | Stream.Stream<Uint8Array>,
78 options?: PutOptions & {
79 onlyIf: Conditional | Headers;
80 contentLength?: number;
81 },
82 ) =>
83 use((raw) => {
84 if (Stream.isStream(value)) {
85 // `contentLength` is ours; everything else (checksums, metadata,
86 // conditions) is R2's and must reach it on every path.
87 const { contentLength, ...r2Options } = options ?? {};
88 const rawStream = getRawStream(value);
89 if (rawStream) {
90 return raw.put(key, rawStream as any, r2Options);
91 } else if (!contentLength) {
92 throw new Error("Content length is required");
93 }
94 // R2 needs a known length, so the stream is piped through a fixed
95 // length stream.
96 const readable = Stream.toReadableStream(value).pipeThrough(
97 new FixedLengthStream(contentLength),
98 );
99 return raw.put(key, readable as any, r2Options);
100 }
101 return raw.put(key, value as any, options);
102 }).pipe(Effect.map(wrapR2ObjectOrBody)) as any,
103 delete: (keys: string | string[]) => use((raw) => raw.delete(keys)),
104 createMultipartUpload: (key: string, options?: MultipartOptions) =>
105 use((raw) => raw.createMultipartUpload(key, options)).pipe(
106 Effect.map(wrapR2MultipartUpload),
107 ),
108 resumeMultipartUpload: (key: string, uploadId: string) =>
109 raw.pipe(
110 Effect.map((raw) => raw.resumeMultipartUpload(key, uploadId)),
111 Effect.map(wrapR2MultipartUpload),
112 ),
113 };
114};