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