feat: create post/put bsos postgres

syncstorage-postgres/src/db/db_impl.rs

@@ -11,15 +11,17 @@ use diesel::{
 };
 use diesel_async::{AsyncConnection, RunQueryDsl, TransactionManager};
 use futures::TryStreamExt;
+use std::collections::HashMap;
 use syncstorage_db_common::{
-    error::DbErrorIntrospect, params, results, util::SyncTimestamp, Db, Sorting,
+    error::DbErrorIntrospect, params, results, util::SyncTimestamp, Db, Sorting, DEFAULT_BSO_TTL,
 };
 use syncstorage_settings::Quota;
 
 use super::PgDb;
 use crate::{
     bsos_query,
     db::CollectionLock,
+    orm_models::BsoChangeset,
     pool::Conn,
     schema::{bsos, user_collections},
     DbError, DbResult,
@@ -353,7 +355,28 @@ impl Db for PgDb {
     }
 
     async fn post_bsos(&mut self, params: params::PostBsos) -> DbResult<results::PostBsos> {
-        todo!()
+        let collection_id = self.get_or_create_collection_id(&params.collection).await?;
+        let modified = self.timestamp();
+
+        for pbso in params.bsos {
+            self.put_bso(params::PutBso {
+                user_id: params.user_id.clone(),
+                collection: params.collection.clone(),
+                id: pbso.id.clone(),
+                payload: pbso.payload,
+                sortindex: pbso.sortindex,
+                ttl: pbso.ttl,
+            })
+            .await?;
+        }
+        self.update_collection(params::UpdateCollection {
+            user_id: params.user_id,
+            collection_id,
+            collection: params.collection,
+        })
+        .await?;
+
+        Ok(modified)
     }
 
     async fn delete_bso(&mut self, params: params::DeleteBso) -> DbResult<results::DeleteBso> {
@@ -413,8 +436,87 @@ impl Db for PgDb {
         Ok(modified)
     }
 
-    async fn put_bso(&mut self, params: params::PutBso) -> DbResult<results::PutBso> {
-        todo!()
+    async fn put_bso(&mut self, bso: params::PutBso) -> DbResult<results::PutBso> {
+        let collection_id = self.get_or_create_collection_id(&bso.collection).await?;
+        let user_id: u64 = bso.user_id.legacy_id;
+        let timestamp = self.timestamp().as_i64();
+        if self.quota.enabled {
+            let usage = self
+                .get_quota_usage(params::GetQuotaUsage {
+                    user_id: bso.user_id.clone(),
+                    collection: bso.collection.clone(),
+                    collection_id,
+                })
+                .await?;
+            if usage.total_bytes >= self.quota.size {
+                let mut tags = HashMap::default();
+                tags.insert("collection".to_owned(), bso.collection.clone());
+                self.metrics.incr_with_tags("storage.quota.at_limit", tags);
+                if self.quota.enforced {
+                    return Err(DbError::quota());
+                } else {
+                    warn!("Quota at limit for user's collection ({} bytes)", usage.total_bytes; "collection"=>bso.collection.clone());
+                }
+            }
+        }
+
+        let payload = bso.payload.as_deref().unwrap_or_default();
+        let sortindex = bso.sortindex;
+        let ttl = bso.ttl.map_or(DEFAULT_BSO_TTL, |ttl| ttl);
+
+        // original required millisecond conversion
+        let expiry_ts = SyncTimestamp::from_i64(timestamp + (i64::from(ttl) * 1000))?;
+        let modified_ts = SyncTimestamp::from_i64(timestamp)?;
+
+        let expiry_dt = expiry_ts.as_naive_datetime()?;
+        let modified_dt = modified_ts.as_naive_datetime()?;
+
+        // The changeset utilizes Diesel's `AsChangeset` trait.
+        // This allows selective updates of fields if and only if they are `Some(<T>)`
+        let changeset = BsoChangeset {
+            sortindex: if bso.sortindex.is_some() {
+                Some(sortindex)
+            } else {
+                None
+            },
+            payload: if bso.payload.is_some() {
+                Some(payload)
+            } else {
+                None
+            },
+            expiry: if bso.ttl.is_some() {
+                Some(expiry_dt)
+            } else {
+                None
+            },
+            modified: if bso.payload.is_some() || bso.sortindex.is_some() {
+                Some(modified_dt)
+            } else {
+                None
+            },
+        };
+        diesel::insert_into(bsos::table)
+            .values((
+                bsos::user_id.eq(user_id as i64),
+                bsos::collection_id.eq(&collection_id),
+                bsos::bso_id.eq(&bso.id),
+                bsos::sortindex.eq(sortindex),
+                bsos::payload.eq(payload),
+                bsos::modified.eq(modified_dt),
+                bsos::expiry.eq(expiry_dt),
+            ))
+            .on_conflict((bsos::user_id, bsos::collection_id, bsos::bso_id))
+            .do_update()
+            .set(changeset)
+            .execute(&mut self.conn)
+            .await?;
+
+        self.update_collection(params::UpdateCollection {
+            user_id: bso.user_id,
+            collection_id,
+            collection: bso.collection,
+        })
+        .await
     }
 
     async fn get_collection_id(&mut self, name: &str) -> DbResult<results::GetCollectionId> {

syncstorage-postgres/src/lib.rs

@@ -2,6 +2,7 @@
 #[macro_use]
 extern crate diesel;
 extern crate diesel_migrations;
+#[macro_use]
 extern crate slog_scope;
 
 mod db;

syncstorage-postgres/src/orm_models.rs

@@ -1,7 +1,7 @@
 use chrono::NaiveDateTime;
 
 use crate::schema::{batch_bsos, batches, bsos, collections, user_collections};
-use diesel::{Identifiable, Queryable};
+use diesel::{AsChangeset, Identifiable, Queryable};
 
 #[allow(clippy::all)]
 #[derive(Queryable, Debug, Identifiable)]
@@ -38,6 +38,15 @@ pub struct Bso {
     pub expiry: NaiveDateTime,
 }
 
+#[derive(AsChangeset)]
+#[diesel(table_name = bsos)]
+pub struct BsoChangeset<'a> {
+    pub sortindex: Option<Option<i32>>,
+    pub payload: Option<&'a str>,
+    pub modified: Option<NaiveDateTime>,
+    pub expiry: Option<NaiveDateTime>,
+}
+
 #[derive(Queryable, Debug, Identifiable)]
 #[diesel(primary_key(collection_id))]
 pub struct Collection {