alibi_sqlx/store/
accounts.rs

1use super::SqlxStore;
2use crate::error::record_not_updated;
3use crate::model::{self, SqlxModel};
4use crate::pool::{Exec, SqlxTransaction};
5use crate::schema::{AuthSchema, SqlxAccountModel};
6use crate::sql::Sql;
7use alibi_core::error::AuthResult;
8use alibi_core::store::AccountStore;
9use alibi_core::store::adapter::cancelled_by_hook;
10use alibi_core::types::{CreateAccount, UpdateAccount};
11use async_trait::async_trait;
12use chrono::Utc;
13
14impl<S> SqlxStore<S>
15where
16    S: AuthSchema,
17    S::Account: SqlxAccountModel,
18{
19    async fn create_account_with_connection(
20        &self,
21        exec: Exec<'_>,
22        tx: Option<&SqlxTransaction>,
23        mut create_account: CreateAccount,
24    ) -> AuthResult<S::Account> {
25        let hook_context = self.hook_context(tx);
26        for hook in self.hooks() {
27            if hook
28                .before_create_account(&mut create_account, &hook_context)
29                .await?
30                .is_cancelled()
31            {
32                return Err(cancelled_by_hook("account creation"));
33            }
34        }
35        let now = Utc::now();
36        let mut fields = std::mem::take(&mut create_account.additional_fields);
37        fields.apply_adapter_transforms_async().await?;
38        let mut active = S::Account::new_active(None, create_account, now);
39        let backend = exec.engine();
40        for (column, value) in S::Account::additional_field_bindings(&fields, backend)? {
41            let value = crate::additional_fields::prepare_value(
42                exec,
43                <S::Account as SqlxModel>::column_kind(column),
44                value,
45            )
46            .await?;
47            S::Account::set_additional_field(&mut active, column, value, backend)?;
48        }
49        let account = model::insert::<S::Account>(exec, &active).await?;
50        if tx.is_none() {
51            for hook in self.hooks() {
52                hook.after_create_account(&account, &hook_context).await?;
53            }
54        }
55        Ok(account)
56    }
57
58    pub(crate) async fn create_account_in_tx(
59        &self,
60        tx: &SqlxTransaction,
61        create_account: CreateAccount,
62    ) -> AuthResult<S::Account> {
63        self.create_account_with_connection(Exec::Tx(tx), Some(tx), create_account)
64            .await
65    }
66
67    async fn find_account_by_id(&self, id: &str) -> AuthResult<Option<S::Account>> {
68        let account_id = <S::Account as SqlxAccountModel>::parse_id(id)?;
69        let mut sql = model::select_model::<S::Account>(self.exec());
70        sql.push(" WHERE ");
71        sql.compare_model::<S::Account>(
72            <S::Account as SqlxModel>::TABLE,
73            S::Account::id_column(),
74            " = ",
75            account_id,
76        );
77        sql.push(" LIMIT 1");
78        self.exec().fetch_optional(sql).await
79    }
80}
81
82#[async_trait]
83impl<S> AccountStore<S> for SqlxStore<S>
84where
85    S: AuthSchema + Send + Sync,
86    S::Account: SqlxAccountModel,
87{
88    async fn provider_token_text(&self, value: &serde_json::Value) -> AuthResult<Option<String>> {
89        if value.is_object() || value.is_array() {
90            return Err(alibi_core::AuthError::internal(
91                "Unsupported provider token SQL parameter",
92            ));
93        }
94        let value = crate::additional_fields::raw_value(&alibi_core::utils::json::JsValue::from(
95            value.clone(),
96        ))?;
97        crate::additional_fields::prepare_string_value(self.exec(), value).await
98    }
99
100    async fn create_account(&self, create_account: CreateAccount) -> AuthResult<S::Account> {
101        self.create_account_with_connection(self.exec(), None, create_account)
102            .await
103    }
104
105    async fn get_account(
106        &self,
107        provider: &str,
108        provider_account_id: &str,
109    ) -> AuthResult<Option<S::Account>> {
110        let table = <S::Account as SqlxModel>::TABLE;
111        let mut sql = model::select_model::<S::Account>(self.exec());
112        sql.push(" WHERE ");
113        sql.compare_model::<S::Account>(table, S::Account::provider_id_column(), " = ", provider);
114        sql.push(" AND ");
115        sql.compare_model::<S::Account>(
116            table,
117            S::Account::account_id_column(),
118            " = ",
119            provider_account_id,
120        );
121        sql.push(" LIMIT ");
122        sql.bind(2_i64);
123        let mut accounts: Vec<S::Account> = self.exec().fetch_all(sql).await?;
124        if accounts.len() > 1 {
125            return Err(alibi_core::AuthError::Database(
126                alibi_core::DatabaseError::AmbiguousAccount {
127                    provider: provider.to_owned(),
128                },
129            ));
130        }
131        Ok(accounts.pop())
132    }
133
134    async fn get_user_accounts(&self, user_id: &str) -> AuthResult<Vec<S::Account>> {
135        let user_id = <S::Account as SqlxAccountModel>::parse_user_id(user_id)?;
136        let mut sql = model::select_model::<S::Account>(self.exec());
137        sql.push(" WHERE ");
138        sql.compare_model::<S::Account>(
139            <S::Account as SqlxModel>::TABLE,
140            S::Account::user_id_column(),
141            " = ",
142            user_id,
143        );
144        self.exec().fetch_all(sql).await
145    }
146
147    async fn update_account(&self, id: &str, mut update: UpdateAccount) -> AuthResult<S::Account> {
148        drop(<S::Account as SqlxAccountModel>::parse_id(id)?);
149        let hook_context = self.hook_context(None);
150        for hook in self.hooks() {
151            if hook
152                .before_update_account(id, &mut update, &hook_context)
153                .await?
154                .is_cancelled()
155            {
156                return Err(cancelled_by_hook("account update"));
157            }
158        }
159        let Some(model) = self.find_account_by_id(id).await? else {
160            return Err(alibi_core::error::AuthError::not_found("Account not found"));
161        };
162
163        let mut active = model.into_active();
164        let mut fields = std::mem::take(&mut update.additional_fields);
165        fields.apply_adapter_transforms_async().await?;
166        S::Account::apply_update(&mut active, update, Utc::now());
167        let backend = self.exec().engine();
168        for (column, value) in S::Account::additional_field_bindings(&fields, backend)? {
169            let value = crate::additional_fields::prepare_value(
170                self.exec(),
171                <S::Account as SqlxModel>::column_kind(column),
172                value,
173            )
174            .await?;
175            S::Account::set_additional_field(&mut active, column, value, backend)?;
176        }
177
178        let account = model::update::<S::Account>(self.exec(), &active)
179            .await?
180            .ok_or_else(record_not_updated)?;
181        for hook in self.hooks() {
182            hook.after_update_account(&account, &hook_context).await?;
183        }
184        Ok(account)
185    }
186
187    async fn delete_account(&self, id: &str) -> AuthResult<()> {
188        let account_id = <S::Account as SqlxAccountModel>::parse_id(id)?;
189        let Some(account_model) = self.find_account_by_id(id).await? else {
190            return Err(alibi_core::error::AuthError::not_found("Account not found"));
191        };
192        let hook_context = self.hook_context(None);
193        for hook in self.hooks() {
194            if hook
195                .before_delete_account(&account_model, &hook_context)
196                .await?
197                .is_cancelled()
198            {
199                return Err(cancelled_by_hook("account deletion"));
200            }
201        }
202        let table = <S::Account as SqlxModel>::TABLE;
203        let mut sql = Sql::with(self.exec().engine(), "DELETE FROM ");
204        sql.ident(table);
205        sql.push(" WHERE ");
206        sql.compare_model::<S::Account>(table, S::Account::id_column(), " = ", account_id);
207        _ = self.exec().execute(sql).await?;
208        for hook in self.hooks() {
209            hook.after_delete_account(&account_model, &hook_context)
210                .await?;
211        }
212        Ok(())
213    }
214}
alibi_sqlx/store/accounts.rs

alibi_sqlx/store/
accounts.rs
