alibi_sqlx/store/
mod.rs

1//! `SQLx`-backed persistence implementation for built-in auth tables.
2
3mod accounts;
4mod api_key_usage_phases;
5mod api_keys;
6mod bundled_schema;
7mod device_codes;
8pub mod entities;
9mod invitations;
10mod jwks;
11mod members;
12pub(crate) mod migrator;
13mod oauth_clients;
14mod organization_roles;
15mod organizations;
16mod passkeys;
17mod sessions;
18mod teams;
19mod two_factor;
20mod users;
21mod verifications;
22mod wallets;
23
24#[doc(hidden)]
25pub mod __private_test_support {
26    pub mod bundled_schema {
27        pub use super::super::bundled_schema::BundledSchema;
28    }
29
30    pub mod migrator {
31        pub use super::super::migrator::run_migrations;
32    }
33}
34
35use crate::OrganizationModels;
36use crate::hooks::{DatabaseHooks, SqlxBackend, SqlxHookContext, current_request_hook_context};
37use crate::pool::{Engine, Exec, SqlxPool, SqlxTransaction};
38use crate::schema::{
39    AuthSchema, SqlxAccountModel, SqlxSessionModel, SqlxUserModel, SqlxVerificationModel,
40};
41use crate::sql::Sql;
42use alibi_core::config::AuthConfig;
43use alibi_core::error::AuthResult;
44use alibi_core::field_policy::FieldValues;
45use alibi_core::store::SchemaMigrator;
46use alibi_core::store::adapter::{AfterHook, AfterHookQueue};
47use alibi_core::store::{
48    AuthTransaction, BoxedTransactionValue, TransactionStore, TransactionWork,
49};
50use alibi_core::types::{AddTeamMemberResult, CreateJwk, Jwk, Member, Team};
51use alibi_core::user_validation::PreparedUserCreation;
52use alibi_core::verification::{
53    VerificationCreation, VerificationPublication, VerificationSnapshot,
54};
55use alibi_core::{
56    CreateAccount, CreateMember, CreatePasskey, CreateSession, CreateUser, CreateVerification,
57    Passkey,
58};
59use async_trait::async_trait;
60use chrono::{DateTime, Utc};
61use std::marker::PhantomData;
62use std::sync::Arc;
63
64#[derive(Clone)]
65pub struct SqlxStore<S: AuthSchema> {
66    config: Arc<AuthConfig>,
67    pool: SqlxPool,
68    hooks: Vec<Arc<dyn DatabaseHooks<S, SqlxBackend>>>,
69    organization_models: OrganizationModels,
70    _schema: PhantomData<S>,
71}
72
73impl<S: AuthSchema> std::fmt::Debug for SqlxStore<S> {
74    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
75        f.debug_struct("SqlxStore").finish_non_exhaustive()
76    }
77}
78
79impl<S: AuthSchema> SqlxStore<S> {
80    #[must_use]
81    pub fn new(config: impl Into<Arc<AuthConfig>>, pool: impl Into<SqlxPool>) -> Self {
82        Self {
83            config: config.into(),
84            pool: pool.into(),
85            hooks: Vec::new(),
86            organization_models: crate::OrganizationModels::default(),
87            _schema: PhantomData,
88        }
89    }
90
91    /// Bind the organization plugin to application-owned tables.
92    #[must_use]
93    pub fn with_organization_models(mut self, models: OrganizationModels) -> Self {
94        self.organization_models = models;
95        self
96    }
97
98    #[must_use]
99    pub fn with_hooks(mut self, hooks: Vec<Arc<dyn DatabaseHooks<S, SqlxBackend>>>) -> Self {
100        self.hooks = hooks;
101        self
102    }
103
104    #[must_use]
105    pub fn hook<H: DatabaseHooks<S, SqlxBackend> + 'static>(mut self, hook: H) -> Self {
106        self.hooks.push(Arc::new(hook));
107        self
108    }
109
110    #[must_use]
111    pub const fn pool(&self) -> &SqlxPool {
112        &self.pool
113    }
114
115    #[must_use]
116    pub const fn config(&self) -> &Arc<AuthConfig> {
117        &self.config
118    }
119
120    pub(crate) fn hooks(&self) -> &[Arc<dyn DatabaseHooks<S, SqlxBackend>>] {
121        &self.hooks
122    }
123
124    pub(crate) async fn generated_id(
125        &self,
126        exec: Exec<'_>,
127        model: &str,
128        table: &str,
129        column: &str,
130    ) -> AuthResult<Option<String>> {
131        let policy = &self.config.advanced.database;
132        if !policy.serial_ids() {
133            return policy.generated_id(model);
134        }
135        let [ddl, allocate] = alibi_core::config::serial_id_statements(
136            table,
137            column,
138            exec.engine() == Engine::Postgres,
139        );
140        _ = exec.execute(Sql::with(exec.engine(), &ddl)).await?;
141        let value = exec
142            .fetch_scalar::<i64>(Sql::with(exec.engine(), &allocate))
143            .await?
144            .ok_or_else(|| alibi_core::AuthError::internal("ID allocation returned no value"))?;
145        Ok(Some(value.to_string()))
146    }
147
148    async fn begin(&self, immediate: bool) -> AuthResult<SqlxTransaction> {
149        let mut transaction = self.pool.begin(immediate).await?;
150        transaction.config = Some(self.config.clone());
151        Ok(transaction)
152    }
153
154    pub(crate) fn exec(&self) -> Exec<'_> {
155        Exec::pool(&self.pool).with_config(self.config.as_ref())
156    }
157
158    pub(crate) fn hook_context<'a>(
159        &'a self,
160        tx: Option<&'a SqlxTransaction>,
161    ) -> SqlxHookContext<'a> {
162        SqlxHookContext {
163            config: self.config.as_ref(),
164            db: &self.pool,
165            tx,
166            request: current_request_hook_context(),
167        }
168    }
169
170    pub(crate) fn find_many_limit(&self) -> i64 {
171        // Mirrors the unsigned page binding of the reference adapter.
172        i64::try_from(self.config.advanced.database.default_find_many_limit).unwrap_or(i64::MAX)
173    }
174
175    /// Run work on a fresh transaction, committing on success. On failure the
176    /// transaction is dropped, which rolls it back, and the work's error returns.
177    pub(crate) async fn in_transaction<T>(
178        &self,
179        immediate: bool,
180        work: impl AsyncFnOnce(&SqlxTransaction) -> AuthResult<T>,
181    ) -> AuthResult<T> {
182        let transaction = self.begin(immediate).await?;
183        let value = work(&transaction).await?;
184        transaction.commit().await?;
185        Ok(value)
186    }
187}
188
189/// Installs the bundled auth schema in `better_auth_migrations`, the ledger
190/// the `SeaORM` adapter also records.
191#[async_trait]
192impl<S: AuthSchema> SchemaMigrator for SqlxStore<S> {
193    async fn migrate(&self) -> AuthResult<()> {
194        migrator::run_migrations_scoped(&self.pool, &self.config).await?;
195        if self.config.advanced.database.serial_ids() {
196            let [ddl, _] = alibi_core::config::serial_id_statements("", "", false);
197            let exec = self.exec();
198            _ = exec.execute(Sql::with(exec.engine(), &ddl)).await?;
199        }
200        if let Some(mapping) = &self.config.advanced.database.two_factor {
201            let commands = mapping.migration_statements()?;
202            self.in_transaction(false, async move |tx| {
203                let exec = Exec::tx(tx);
204                if migrator::has_table(exec, "twoFactor").await? {
205                    for command in commands {
206                        _ = exec.execute(Sql::with(exec.engine(), &command)).await?;
207                    }
208                }
209                Ok(())
210            })
211            .await?;
212        }
213        if let Some(mapping) = &self.config.advanced.database.sso_provider {
214            let commands = mapping.migration_statements()?;
215            self.in_transaction(false, async move |tx| {
216                let exec = Exec::tx(tx);
217                if migrator::has_table(exec, "ssoProvider").await? {
218                    for command in commands {
219                        _ = exec.execute(Sql::with(exec.engine(), &command)).await?;
220                    }
221                }
222                Ok(())
223            })
224            .await?;
225        }
226
227        let storage = &self.config.advanced.database;
228        alibi_core::store::validate_sso_fields(storage)?;
229        if !storage.sso_provider_fields.0.is_empty() {
230            let table = storage
231                .sso_provider
232                .as_ref()
233                .map_or("ssoProvider", |mapping| mapping.table_name.as_str());
234            self.in_transaction(false, async move |tx| {
235                let exec = Exec::tx(tx);
236                for (name, field) in &storage.sso_provider_fields.0 {
237                    let column = alibi_core::store::sso_field_column(name, field);
238                    if !migrator::has_column(exec, table, column).await? {
239                        let definition = alibi_core::store::sso_field_definition(
240                            name,
241                            field,
242                            exec.engine() == crate::pool::Engine::Postgres,
243                        )?;
244                        let mut ddl = Sql::with(exec.engine(), "ALTER TABLE ");
245                        ddl.ident(table);
246                        ddl.push(" ADD COLUMN ");
247                        ddl.push(&definition);
248                        _ = exec.execute(ddl).await?;
249                    }
250                }
251                Ok(())
252            })
253            .await?;
254        }
255
256        Ok(())
257    }
258}
259
260struct SqlxStoreTransaction<'a, S: AuthSchema> {
261    store: &'a SqlxStore<S>,
262    tx: &'a SqlxTransaction,
263    pending_after: AfterHookQueue<S>,
264}
265
266#[async_trait]
267impl<S> AuthTransaction<S> for SqlxStoreTransaction<'_, S>
268where
269    S: AuthSchema,
270    S::User: SqlxUserModel,
271    S::Account: SqlxAccountModel,
272    S::Session: SqlxSessionModel,
273    S::Verification: SqlxVerificationModel,
274{
275    async fn get_scim_connection_binding(
276        &self,
277        id: &str,
278    ) -> AuthResult<Option<alibi_core::store::ScimConnectionBinding>> {
279        self.store
280            .get_scim_connection_binding_with_exec(Exec::tx(self.tx), id)
281            .await
282    }
283    async fn scim_projection_candidates(
284        &self,
285        domain: &str,
286        cursor: Option<&str>,
287    ) -> AuthResult<Vec<alibi_core::store::ScimProjectionCandidate>> {
288        self.store
289            .scim_projection_candidates_with_exec(Exec::tx(self.tx), domain, cursor)
290            .await
291    }
292    async fn renew_scim_decommission_lease(
293        &self,
294        binding: &alibi_core::store::ScimConnectionBinding,
295        lease: &str,
296    ) -> AuthResult<bool> {
297        self.store
298            .renew_scim_decommission_lease_with_exec(Exec::tx(self.tx), binding, lease)
299            .await
300    }
301    async fn advance_scim_decommission(
302        &self,
303        binding: &alibi_core::store::ScimConnectionBinding,
304        lease: &str,
305        cursor: Option<&str>,
306        count: usize,
307        more: bool,
308    ) -> AuthResult<Option<alibi_core::store::ScimConnectionBinding>> {
309        self.store
310            .advance_scim_decommission_with_exec(
311                Exec::tx(self.tx),
312                binding,
313                lease,
314                cursor,
315                count,
316                more,
317            )
318            .await
319    }
320    async fn reconcile_scim_identity(
321        &self,
322        user_id: &str,
323    ) -> AuthResult<alibi_core::store::ScimIdentityState> {
324        self.store
325            .reconcile_scim_identity_with_exec(Exec::tx(self.tx), user_id)
326            .await
327    }
328    async fn delete_user_sessions(&self, user_id: &str) -> AuthResult<()> {
329        use crate::model::SqlxModel;
330        let sessions = self.get_user_sessions(user_id).await?;
331        let context = self.store.hook_context(Some(self.tx));
332        for session in &sessions {
333            for hook in self.store.hooks() {
334                if hook
335                    .before_delete_session(session, &context)
336                    .await?
337                    .is_cancelled()
338                {
339                    return Ok(());
340                }
341            }
342        }
343        let exec = Exec::tx(self.tx);
344        let table = <S::Session as SqlxModel>::TABLE;
345        let id = <S::Session as SqlxSessionModel>::parse_user_id(user_id)?;
346        let mut sql = Sql::with(exec.engine(), "DELETE FROM ");
347        sql.ident(table);
348        sql.push(" WHERE ");
349        sql.compare_model::<S::Session>(table, S::Session::user_id_column(), " = ", id);
350        _ = exec.execute(sql).await?;
351        for session in sessions {
352            self.pending_after
353                .push(AfterHook::SessionDeleted(session))
354                .await;
355        }
356        Ok(())
357    }
358    async fn get_user_sessions(&self, user_id: &str) -> AuthResult<Vec<S::Session>> {
359        use crate::model::SqlxModel;
360        let exec = Exec::tx(self.tx);
361        let table = <S::Session as SqlxModel>::TABLE;
362        let mut sql = crate::model::select_model::<S::Session>(exec);
363        sql.push(" WHERE ");
364        sql.compare_model::<S::Session>(
365            table,
366            S::Session::user_id_column(),
367            " = ",
368            S::Session::parse_user_id(user_id)?,
369        );
370        sql.push(" LIMIT ");
371        sql.bind(self.store.find_many_limit());
372        exec.fetch_all::<S::Session>(sql).await
373    }
374    async fn list_oauth_tokens_for_session(
375        &self,
376        kind: alibi_core::store::OAuthTokenKind,
377        session_id: &str,
378    ) -> AuthResult<Vec<alibi_core::store::OAuthToken>> {
379        self.store
380            .session_oauth_tokens(Exec::tx(self.tx), kind, session_id)
381            .await
382    }
383    async fn get_oauth_client(
384        &self,
385        client_id: &str,
386    ) -> AuthResult<Option<alibi_core::store::OAuthClient>> {
387        self.store
388            .oauth_client_with_exec(Exec::tx(self.tx), client_id)
389            .await
390    }
391    async fn queue_session_deletion_work(
392        &self,
393        session_id: &str,
394        works: Vec<Box<dyn alibi_core::store::SessionDeletionWork<S>>>,
395        context: Arc<alibi_core::AuthContext<S>>,
396    ) -> AuthResult<()> {
397        self.pending_after
398            .attach_session_retirement(session_id, works, context)
399            .await;
400        Ok(())
401    }
402    async fn check_scim_user_attributes(
403        &self,
404        id: &str,
405        connection: &str,
406        attributes: &alibi_core::store::ScimUserAttributes,
407    ) -> AuthResult<()> {
408        self.store
409            .assert_scim_user_keys_with_exec(
410                Exec::tx(self.tx),
411                connection,
412                &attributes.user_name_key,
413                attributes.external_id_key.as_deref(),
414                None,
415                Some(id),
416            )
417            .await
418    }
419    async fn lock_scim_user(
420        &self,
421        id: &str,
422        connection: &str,
423        domain: &str,
424        attributes: Option<&alibi_core::store::ScimUserAttributes>,
425        now: chrono::DateTime<chrono::Utc>,
426    ) -> AuthResult<alibi_core::store::ScimUserMutationContext> {
427        self.store
428            .lock_scim_user_with_exec(Exec::tx(self.tx), id, connection, domain, attributes, now)
429            .await
430    }
431    async fn replace_scim_user(
432        &self,
433        id: &str,
434        data: alibi_core::store::ReplaceScimUser,
435        roles: bool,
436    ) -> AuthResult<(
437        alibi_core::store::ScimUserRecord,
438        Vec<alibi_core::store::ScimProjectionPreparedUser>,
439    )> {
440        self.store
441            .replace_scim_user_with_exec(Exec::tx(self.tx), id, data, roles)
442            .await
443    }
444    async fn create_scim_user(
445        &self,
446        data: alibi_core::store::CreateScimUser,
447        roles: bool,
448    ) -> AuthResult<(
449        alibi_core::store::ScimUserRecord,
450        Vec<alibi_core::store::ScimProjectionPreparedUser>,
451    )> {
452        self.store
453            .create_scim_user_with_exec(Exec::tx(self.tx), data, roles)
454            .await
455    }
456    async fn delete_scim_user(
457        &self,
458        id: &str,
459        connection: &str,
460        domain: &str,
461        roles: bool,
462    ) -> AuthResult<(String, Vec<alibi_core::store::ScimProjectionPreparedUser>)> {
463        self.store
464            .delete_scim_user_with_exec(Exec::tx(self.tx), id, connection, domain, roles)
465            .await
466    }
467    async fn patch_scim_group(
468        &self,
469        id: &str,
470        connection: &str,
471        domain: &str,
472        patch: alibi_core::store::ScimGroupPatch,
473        roles: bool,
474    ) -> AuthResult<(
475        alibi_core::store::ScimGroupRecord,
476        Vec<alibi_core::store::ScimProjectionPreparedUser>,
477    )> {
478        self.store
479            .patch_scim_group_with_exec(Exec::tx(self.tx), id, connection, domain, patch, roles)
480            .await
481    }
482    async fn replace_scim_group(
483        &self,
484        id: &str,
485        data: alibi_core::store::ReplaceScimGroup,
486        roles: bool,
487    ) -> AuthResult<(
488        alibi_core::store::ScimGroupRecord,
489        Vec<alibi_core::store::ScimProjectionPreparedUser>,
490    )> {
491        self.store
492            .replace_scim_group_with_exec(Exec::tx(self.tx), id, data, roles)
493            .await
494    }
495    async fn delete_scim_group(
496        &self,
497        id: &str,
498        connection: &str,
499        domain: &str,
500        roles: bool,
501    ) -> AuthResult<Vec<alibi_core::store::ScimProjectionPreparedUser>> {
502        self.store
503            .delete_scim_group_with_exec(Exec::tx(self.tx), id, connection, domain, roles)
504            .await
505    }
506    async fn create_scim_group(
507        &self,
508        data: alibi_core::store::CreateScimGroup,
509        roles: bool,
510    ) -> AuthResult<(
511        alibi_core::store::ScimGroupRecord,
512        Vec<alibi_core::store::ScimProjectionPreparedUser>,
513    )> {
514        self.store
515            .create_scim_group_with_exec(Exec::tx(self.tx), data, roles)
516            .await
517    }
518    async fn fence_scim_connection(&self, id: &str, key: &str) -> AuthResult<()> {
519        self.store
520            .fence_scim_connection_with_exec(Exec::tx(self.tx), id, key)
521            .await
522    }
523    async fn prepare_scim_projection(
524        &self,
525        domain: &str,
526        ids: &[String],
527        roles: bool,
528    ) -> AuthResult<Vec<alibi_core::store::ScimProjectionPreparedUser>> {
529        self.store
530            .prepare_scim_projection_with_exec(Exec::tx(self.tx), domain, ids, roles)
531            .await
532    }
533    async fn create_scim_projection_grant(
534        &self,
535        data: alibi_core::store::CreateScimProjectionGrant,
536    ) -> AuthResult<()> {
537        self.store
538            .create_scim_projection_grant_with_exec(Exec::tx(self.tx), data)
539            .await
540    }
541    async fn remove_scim_projection_grants(
542        &self,
543        domain: &str,
544        user_id: &str,
545        ids: &[String],
546    ) -> AuthResult<()> {
547        self.store
548            .remove_scim_projection_grants_with_exec(Exec::tx(self.tx), domain, user_id, ids)
549            .await
550    }
551    async fn get_user_by_email(&self, email: &str) -> AuthResult<Option<S::User>> {
552        self.store
553            .get_user_by_email_with_connection(Exec::tx(self.tx), email)
554            .await
555    }
556    async fn get_account(
557        &self,
558        provider: &str,
559        account_id: &str,
560    ) -> AuthResult<Option<S::Account>> {
561        self.store
562            .get_account_with_connection(Exec::tx(self.tx), provider, account_id)
563            .await
564    }
565    async fn update_user(&self, id: &str, update: alibi_core::UpdateUser) -> AuthResult<S::User> {
566        let user = self
567            .store
568            .update_user_with_connection(Exec::tx(self.tx), Some(self.tx), id, update)
569            .await?;
570        self.pending_after
571            .push(AfterHook::UserUpdated(user.clone()))
572            .await;
573        Ok(user)
574    }
575    async fn update_account(
576        &self,
577        id: &str,
578        update: alibi_core::UpdateAccount,
579    ) -> AuthResult<S::Account> {
580        let account = self
581            .store
582            .update_account_with_connection(Exec::tx(self.tx), Some(self.tx), id, update)
583            .await?;
584        self.pending_after
585            .push(AfterHook::AccountUpdated(account.clone()))
586            .await;
587        Ok(account)
588    }
589    async fn lock_sso_provider(
590        &self,
591        id: &str,
592        provider_id: &str,
593    ) -> AuthResult<Option<alibi_core::store::SsoProvider>> {
594        self.store
595            .lock_sso_provider_with(Exec::tx(self.tx), id, provider_id)
596            .await
597    }
598    async fn update_sso_provider(
599        &self,
600        update: alibi_core::store::SsoProviderUpdate,
601    ) -> AuthResult<Option<alibi_core::store::SsoProvider>> {
602        self.store
603            .update_sso_provider_with(Exec::tx(self.tx), update)
604            .await
605    }
606    async fn delete_sso_provider_and_accounts(
607        &self,
608        id: &str,
609        provider_id: &str,
610    ) -> AuthResult<bool> {
611        self.store
612            .delete_sso_provider_with(Exec::tx(self.tx), id, provider_id)
613            .await
614    }
615    async fn list_jwks(&self) -> AuthResult<Vec<Jwk>> {
616        self.store.list_jwks_with(Exec::tx(self.tx)).await
617    }
618    async fn get_jwk_by_id(&self, id: &str) -> AuthResult<Option<Jwk>> {
619        self.store.get_jwk_with(Exec::tx(self.tx), id).await
620    }
621    async fn create_jwk(&self, data: CreateJwk) -> AuthResult<Jwk> {
622        self.store.create_jwk_with(Exec::tx(self.tx), data).await
623    }
624
625    async fn get_team(&self, organization_id: &str, team_id: &str) -> AuthResult<Option<Team>> {
626        self.store
627            .get_team_with_connection(Exec::tx(self.tx), Some(organization_id), team_id)
628            .await
629    }
630    async fn add_team_member(
631        &self,
632        team_id: &str,
633        user_id: &str,
634        maximum: Option<f64>,
635    ) -> AuthResult<AddTeamMemberResult> {
636        self.store
637            .add_team_member_in_tx(self.tx, team_id, user_id, maximum)
638            .await
639    }
640    async fn create_member(&self, member: CreateMember) -> AuthResult<Member> {
641        self.store
642            .create_member_with_connection(Exec::tx(self.tx), member)
643            .await
644    }
645    async fn update_session_active_team(
646        &self,
647        token: &str,
648        team_id: Option<&str>,
649    ) -> AuthResult<S::Session> {
650        self.store
651            .update_session_scope_with_connection(
652                Exec::tx(self.tx),
653                token,
654                sessions::SessionScope::Team(team_id),
655            )
656            .await
657    }
658    async fn update_session_active_organization(
659        &self,
660        token: &str,
661        organization_id: Option<&str>,
662    ) -> AuthResult<S::Session> {
663        self.store
664            .update_session_scope_with_connection(
665                Exec::tx(self.tx),
666                token,
667                sessions::SessionScope::Organization(organization_id),
668            )
669            .await
670    }
671    async fn provider_verification_output(
672        &self,
673        id: &str,
674    ) -> AuthResult<Option<serde_json::Value>> {
675        users::provider_verification_output::<S::User>(Exec::tx(self.tx), id).await
676    }
677    async fn get_user_by_id(&self, id: &str) -> AuthResult<Option<S::User>> {
678        users::find_user_by_id::<S::User>(Exec::tx(self.tx), id, users::Lock::None).await
679    }
680    async fn create_passkey(&self, data: CreatePasskey) -> AuthResult<Passkey> {
681        self.store
682            .create_passkey_with_connection(Exec::tx(self.tx), data)
683            .await
684    }
685    async fn create_user(&self, create_user: CreateUser) -> AuthResult<S::User> {
686        let user = self.store.create_user_in_tx(self.tx, create_user).await?;
687        self.pending_after
688            .push(AfterHook::UserCreated(user.clone()))
689            .await;
690        Ok(user)
691    }
692
693    async fn create_user_prepared(&self, prepared: PreparedUserCreation) -> AuthResult<S::User> {
694        let user = self
695            .store
696            .create_user_prepared_in_tx(self.tx, prepared)
697            .await?;
698        self.pending_after
699            .push(AfterHook::UserCreated(user.clone()))
700            .await;
701        Ok(user)
702    }
703
704    async fn create_account(&self, create_account: CreateAccount) -> AuthResult<S::Account> {
705        let account = self
706            .store
707            .create_account_in_tx(self.tx, create_account)
708            .await?;
709        self.pending_after
710            .push(AfterHook::AccountCreated(account.clone()))
711            .await;
712        Ok(account)
713    }
714
715    async fn prepare_secondary_session_update(
716        &self,
717        session: S::Session,
718        expires_at: Option<DateTime<Utc>>,
719        fields: FieldValues,
720    ) -> AuthResult<Option<(S::Session, FieldValues)>> {
721        self.store
722            .prepare_secondary_update_with_connection(
723                Exec::tx(self.tx),
724                Some(self.tx),
725                session,
726                expires_at,
727                fields,
728            )
729            .await
730    }
731    async fn complete_secondary_session_update(
732        &self,
733        session: S::Session,
734        expires_at: Option<DateTime<Utc>>,
735        fields: FieldValues,
736        persist: bool,
737    ) -> AuthResult<Option<S::Session>> {
738        use alibi_core::AuthSession;
739        let token = session.token().to_owned();
740        let result = self
741            .store
742            .complete_secondary_update_with_connection(
743                Exec::tx(self.tx),
744                Some(self.tx),
745                session,
746                expires_at,
747                fields,
748                persist,
749            )
750            .await?;
751        let event = result.as_ref().map_or_else(
752            || AfterHook::SessionUpdateMissing(token),
753            |model| AfterHook::SessionUpdated(model.clone()),
754        );
755        self.pending_after.push(event).await;
756        Ok(result)
757    }
758    async fn prepare_secondary_session_creation(
759        &self,
760        input: CreateSession,
761        persist: bool,
762    ) -> AuthResult<S::Session> {
763        let session = self
764            .store
765            .prepare_secondary_session_in_tx(self.tx, input, persist)
766            .await?;
767        self.pending_after
768            .push(AfterHook::SessionCreated(session.clone()))
769            .await;
770        Ok(session)
771    }
772    async fn create_session(&self, create_session: CreateSession) -> AuthResult<S::Session> {
773        let session = self
774            .store
775            .create_session_in_tx(self.tx, create_session)
776            .await?;
777        self.pending_after
778            .push(AfterHook::SessionCreated(session.clone()))
779            .await;
780        Ok(session)
781    }
782    async fn create_verification_record(
783        &self,
784        data: VerificationCreation,
785        publication: VerificationPublication,
786    ) -> AuthResult<Option<VerificationSnapshot>> {
787        let snapshot = self
788            .store
789            .create_verification_record_with_connection(
790                Exec::tx(self.tx),
791                Some(self.tx),
792                data,
793                publication,
794            )
795            .await?;
796        if let Some(snapshot) = &snapshot {
797            self.pending_after
798                .push(AfterHook::VerificationRecordCreated(snapshot.clone()))
799                .await;
800        }
801        Ok(snapshot)
802    }
803
804    async fn create_verification(
805        &self,
806        verification: CreateVerification,
807    ) -> AuthResult<S::Verification> {
808        let verification = self
809            .store
810            .create_verification_in_tx(self.tx, verification)
811            .await?;
812        self.pending_after
813            .push(AfterHook::VerificationCreated(verification.clone()))
814            .await;
815        Ok(verification)
816    }
817}
818
819#[async_trait]
820impl<S> TransactionStore<S> for SqlxStore<S>
821where
822    S: AuthSchema,
823    S::User: SqlxUserModel,
824    S::Account: SqlxAccountModel,
825    S::Session: SqlxSessionModel,
826    S::Verification: SqlxVerificationModel,
827{
828    async fn transaction_boxed(
829        &self,
830        work: Box<TransactionWork<S>>,
831    ) -> AuthResult<BoxedTransactionValue> {
832        self.transaction_boxed_with_policy(
833            work,
834            alibi_core::store::AfterCommitErrorPolicy::Propagate,
835        )
836        .await
837    }
838    async fn transaction_boxed_with_policy(
839        &self,
840        work: Box<TransactionWork<S>>,
841        policy: alibi_core::store::AfterCommitErrorPolicy,
842    ) -> AuthResult<BoxedTransactionValue> {
843        let tx = self.begin(false).await?;
844        let tx_store = SqlxStoreTransaction {
845            store: self,
846            tx: &tx,
847            pending_after: AfterHookQueue::new(),
848        };
849        let outcome = work(&tx_store).await;
850        let pending_after = tx_store.pending_after;
851        match outcome {
852            Ok(value) => {
853                tx.commit().await?;
854                // After hooks run only once the auth writes are durable, in
855                // operation order; rollback drops them.
856                pending_after
857                    .run_with_policy(self.hooks(), &self.hook_context(None), policy)
858                    .await?;
859                Ok(value)
860            }
861            Err(err) => {
862                tx.rollback().await?;
863                Err(err)
864            }
865        }
866    }
867}
868
869/// `FOR UPDATE` on PostgreSQL; SQLite relies on its `BEGIN IMMEDIATE` writer lock.
870pub(crate) fn lock_exclusive(sql: &mut Sql) {
871    if sql.engine() == Engine::Postgres {
872        sql.push(" FOR UPDATE");
873    }
874}
875
876/// `FOR SHARE` on PostgreSQL; SQLite relies on its `BEGIN IMMEDIATE` writer lock.
877pub(crate) fn lock_shared(sql: &mut Sql) {
878    if sql.engine() == Engine::Postgres {
879        sql.push(" FOR SHARE");
880    }
881}
882
883/// Bind raw numeric `LIMIT`/`OFFSET` values; the database validates them.
884pub(crate) fn bind_page(sql: &mut crate::sql::Sql, limit: Option<f64>, offset: Option<f64>) {
885    for (clause, number) in [(" LIMIT ", limit), (" OFFSET ", offset)] {
886        if let Some(number) = number {
887            sql.push(clause);
888            if sql.engine() == Engine::Postgres {
889                // node-postgres sends Number parameters as text. Binding float8
890                // instead lets PostgreSQL round fractional pages to bigint.
891                sql.bind(ryu_js::Buffer::new().format(number).to_owned());
892                sql.push("::bigint");
893            } else {
894                sql.bind(number);
895            }
896        }
897    }
898}
899
900mod sso;
901
902mod scim;
903mod scim_decommission;
904mod scim_groups;
905mod scim_projection;
906mod scim_user_create;
907mod scim_user_replace;
908mod scim_users;
alibi_sqlx/store/mod.rs

alibi_sqlx/store/
mod.rs
