Merge pull request #2 from hcengineering/feat-write-recovery

write recovery data to s3

server/src/blob.rs

@@ -11,6 +11,7 @@ use tracing::*;
 
 use crate::handlers::ApiError;
 use crate::postgres::DbError;
+use crate::recovery;
 use crate::s3::S3Client;
 use crate::{
     config::CONFIG,
@@ -20,6 +21,7 @@ use crate::{
 #[derive(Debug)]
 pub struct Blob {
     pub s3_key: String,
+    pub hash: String,
     pub length: usize,
     pub inline: Option<Bytes>,
     pub parts_count: Option<usize>,
@@ -66,7 +68,7 @@ where
 
         let buffer = buffer.freeze();
 
-        let hash = hash.update(&buffer).finalize().to_hex();
+        let hash = hash.update(&buffer).finalize().to_hex().to_string();
         let length = buffer.len();
 
         let inline = Some(buffer.clone());
@@ -108,6 +110,7 @@ where
             };
 
         Blob {
+            hash,
             s3_key,
             length,
             inline,
@@ -150,6 +153,7 @@ where
         };
 
         Blob {
+            hash,
             s3_key,
             length: upload.length,
             inline: None,
@@ -158,6 +162,10 @@ where
         }
     };
 
+    if !blob.deduplicated {
+        recovery::set_blob(&s3, &blob.s3_key, &blob.hash).await?;
+    }
+
     Ok(blob)
 }
 

server/src/handlers.rs

@@ -17,13 +17,13 @@ use size::Size;
 use tracing::*;
 use uuid::Uuid;
 
-use crate::merge::MergeStrategy;
 use crate::s3::S3Client;
 use crate::{blob, merge};
 use crate::{
     config::CONFIG,
     postgres::{self, Pool},
 };
+use crate::{merge::MergeStrategy, recovery};
 
 #[derive(Deserialize, Debug)]
 pub struct ObjectPath {
@@ -212,6 +212,9 @@ pub async fn put(request: HttpRequest, payload: Payload) -> HandlerResult<HttpRe
         }
     });
 
+    let obj_parts = vec![&part_data];
+    recovery::set_object(&s3, path.workspace, &part_data.key, obj_parts).await?;
+
     postgres::set_part(&pool, path.workspace, &part_data.key, inline, &part_data).await?;
 
     let mut response = HttpResponse::Created();
@@ -280,6 +283,13 @@ pub async fn patch(request: HttpRequest, payload: Payload) -> HandlerResult<Http
             merge_strategy: None,
         };
 
+        let obj_parts = parts
+            .iter()
+            .map(|p| &p.data)
+            .chain(std::iter::once(&part_data))
+            .collect::<Vec<&PartData>>();
+        recovery::set_object(&s3, path.workspace, &part_data.key, obj_parts).await?;
+
         postgres::append_part(
             &pool,
             path.workspace,

server/src/main.rs

@@ -18,6 +18,7 @@ mod blob;
 mod config;
 mod handlers;
 mod merge;
+mod recovery;
 mod patch;
 mod postgres;
 mod s3;

server/src/recovery.rs

@@ -0,0 +1,43 @@
+use bytes::Bytes;
+
+use crate::config::CONFIG;
+use crate::{handlers::PartData, s3::S3Client};
+
+pub async fn set_object(
+    s3: &S3Client,
+    workspace: uuid::Uuid,
+    key: &str,
+    parts: Vec<&PartData>,
+) -> anyhow::Result<()> {
+    let s3_bucket = &CONFIG.s3_bucket;
+
+    let key = format!("blob/{}/{}", workspace, key);
+    let body = Bytes::from(serde_json::to_string(&parts)?);
+
+    s3.put_object()
+        .bucket(s3_bucket)
+        .key(key)
+        .body(body.into())
+        .content_type("application/json")
+        .send()
+        .await?;
+
+    Ok(())
+}
+
+pub async fn set_blob(s3: &S3Client, key: &str, hash: &str) -> anyhow::Result<()> {
+    let s3_bucket = &CONFIG.s3_bucket;
+
+    let key = format!("hash/{}", key);
+    let body = Bytes::from(hash.to_string());
+
+    s3.put_object()
+        .bucket(s3_bucket)
+        .key(key)
+        .body(body.into())
+        .content_type("text/plain")
+        .send()
+        .await?;
+
+    Ok(())
+}