1 files changed, 33 insertions, 22 deletions

diff --git a/src/mail/user.rs b/src/mail/user.rs
index 5523c2a..da0d509 100644
--- a/src/mail/user.rs
+++ b/src/mail/user.rs
@@ -2,18 +2,18 @@ use std::collections::{BTreeMap, HashMap};
 use std::sync::{Arc, Weak};
 
 use anyhow::{anyhow, bail, Result};
-use k2v_client::{CausalityToken, K2vClient, K2vValue};
 use lazy_static::lazy_static;
 use serde::{Deserialize, Serialize};
 use tokio::sync::watch;
 
 use crate::cryptoblob::{open_deserialize, seal_serialize};
-use crate::login::{Credentials, StorageCredentials};
+use crate::login::Credentials;
 use crate::mail::incoming::incoming_mail_watch_process;
 use crate::mail::mailbox::Mailbox;
 use crate::mail::uidindex::ImapUidvalidity;
 use crate::mail::unique_ident::{gen_ident, UniqueIdent};
-use crate::time::now_msec;
+use crate::storage;
+use crate::timestamp::now_msec;
 
 pub const MAILBOX_HIERARCHY_DELIMITER: char = '.';
 
@@ -33,7 +33,7 @@ const MAILBOX_LIST_SK: &str = "list";
 pub struct User {
     pub username: String,
     pub creds: Credentials,
-    pub k2v: K2vClient,
+    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();
@@ -165,6 +165,7 @@ impl User {
                     list.rename_mailbox(name, &nnew)?;
                 }
             }
+
             self.save_mailbox_list(&list, ct).await?;
         }
         Ok(())
@@ -173,14 +174,14 @@ impl User {
     // ---- Internal user & mailbox management ----
 
     async fn open(username: String, creds: Credentials) -> Result<Arc<Self>> {
-        let k2v = creds.k2v_client()?;
+        let storage = creds.storage.build().await?;
 
         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()),
         });
@@ -223,32 +224,42 @@ impl User {
 
     // ---- Mailbox list management ----
 
-    async fn load_mailbox_list(&self) -> Result<(MailboxList, Option<CausalityToken>)> {
-        let (mut list, ct) = match self.k2v.read_item(MAILBOX_LIST_PK, MAILBOX_LIST_SK).await {
-            Err(k2v_client::Error::NotFound) => (MailboxList::new(), None),
+    async fn load_mailbox_list(&self) -> Result<(MailboxList, Option<storage::RowRef>)> {
+        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(cv) => {
+            Ok(rv) => {
                 let mut list = MailboxList::new();
-                for v in cv.value {
-                    if let K2vValue::Value(vbytes) = v {
+                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(cv.causality))
+                (list, Some(row_ref))
             }
         };
 
-        self.ensure_inbox_exists(&mut list, &ct).await?;
+        self.ensure_inbox_exists(&mut list, &row).await?;
 
-        Ok((list, ct))
+        Ok((list, row))
     }
 
     async fn ensure_inbox_exists(
         &self,
         list: &mut MailboxList,
-        ct: &Option<CausalityToken>,
+        ct: &Option<storage::RowRef>,
     ) -> Result<bool> {
         // If INBOX doesn't exist, create a new mailbox with that name
         // and save new mailbox list.
@@ -277,12 +288,12 @@ impl User {
     async fn save_mailbox_list(
         &self,
         list: &MailboxList,
-        ct: Option<CausalityToken>,
+        ct: Option<storage::RowRef>,
     ) -> Result<()> {
         let list_blob = seal_serialize(list, &self.creds.keys.master)?;
-        self.k2v
-            .insert_item(MAILBOX_LIST_PK, MAILBOX_LIST_SK, list_blob, ct)
-            .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(())
     }
 }
@@ -455,6 +466,6 @@ enum CreatedMailbox {
 // ---- User cache ----
 
 lazy_static! {
-    static ref USER_CACHE: std::sync::Mutex<HashMap<(String, StorageCredentials), Weak<User>>> =
+    static ref USER_CACHE: std::sync::Mutex<HashMap<(String, storage::UnicityBuffer), Weak<User>>> =
         std::sync::Mutex::new(HashMap::new());
 }