opendal/services/vercel_blob/
core.rs1use std::fmt::Debug;
19use std::fmt::Formatter;
20use std::sync::Arc;
21
22use bytes::Buf;
23use bytes::Bytes;
24use http::header;
25use http::request;
26use http::Request;
27use http::Response;
28use http::StatusCode;
29use serde::Deserialize;
30use serde::Serialize;
31use serde_json::json;
32
33use self::constants::*;
34use super::error::parse_error;
35use crate::raw::*;
36use crate::*;
37
38pub(super) mod constants {
39 pub const X_VERCEL_BLOB_CONTENT_TYPE: &str = "x-content-type";
42 pub const X_VERCEL_BLOB_ADD_RANDOM_SUFFIX: &str = "x-add-random-suffix";
46 pub const X_VERCEL_BLOB_MPU_ACTION: &str = "x-mpu-action";
53 pub const X_VERCEL_BLOB_MPU_KEY: &str = "x-mpu-key";
54 pub const X_VERCEL_BLOB_MPU_PART_NUMBER: &str = "x-mpu-part-number";
55 pub const X_VERCEL_BLOB_MPU_UPLOAD_ID: &str = "x-mpu-upload-id";
56}
57
58#[derive(Clone)]
59pub struct VercelBlobCore {
60 pub info: Arc<AccessorInfo>,
61 pub root: String,
63 pub token: String,
65}
66
67impl Debug for VercelBlobCore {
68 fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result {
69 f.debug_struct("Backend")
70 .field("root", &self.root)
71 .finish_non_exhaustive()
72 }
73}
74
75impl VercelBlobCore {
76 #[inline]
77 pub async fn send(&self, req: Request<Buffer>) -> Result<Response<Buffer>> {
78 self.info.http_client().send(req).await
79 }
80
81 pub fn sign(&self, req: request::Builder) -> request::Builder {
82 req.header(header::AUTHORIZATION, format!("Bearer {}", self.token))
83 }
84}
85
86impl VercelBlobCore {
87 pub async fn download(
88 &self,
89 path: &str,
90 range: BytesRange,
91 _: &OpRead,
92 ) -> Result<Response<HttpBody>> {
93 let p = build_abs_path(&self.root, path);
94 let resp = self.list(&p, Some(1)).await?;
97
98 let url = resolve_blob(resp.blobs, p);
100
101 if url.is_empty() {
102 return Err(Error::new(ErrorKind::NotFound, "Blob not found"));
103 }
104
105 let mut req = Request::get(url);
106
107 if !range.is_full() {
108 req = req.header(header::RANGE, range.to_header());
109 }
110
111 let req = req
113 .extension(Operation::Read)
114 .body(Buffer::new())
115 .map_err(new_request_build_error)?;
116
117 self.info.http_client().fetch(req).await
118 }
119
120 pub async fn upload(
121 &self,
122 path: &str,
123 size: u64,
124 args: &OpWrite,
125 body: Buffer,
126 ) -> Result<Response<Buffer>> {
127 let p = build_abs_path(&self.root, path);
128
129 let url = format!(
130 "https://blob.vercel-storage.com/{}",
131 percent_encode_path(&p)
132 );
133
134 let mut req = Request::put(&url);
135
136 req = req.header(X_VERCEL_BLOB_ADD_RANDOM_SUFFIX, "0");
137
138 req = req.header(header::CONTENT_LENGTH, size.to_string());
139
140 if let Some(mime) = args.content_type() {
141 req = req.header(X_VERCEL_BLOB_CONTENT_TYPE, mime)
142 }
143
144 let req = self.sign(req);
145
146 let req = req
148 .extension(Operation::Write)
149 .body(body)
150 .map_err(new_request_build_error)?;
151
152 self.send(req).await
153 }
154
155 pub async fn head(&self, path: &str) -> Result<Response<Buffer>> {
156 let p = build_abs_path(&self.root, path);
157
158 let resp = self.list(&p, Some(1)).await?;
159
160 let url = resolve_blob(resp.blobs, p);
161
162 if url.is_empty() {
163 return Err(Error::new(ErrorKind::NotFound, "Blob not found"));
164 }
165
166 let req = Request::get(format!(
167 "https://blob.vercel-storage.com?url={}",
168 percent_encode_path(&url)
169 ));
170
171 let req = self.sign(req);
172
173 let req = req
175 .extension(Operation::Stat)
176 .body(Buffer::new())
177 .map_err(new_request_build_error)?;
178
179 self.send(req).await
180 }
181
182 pub async fn copy(&self, from: &str, to: &str) -> Result<Response<Buffer>> {
183 let from = build_abs_path(&self.root, from);
184
185 let resp = self.list(&from, Some(1)).await?;
186
187 let from_url = resolve_blob(resp.blobs, from);
188
189 if from_url.is_empty() {
190 return Err(Error::new(ErrorKind::NotFound, "Blob not found"));
191 }
192
193 let to = build_abs_path(&self.root, to);
194
195 let to_url = format!(
196 "https://blob.vercel-storage.com/{}?fromUrl={}",
197 percent_encode_path(&to),
198 percent_encode_path(&from_url),
199 );
200
201 let req = Request::put(&to_url);
202
203 let req = self.sign(req);
204
205 let req = req
207 .extension(Operation::Copy)
208 .body(Buffer::new())
209 .map_err(new_request_build_error)?;
210
211 self.send(req).await
212 }
213
214 pub async fn list(&self, prefix: &str, limit: Option<usize>) -> Result<ListResponse> {
215 let prefix = if prefix == "/" { "" } else { prefix };
216
217 let mut url = format!(
218 "https://blob.vercel-storage.com?prefix={}",
219 percent_encode_path(prefix)
220 );
221
222 if let Some(limit) = limit {
223 url.push_str(&format!("&limit={}", limit))
224 }
225
226 let req = Request::get(&url);
227
228 let req = self.sign(req);
229
230 let req = req
232 .extension(Operation::List)
233 .body(Buffer::new())
234 .map_err(new_request_build_error)?;
235
236 let resp = self.send(req).await?;
237
238 let status = resp.status();
239
240 if status != StatusCode::OK {
241 return Err(parse_error(resp));
242 }
243
244 let body = resp.into_body();
245
246 let resp: ListResponse =
247 serde_json::from_reader(body.reader()).map_err(new_json_deserialize_error)?;
248
249 Ok(resp)
250 }
251
252 pub async fn initiate_multipart_upload(
253 &self,
254 path: &str,
255 args: &OpWrite,
256 ) -> Result<Response<Buffer>> {
257 let p = build_abs_path(&self.root, path);
258
259 let url = format!(
260 "https://blob.vercel-storage.com/mpu/{}",
261 percent_encode_path(&p)
262 );
263
264 let req = Request::post(&url);
265
266 let mut req = self.sign(req);
267
268 req = req.header(X_VERCEL_BLOB_MPU_ACTION, "create");
269 req = req.header(X_VERCEL_BLOB_ADD_RANDOM_SUFFIX, "0");
270
271 if let Some(mime) = args.content_type() {
272 req = req.header(X_VERCEL_BLOB_CONTENT_TYPE, mime);
273 };
274
275 let req = req
277 .extension(Operation::Write)
278 .body(Buffer::new())
279 .map_err(new_request_build_error)?;
280
281 self.send(req).await
282 }
283
284 pub async fn upload_part(
285 &self,
286 path: &str,
287 upload_id: &str,
288 part_number: usize,
289 size: u64,
290 body: Buffer,
291 ) -> Result<Response<Buffer>> {
292 let p = build_abs_path(&self.root, path);
293
294 let url = format!(
295 "https://blob.vercel-storage.com/mpu/{}",
296 percent_encode_path(&p)
297 );
298
299 let mut req = Request::post(&url);
300
301 req = req.header(header::CONTENT_LENGTH, size);
302 req = req.header(X_VERCEL_BLOB_MPU_ACTION, "upload");
303 req = req.header(X_VERCEL_BLOB_MPU_KEY, p);
304 req = req.header(X_VERCEL_BLOB_MPU_UPLOAD_ID, upload_id);
305 req = req.header(X_VERCEL_BLOB_MPU_PART_NUMBER, part_number);
306
307 let req = self.sign(req);
308
309 let req = req
311 .extension(Operation::Write)
312 .body(body)
313 .map_err(new_request_build_error)?;
314
315 self.send(req).await
316 }
317
318 pub async fn complete_multipart_upload(
319 &self,
320 path: &str,
321 upload_id: &str,
322 parts: Vec<Part>,
323 ) -> Result<Response<Buffer>> {
324 let p = build_abs_path(&self.root, path);
325
326 let url = format!(
327 "https://blob.vercel-storage.com/mpu/{}",
328 percent_encode_path(&p)
329 );
330
331 let mut req = Request::post(&url);
332
333 req = req.header(X_VERCEL_BLOB_MPU_ACTION, "complete");
334 req = req.header(X_VERCEL_BLOB_MPU_KEY, p);
335 req = req.header(X_VERCEL_BLOB_MPU_UPLOAD_ID, upload_id);
336
337 let req = self.sign(req);
338
339 let parts_json = json!(parts);
340
341 let req = req
342 .header(header::CONTENT_TYPE, "application/json")
343 .extension(Operation::Write)
344 .body(Buffer::from(Bytes::from(parts_json.to_string())))
345 .map_err(new_request_build_error)?;
346
347 self.send(req).await
348 }
349}
350
351pub fn parse_blob(blob: &Blob) -> Result<Metadata> {
352 let mode = if blob.pathname.ends_with('/') {
353 EntryMode::DIR
354 } else {
355 EntryMode::FILE
356 };
357 let mut md = Metadata::new(mode);
358 if let Some(content_type) = blob.content_type.clone() {
359 md.set_content_type(&content_type);
360 }
361 md.set_content_length(blob.size);
362 md.set_last_modified(parse_datetime_from_rfc3339(&blob.uploaded_at)?);
363 md.set_content_disposition(&blob.content_disposition);
364 Ok(md)
365}
366
367fn resolve_blob(blobs: Vec<Blob>, path: String) -> String {
368 for blob in blobs {
369 if blob.pathname == path {
370 return blob.url;
371 }
372 }
373 "".to_string()
374}
375
376#[derive(Default, Debug, Clone, PartialEq, Deserialize)]
377#[serde(rename_all = "camelCase")]
378pub struct ListResponse {
379 pub cursor: Option<String>,
380 pub has_more: bool,
381 pub blobs: Vec<Blob>,
382}
383
384#[derive(Default, Debug, Clone, PartialEq, Deserialize)]
385#[serde(rename_all = "camelCase")]
386pub struct Blob {
387 pub url: String,
388 pub pathname: String,
389 pub size: u64,
390 pub uploaded_at: String,
391 pub content_disposition: String,
392 pub content_type: Option<String>,
393}
394
395#[derive(Default, Debug, Clone, PartialEq, Serialize)]
396#[serde(rename_all = "camelCase")]
397pub struct Part {
398 pub part_number: usize,
399 pub etag: String,
400}
401
402#[derive(Default, Debug, Clone, PartialEq, Deserialize)]
403#[serde(rename_all = "camelCase")]
404pub struct InitiateMultipartUploadResponse {
405 pub upload_id: String,
406 pub key: String,
407}
408
409#[derive(Default, Debug, Clone, PartialEq, Deserialize)]
410#[serde(rename_all = "camelCase")]
411pub struct UploadPartResponse {
412 pub etag: String,
413}