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 toR2Error,
11 type HttpMetadata,
12 type R2Auth,
13} from "./BucketHttp.ts";
14import { ReadBucket, type ReadBucketClient } from "./ReadBucket.ts";
15import {
16 R2Error,
17 type GetOptions,
18 type ListOptions,
19 type Objects,
20} from "./BucketTypes.ts";
21
22/**
23 * HTTP-backed implementation of the {@link ReadBucket} binding.
24 *
25 * It creates a scoped {@link AccountApiToken} with the `Workers R2 Storage Read` and `Workers R2 Storage Write` permissions.
26 */
27export const ReadBucketHttp = Layer.effect(
28 ReadBucket,
29 Effect.suspend(() =>
30 makeHttpBucketBinding({
31 permissionGroups: ["Workers R2 Storage Read"],
32 makeClient: (token, bucketName, jurisdiction) =>
33 makeReadR2HttpClient(
34 { authorize: authorizeWith(token), accountId: token.accountId },
35 bucketName,
36 jurisdiction,
37 ),
38 }),
39 ),
40);
41
42export const makeReadR2HttpClient = (
43 auth: R2Auth,
44 bucketName: Effect.Effect<string>,
45 jurisdiction: Effect.Effect<string>,
46): ReadBucketClient => {
47 const authorize = auth.authorize;
48 const scope = makeR2HttpScope(auth.accountId, bucketName, jurisdiction);
49
50 return {
51 raw: Effect.die(
52 new R2Error({
53 message:
54 "R2BucketBindingHttp does not expose a native `raw` Bucket; use the binary HTTP methods instead.",
55 cause: new Error("unsupported"),
56 }),
57 ),
58 head: (key: string) =>
59 scope.pipe(
60 Effect.flatMap(({ accountId, bucketName, cfR2Jurisdiction }) =>
61 authorize(
62 r2.getObject({
63 accountId,
64 bucketName,
65 objectName: key,
66 cfR2Jurisdiction,
67 }),
68 ),
69 ),
70 // The HTTP body is lazy, so reading headers does not download it.
71 Effect.map((res) =>
72 baseObject(key, httpMetadataOf(res), {
73 size: res.contentLength,
74 etag: res.etag,
75 uploaded: res.lastModified ? new Date(res.lastModified) : undefined,
76 storageClass: res.cfR2StorageClass,
77 }),
78 ),
79 // Native R2 `head` resolves to `null` for a missing object rather
80 // than failing — mirror that for the HTTP-backed client.
81 Effect.catchTag("NoSuchKey", () => Effect.succeed(null)),
82 Effect.mapError(toR2Error),
83 ),
84 get: ((key: string, _options?: GetOptions) =>
85 scope.pipe(
86 Effect.flatMap(({ accountId, bucketName, cfR2Jurisdiction }) =>
87 authorize(
88 r2.getObject({
89 accountId,
90 bucketName,
91 objectName: key,
92 cfR2Jurisdiction,
93 }),
94 ),
95 ),
96 Effect.map((res) => objectBodyFromResponse(key, res)),
97 // Native R2 `get` resolves to `null` for a missing object.
98 Effect.catchTag("NoSuchKey", () => Effect.succeed(null)),
99 Effect.mapError(toR2Error),
100 )) as any,
101 list: (options?: ListOptions) =>
102 scope.pipe(
103 Effect.flatMap(({ accountId, bucketName, cfR2Jurisdiction }) =>
104 authorize(
105 r2.listObjects({
106 accountId,
107 bucketName,
108 cfR2Jurisdiction,
109 prefix: options?.prefix,
110 delimiter: options?.delimiter,
111 cursor: options?.cursor,
112 startAfter: options?.startAfter,
113 perPage: options?.limit,
114 }),
115 ),
116 ),
117 Effect.mapError(toR2Error),
118 Effect.map((res) => {
119 const objects = res.result.map((o) =>
120 baseObject(
121 o.key ?? "",
122 {},
123 {
124 size: o.size ?? undefined,
125 etag: o.etag ?? undefined,
126 uploaded: o.lastModified ? new Date(o.lastModified) : undefined,
127 storageClass: o.storageClass ?? undefined,
128 customMetadata:
129 (o.customMetadata as Record<string, string> | null) ??
130 undefined,
131 },
132 ),
133 );
134 const cursor = res.resultInfo?.cursor ?? undefined;
135 return (
136 cursor
137 ? { objects, delimitedPrefixes: [], truncated: true, cursor }
138 : { objects, delimitedPrefixes: [], truncated: false }
139 ) as Objects;
140 }),
141 ),
142 };
143};
144
145const objectBodyFromResponse = (
146 key: string,
147 res: r2.GetObjectResponse,
148): R2ObjectBody => {
149 const meta = httpMetadataOf(res);
150 // The HTTP body is a single-consumption stream — expose it both as a Stream
151 // and via the buffering accessors, but the caller may only read it once.
152 let response: Response | undefined;
153 const getResponse = () =>
154 (response ??= new Response(Stream.toReadableStream(res.body) as any));
155 const consume = <T>(
156 fn: (r: Response) => Promise<T>,
157 ): Effect.Effect<T, R2Error> =>
158 Effect.tryPromise({ try: () => fn(getResponse()), catch: toR2Error });
159 return {
160 ...baseObject(key, meta, {
161 size: res.contentLength,
162 etag: res.etag,
163 uploaded: res.lastModified ? new Date(res.lastModified) : undefined,
164 storageClass: res.cfR2StorageClass,
165 }),
166 body: Stream.fromReadableStream({
167 evaluate: () => getResponse().body as ReadableStream<Uint8Array>,
168 onError: toR2Error,
169 }),
170 bodyUsed: false,
171 arrayBuffer: () => consume((r) => r.arrayBuffer()),
172 bytes: () => consume((r) => r.arrayBuffer().then((b) => new Uint8Array(b))),
173 text: () => consume((r) => r.text()),
174 json: <T>() => consume((r) => r.json() as Promise<T>),
175 blob: () => consume((r) => r.blob()),
176 } as unknown as R2ObjectBody;
177};
178
179const httpMetadataOf = (res: r2.GetObjectResponse): HttpMetadata => ({
180 contentType: res.contentType,
181 contentEncoding: res.contentEncoding,
182 contentDisposition: res.contentDisposition,
183 contentLanguage: res.contentLanguage,
184 cacheControl: res.cacheControl,
185 cacheExpiry: res.expires ? new Date(res.expires) : undefined,
186});