origin_accounts/
service.rs

1use crate::store::AccountStore;
2use origin_auth::{TokenSet, TokenStore};
3use origin_domain::{Account, AccountId, AccountStatus, AppError, Clock, ConnectorId, Result};
4use origin_events::{AccountExpired, EventBus, PlatformEvent};
5use origin_storage::{Storage, namespace};
6use std::sync::Arc;
7
8/// Connecting, listing and disconnecting accounts.
9///
10/// Keeps the two halves of an account consistent: the record in storage and the
11/// credentials in the credential store.
12#[derive(Debug, Clone)]
13pub struct AccountService {
14    accounts: AccountStore,
15    tokens: TokenStore,
16    events: EventBus,
17    storage: Arc<dyn Storage>,
18    clock: Arc<dyn Clock>,
19}
20
21impl AccountService {
22    pub fn new(
23        accounts: AccountStore,
24        tokens: TokenStore,
25        events: EventBus,
26        storage: Arc<dyn Storage>,
27        clock: Arc<dyn Clock>,
28    ) -> Self {
29        Self {
30            accounts,
31            tokens,
32            events,
33            storage,
34            clock,
35        }
36    }
37
38    /// Register a freshly authorized account.
39    ///
40    /// Credentials are written first so no visible account can exist without them. If
41    /// persisting the account fails, the credential write is rolled back.
42    pub async fn connect(
43        &self,
44        connector: &ConnectorId,
45        display_name: impl Into<String>,
46        tokens: &TokenSet,
47    ) -> Result<Account> {
48        let account = Account {
49            id: AccountId::generate(),
50            connector: connector.clone(),
51            display_name: display_name.into(),
52            status: AccountStatus::Active,
53            connected_at: self.clock.now(),
54        };
55
56        self.tokens.save(connector, &account.id, tokens).await?;
57        if let Err(save_error) = self.accounts.save(&account).await {
58            if let Err(cleanup_error) = self.tokens.delete(connector, &account.id).await {
59                return Err(AppError::storage(format!(
60                    "{save_error}; rolling back credentials also failed: {cleanup_error}"
61                )));
62            }
63            return Err(save_error);
64        }
65
66        tracing::info!(
67            %connector,
68            account = %account.id,
69            "account connected"
70        );
71        Ok(account)
72    }
73
74    /// Register an account from a pasted token instead of an OAuth flow (B7/C1).
75    ///
76    /// The caller is expected to have verified the token against the service first —
77    /// this method only records it. Same credential-first ordering and rollback as
78    /// [`AccountService::connect`].
79    pub async fn connect_with_token(
80        &self,
81        connector: &ConnectorId,
82        display_name: impl Into<String>,
83        token: impl Into<String>,
84        scopes: Vec<String>,
85    ) -> Result<Account> {
86        let tokens = TokenSet::personal_access_token(token, scopes);
87        self.connect(connector, display_name, &tokens).await
88    }
89
90    pub async fn list(&self) -> Result<Vec<Account>> {
91        self.accounts.list().await
92    }
93    pub async fn list_for(&self, connector: &ConnectorId) -> Result<Vec<Account>> {
94        self.accounts.list_for(connector).await
95    }
96
97    pub async fn get(&self, account: &AccountId) -> Result<Account> {
98        self.accounts
99            .get(account)
100            .await?
101            .ok_or_else(|| AppError::validation(format!("unknown account {account}")))
102    }
103
104    /// Remove an account, its credentials and everything stored under it.
105    ///
106    /// The namespace convention (ADR-0019) is what makes the last part mechanical: all
107    /// account data lives under `acct.<connector>.<account>.`, so no module has to
108    /// register which namespaces it wrote.
109    pub async fn disconnect(&self, account: &AccountId) -> Result<()> {
110        let record = self.get(account).await?;
111
112        // Credentials go first: if a later step fails, the worst case is a
113        // disconnected account still listed, not a live token nobody can see.
114        self.tokens.delete(&record.connector, account).await?;
115
116        let prefix = namespace::account_prefix(&record.connector, account);
117        let removed = self.storage.clear_prefix(&prefix).await?;
118
119        self.accounts.remove(account).await?;
120
121        tracing::info!(
122            connector = %record.connector,
123            %account,
124            records_removed = removed,
125            "account disconnected"
126        );
127        Ok(())
128    }
129
130    /// Mark an account as needing re-authentication.
131    ///
132    /// Called when a connector reports that credentials are no longer valid. The
133    /// credentials are kept: a provider outage can look like an expired token, and
134    /// deleting them would force an unnecessary reconnect.
135    pub async fn mark_expired(&self, account: &AccountId) -> Result<()> {
136        let mut record = self.get(account).await?;
137
138        if record.status == AccountStatus::Expired {
139            return Ok(());
140        }
141
142        record.status = AccountStatus::Expired;
143        self.accounts.save(&record).await?;
144
145        let _ = self
146            .events
147            .publish(PlatformEvent::AccountExpired(AccountExpired {
148                account: account.clone(),
149                connector: record.connector.clone(),
150            }));
151
152        tracing::warn!(connector = %record.connector, %account, "account marked as expired");
153        Ok(())
154    }
155
156    /// Mark an account usable again after a successful verification.
157    pub async fn mark_active(&self, account: &AccountId) -> Result<()> {
158        let mut record = self.get(account).await?;
159
160        if record.status == AccountStatus::Active {
161            return Ok(());
162        }
163
164        record.status = AccountStatus::Active;
165        self.accounts.save(&record).await
166    }
167}
origin_accounts/service.rs

origin_accounts/
service.rs
