1import * as r2 from "@distilled.cloud/cloudflare/r2";
2import * as Effect from "effect/Effect";
3import * as Layer from "effect/Layer";
4import * as Stream from "effect/Stream";
5import { authorizeWith } from "../HttpClientUtils.ts";
6import {
7 baseObject,
8 makeHttpBucketBinding,
9 makeR2HttpScope,
10 readHttpMetadata,
11 toBody,
12 toR2Error,
13 type R2Auth,
14} from "./BucketHttp.ts";
15import { R2Error, type PutOptions } from "./BucketTypes.ts";
16import { WriteBucket, type WriteBucketClient } from "./WriteBucket.ts";
17
18/**
19 * HTTP-backed implementation of the {@link WriteBucket} binding.
20 *
21 * It creates a scoped {@link AccountApiToken} with the `Workers R2 Storage Read` and `Workers R2 Storage Write` permissions.
22 */
23export const WriteBucketHttp = Layer.effect(
24 WriteBucket,
25 Effect.suspend(() =>
26 makeHttpBucketBinding({
27 permissionGroups: ["Workers R2 Storage Write"],
28 makeClient: (token, bucketName, jurisdiction) =>
29 makeWriteR2HttpClient(
30 { authorize: authorizeWith(token), accountId: token.accountId },
31 bucketName,
32 jurisdiction,
33 ),
34 }),
35 ),
36);
37
38/** Build the write half of the HTTP-backed {@link ReadWrite} client. */
39export const makeWriteR2HttpClient = (
40 auth: R2Auth,
41 bucketName: Effect.Effect<string>,
42 jurisdiction: Effect.Effect<string>,
43): WriteBucketClient => {
44 const authorize = auth.authorize;
45 const scope = makeR2HttpScope(auth.accountId, bucketName, jurisdiction);
46
47 return {
48 put: ((
49 key: string,
50 value:
51 | ReadableStream
52 | ArrayBuffer
53 | ArrayBufferView
54 | string
55 | null
56 | Blob
57 | Stream.Stream<Uint8Array, unknown>,
58 options?: PutOptions,
59 ) =>
60 scope.pipe(
61 Effect.flatMap(({ accountId, bucketName, cfR2Jurisdiction }) =>
62 toBody(value).pipe(
63 Effect.flatMap(({ body, contentLength }) => {
64 const meta = readHttpMetadata(options);
65 return authorize(
66 r2.putObject({
67 accountId,
68 bucketName,
69 objectName: key,
70 cfR2Jurisdiction,
71 body,
72 contentType: meta?.contentType,
73 contentEncoding: meta?.contentEncoding,
74 contentDisposition: meta?.contentDisposition,
75 contentLanguage: meta?.contentLanguage,
76 cacheControl: meta?.cacheControl,
77 contentLength:
78 options?.contentLength != null
79 ? String(options.contentLength)
80 : contentLength != null
81 ? String(contentLength)
82 : undefined,
83 cfR2StorageClass: (
84 options as { storageClass?: string } | undefined
85 )?.storageClass,
86 }),
87 ).pipe(
88 Effect.map(() =>
89 baseObject(key, meta ?? {}, {
90 size: contentLength,
91 customMetadata: (
92 options as
93 | { customMetadata?: Record<string, string> }
94 | undefined
95 )?.customMetadata,
96 storageClass: (
97 options as { storageClass?: string } | undefined
98 )?.storageClass,
99 uploaded: new Date(),
100 }),
101 ),
102 );
103 }),
104 ),
105 ),
106 Effect.mapError(toR2Error),
107 )) as any,
108 delete: (keys: string | string[]) =>
109 scope.pipe(
110 Effect.flatMap(({ accountId, bucketName, cfR2Jurisdiction }) =>
111 Array.isArray(keys)
112 ? authorize(
113 r2.deleteObjects({
114 accountId,
115 bucketName,
116 cfR2Jurisdiction,
117 body: keys,
118 }),
119 ).pipe(Effect.asVoid, Effect.mapError(toR2Error))
120 : authorize(
121 r2.deleteObject({
122 accountId,
123 bucketName,
124 objectName: keys,
125 cfR2Jurisdiction,
126 }),
127 ).pipe(
128 Effect.asVoid,
129 // The native binding's `delete` is idempotent — deleting a
130 // key that isn't there resolves. Keep the HTTP client at
131 // parity instead of surfacing R2's `NoSuchKey`.
132 Effect.catchTag("NoSuchKey", () => Effect.void),
133 Effect.mapError(toR2Error),
134 ),
135 ),
136 ),
137 createMultipartUpload: () =>
138 Effect.die(
139 new R2Error({
140 message:
141 "R2BucketBindingHttp does not support multipart uploads over the HTTP API.",
142 cause: new Error("unsupported"),
143 }),
144 ),
145 resumeMultipartUpload: () =>
146 Effect.die(
147 new R2Error({
148 message:
149 "R2BucketBindingHttp does not support multipart uploads over the HTTP API.",
150 cause: new Error("unsupported"),
151 }),
152 ),
153 };
154};