feat(services/s3): Add object versioning for S3

core/src/services/s3/backend.rs

@@ -1093,7 +1093,16 @@ impl Access for S3Backend {
         let status = resp.status();
 
         match status {
-            StatusCode::OK => parse_into_metadata(path, resp.headers()).map(RpStat::new),
+            StatusCode::OK => {
+                let headers = resp.headers();
+                let mut meta = parse_into_metadata(path, headers)?;
+
+                if let Some(v) = parse_header_to_str(headers, "x-amz-version-id")? {
+                    meta.set_version(v);
+                }
+
+                Ok(RpStat::new(meta))
+            }
             _ => Err(parse_error(resp)),
         }
     }
@@ -1124,13 +1133,13 @@ impl Access for S3Backend {
         Ok((RpWrite::default(), w))
     }
 
-    async fn delete(&self, path: &str, _: OpDelete) -> Result<RpDelete> {
+    async fn delete(&self, path: &str, args: OpDelete) -> Result<RpDelete> {
         // This would delete the bucket, do not perform
         if self.core.root == "/" && path == "/" {
             return Ok(RpDelete::default());
         }
 
-        let resp = self.core.s3_delete_object(path).await?;
+        let resp = self.core.s3_delete_object(path, &args).await?;
 
         let status = resp.status();
 

core/src/services/s3/core.rs

@@ -71,6 +71,8 @@ mod constants {
     pub const RESPONSE_CONTENT_DISPOSITION: &str = "response-content-disposition";
     pub const RESPONSE_CONTENT_TYPE: &str = "response-content-type";
     pub const RESPONSE_CACHE_CONTROL: &str = "response-cache-control";
+
+    pub const S3_QUERY_VERSION_ID: &str = "versionId";
 }
 
 pub struct S3Core {
@@ -312,6 +314,13 @@ impl S3Core {
                 percent_encode_path(override_cache_control)
             ))
         }
+        if let Some(version) = args.version() {
+            query_args.push(format!(
+                "{}={}",
+                constants::S3_QUERY_VERSION_ID,
+                percent_decode_path(version)
+            ))
+        }
         if !query_args.is_empty() {
             url.push_str(&format!("?{}", query_args.join("&")));
         }
@@ -367,6 +376,13 @@ impl S3Core {
                 percent_encode_path(override_cache_control)
             ))
         }
+        if let Some(version) = args.version() {
+            query_args.push(format!(
+                "{}={}",
+                constants::S3_QUERY_VERSION_ID,
+                percent_decode_path(version)
+            ))
+        }
         if !query_args.is_empty() {
             url.push_str(&format!("?{}", query_args.join("&")));
         }
@@ -463,10 +479,24 @@ impl S3Core {
         self.send(req).await
     }
 
-    pub async fn s3_delete_object(&self, path: &str) -> Result<Response<Buffer>> {
+    pub async fn s3_delete_object(&self, path: &str, args: &OpDelete) -> Result<Response<Buffer>> {
         let p = build_abs_path(&self.root, path);
 
-        let url = format!("{}/{}", self.endpoint, percent_encode_path(&p));
+        let mut url = format!("{}/{}", self.endpoint, percent_encode_path(&p));
+
+        let mut query_args = Vec::new();
+
+        if let Some(version) = args.version() {
+            query_args.push(format!(
+                "{}={}",
+                constants::S3_QUERY_VERSION_ID,
+                percent_encode_path(version)
+            ))
+        }
+
+        if !query_args.is_empty() {
+            url.push_str(&format!("?{}", query_args.join("&")));
+        }
 
         let mut req = Request::delete(&url)
             .body(Buffer::new())