origin_accounts/
store.rs

1use crate::ACCOUNTS_NAMESPACE;
2use origin_domain::{Account, AccountId, AppError, Clock, ConnectorId, Result};
3use origin_storage::{Record, Storage, StorageKey};
4use std::sync::Arc;
5
6/// Persistence for the account list.
7#[derive(Debug, Clone)]
8pub struct AccountStore {
9    storage: Arc<dyn Storage>,
10    clock: Arc<dyn Clock>,
11}
12
13impl AccountStore {
14    pub fn new(storage: Arc<dyn Storage>, clock: Arc<dyn Clock>) -> Self {
15        Self { storage, clock }
16    }
17
18    fn key(account: &AccountId) -> StorageKey {
19        StorageKey::new(ACCOUNTS_NAMESPACE, account.as_str())
20    }
21
22    pub async fn get(&self, account: &AccountId) -> Result<Option<Account>> {
23        let Some(record) = self.storage.get(&Self::key(account)).await? else {
24            return Ok(None);
25        };
26
27        serde_json::from_str(&record.value)
28            .map(Some)
29            .map_err(|error| AppError::storage(format!("account {account} is unreadable: {error}")))
30    }
31
32    pub async fn save(&self, account: &Account) -> Result<()> {
33        let encoded = serde_json::to_string(account)
34            .map_err(|error| AppError::storage(format!("cannot encode account: {error}")))?;
35
36        // Accounts are user data, not cache: stored without an expiry so no cache sweep
37        // can remove them.
38        self.storage
39            .put(
40                &Self::key(&account.id),
41                Record::new(encoded, self.clock.now()),
42            )
43            .await
44    }
45
46    pub async fn remove(&self, account: &AccountId) -> Result<()> {
47        self.storage.delete(&Self::key(account)).await
48    }
49
50    /// Every account, across all connectors.
51    pub async fn list(&self) -> Result<Vec<Account>> {
52        let mut accounts = Vec::new();
53
54        for key in self.storage.keys(ACCOUNTS_NAMESPACE).await? {
55            let id = AccountId::new(key.key());
56            match self.get(&id).await {
57                Ok(Some(account)) => accounts.push(account),
58                Ok(None) => {}
59                // One unreadable record must not hide every other account; the user
60                // would see an empty list and reconnect everything.
61                Err(error) => tracing::warn!(%id, %error, "skipping unreadable account record"),
62            }
63        }
64
65        accounts.sort_by(|a, b| a.display_name.cmp(&b.display_name));
66        Ok(accounts)
67    }
68
69    /// Accounts belonging to one connector.
70    pub async fn list_for(&self, connector: &ConnectorId) -> Result<Vec<Account>> {
71        Ok(self
72            .list()
73            .await?
74            .into_iter()
75            .filter(|account| &account.connector == connector)
76            .collect())
77    }
78}
origin_accounts/store.rs

origin_accounts/
store.rs
