Storage trait new implementation

10 files changed, 311 insertions, 369 deletions

diff --git a/src/bayou.rs b/src/bayou.rs
index 3042f94..afe3c75 100644
--- a/src/bayou.rs
+++ b/src/bayou.rs
@@ -45,8 +45,7 @@ pub struct Bayou<S: BayouState> {
     path: String,
     key: Key,
 
-    k2v: storage::RowStore,
-    s3: storage::BlobStore,
+    storage: storage::Store,
 
     checkpoint: (Timestamp, S),
     history: Vec<(Timestamp, S::Op, Option<S>)>,
@@ -60,17 +59,16 @@ pub struct Bayou<S: BayouState> {
 
 impl<S: BayouState> Bayou<S> {
     pub fn new(creds: &Credentials, path: String) -> Result<Self> {
-        let k2v_client = creds.row_client()?;
-        let s3_client = creds.blob_client()?;
+        let storage = creds.storage.build()?;
 
-        let target = k2v_client.row(&path, WATCH_SK);
-        let watch = K2vWatch::new(creds, path.clone(), WATCH_SK.to_string())?;
+        //let target = k2v_client.row(&path, WATCH_SK);
+        let target = storage::RowRef::new(&path, WATCH_SK);
+        let watch = K2vWatch::new(creds, target.clone())?;
 
         Ok(Self {
             path,
+            storage,
             key: creds.keys.master.clone(),
-            k2v: k2v_client,
-            s3: s3_client,
             checkpoint: (Timestamp::zero(), S::default()),
             history: vec![],
             last_sync: None,
@@ -96,9 +94,7 @@ impl<S: BayouState> Bayou<S> {
             } else {
                 debug!("(sync) loading checkpoint: {}", key);
 
-                let obj_res = self.s3.blob(key).fetch().await?;
-                let buf = obj_res.content().ok_or(anyhow!("object can't be empty"))?;
-
+                let buf = self.storage.blob_fetch(&storage::BlobRef(key.to_string())).await?.value;
                 debug!("(sync) checkpoint body length: {}", buf.len());
 
                 let ck = open_deserialize::<S>(&buf, &self.key)?;
@@ -129,42 +125,26 @@ impl<S: BayouState> Bayou<S> {
         // 3. List all operations starting from checkpoint
         let ts_ser = self.checkpoint.0.to_string();
         debug!("(sync) looking up operations starting at {}", ts_ser);
-        let ops_map = self.k2v.select(storage::Selector::Range { shard_key: &self.path, begin: &ts_ser, end: WATCH_SK }).await?;
-        /*let ops_map = self
-            .k2v
-            .read_batch(&[BatchReadOp {
-                partition_key: &self.path,
-                filter: Filter {
-                    start: Some(&ts_ser),
-                    end: Some(WATCH_SK),
-                    prefix: None,
-                    limit: None,
-                    reverse: false,
-                },
-                single_item: false,
-                conflicts_only: false,
-                tombstones: false,
-            }])
-            .await?
-            .into_iter()
-            .next()
-            .ok_or(anyhow!("Missing K2V result"))?
-            .items;*/
+        let ops_map = self.storage.row_fetch(&storage::Selector::Range { 
+            shard: &self.path, 
+            sort_begin: &ts_ser, 
+            sort_end: WATCH_SK 
+        }).await?;
 
         let mut ops = vec![];
         for row_value in ops_map {
-            let row = row_value.to_ref();
-            let sort_key = row.key().1;
+            let row = row_value.row_ref;
+            let sort_key = row.uid.sort;
             let ts = sort_key.parse::<Timestamp>().map_err(|_| anyhow!("Invalid operation timestamp: {}", sort_key))?;
 
-            let val = row_value.content();
+            let val = row_value.value;
             if val.len() != 1 {
-                bail!("Invalid operation, has {} values", row_value.content().len());
+                bail!("Invalid operation, has {} values", val.len());
             }
             match &val[0] {
                 storage::Alternative::Value(v) => {
                     let op = open_deserialize::<S::Op>(v, &self.key)?;
-                    debug!("(sync) operation {}: {} {:?}", sort_key, base64::encode(v), op);
+                    debug!("(sync) operation {}: {:?}", sort_key, op);
                     ops.push((ts, op));
                 }
                 storage::Alternative::Tombstone => {
@@ -231,7 +211,7 @@ impl<S: BayouState> Bayou<S> {
 
         // Save info that sync has been done
         self.last_sync = new_last_sync;
-        self.last_sync_watch_ct = self.k2v.from_orphan(new_last_sync_watch_ct).expect("Source & target storage must be compatible");
+        self.last_sync_watch_ct = new_last_sync_watch_ct; 
         Ok(())
     }
 
@@ -243,7 +223,7 @@ impl<S: BayouState> Bayou<S> {
             Some(t) => Instant::now() > t + (CHECKPOINT_INTERVAL / 5),
             _ => true,
         };
-        let changed = self.last_sync_watch_ct.to_orphan() != *self.watch.rx.borrow();
+        let changed = self.last_sync_watch_ct != *self.watch.rx.borrow();
         if too_old || changed {
             self.sync().await?;
         }
@@ -263,12 +243,12 @@ impl<S: BayouState> Bayou<S> {
                 .map(|(ts, _, _)| ts)
                 .unwrap_or(&self.checkpoint.0),
         );
-        self.k2v
-            .row(&self.path, &ts.to_string())
-            .set_value(&seal_serialize(&op, &self.key)?)
-            .push()
-            .await?;
 
+        let row_val = storage::RowVal::new(
+            storage::RowRef::new(&self.path, &ts.to_string()),
+            seal_serialize(&op, &self.key)?,
+        );
+        self.storage.row_insert(vec![row_val]).await?;
         self.watch.notify.notify_one();
 
         let new_state = self.state().apply(&op);
@@ -368,12 +348,11 @@ impl<S: BayouState> Bayou<S> {
         let cryptoblob = seal_serialize(&state_cp, &self.key)?;
         debug!("(cp) checkpoint body length: {}", cryptoblob.len());
 
-        self.s3
-            .blob(format!("{}/checkpoint/{}", self.path, ts_cp.to_string()).as_str())
-            .set_value(cryptoblob.into())
-            .push()
-            .await?;
-
+        let blob_val = storage::BlobVal::new(
+            storage::BlobRef(format!("{}/checkpoint/{}", self.path, ts_cp.to_string())),
+            cryptoblob.into(),
+        );
+        self.storage.blob_insert(&blob_val).await?;
 
         // Drop old checkpoints (but keep at least CHECKPOINTS_TO_KEEP of them)
         let ecp_len = existing_checkpoints.len();
@@ -383,22 +362,16 @@ impl<S: BayouState> Bayou<S> {
             // Delete blobs
             for (_ts, key) in existing_checkpoints[..last_to_keep].iter() {
                 debug!("(cp) drop old checkpoint {}", key);
-                self.s3
-                    .blob(key)
-                    .rm()
-                    .await?;
+                self.storage.blob_rm(&storage::BlobRef(key.to_string())).await?;
             }
 
             // Delete corresponding range of operations
             let ts_ser = existing_checkpoints[last_to_keep].0.to_string();
-            self.k2v
-                .rm(storage::Selector::Range{
-                    shard_key: &self.path, 
-                    begin: "", 
-                    end: &ts_ser
-                })
-                .await?;
-
+            self.storage.row_rm(&storage::Selector::Range {
+                shard: &self.path, 
+                sort_begin: "", 
+                sort_end: &ts_ser
+            }).await?
         }
 
         Ok(())
@@ -417,11 +390,11 @@ impl<S: BayouState> Bayou<S> {
     async fn list_checkpoints(&self) -> Result<Vec<(Timestamp, String)>> {
         let prefix = format!("{}/checkpoint/", self.path);
 
-        let checkpoints_res = self.s3.list(&prefix).await?;
+        let checkpoints_res = self.storage.blob_list(&prefix).await?;
 
         let mut checkpoints = vec![];
         for object in checkpoints_res {
-            let key = object.key();
+            let key = object.0;
             if let Some(ckid) = key.strip_prefix(&prefix) {
                 if let Ok(ts) = ckid.parse::<Timestamp>() {
                     checkpoints.push((ts, key.into()));
@@ -436,9 +409,8 @@ impl<S: BayouState> Bayou<S> {
 // ---- Bayou watch in K2V ----
 
 struct K2vWatch {
-    pk: String,
-    sk: String,
-    rx: watch::Receiver<storage::OrphanRowRef>,
+    target: storage::RowRef,
+    rx: watch::Receiver<storage::RowRef>,
     notify: Notify,
 }
 
@@ -446,17 +418,17 @@ impl K2vWatch {
     /// Creates a new watch and launches subordinate threads.
     /// These threads hold Weak pointers to the struct;
     /// they exit when the Arc is dropped.
-    fn new(creds: &Credentials, pk: String, sk: String) -> Result<Arc<Self>> {
-        let row_client = creds.row_client()?;
+    fn new(creds: &Credentials, target: storage::RowRef) -> Result<Arc<Self>> {
+        let storage = creds.storage.build()?;
 
-        let (tx, rx) = watch::channel::<storage::OrphanRowRef>(row_client.row(&pk, &sk).to_orphan());
+        let (tx, rx) = watch::channel::<storage::RowRef>(target.clone());
         let notify = Notify::new();
 
-        let watch = Arc::new(K2vWatch { pk, sk, rx, notify });
+        let watch = Arc::new(K2vWatch { target, rx, notify });
 
         tokio::spawn(Self::background_task(
             Arc::downgrade(&watch),
-            row_client,
+            storage,
             tx,
         ));
 
@@ -465,11 +437,11 @@ impl K2vWatch {
 
     async fn background_task(
         self_weak: Weak<Self>,
-        k2v: storage::RowStore,
-        tx: watch::Sender<storage::OrphanRowRef>,
+        storage: storage::Store,
+        tx: watch::Sender<storage::RowRef>,
     ) {
         let mut row = match Weak::upgrade(&self_weak) {
-            Some(this) => k2v.row(&this.pk, &this.sk),
+            Some(this) => this.target.clone(),
             None => {
                 error!("can't start loop");
                 return
@@ -479,20 +451,19 @@ impl K2vWatch {
         while let Some(this) = Weak::upgrade(&self_weak) {
             debug!(
                 "bayou k2v watch bg loop iter ({}, {})",
-                this.pk, this.sk
+                this.target.uid.shard, this.target.uid.sort
             );
             tokio::select!(
                 _ = tokio::time::sleep(Duration::from_secs(60)) => continue,
-                update = row.poll() => {
-                //update = k2v_wait_value_changed(&k2v, &this.pk, &this.sk, &ct) => {
+                update = storage.row_poll(&row) => {
                     match update {
                         Err(e) => {
                             error!("Error in bayou k2v wait value changed: {}", e);
                             tokio::time::sleep(Duration::from_secs(30)).await;
                         }
                         Ok(new_value) => {
-                            row = new_value.to_ref();
-                            if tx.send(row.to_orphan()).is_err() {
+                            row = new_value.row_ref;
+                            if tx.send(row.clone()).is_err() {
                                 break;
                             }
                         }
@@ -500,7 +471,8 @@ impl K2vWatch {
                 }
                 _ = this.notify.notified() => {
                     let rand = u128::to_be_bytes(thread_rng().gen()).to_vec();
-                    if let Err(e) = row.set_value(&rand).push().await
+                    let row_val = storage::RowVal::new(row.clone(), rand);
+                    if let Err(e) = storage.row_insert(vec![row_val]).await
                     {
                         error!("Error in bayou k2v watch updater loop: {}", e);
                         tokio::time::sleep(Duration::from_secs(30)).await;
diff --git a/src/login/ldap_provider.rs b/src/login/ldap_provider.rs
index 4e3af3c..009605d 100644
--- a/src/login/ldap_provider.rs
+++ b/src/login/ldap_provider.rs
@@ -85,11 +85,11 @@ impl LdapLoginProvider {
         })
     }
 
-    fn storage_creds_from_ldap_user(&self, user: &SearchEntry) -> Result<Builders> {
-        let storage: Builders = match &self.storage_specific {
-            StorageSpecific::InMemory => Box::new(storage::in_memory::FullMem::new(
+    fn storage_creds_from_ldap_user(&self, user: &SearchEntry) -> Result<Builder> {
+        let storage: Builder = match &self.storage_specific {
+            StorageSpecific::InMemory => storage::in_memory::MemBuilder::new(
                 &get_attr(user, &self.username_attr)?
-            )),
+            ),
             StorageSpecific::Garage { from_config, bucket_source }  => {
                 let aws_access_key_id = get_attr(user, &from_config.aws_access_key_id_attr)?;
                 let aws_secret_access_key = get_attr(user, &from_config.aws_secret_access_key_attr)?;
@@ -99,14 +99,14 @@ impl LdapLoginProvider {
                 };
 
 
-                Box::new(storage::garage::GrgCreds {
+                storage::garage::GarageBuilder::new(storage::garage::GarageConf {
                     region: from_config.aws_region.clone(), 
                     s3_endpoint: from_config.s3_endpoint.clone(),
                     k2v_endpoint: from_config.k2v_endpoint.clone(),
                     aws_access_key_id, 
                     aws_secret_access_key, 
-                    bucket
-                })
+                    bucket,
+                })?
             },
         };
 
diff --git a/src/login/mod.rs b/src/login/mod.rs
index 9e0c437..d331522 100644
--- a/src/login/mod.rs
+++ b/src/login/mod.rs
@@ -33,23 +33,15 @@ pub type ArcLoginProvider = Arc<dyn LoginProvider + Send + Sync>;
 #[derive(Clone, Debug)]
 pub struct Credentials {
     /// The storage credentials are used to authenticate access to the underlying storage (S3, K2V)
-    pub storage: Builders,
+    pub storage: Builder,
     /// The cryptographic keys are used to encrypt and decrypt data stored in S3 and K2V
     pub keys: CryptoKeys,
 }
-impl Credentials {
-    pub fn row_client(&self) -> Result<RowStore> {
-        Ok(self.storage.row_store()?)
-    }
-    pub fn blob_client(&self) -> Result<BlobStore> {
-        Ok(self.storage.blob_store()?)
-    }
-}
 
 #[derive(Clone, Debug)]
 pub struct PublicCredentials {
     /// The storage credentials are used to authenticate access to the underlying storage (S3, K2V)
-    pub storage: Builders,
+    pub storage: Builder,
     pub public_key: PublicKey,
 }
 
diff --git a/src/login/static_provider.rs b/src/login/static_provider.rs
index 788a4c5..5896f16 100644
--- a/src/login/static_provider.rs
+++ b/src/login/static_provider.rs
@@ -88,16 +88,16 @@ impl LoginProvider for StaticLoginProvider {
         }
 
         tracing::debug!(user=%username, "fetch keys");
-        let storage: storage::Builders = match &user.config.storage {
-            StaticStorage::InMemory => Box::new(storage::in_memory::FullMem::new(username)),
-            StaticStorage::Garage(grgconf) => Box::new(storage::garage::GrgCreds {
+        let storage: storage::Builder = match &user.config.storage {
+            StaticStorage::InMemory => storage::in_memory::MemBuilder::new(username),
+            StaticStorage::Garage(grgconf) => storage::garage::GarageBuilder::new(storage::garage::GarageConf {
                 region: grgconf.aws_region.clone(),
                 k2v_endpoint: grgconf.k2v_endpoint.clone(),
                 s3_endpoint: grgconf.s3_endpoint.clone(),
                 aws_access_key_id: grgconf.aws_access_key_id.clone(),
                 aws_secret_access_key: grgconf.aws_secret_access_key.clone(),
                 bucket: grgconf.bucket.clone(),
-            }),
+            })?,
         };
 
         let cr = CryptoRoot(user.config.crypto_root.clone());
@@ -114,16 +114,16 @@ impl LoginProvider for StaticLoginProvider {
             Some(u) => u,
         };
 
-        let storage: storage::Builders = match &user.config.storage {
-            StaticStorage::InMemory => Box::new(storage::in_memory::FullMem::new(&user.username)),
-            StaticStorage::Garage(grgconf) => Box::new(storage::garage::GrgCreds {
+        let storage: storage::Builder = match &user.config.storage {
+            StaticStorage::InMemory => storage::in_memory::MemBuilder::new(&user.username),
+            StaticStorage::Garage(grgconf) => storage::garage::GarageBuilder::new(storage::garage::GarageConf {
                 region: grgconf.aws_region.clone(),
                 k2v_endpoint: grgconf.k2v_endpoint.clone(),
                 s3_endpoint: grgconf.s3_endpoint.clone(),
                 aws_access_key_id: grgconf.aws_access_key_id.clone(),
                 aws_secret_access_key: grgconf.aws_secret_access_key.clone(),
                 bucket: grgconf.bucket.clone(),
-            }),
+            })?,
         };
 
         let cr = CryptoRoot(user.config.crypto_root.clone());
diff --git a/src/mail/incoming.rs b/src/mail/incoming.rs
index e3c729f..f6b831d 100644
--- a/src/mail/incoming.rs
+++ b/src/mail/incoming.rs
@@ -5,6 +5,7 @@ use std::sync::{Arc, Weak};
 use std::time::Duration;
 
 use anyhow::{anyhow, bail, Result};
+use base64::Engine;
 use futures::{future::BoxFuture, FutureExt};
 //use tokio::io::AsyncReadExt;
 use tokio::sync::watch;
@@ -50,13 +51,11 @@ async fn incoming_mail_watch_process_internal(
     creds: Credentials,
     mut rx_inbox_id: watch::Receiver<Option<(UniqueIdent, ImapUidvalidity)>>,
 ) -> Result<()> {
-    let mut lock_held = k2v_lock_loop(creds.row_client()?, INCOMING_PK, INCOMING_LOCK_SK);
-
-    let k2v = creds.row_client()?;
-    let s3 = creds.blob_client()?;
+    let mut lock_held = k2v_lock_loop(creds.storage.build()?, storage::RowRef::new(INCOMING_PK, INCOMING_LOCK_SK));
+    let storage = creds.storage.build()?;
 
     let mut inbox: Option<Arc<Mailbox>> = None;
-    let mut incoming_key = k2v.row(INCOMING_PK, INCOMING_WATCH_SK);
+    let mut incoming_key = storage::RowRef::new(INCOMING_PK, INCOMING_WATCH_SK);
 
     loop {
         let maybe_updated_incoming_key = if *lock_held.borrow() {
@@ -64,9 +63,9 @@ async fn incoming_mail_watch_process_internal(
 
             let wait_new_mail = async {
                 loop {
-                    match incoming_key.poll().await
+                    match storage.row_poll(&incoming_key).await
                     {
-                        Ok(row_val) => break row_val.to_ref(),
+                        Ok(row_val) => break row_val.row_ref,
                         Err(e) => {
                             error!("Error in wait_new_mail: {}", e);
                             tokio::time::sleep(Duration::from_secs(30)).await;
@@ -77,7 +76,7 @@ async fn incoming_mail_watch_process_internal(
 
             tokio::select! {
                 inc_k = wait_new_mail => Some(inc_k),
-                _     = tokio::time::sleep(MAIL_CHECK_INTERVAL) => Some(k2v.from_orphan(incoming_key.to_orphan()).expect("Incompatible source & target storage")),
+                _     = tokio::time::sleep(MAIL_CHECK_INTERVAL) => Some(incoming_key.clone()),
                 _     = lock_held.changed() => None,
                 _     = rx_inbox_id.changed() => None,
             }
@@ -119,7 +118,7 @@ async fn incoming_mail_watch_process_internal(
         // If we were able to open INBOX, and we have mail,
         // fetch new mail
         if let (Some(inbox), Some(updated_incoming_key)) = (&inbox, maybe_updated_incoming_key) {
-            match handle_incoming_mail(&user, &s3, inbox, &lock_held).await {
+            match handle_incoming_mail(&user, &storage, inbox, &lock_held).await {
                 Ok(()) => {
                     incoming_key = updated_incoming_key;
                 }
@@ -136,20 +135,20 @@ async fn incoming_mail_watch_process_internal(
 
 async fn handle_incoming_mail(
     user: &Arc<User>,
-    blobs: &storage::BlobStore,
+    storage: &storage::Store,
     inbox: &Arc<Mailbox>,
     lock_held: &watch::Receiver<bool>,
 ) -> Result<()> {
-    let mails_res = blobs.list("incoming/").await?;
+    let mails_res = storage.blob_list("incoming/").await?;
 
     for object in mails_res {
         if !*lock_held.borrow() {
             break;
         }
-        let key = object.key();
+        let key = object.0;
         if let Some(mail_id) = key.strip_prefix("incoming/") {
             if let Ok(mail_id) = mail_id.parse::<UniqueIdent>() {
-                move_incoming_message(user, blobs, inbox, mail_id).await?;
+                move_incoming_message(user, storage, inbox, mail_id).await?;
             }
         }
     }
@@ -159,7 +158,7 @@ async fn handle_incoming_mail(
 
 async fn move_incoming_message(
     user: &Arc<User>,
-    s3: &storage::BlobStore,
+    storage: &storage::Store,
     inbox: &Arc<Mailbox>,
     id: UniqueIdent,
 ) -> Result<()> {
@@ -168,14 +167,15 @@ async fn move_incoming_message(
     let object_key = format!("incoming/{}", id);
 
     // 1. Fetch message from S3
-    let object = s3.blob(&object_key).fetch().await?;
+    let object = storage.blob_fetch(&storage::BlobRef(object_key)).await?;
 
     // 1.a decrypt message key from headers
     //info!("Object metadata: {:?}", get_result.metadata);
     let key_encrypted_b64 = object
-        .get_meta(MESSAGE_KEY)
+        .meta
+        .get(MESSAGE_KEY)
         .ok_or(anyhow!("Missing key in metadata"))?;
-    let key_encrypted = base64::decode(key_encrypted_b64)?;
+    let key_encrypted = base64::engine::general_purpose::STANDARD.decode(key_encrypted_b64)?;
     let message_key = sodiumoxide::crypto::sealedbox::open(
         &key_encrypted,
         &user.creds.keys.public,
@@ -186,28 +186,28 @@ async fn move_incoming_message(
         cryptoblob::Key::from_slice(&message_key).ok_or(anyhow!("Invalid message key"))?;
 
     // 1.b retrieve message body
-    let obj_body = object.content().ok_or(anyhow!("Missing object body"))?;
+    let obj_body = object.value;
     let plain_mail = cryptoblob::open(&obj_body, &message_key)
         .map_err(|_| anyhow!("Cannot decrypt email content"))?;
 
     // 2 parse mail and add to inbox
     let msg = IMF::try_from(&plain_mail[..]).map_err(|_| anyhow!("Invalid email body"))?;
     inbox
-        .append_from_s3(msg, id, object.to_ref(), message_key)
+        .append_from_s3(msg, id, object.blob_ref.clone(), message_key)
         .await?;
 
     // 3 delete from incoming
-    object.to_ref().rm().await?;
+    storage.blob_rm(&object.blob_ref).await?;
 
     Ok(())
 }
 
 // ---- UTIL: K2V locking loop, use this to try to grab a lock using a K2V entry as a signal ----
 
-fn k2v_lock_loop(k2v: storage::RowStore, pk: &'static str, sk: &'static str) -> watch::Receiver<bool> {
+fn k2v_lock_loop(storage: storage::Store, row_ref: storage::RowRef) -> watch::Receiver<bool> {
     let (held_tx, held_rx) = watch::channel(false);
 
-    tokio::spawn(k2v_lock_loop_internal(k2v, pk, sk, held_tx));
+    tokio::spawn(k2v_lock_loop_internal(storage, row_ref, held_tx));
 
     held_rx
 }
@@ -216,13 +216,12 @@ fn k2v_lock_loop(k2v: storage::RowStore, pk: &'static str, sk: &'static str) ->
 enum LockState {
     Unknown,
     Empty,
-    Held(UniqueIdent, u64, storage::OrphanRowRef),
+    Held(UniqueIdent, u64, storage::RowRef),
 }
 
 async fn k2v_lock_loop_internal(
-    k2v: storage::RowStore,
-    pk: &'static str,
-    sk: &'static str,
+    storage: storage::Store,
+    row_ref: storage::RowRef,
     held_tx: watch::Sender<bool>,
 ) {
     let (state_tx, mut state_rx) = watch::channel::<LockState>(LockState::Unknown);
@@ -232,10 +231,10 @@ async fn k2v_lock_loop_internal(
 
     // Loop 1: watch state of lock in K2V, save that in corresponding watch channel
     let watch_lock_loop: BoxFuture<Result<()>> = async {
-        let mut ct = k2v.row(pk, sk);
+        let mut ct = row_ref.clone();
         loop {
             info!("k2v watch lock loop iter: ct = {:?}", ct);
-            match ct.poll().await {
+            match storage.row_poll(&ct).await {
                 Err(e) => {
                     error!(
                         "Error in k2v wait value changed: {} ; assuming we no longer hold lock.",
@@ -246,7 +245,7 @@ async fn k2v_lock_loop_internal(
                 }
                 Ok(cv) => {
                     let mut lock_state = None;
-                    for v in cv.content().iter() {
+                    for v in cv.value.iter() {
                         if let storage::Alternative::Value(vbytes) = v {
                             if vbytes.len() == 32 {
                                 let ts = u64::from_be_bytes(vbytes[..8].try_into().unwrap());
@@ -260,7 +259,7 @@ async fn k2v_lock_loop_internal(
                             }
                         }
                     }
-                    let new_ct = cv.to_ref();
+                    let new_ct = cv.row_ref;
 
                     info!(
                         "k2v watch lock loop: changed, old ct = {:?}, new ct = {:?}, v = {:?}",
@@ -268,7 +267,7 @@ async fn k2v_lock_loop_internal(
                     );
                     state_tx.send(
                         lock_state
-                            .map(|(pid, ts)| LockState::Held(pid, ts, new_ct.to_orphan()))
+                            .map(|(pid, ts)| LockState::Held(pid, ts, new_ct.clone()))
                             .unwrap_or(LockState::Empty),
                     )?;
                     ct = new_ct;
@@ -358,10 +357,10 @@ async fn k2v_lock_loop_internal(
             ));
             lock[8..].copy_from_slice(&our_pid.0);
             let row = match ct {
-                Some(orphan) => k2v.from_orphan(orphan).expect("Source & target must be storage compatible"),
-                None => k2v.row(pk, sk),
+                Some(existing) => existing,
+                None => row_ref.clone(),
             };
-            if let Err(e) = row.set_value(&lock).push().await {
+            if let Err(e) = storage.row_insert(vec![storage::RowVal::new(row, lock)]).await {
                 error!("Could not take lock: {}", e);
                 tokio::time::sleep(Duration::from_secs(30)).await;
             }
@@ -377,7 +376,7 @@ async fn k2v_lock_loop_internal(
     info!("lock loop exited, releasing");
 
     if !held_tx.is_closed() {
-        warn!("wierd...");
+        warn!("weird...");
         let _ = held_tx.send(false);
     }
 
@@ -387,8 +386,10 @@ async fn k2v_lock_loop_internal(
         _ => None,
     };
     if let Some(ct) = release {
-        let row = k2v.from_orphan(ct).expect("Incompatible source & target storage");
-        let _ = row.rm().await;
+        match storage.row_rm(&storage::Selector::Single(&ct)).await {
+            Err(e) => warn!("Unable to release lock {:?}: {}", ct, e),
+            Ok(_) => (),
+        };
     }
 }
 
@@ -410,30 +411,32 @@ impl EncryptedMessage {
     }
 
     pub async fn deliver_to(self: Arc<Self>, creds: PublicCredentials) -> Result<()> {
-        let s3_client = creds.storage.blob_store()?;
-        let k2v_client = creds.storage.row_store()?;
+        let storage = creds.storage.build()?;
 
         // Get causality token of previous watch key
-        let query = k2v_client.row(INCOMING_PK, INCOMING_WATCH_SK);
-        let watch_ct = match query.fetch().await {
+        let query = storage::RowRef::new(INCOMING_PK, INCOMING_WATCH_SK);
+        let watch_ct = match storage.row_fetch(&storage::Selector::Single(&query)).await {
             Err(_) => query,
-            Ok(cv) => cv.to_ref(),
+            Ok(cv) => cv.into_iter().next().map(|v| v.row_ref).unwrap_or(query),
         };
 
         // Write mail to encrypted storage
         let encrypted_key =
             sodiumoxide::crypto::sealedbox::seal(self.key.as_ref(), &creds.public_key);
-        let key_header = base64::encode(&encrypted_key);
+        let key_header = base64::engine::general_purpose::STANDARD.encode(&encrypted_key);
 
-        let mut  send = s3_client
-            .blob(&format!("incoming/{}", gen_ident()))
-            .set_value(self.encrypted_body.clone().into());
-        send.set_meta(MESSAGE_KEY, &key_header);
-        send.push().await?;
+        let blob_val = storage::BlobVal::new(
+            storage::BlobRef(format!("incoming/{}", gen_ident())),
+            self.encrypted_body.clone().into(),
+        ).with_meta(MESSAGE_KEY.to_string(), key_header);
+        storage.blob_insert(&blob_val).await?;
 
         // Update watch key to signal new mail
-        watch_ct.set_value(gen_ident().0.as_ref()).push().await?;
-
+        let watch_val = storage::RowVal::new(
+            watch_ct.clone(),
+            gen_ident().0.to_vec(),
+        );
+        storage.row_insert(vec![watch_val]).await?;
         Ok(())
     }
 }
diff --git a/src/mail/mailbox.rs b/src/mail/mailbox.rs
index 060267a..b4afd5e 100644
--- a/src/mail/mailbox.rs
+++ b/src/mail/mailbox.rs
@@ -8,7 +8,7 @@ use crate::login::Credentials;
 use crate::mail::uidindex::*;
 use crate::mail::unique_ident::*;
 use crate::mail::IMF;
-use crate::storage::{RowStore, BlobStore, self};
+use crate::storage::{Store, RowRef, RowVal, BlobRef, BlobVal, Selector, self};
 use crate::timestamp::now_msec;
 
 pub struct Mailbox {
@@ -44,8 +44,7 @@ impl Mailbox {
         let mbox = RwLock::new(MailboxInternal {
             id,
             encryption_key: creds.keys.master.clone(),
-            k2v: creds.storage.row_store()?,
-            s3: creds.storage.blob_store()?,
+            storage: creds.storage.build()?,
             uid_index,
             mail_path,
         });
@@ -178,10 +177,7 @@ struct MailboxInternal {
     id: UniqueIdent,
     mail_path: String,
     encryption_key: Key,
-
-    k2v: RowStore,
-    s3: BlobStore,
-
+    storage: Store,
     uid_index: Bayou<UidIndex>,
 }
 
@@ -200,15 +196,15 @@ impl MailboxInternal {
 
     async fn fetch_meta(&self, ids: &[UniqueIdent]) -> Result<Vec<MailMeta>> {
         let ids = ids.iter().map(|x| x.to_string()).collect::<Vec<_>>();
-        let ops = ids.iter().map(|id| (self.mail_path.as_str(), id.as_str())).collect::<Vec<_>>();
-        let res_vec = self.k2v.select(storage::Selector::List(ops)).await?;
+        let ops = ids.iter().map(|id| RowRef::new(self.mail_path.as_str(), id.as_str())).collect::<Vec<_>>();
+        let res_vec = self.storage.row_fetch(&Selector::List(ops)).await?;
 
         let mut meta_vec = vec![];
         for res in res_vec.into_iter() {
             let mut meta_opt = None;
 
             // Resolve conflicts
-            for v in res.content().iter() {
+            for v in res.value.iter() {
                 match v {
                     storage::Alternative::Tombstone => (),
                     storage::Alternative::Value(v) => {
@@ -227,7 +223,7 @@ impl MailboxInternal {
             if let Some(meta) = meta_opt {
                 meta_vec.push(meta);
             } else {
-                bail!("No valid meta value in k2v for {:?}", res.to_ref().key());
+                bail!("No valid meta value in k2v for {:?}", res.row_ref);
             }
         }
 
@@ -235,9 +231,9 @@ impl MailboxInternal {
     }
 
     async fn fetch_full(&self, id: UniqueIdent, message_key: &Key) -> Result<Vec<u8>> {
-        let obj_res = self.s3.blob(&format!("{}/{}", self.mail_path, id)).fetch().await?;
-        let body = obj_res.content().ok_or(anyhow!("missing body"))?;
-        cryptoblob::open(body, message_key)
+        let obj_res = self.storage.blob_fetch(&BlobRef(format!("{}/{}", self.mail_path, id))).await?;
+        let body = obj_res.value;
+        cryptoblob::open(&body, message_key)
     }
 
     // ---- Functions for changing the mailbox ----
@@ -270,7 +266,10 @@ impl MailboxInternal {
             async {
                 // Encrypt and save mail body
                 let message_blob = cryptoblob::seal(mail.raw, &message_key)?;
-                self.s3.blob(&format!("{}/{}", self.mail_path, ident)).set_value(message_blob).push().await?;
+                self.storage.blob_insert(&BlobVal::new(
+                    BlobRef(format!("{}/{}", self.mail_path, ident)),
+                    message_blob,
+                )).await?;
                 Ok::<_, anyhow::Error>(())
             },
             async {
@@ -282,7 +281,10 @@ impl MailboxInternal {
                     rfc822_size: mail.raw.len(),
                 };
                 let meta_blob = seal_serialize(&meta, &self.encryption_key)?;
-                self.k2v.row(&self.mail_path, &ident.to_string()).set_value(&meta_blob).push().await?;
+                self.storage.row_insert(vec![RowVal::new(
+                    RowRef::new(&self.mail_path, &ident.to_string()),
+                    meta_blob,
+                )]).await?;
                 Ok::<_, anyhow::Error>(())
             },
             self.uid_index.opportunistic_sync()
@@ -307,14 +309,14 @@ impl MailboxInternal {
         &mut self,
         mail: IMF<'a>,
         ident: UniqueIdent,
-        blob_ref: storage::BlobRef,
+        blob_src: storage::BlobRef,
         message_key: Key,
     ) -> Result<()> {
         futures::try_join!(
             async {
                 // Copy mail body from previous location
-                let dst = self.s3.blob(&format!("{}/{}", self.mail_path, ident));
-                blob_ref.copy(&dst).await?;
+                let blob_dst = BlobRef(format!("{}/{}", self.mail_path, ident));
+                self.storage.blob_copy(&blob_src, &blob_dst).await?;
                 Ok::<_, anyhow::Error>(())
             },
             async {
@@ -326,7 +328,10 @@ impl MailboxInternal {
                     rfc822_size: mail.raw.len(),
                 };
                 let meta_blob = seal_serialize(&meta, &self.encryption_key)?;
-                self.k2v.row(&self.mail_path, &ident.to_string()).set_value(&meta_blob).push().await?;
+                self.storage.row_insert(vec![RowVal::new(
+                        RowRef::new(&self.mail_path, &ident.to_string()),
+                        meta_blob,
+                )]).await?;
                 Ok::<_, anyhow::Error>(())
             },
             self.uid_index.opportunistic_sync()
@@ -350,13 +355,13 @@ impl MailboxInternal {
         futures::try_join!(
             async {
                 // Delete mail body from S3
-                self.s3.blob(&format!("{}/{}", self.mail_path, ident)).rm().await?;
+                self.storage.blob_rm(&BlobRef(format!("{}/{}", self.mail_path, ident))).await?;
                 Ok::<_, anyhow::Error>(())
             },
             async {
                 // Delete mail meta from K2V
                 let sk = ident.to_string();
-                self.k2v.row(&self.mail_path, &sk).fetch().await?.to_ref().rm().await?;
+                self.storage.row_rm(&Selector::Single(&RowRef::new(&self.mail_path, &sk))).await?;
                 Ok::<_, anyhow::Error>(())
             }
         )?;
@@ -402,15 +407,19 @@ impl MailboxInternal {
 
         futures::try_join!(
             async {
-                let dst = self.s3.blob(&format!("{}/{}", self.mail_path, new_id));
-                self.s3.blob(&format!("{}/{}", from.mail_path, source_id)).copy(&dst).await?;
+                let dst = BlobRef(format!("{}/{}", self.mail_path, new_id));
+                let src = BlobRef(format!("{}/{}", from.mail_path, source_id));
+                self.storage.blob_copy(&src, &dst).await?;
                 Ok::<_, anyhow::Error>(())
             },
             async {
                 // Copy mail meta in K2V
                 let meta = &from.fetch_meta(&[source_id]).await?[0];
                 let meta_blob = seal_serialize(meta, &self.encryption_key)?;
-                self.k2v.row(&self.mail_path, &new_id.to_string()).set_value(&meta_blob).push().await?;
+                self.storage.row_insert(vec![RowVal::new(
+                    RowRef::new(&self.mail_path, &new_id.to_string()),
+                    meta_blob,
+                )]).await?;
                 Ok::<_, anyhow::Error>(())
             },
             self.uid_index.opportunistic_sync(),
diff --git a/src/mail/user.rs b/src/mail/user.rs
index bdfb30c..8413cbf 100644
--- a/src/mail/user.rs
+++ b/src/mail/user.rs
@@ -33,7 +33,7 @@ const MAILBOX_LIST_SK: &str = "list";
 pub struct User {
     pub username: String,
     pub creds: Credentials,
-    pub k2v: storage::RowStore,
+    pub storage: storage::Store,
     pub mailboxes: std::sync::Mutex<HashMap<UniqueIdent, Weak<Mailbox>>>,
 
     tx_inbox_id: watch::Sender<Option<(UniqueIdent, ImapUidvalidity)>>,
@@ -41,7 +41,7 @@ pub struct User {
 
 impl User {
     pub async fn new(username: String, creds: Credentials) -> Result<Arc<Self>> {
-        let cache_key = (username.clone(), creds.storage.clone());
+        let cache_key = (username.clone(), creds.storage.unique());
 
         {
             let cache = USER_CACHE.lock().unwrap();
@@ -81,11 +81,7 @@ impl User {
             let mb_uidvalidity = mb.current_uid_index().await.uidvalidity;
             if mb_uidvalidity > uidvalidity {
                 list.update_uidvalidity(name, mb_uidvalidity);
-                let orphan = match ct {
-                    Some(x) => Some(x.to_orphan()),
-                    None => None,
-                };
-                self.save_mailbox_list(&list, orphan).await?;
+                self.save_mailbox_list(&list, ct).await?;
             }
             Ok(Some(mb))
         } else {
@@ -108,11 +104,7 @@ impl User {
         let (mut list, ct) = self.load_mailbox_list().await?;
         match list.create_mailbox(name) {
             CreatedMailbox::Created(_, _) => {
-                let orphan = match ct {
-                    Some(x) => Some(x.to_orphan()),
-                    None => None,
-                };
-                self.save_mailbox_list(&list, orphan).await?;
+                self.save_mailbox_list(&list, ct).await?;
                 Ok(())
             }
             CreatedMailbox::Existed(_, _) => Err(anyhow!("Mailbox {} already exists", name)),
@@ -129,11 +121,7 @@ impl User {
         if list.has_mailbox(name) {
             // TODO: actually delete mailbox contents
             list.set_mailbox(name, None);
-            let orphan = match ct {
-                Some(x) => Some(x.to_orphan()),
-                None => None,
-            };
-            self.save_mailbox_list(&list, orphan).await?;
+            self.save_mailbox_list(&list, ct).await?;
             Ok(())
         } else {
             bail!("Mailbox {} does not exist", name);
@@ -154,11 +142,7 @@ impl User {
         if old_name == INBOX {
             list.rename_mailbox(old_name, new_name)?;
             if !self.ensure_inbox_exists(&mut list, &ct).await? {
-                let orphan = match ct {
-                    Some(x) => Some(x.to_orphan()),
-                    None => None,
-                };
-                self.save_mailbox_list(&list, orphan).await?;
+                self.save_mailbox_list(&list, ct).await?;
             }
         } else {
             let names = list.existing_mailbox_names();
@@ -182,11 +166,7 @@ impl User {
                 }
             }
 
-            let orphan = match ct {
-                Some(x) => Some(x.to_orphan()),
-                None => None,
-            };
-            self.save_mailbox_list(&list, orphan).await?;
+            self.save_mailbox_list(&list, ct).await?;
         }
         Ok(())
     }
@@ -194,14 +174,14 @@ impl User {
     // ---- Internal user & mailbox management ----
 
     async fn open(username: String, creds: Credentials) -> Result<Arc<Self>> {
-        let k2v = creds.row_client()?;
+        let storage = creds.storage.build()?;
 
         let (tx_inbox_id, rx_inbox_id) = watch::channel(None);
 
         let user = Arc::new(Self {
             username,
             creds: creds.clone(),
-            k2v,
+            storage,
             tx_inbox_id,
             mailboxes: std::sync::Mutex::new(HashMap::new()),
         });
@@ -245,19 +225,25 @@ impl User {
     // ---- Mailbox list management ----
 
     async fn load_mailbox_list(&self) -> Result<(MailboxList, Option<storage::RowRef>)> {
-        let (mut list, row) = match self.k2v.row(MAILBOX_LIST_PK, MAILBOX_LIST_SK).fetch().await {
+        let row_ref = storage::RowRef::new(MAILBOX_LIST_PK, MAILBOX_LIST_SK);
+        let (mut list, row) = match self.storage.row_fetch(&storage::Selector::Single(&row_ref)).await {
             Err(storage::StorageError::NotFound) => (MailboxList::new(), None),
             Err(e) => return Err(e.into()),
             Ok(rv) => {
                 let mut list = MailboxList::new();
-                for v in rv.content() {
+                let (row_ref, row_vals) = match rv.into_iter().next() {
+                    Some(row_val) => (row_val.row_ref, row_val.value),
+                    None => (row_ref, vec![]),
+                };
+
+                for v in row_vals {
                     if let storage::Alternative::Value(vbytes) = v {
                         let list2 =
                             open_deserialize::<MailboxList>(&vbytes, &self.creds.keys.master)?;
                         list.merge(list2);
                     }
                 }
-                (list, Some(rv.to_ref()))
+                (list, Some(row_ref))
             }
         };
 
@@ -278,11 +264,7 @@ impl User {
         let saved;
         let (inbox_id, inbox_uidvalidity) = match list.create_mailbox(INBOX) {
             CreatedMailbox::Created(i, v) => {
-                let orphan = match ct {
-                    Some(x) => Some(x.to_orphan()),
-                    None => None,
-                };
-                self.save_mailbox_list(list, orphan).await?;
+                self.save_mailbox_list(list, ct.clone()).await?;
                 saved = true;
                 (i, v)
             }
@@ -302,14 +284,12 @@ impl User {
     async fn save_mailbox_list(
         &self,
         list: &MailboxList,
-        ct: Option<storage::OrphanRowRef>,
+        ct: Option<storage::RowRef>,
     ) -> Result<()> {
         let list_blob = seal_serialize(list, &self.creds.keys.master)?;
-        let rref = match ct {
-            Some(x) => self.k2v.from_orphan(x).expect("Source & target must be same storage"),
-            None => self.k2v.row(MAILBOX_LIST_PK, MAILBOX_LIST_SK),
-        };
-        rref.set_value(&list_blob).push().await?;
+        let rref = ct.unwrap_or(storage::RowRef::new(MAILBOX_LIST_PK, MAILBOX_LIST_SK));
+        let row_val = storage::RowVal::new(rref, list_blob);
+        self.storage.row_insert(vec![row_val]).await?;
         Ok(())
     }
 }
@@ -482,6 +462,6 @@ enum CreatedMailbox {
 // ---- User cache ----
 
 lazy_static! {
-    static ref USER_CACHE: std::sync::Mutex<HashMap<(String, storage::Builders), Weak<User>>> =
+    static ref USER_CACHE: std::sync::Mutex<HashMap<(String, storage::UnicityBuffer), Weak<User>>> =
         std::sync::Mutex::new(HashMap::new());
 }
diff --git a/src/storage/garage.rs b/src/storage/garage.rs
index 8276f70..ff37287 100644
--- a/src/storage/garage.rs
+++ b/src/storage/garage.rs
@@ -1,7 +1,8 @@
 use crate::storage::*;
+use serde::Serialize;
 
-#[derive(Clone, Debug, Hash)]
-pub struct GarageBuilder {
+#[derive(Clone, Debug, Serialize)]
+pub struct GarageConf {
     pub region: String,
     pub s3_endpoint: String,
     pub k2v_endpoint: String,
@@ -10,10 +11,28 @@ pub struct GarageBuilder {
     pub bucket: String,
 }
 
+#[derive(Clone, Debug)]
+pub struct GarageBuilder {
+    conf: GarageConf,
+    unicity: Vec<u8>,
+}
+
+impl GarageBuilder {
+    pub fn new(conf: GarageConf) -> anyhow::Result<Arc<Self>> {
+        let mut unicity: Vec<u8> = vec![];
+        unicity.extend_from_slice(file!().as_bytes());
+        unicity.append(&mut rmp_serde::to_vec(&conf)?);
+        Ok(Arc::new(Self { conf, unicity }))
+    } 
+}
+
 impl IBuilder for GarageBuilder {
-    fn build(&self) -> Box<dyn IStore> {
+    fn build(&self) -> Result<Store, StorageError> {
         unimplemented!();
     }
+    fn unique(&self) -> UnicityBuffer {
+        UnicityBuffer(self.unicity.clone())
+    }
 }
 
 pub struct GarageStore {
@@ -33,7 +52,7 @@ impl IStore for GarageStore {
         unimplemented!();
 
     }
-    async fn row_poll(&self, value: RowRef) -> Result<RowVal, StorageError> {
+    async fn row_poll(&self, value: &RowRef) -> Result<RowVal, StorageError> {
         unimplemented!();
     }
 
@@ -41,6 +60,9 @@ impl IStore for GarageStore {
         unimplemented!();
 
     }
+    async fn blob_insert(&self, blob_val: &BlobVal) -> Result<BlobVal, StorageError> {
+        unimplemented!();
+    }
     async fn blob_copy(&self, src: &BlobRef, dst: &BlobRef) -> Result<BlobVal, StorageError> {
         unimplemented!();
 
diff --git a/src/storage/in_memory.rs b/src/storage/in_memory.rs
index 09c6763..6d0460f 100644
--- a/src/storage/in_memory.rs
+++ b/src/storage/in_memory.rs
@@ -14,18 +14,36 @@ pub type ArcBlob = Arc<RwLock<HashMap<String, Vec<u8>>>>;
 #[derive(Clone, Debug)]
 pub struct MemBuilder {
     user: String,
-    url: String,
+    unicity: Vec<u8>,
     row: ArcRow,
     blob: ArcBlob,
 }
 
+impl MemBuilder {
+    pub fn new(user: &str) -> Arc<Self> {
+        let mut unicity: Vec<u8> = vec![];
+        unicity.extend_from_slice(file!().as_bytes());
+        unicity.extend_from_slice(user.as_bytes());
+        Arc::new(Self {
+            user: user.to_string(),
+            unicity,
+            row: Arc::new(RwLock::new(HashMap::new())),
+            blob: Arc::new(RwLock::new(HashMap::new())),
+        })
+    }
+}
+
 impl IBuilder for MemBuilder {
-    fn build(&self) -> Box<dyn IStore> {
-        Box::new(MemStore {
+    fn build(&self) -> Result<Store, StorageError> {
+        Ok(Box::new(MemStore {
             row: self.row.clone(),
             blob: self.blob.clone(),
-        })
+        }))
     } 
+
+    fn unique(&self) -> UnicityBuffer {
+        UnicityBuffer(self.unicity.clone())
+    }
 }
 
 pub struct MemStore {
@@ -56,7 +74,7 @@ impl IStore for MemStore {
                     .or(Err(StorageError::Internal))?
                     .get(*shard)
                     .ok_or(StorageError::NotFound)?
-                    .range((Included(sort_begin.to_string()), Included(sort_end.to_string())))
+                    .range((Included(sort_begin.to_string()), Excluded(sort_end.to_string())))
                     .map(|(k, v)| RowVal {
                         row_ref: RowRef { uid: RowUid { shard: shard.to_string(), sort: k.to_string() }, causality: Some("c".to_string()) },
                         value: vec![Alternative::Value(v.clone())],
@@ -100,7 +118,7 @@ impl IStore for MemStore {
             },
             Selector::Single(row_ref) => {
                 let bytes = self.inner_fetch(row_ref)?;
-                Ok(vec![RowVal{ row_ref: row_ref.clone(), value: vec![Alternative::Value(bytes)]}])
+                Ok(vec![RowVal{ row_ref: (*row_ref).clone(), value: vec![Alternative::Value(bytes)]}])
             }
         }
     }
@@ -113,7 +131,7 @@ impl IStore for MemStore {
         unimplemented!();
 
     }
-    async fn row_poll(&self, value: RowRef) -> Result<RowVal, StorageError> {
+    async fn row_poll(&self, value: &RowRef) -> Result<RowVal, StorageError> {
         unimplemented!();
     }
 
@@ -121,6 +139,9 @@ impl IStore for MemStore {
         unimplemented!();
 
     }
+    async fn blob_insert(&self, blob_val: &BlobVal) -> Result<BlobVal, StorageError> {
+        unimplemented!();
+    }
     async fn blob_copy(&self, src: &BlobRef, dst: &BlobRef) -> Result<BlobVal, StorageError> {
         unimplemented!();
 
diff --git a/src/storage/mod.rs b/src/storage/mod.rs
index cb66d58..8004ac5 100644
--- a/src/storage/mod.rs
+++ b/src/storage/mod.rs
@@ -11,9 +11,9 @@
 pub mod in_memory;
 pub mod garage;
 
-use std::hash::{Hash, Hasher};
+use std::sync::Arc;
+use std::hash::Hash;
 use std::collections::HashMap;
-use futures::future::BoxFuture;
 use async_trait::async_trait;
 
 #[derive(Debug, Clone)]
@@ -23,45 +23,95 @@ pub enum Alternative {
 }
 type ConcurrentValues = Vec<Alternative>;
 
-#[derive(Debug)]
+#[derive(Debug, Clone)]
 pub enum StorageError {
     NotFound,
     Internal,
 }
+impl std::fmt::Display for StorageError {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.write_str("Storage Error: ")?;
+        match self {
+            Self::NotFound => f.write_str("Item not found"),
+            Self::Internal => f.write_str("An internal error occured"),
+        }
+    }
+}
+impl std::error::Error for StorageError {}
 
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, PartialEq)]
 pub struct RowUid {
-    shard: String,
-    sort: String,
+    pub shard: String,
+    pub sort: String,
 }
 
-#[derive(Debug, Clone)]
+#[derive(Debug, Clone, PartialEq)]
 pub struct RowRef {
-    uid: RowUid,
-    causality: Option<String>,
+    pub uid: RowUid,
+    pub causality: Option<String>,
+}
+
+impl RowRef {
+    pub fn new(shard: &str, sort: &str) -> Self {
+        Self {
+            uid: RowUid { 
+                shard: shard.to_string(),
+                sort: sort.to_string(),
+            },
+            causality: None,
+        }
+    }
 }
 
 #[derive(Debug, Clone)]
 pub struct RowVal {
-    row_ref: RowRef,
-    value: ConcurrentValues,
+    pub row_ref: RowRef,
+    pub value: ConcurrentValues,
 }
 
+impl RowVal {
+    pub fn new(row_ref: RowRef, value: Vec<u8>) -> Self {
+        Self {
+            row_ref,
+            value: vec![Alternative::Value(value)],
+        }
+    }
+}
+
+
 #[derive(Debug, Clone)]
-pub struct BlobRef(String);
+pub struct BlobRef(pub String);
+impl BlobRef {
+    pub fn new(key: &str) -> Self {
+        Self(key.to_string())
+    }
+}
 
 #[derive(Debug, Clone)]
 pub struct BlobVal {
-    blob_ref: BlobRef,
-    meta: HashMap<String, String>,
-    value: Vec<u8>,
+    pub blob_ref: BlobRef,
+    pub meta: HashMap<String, String>,
+    pub value: Vec<u8>,
+}
+impl BlobVal {
+    pub fn new(blob_ref: BlobRef, value: Vec<u8>) -> Self {
+        Self {
+            blob_ref, value,
+            meta: HashMap::new(),
+        }
+    }
+
+    pub fn with_meta(mut self, k: String, v: String) -> Self {
+        self.meta.insert(k, v);
+        self
+    }
 }
 
 pub enum Selector<'a> {
     Range { shard: &'a str, sort_begin: &'a str, sort_end: &'a str },
     List (Vec<RowRef>), // list of (shard_key, sort_key)
     Prefix { shard: &'a str, sort_prefix: &'a str },
-    Single(RowRef),
+    Single(&'a RowRef),
 }
 
 #[async_trait]
@@ -69,131 +119,24 @@ pub trait IStore {
     async fn row_fetch<'a>(&self, select: &Selector<'a>) -> Result<Vec<RowVal>, StorageError>;
     async fn row_rm<'a>(&self, select: &Selector<'a>) -> Result<(), StorageError>;
     async fn row_insert(&self, values: Vec<RowVal>) -> Result<(), StorageError>;
-    async fn row_poll(&self, value: RowRef) -> Result<RowVal, StorageError>;
+    async fn row_poll(&self, value: &RowRef) -> Result<RowVal, StorageError>;
 
     async fn blob_fetch(&self, blob_ref: &BlobRef) -> Result<BlobVal, StorageError>;
+    async fn blob_insert(&self, blob_val: &BlobVal) -> Result<BlobVal, StorageError>;
     async fn blob_copy(&self, src: &BlobRef, dst: &BlobRef) -> Result<BlobVal, StorageError>;
     async fn blob_list(&self, prefix: &str) -> Result<Vec<BlobRef>, StorageError>;
     async fn blob_rm(&self, blob_ref: &BlobRef) -> Result<(), StorageError>;
 }
 
-pub trait IBuilder {
-    fn build(&self) -> Box<dyn IStore>;
-}
-
-
-
-
-
-
-/*
-#[derive(Clone, Debug, PartialEq)]
-pub enum OrphanRowRef {
-    Garage(garage::GrgOrphanRowRef),
-    Memory(in_memory::MemOrphanRowRef),
-}
-
+#[derive(Clone,Debug,PartialEq,Eq,Hash)]
+pub struct UnicityBuffer(Vec<u8>);
 
+pub trait IBuilder: std::fmt::Debug {
+    fn build(&self) -> Result<Store, StorageError>;
 
-
-impl std::fmt::Display for StorageError {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        f.write_str("Storage Error: ")?;
-        match self {
-            Self::NotFound => f.write_str("Item not found"),
-            Self::Internal => f.write_str("An internal error occured"),
-            Self::IncompatibleOrphan => f.write_str("Incompatible orphan"),
-        }
-    }
+    /// Returns an opaque buffer that uniquely identifies this builder
+    fn unique(&self) -> UnicityBuffer;
 }
-impl std::error::Error for StorageError {}
 
-// Utils
-pub type AsyncResult<'a, T> = BoxFuture<'a, Result<T, StorageError>>;
-
-// ----- Builders
-pub trait IBuilders {
-    fn box_clone(&self) -> Builders;
-    fn row_store(&self) -> Result<RowStore, StorageError>;
-    fn blob_store(&self) -> Result<BlobStore, StorageError>;
-    fn url(&self) -> &str;
-}
-pub type Builders = Box<dyn IBuilders + Send + Sync>;
-impl Clone for Builders {
-    fn clone(&self) -> Self {
-        self.box_clone()
-    }
-}
-impl std::fmt::Debug for Builders {
-    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
-        f.write_str("aerogramme::storage::Builder")
-    }
-}
-impl PartialEq for Builders {
-    fn eq(&self, other: &Self) -> bool {
-        self.url() == other.url()
-    }
-}
-impl Eq for Builders {}
-impl Hash for Builders {
-    fn hash<H: Hasher>(&self, state: &mut H) {
-        self.url().hash(state);
-    }
-}
-
-// ------ Row 
-pub trait IRowStore
-{
-    fn row(&self, partition: &str, sort: &str) -> RowRef;
-    fn select(&self, selector: Selector) -> AsyncResult<Vec<RowValue>>;
-    fn rm(&self, selector: Selector) -> AsyncResult<()>;
-    fn from_orphan(&self, orphan: OrphanRowRef) -> Result<RowRef, StorageError>;
-}
-pub type RowStore = Box<dyn IRowStore + Sync + Send>;
-
-pub trait IRowRef: std::fmt::Debug
-{
-    fn to_orphan(&self) -> OrphanRowRef;
-    fn key(&self) -> (&str, &str);
-    fn set_value(&self, content: &[u8]) -> RowValue;
-    fn fetch(&self) -> AsyncResult<RowValue>;
-    fn rm(&self) -> AsyncResult<()>;
-    fn poll(&self) -> AsyncResult<RowValue>;
-}
-pub type RowRef<'a> = Box<dyn IRowRef + Send + Sync + 'a>;
-
-pub trait IRowValue: std::fmt::Debug
-{
-    fn to_ref(&self) -> RowRef;
-    fn content(&self) -> ConcurrentValues;
-    fn push(&self) -> AsyncResult<()>;
-}
-pub type RowValue = Box<dyn IRowValue + Send + Sync>;
-
-// ------- Blob 
-pub trait IBlobStore
-{
-    fn blob(&self, key: &str) -> BlobRef;
-    fn list(&self, prefix: &str) -> AsyncResult<Vec<BlobRef>>;
-}
-pub type BlobStore = Box<dyn IBlobStore + Send + Sync>;
-
-pub trait IBlobRef
-{
-    fn set_value(&self, content: Vec<u8>) -> BlobValue;
-    fn key(&self) -> &str;
-    fn fetch(&self) -> AsyncResult<BlobValue>;
-    fn copy(&self, dst: &BlobRef) -> AsyncResult<()>;
-    fn rm(&self) -> AsyncResult<()>;
-}
-pub type BlobRef = Box<dyn IBlobRef + Send + Sync>;
-
-pub trait IBlobValue {
-    fn to_ref(&self) -> BlobRef;
-    fn get_meta(&self, key: &str) -> Option<&[u8]>;
-    fn set_meta(&mut self, key: &str, val: &str);
-    fn content(&self) -> Option<&[u8]>;
-    fn push(&self) -> AsyncResult<()>;
-}
-pub type BlobValue = Box<dyn IBlobValue + Send + Sync>;
-*/
+pub type Builder = Arc<dyn IBuilder + Send + Sync>;
+pub type Store = Box<dyn IStore + Send + Sync>;