hoslo commented on code in PR #4103: URL: https://github.com/apache/opendal/pull/4103#discussion_r1472988557
########## core/src/services/vercel_blob/core.rs: ########## @@ -0,0 +1,433 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +use std::fmt::Debug; +use std::fmt::Formatter; + +use bytes::Bytes; +use http::header; +use http::request; +use http::Request; +use http::Response; +use http::StatusCode; +use serde::Deserialize; +use serde::Serialize; +use serde_json::json; + +use crate::raw::*; +use crate::*; + +use super::error::parse_error; + +#[derive(Clone)] +pub struct VercelBlobCore { + /// The root of this core. + pub root: String, + /// Vercel Blob token. + pub token: String, + + pub client: HttpClient, +} + +impl Debug for VercelBlobCore { + fn fmt(&self, f: &mut Formatter<'_>) -> std::fmt::Result { + f.debug_struct("Backend") + .field("root", &self.root) + .finish_non_exhaustive() + } +} + +impl VercelBlobCore { + #[inline] + pub async fn send(&self, req: Request<AsyncBody>) -> Result<Response<IncomingAsyncBody>> { + self.client.send(req).await + } + + pub fn sign(&self, req: request::Builder) -> request::Builder { + req.header(header::AUTHORIZATION, format!("Bearer {}", self.token)) + } +} + +impl VercelBlobCore { + pub async fn download(&self, path: &str, args: OpRead) -> Result<Response<IncomingAsyncBody>> { + let p = build_abs_path(&self.root, path); + // Vercel blob use an unguessable random id url to download the file + // So we use list to get the url of the file and then use it to download the file + let resp = self.list(&p, Some(1)).await?; + + let blobs = resp.blobs; + + if blobs.is_empty() { + return Err(Error::new(ErrorKind::NotFound, "Blob not found")); + } + + // Use the first url to download the file + let url = &blobs[0].url; + + let mut req = Request::get(url); + + let range = args.range(); + if !range.is_full() { + req = req.header(http::header::RANGE, range.to_header()); + } + + // Set body + let req = req + .body(AsyncBody::Empty) + .map_err(new_request_build_error)?; + + self.send(req).await + } + + pub async fn get_put_request( + &self, + path: &str, + size: Option<u64>, + args: &OpWrite, + body: AsyncBody, + ) -> Result<Request<AsyncBody>> { + let p = build_abs_path(&self.root, path); + + let url = format!( + "https://blob.vercel-storage.com/{}";, + percent_encode_path(&p) + ); + + let mut req = Request::put(&url); + + // x-add-random-suffix specifying whether to add a random suffix to the pathname + // Default value is 1, which means to add a random suffix. + // Set it to 0 to disable the random suffix. + req = req.header("x-add-random-suffix", "0"); + + if let Some(size) = size { + req = req.header(header::CONTENT_LENGTH, size.to_string()) + } + + // https://github.com/vercel/storage/blob/main/packages/blob/src/put.ts#L16 + // x-content-type specifies the MIME type of the file being uploaded. + if let Some(mime) = args.content_type() { + req = req.header("x-content-type", mime) + } + + let req = self.sign(req); + + // Set body + let req = req.body(body).map_err(new_request_build_error)?; + + Ok(req) + } + + pub async fn delete(&self, path: &str) -> Result<()> { + let p = build_abs_path(&self.root, path); + + let resp = self.list(&p, Some(1)).await?; + + let blobs = resp.blobs; + + if blobs.is_empty() { + return Ok(()); + } + + let url = &blobs[0].url; + + let req = Request::post("https://blob.vercel-storage.com/delete";); + + let req = self.sign(req); + + let req_body = &json!({ + "urls": vec![url] + }); + + let req = req + .header(header::CONTENT_TYPE, "application/json") + .body(AsyncBody::Bytes(Bytes::from(req_body.to_string()))) + .map_err(new_request_build_error)?; + + let resp = self.send(req).await?; + + let status = resp.status(); + + match status { + StatusCode::OK => Ok(()), + _ => Err(parse_error(resp).await?), + } + } + + pub async fn head(&self, path: &str) -> Result<Response<IncomingAsyncBody>> { + let p = build_abs_path(&self.root, path); + + let resp = self.list(&p, Some(1)).await?; + + let blobs = resp.blobs; + + if blobs.is_empty() { + return Err(Error::new(ErrorKind::NotFound, "Blob not found")); + } + + let url = &blobs[0].url; + + let req = Request::get(format!( + "https://blob.vercel-storage.com?url={}";, + percent_encode_path(url) + )); + + let req = self.sign(req); + + // Set body + let req = req + .body(AsyncBody::Empty) + .map_err(new_request_build_error)?; + + self.send(req).await + } + + pub async fn copy(&self, from: &str, to: &str) -> Result<Response<IncomingAsyncBody>> { + let from = build_abs_path(&self.root, from); + + let resp = self.list(&from, Some(1)).await?; + + let blobs = resp.blobs; + + if blobs.is_empty() { + return Err(Error::new(ErrorKind::NotFound, "Blob not found")); + } + + let from_url = &blobs[0].url; + + let to = build_abs_path(&self.root, to); + + let to_url = format!( + "https://blob.vercel-storage.com/{}?fromUrl={}";, + percent_encode_path(&to), + percent_encode_path(from_url), + ); + + let req = Request::put(&to_url); + + let req = self.sign(req); + + // Set body + let req = req + .body(AsyncBody::Empty) + .map_err(new_request_build_error)?; + + self.send(req).await + } + + pub async fn list(&self, prefix: &str, limit: Option<usize>) -> Result<ListResponse> { + let prefix = if prefix == "/" { "" } else { prefix }; + + let mut url = format!( + "https://blob.vercel-storage.com?prefix={}";, + percent_encode_path(prefix) + ); + + if let Some(limit) = limit { + url.push_str(&format!("&limit={}", limit)) + } + + let req = Request::get(&url); + + let req = self.sign(req); + + // Set body + let req = req + .body(AsyncBody::Empty) + .map_err(new_request_build_error)?; + + let resp = self.send(req).await?; + + let status = resp.status(); + + match status { + StatusCode::OK => { + let body = resp.into_body().bytes().await?; + + let resp: ListResponse = + serde_json::from_slice(&body).map_err(new_json_deserialize_error)?; + + Ok(resp) + } + _ => Err(parse_error(resp).await?), + } + } + + pub async fn initiate_multipart_upload( + &self, + path: &str, + args: &OpWrite, + ) -> Result<Response<IncomingAsyncBody>> { + let p = build_abs_path(&self.root, path); + + let url = format!( + "https://blob.vercel-storage.com/mpu/{}";, + percent_encode_path(&p) + ); + + let req = Request::post(&url); + + let mut req = self.sign(req); + + // https://github.com/vercel/storage/blob/main/packages/blob/src/put-multipart.ts#L130 + // x-mpu-action specifies the action to perform on the MPU. + // In this case, we want to create a new MPU. + req = req.header("x-mpu-action", "create"); + + req = req.header("x-add-random-suffix", "0"); Review Comment: fixed -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
