alibi-seaorm 0.11.0

//! SeaORM-backed persistence implementation for built-in auth tables.

mod accounts;
mod api_key_usage_phases;
mod api_keys;
mod bundled_schema;
mod device_codes;
pub mod entities;
mod invitations;
mod jwks;
mod members;
mod migrator;
mod oauth_clients;
mod organization_roles;
mod organizations;
mod passkeys;
mod scoped_connection;
mod sessions;
mod teams;
mod two_factor;
mod users;
mod verifications;
mod wallets;

#[doc(hidden)]
pub mod __private_test_support {
    pub mod bundled_schema {
        pub use super::super::bundled_schema::BundledSchema;
    }

    pub mod migrator {
        pub use super::super::migrator::{AuthMigrator, run_migrations};
    }
}

use crate::hooks::{DatabaseHooks, SeaOrmBackend, SeaOrmHookContext, current_request_hook_context};
use crate::schema::{
    AuthSchema, SeaOrmAccountModel, SeaOrmSessionModel, SeaOrmUserModel, SeaOrmVerificationModel,
};
use alibi_core::config::AuthConfig;
use alibi_core::error::{AuthError, AuthResult, DatabaseError};
use alibi_core::field_policy::FieldValues;
use alibi_core::store::SchemaMigrator;
use alibi_core::store::adapter::{AfterHook, AfterHookQueue};
use alibi_core::store::{
    AuthTransaction, BoxedTransactionValue, TransactionStore, TransactionWork,
};
use alibi_core::types::{AddTeamMemberResult, CreateJwk, Jwk, Member, Team};
use alibi_core::user_validation::PreparedUserCreation;
use alibi_core::verification::{
    VerificationCreation, VerificationPublication, VerificationSnapshot,
};
use alibi_core::{
    CreateAccount, CreateMember, CreatePasskey, CreateSession, CreateUser, CreateVerification,
    Passkey,
};
use async_trait::async_trait;
use chrono::{DateTime, Utc};
use scoped_connection::{ScopedConnection, ScopedTransaction};
use sea_orm::{DatabaseConnection, DbErr, EntityTrait, SqlErr, TransactionTrait};
use std::marker::PhantomData;
use std::sync::Arc;

#[derive(Clone)]
pub struct SeaOrmStore<S: AuthSchema> {
    config: Arc<AuthConfig>,
    db: ScopedConnection,
    hooks: Vec<Arc<dyn DatabaseHooks<S, SeaOrmBackend>>>,
    _schema: PhantomData<S>,
}

impl<S: AuthSchema> std::fmt::Debug for SeaOrmStore<S> {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        f.debug_struct("SeaOrmStore").finish_non_exhaustive()
    }
}

impl<S: AuthSchema> SeaOrmStore<S> {
    pub(crate) async fn generated_id<C: sea_orm::ConnectionTrait>(
        &self,
        db: &C,
        model: &str,
        table: &str,
        column: &str,
    ) -> AuthResult<Option<String>> {
        let policy = &self.config.advanced.database;
        if !policy.serial_ids() {
            return policy.generated_id(model);
        }
        let backend = db.get_database_backend();
        let [ddl, allocate] = alibi_core::config::serial_id_statements(
            table,
            column,
            backend == sea_orm::DbBackend::Postgres,
        );
        _ = db.execute_unprepared(&ddl).await.map_err(map_db_err)?;
        let row = db
            .query_one_raw(sea_orm::Statement::from_string(backend, allocate))
            .await
            .map_err(map_db_err)?
            .ok_or_else(|| AuthError::internal("ID allocation returned no value"))?;
        let value: i64 = row.try_get("", "value").map_err(map_db_err)?;
        Ok(Some(value.to_string()))
    }

    /// `generated_id` for a model entity, naming its table and ID column.
    pub(crate) async fn generated_entity_id<E: EntityTrait, C: sea_orm::ConnectionTrait>(
        &self,
        db: &C,
        model: &str,
        id_column: impl sea_orm::Iden,
    ) -> AuthResult<Option<String>> {
        let table = E::default().table_name().to_owned();
        self.generated_id(db, model, &table, &id_column.to_string())
            .await
    }

    #[must_use]
    pub fn new(config: impl Into<Arc<AuthConfig>>, db: DatabaseConnection) -> Self {
        let config = config.into();
        let schema = config.advanced.database.schema_name.clone();
        let factor = config.advanced.database.two_factor.clone();
        let sso_provider = config.advanced.database.sso_provider.clone();
        Self {
            config,
            db: ScopedConnection {
                inner: db,
                schema,
                factor,
                sso_provider,
            },
            hooks: Vec::new(),
            _schema: PhantomData,
        }
    }

    #[must_use]
    pub fn with_hooks(mut self, hooks: Vec<Arc<dyn DatabaseHooks<S, SeaOrmBackend>>>) -> Self {
        self.hooks = hooks;
        self
    }

    #[must_use]
    pub fn hook<H: DatabaseHooks<S, SeaOrmBackend> + 'static>(mut self, hook: H) -> Self {
        self.hooks.push(Arc::new(hook));
        self
    }

    #[must_use]
    pub(super) const fn scoped_connection(&self) -> &ScopedConnection {
        &self.db
    }

    pub const fn connection(&self) -> &DatabaseConnection {
        &self.db.inner
    }

    #[must_use]
    pub const fn config(&self) -> &Arc<AuthConfig> {
        &self.config
    }

    pub(crate) fn hooks(&self) -> &[Arc<dyn DatabaseHooks<S, SeaOrmBackend>>] {
        &self.hooks
    }

    pub(crate) fn hook_context<'a>(
        &'a self,
        tx: Option<&'a ScopedTransaction>,
    ) -> SeaOrmHookContext<'a> {
        SeaOrmHookContext {
            config: self.config.as_ref(),
            db: &self.db.inner,
            tx: tx.map(|tx| &tx.inner),
            request: current_request_hook_context(),
        }
    }
}

impl<S: AuthSchema> SeaOrmStore<S> {
    async fn migrate_bundled_schema(&self) -> AuthResult<()> {
        if self.db.get_database_backend() == sea_orm::DbBackend::Postgres
            && let Some(schema) = &self.db.schema
        {
            use sea_orm::{ConnectionTrait, Statement};
            use sea_orm_migration::MigratorTrait;
            let quoted = format!("\"{}\"", schema.replace('"', "\"\""));
            _ = self
                .db
                .inner
                .execute_unprepared(&format!("CREATE SCHEMA IF NOT EXISTS {quoted}"))
                .await
                .map_err(map_db_err)?;
            // Only the migration transaction has a local DDL namespace.
            // Runtime connections and statements retain statement qualification.
            let transaction = self.db.inner.begin().await.map_err(map_db_err)?;
            _ = transaction
                .execute_raw(Statement::from_string(
                    sea_orm::DbBackend::Postgres,
                    format!("SET LOCAL search_path TO {quoted}"),
                ))
                .await
                .map_err(map_db_err)?;
            migrator::AuthMigrator::up(&transaction, None)
                .await
                .map_err(map_db_err)?;
            return transaction.commit().await.map_err(map_db_err);
        }
        migrator::run_migrations(&self.db.inner)
            .await
            .map_err(map_db_err)
    }
}

#[async_trait]
impl<S: AuthSchema> SchemaMigrator for SeaOrmStore<S> {
    async fn migrate(&self) -> AuthResult<()> {
        self.migrate_bundled_schema().await?;
        if self.config.advanced.database.serial_ids() {
            use sea_orm::ConnectionTrait;
            let [ddl, _] = alibi_core::config::serial_id_statements("", "", false);
            _ = self.db.execute_unprepared(&ddl).await.map_err(map_db_err)?;
        }
        if let Some(mapping) = &self.config.advanced.database.two_factor {
            use sea_orm::ConnectionTrait;
            let commands = mapping.migration_statements()?;
            let transaction = self.db.begin().await.map_err(map_db_err)?;
            if transaction
                .has_table("twoFactor")
                .await
                .map_err(map_db_err)?
            {
                for command in commands {
                    _ = transaction
                        .execute_unprepared(&command)
                        .await
                        .map_err(map_db_err)?;
                }
            }
            transaction.commit().await.map_err(map_db_err)?;
        }
        if let Some(mapping) = &self.config.advanced.database.sso_provider {
            use sea_orm::ConnectionTrait;
            let commands = mapping.migration_statements()?;
            let transaction = self.db.begin().await.map_err(map_db_err)?;
            if transaction
                .has_table("ssoProvider")
                .await
                .map_err(map_db_err)?
            {
                for command in commands {
                    _ = transaction
                        .execute_unprepared(&command)
                        .await
                        .map_err(map_db_err)?;
                }
            }
            transaction.commit().await.map_err(map_db_err)?;
        }

        let storage = &self.config.advanced.database;
        alibi_core::store::validate_sso_fields(storage)?;
        if !storage.sso_provider_fields.0.is_empty() {
            use sea_orm::ConnectionTrait;
            let table = storage
                .sso_provider
                .as_ref()
                .map_or("ssoProvider", |mapping| mapping.table_name.as_str());
            let transaction = self.db.begin().await.map_err(map_db_err)?;
            for (name, field) in &storage.sso_provider_fields.0 {
                let column = alibi_core::store::sso_field_column(name, field);
                if !transaction
                    .has_column(table, column)
                    .await
                    .map_err(map_db_err)?
                {
                    let definition = alibi_core::store::sso_field_definition(
                        name,
                        field,
                        transaction.get_database_backend() == sea_orm::DbBackend::Postgres,
                    )?;
                    let table = format!("\"{}\"", table.replace('"', "\"\""));
                    _ = transaction
                        .execute_unprepared(&format!("ALTER TABLE {table} ADD COLUMN {definition}"))
                        .await
                        .map_err(map_db_err)?;
                }
            }
            transaction.commit().await.map_err(map_db_err)?;
        }

        Ok(())
    }
}

struct SeaOrmStoreTransaction<'a, S: AuthSchema> {
    store: &'a SeaOrmStore<S>,
    tx: &'a ScopedTransaction,
    pending_after: AfterHookQueue<S>,
}

#[async_trait]
impl<S> AuthTransaction<S> for SeaOrmStoreTransaction<'_, S>
where
    S: AuthSchema,
    S::User: SeaOrmUserModel,
    S::Account: SeaOrmAccountModel,
    S::Session: SeaOrmSessionModel,
    S::Verification: SeaOrmVerificationModel,
{
    async fn get_scim_connection_binding(
        &self,
        id: &str,
    ) -> AuthResult<Option<alibi_core::store::ScimConnectionBinding>> {
        self.store
            .get_scim_connection_binding_with_connection(self.tx, id)
            .await
    }
    async fn scim_projection_candidates(
        &self,
        domain: &str,
        cursor: Option<&str>,
    ) -> AuthResult<Vec<alibi_core::store::ScimProjectionCandidate>> {
        self.store
            .scim_projection_candidates_with_connection(self.tx, domain, cursor)
            .await
    }
    async fn renew_scim_decommission_lease(
        &self,
        binding: &alibi_core::store::ScimConnectionBinding,
        lease: &str,
    ) -> AuthResult<bool> {
        self.store
            .renew_scim_decommission_lease_with_connection(self.tx, binding, lease)
            .await
    }
    async fn advance_scim_decommission(
        &self,
        binding: &alibi_core::store::ScimConnectionBinding,
        lease: &str,
        cursor: Option<&str>,
        count: usize,
        more: bool,
    ) -> AuthResult<Option<alibi_core::store::ScimConnectionBinding>> {
        self.store
            .advance_scim_decommission_with_connection(self.tx, binding, lease, cursor, count, more)
            .await
    }
    async fn reconcile_scim_identity(
        &self,
        user_id: &str,
    ) -> AuthResult<alibi_core::store::ScimIdentityState> {
        self.store
            .reconcile_scim_identity_with_connection(self.tx, user_id)
            .await
    }
    async fn delete_user_sessions(&self, user_id: &str) -> AuthResult<()> {
        use sea_orm::{ColumnTrait, QueryFilter};
        let sessions = self.get_user_sessions(user_id).await?;
        let context = self.store.hook_context(Some(self.tx));
        for session in &sessions {
            for hook in self.store.hooks() {
                if hook
                    .before_delete_session(session, &context)
                    .await?
                    .is_cancelled()
                {
                    return Ok(());
                }
            }
        }
        _ = <S::Session as SeaOrmSessionModel>::Entity::delete_many()
            .filter(
                S::Session::user_id_column()
                    .eq(<S::Session as SeaOrmSessionModel>::parse_user_id(user_id)?),
            )
            .exec(self.tx)
            .await
            .map_err(map_db_err)?;
        for session in sessions {
            self.pending_after
                .push(AfterHook::SessionDeleted(session))
                .await;
        }
        Ok(())
    }
    async fn get_user_sessions(&self, user_id: &str) -> AuthResult<Vec<S::Session>> {
        use sea_orm::{ColumnTrait, QueryFilter, QuerySelect};
        <S::Session as SeaOrmSessionModel>::Entity::find()
            .filter(S::Session::user_id_column().eq(S::Session::parse_user_id(user_id)?))
            .limit(
                u64::try_from(
                    self.store
                        .config()
                        .advanced
                        .database
                        .default_find_many_limit,
                )
                .unwrap_or(u64::MAX),
            )
            .all(self.tx)
            .await
            .map_err(map_db_err)
    }
    async fn list_oauth_tokens_for_session(
        &self,
        kind: alibi_core::store::OAuthTokenKind,
        session_id: &str,
    ) -> AuthResult<Vec<alibi_core::store::OAuthToken>> {
        self.store
            .session_oauth_tokens(self.tx, kind, session_id)
            .await
    }
    async fn get_oauth_client(
        &self,
        client_id: &str,
    ) -> AuthResult<Option<alibi_core::store::OAuthClient>> {
        self.store
            .oauth_client_with_connection(self.tx, client_id)
            .await
    }
    async fn queue_session_deletion_work(
        &self,
        session_id: &str,
        works: Vec<Box<dyn alibi_core::store::SessionDeletionWork<S>>>,
        context: Arc<alibi_core::AuthContext<S>>,
    ) -> AuthResult<()> {
        self.pending_after
            .attach_session_retirement(session_id, works, context)
            .await;
        Ok(())
    }
    async fn check_scim_user_attributes(
        &self,
        id: &str,
        connection: &str,
        attributes: &alibi_core::store::ScimUserAttributes,
    ) -> AuthResult<()> {
        self.store
            .assert_scim_user_keys_with_connection(
                self.tx,
                connection,
                &attributes.user_name_key,
                attributes.external_id_key.as_deref(),
                None,
                Some(id),
            )
            .await
    }
    async fn lock_scim_user(
        &self,
        id: &str,
        connection: &str,
        domain: &str,
        attributes: Option<&alibi_core::store::ScimUserAttributes>,
        now: chrono::DateTime<chrono::Utc>,
    ) -> AuthResult<alibi_core::store::ScimUserMutationContext> {
        self.store
            .lock_scim_user_with_connection(self.tx, id, connection, domain, attributes, now)
            .await
    }
    async fn replace_scim_user(
        &self,
        id: &str,
        data: alibi_core::store::ReplaceScimUser,
        roles: bool,
    ) -> AuthResult<(
        alibi_core::store::ScimUserRecord,
        Vec<alibi_core::store::ScimProjectionPreparedUser>,
    )> {
        self.store
            .replace_scim_user_with_connection(self.tx, id, data, roles)
            .await
    }
    async fn create_scim_user(
        &self,
        data: alibi_core::store::CreateScimUser,
        roles: bool,
    ) -> AuthResult<(
        alibi_core::store::ScimUserRecord,
        Vec<alibi_core::store::ScimProjectionPreparedUser>,
    )> {
        self.store
            .create_scim_user_with_connection(self.tx, data, roles)
            .await
    }
    async fn delete_scim_user(
        &self,
        id: &str,
        connection: &str,
        domain: &str,
        roles: bool,
    ) -> AuthResult<(String, Vec<alibi_core::store::ScimProjectionPreparedUser>)> {
        self.store
            .delete_scim_user_with_connection(self.tx, id, connection, domain, roles)
            .await
    }
    async fn patch_scim_group(
        &self,
        id: &str,
        connection: &str,
        domain: &str,
        patch: alibi_core::store::ScimGroupPatch,
        roles: bool,
    ) -> AuthResult<(
        alibi_core::store::ScimGroupRecord,
        Vec<alibi_core::store::ScimProjectionPreparedUser>,
    )> {
        self.store
            .patch_scim_group_with_connection(self.tx, id, connection, domain, patch, roles)
            .await
    }
    async fn replace_scim_group(
        &self,
        id: &str,
        data: alibi_core::store::ReplaceScimGroup,
        roles: bool,
    ) -> AuthResult<(
        alibi_core::store::ScimGroupRecord,
        Vec<alibi_core::store::ScimProjectionPreparedUser>,
    )> {
        self.store
            .replace_scim_group_with_connection(self.tx, id, data, roles)
            .await
    }
    async fn delete_scim_group(
        &self,
        id: &str,
        connection: &str,
        domain: &str,
        roles: bool,
    ) -> AuthResult<Vec<alibi_core::store::ScimProjectionPreparedUser>> {
        self.store
            .delete_scim_group_with_connection(self.tx, id, connection, domain, roles)
            .await
    }
    async fn create_scim_group(
        &self,
        data: alibi_core::store::CreateScimGroup,
        roles: bool,
    ) -> AuthResult<(
        alibi_core::store::ScimGroupRecord,
        Vec<alibi_core::store::ScimProjectionPreparedUser>,
    )> {
        self.store
            .create_scim_group_with_connection(self.tx, data, roles)
            .await
    }
    async fn fence_scim_connection(&self, id: &str, key: &str) -> AuthResult<()> {
        self.store
            .fence_scim_connection_with_connection(self.tx, id, key)
            .await
    }
    async fn prepare_scim_projection(
        &self,
        domain: &str,
        ids: &[String],
        roles: bool,
    ) -> AuthResult<Vec<alibi_core::store::ScimProjectionPreparedUser>> {
        self.store
            .prepare_scim_projection_with_connection(self.tx, domain, ids, roles)
            .await
    }
    async fn create_scim_projection_grant(
        &self,
        data: alibi_core::store::CreateScimProjectionGrant,
    ) -> AuthResult<()> {
        self.store
            .create_scim_projection_grant_with_connection(self.tx, data)
            .await
    }
    async fn remove_scim_projection_grants(
        &self,
        domain: &str,
        user_id: &str,
        ids: &[String],
    ) -> AuthResult<()> {
        self.store
            .remove_scim_projection_grants_with_connection(self.tx, domain, user_id, ids)
            .await
    }
    async fn get_user_by_email(&self, email: &str) -> AuthResult<Option<S::User>> {
        self.store
            .get_user_by_email_with_connection(self.tx, email)
            .await
    }
    async fn get_account(
        &self,
        provider: &str,
        account_id: &str,
    ) -> AuthResult<Option<S::Account>> {
        self.store
            .get_account_with_connection(self.tx, provider, account_id)
            .await
    }
    async fn update_user(&self, id: &str, update: alibi_core::UpdateUser) -> AuthResult<S::User> {
        let user = self
            .store
            .update_user_with_connection(self.tx, Some(self.tx), id, update)
            .await?;
        self.pending_after
            .push(AfterHook::UserUpdated(user.clone()))
            .await;
        Ok(user)
    }
    async fn update_account(
        &self,
        id: &str,
        update: alibi_core::UpdateAccount,
    ) -> AuthResult<S::Account> {
        let account = self
            .store
            .update_account_with_connection(self.tx, Some(self.tx), id, update)
            .await?;
        self.pending_after
            .push(AfterHook::AccountUpdated(account.clone()))
            .await;
        Ok(account)
    }
    async fn lock_sso_provider(
        &self,
        id: &str,
        provider_id: &str,
    ) -> AuthResult<Option<alibi_core::store::SsoProvider>> {
        self.store
            .lock_sso_provider_with(self.tx, id, provider_id)
            .await
    }
    async fn update_sso_provider(
        &self,
        update: alibi_core::store::SsoProviderUpdate,
    ) -> AuthResult<Option<alibi_core::store::SsoProvider>> {
        self.store.update_sso_provider_with(self.tx, update).await
    }
    async fn delete_sso_provider_and_accounts(
        &self,
        id: &str,
        provider_id: &str,
    ) -> AuthResult<bool> {
        self.store
            .delete_sso_provider_with(self.tx, id, provider_id)
            .await
    }
    async fn list_jwks(&self) -> AuthResult<Vec<Jwk>> {
        self.store.list_jwks_with_connection(self.tx).await
    }
    async fn get_jwk_by_id(&self, id: &str) -> AuthResult<Option<Jwk>> {
        self.store.get_jwk_with_connection(self.tx, id).await
    }
    async fn create_jwk(&self, data: CreateJwk) -> AuthResult<Jwk> {
        self.store.create_jwk_with_connection(self.tx, data).await
    }

    async fn get_team(&self, organization_id: &str, team_id: &str) -> AuthResult<Option<Team>> {
        self.store
            .get_team_with_connection(self.tx, Some(organization_id), team_id)
            .await
    }
    async fn add_team_member(
        &self,
        team_id: &str,
        user_id: &str,
        maximum: Option<f64>,
    ) -> AuthResult<AddTeamMemberResult> {
        self.store
            .add_team_member_in_tx(self.tx, team_id, user_id, maximum)
            .await
    }
    async fn create_member(&self, member: CreateMember) -> AuthResult<Member> {
        self.store
            .create_member_with_connection(self.tx, member)
            .await
    }
    async fn update_session_active_team(
        &self,
        token: &str,
        team_id: Option<&str>,
    ) -> AuthResult<S::Session> {
        self.store
            .update_session_scope_with_connection(
                self.tx,
                token,
                sessions::SessionScope::Team(team_id),
            )
            .await
    }
    async fn update_session_active_organization(
        &self,
        token: &str,
        organization_id: Option<&str>,
    ) -> AuthResult<S::Session> {
        self.store
            .update_session_scope_with_connection(
                self.tx,
                token,
                sessions::SessionScope::Organization(organization_id),
            )
            .await
    }
    async fn provider_verification_output(
        &self,
        id: &str,
    ) -> AuthResult<Option<serde_json::Value>> {
        users::provider_verification_output::<S::User, _>(self.tx, id).await
    }
    async fn get_user_by_id(&self, id: &str) -> AuthResult<Option<S::User>> {
        use sea_orm::{ColumnTrait, QueryFilter};
        let id = S::User::parse_id(id)?;
        users::user_query::<S::User>(self.tx.get_database_backend())
            .filter(S::User::id_column().eq(id))
            .one(self.tx)
            .await
            .map_err(map_db_err)
    }
    async fn create_passkey(&self, data: CreatePasskey) -> AuthResult<Passkey> {
        self.store
            .create_passkey_with_connection(self.tx, data)
            .await
    }
    async fn create_user(&self, create_user: CreateUser) -> AuthResult<S::User> {
        let user = self.store.create_user_in_tx(self.tx, create_user).await?;
        self.pending_after
            .push(AfterHook::UserCreated(user.clone()))
            .await;
        Ok(user)
    }

    async fn create_user_prepared(&self, prepared: PreparedUserCreation) -> AuthResult<S::User> {
        let user = self
            .store
            .create_user_prepared_in_tx(self.tx, prepared)
            .await?;
        self.pending_after
            .push(AfterHook::UserCreated(user.clone()))
            .await;
        Ok(user)
    }

    async fn create_account(&self, create_account: CreateAccount) -> AuthResult<S::Account> {
        let account = self
            .store
            .create_account_in_tx(self.tx, create_account)
            .await?;
        self.pending_after
            .push(AfterHook::AccountCreated(account.clone()))
            .await;
        Ok(account)
    }

    async fn prepare_secondary_session_update(
        &self,
        session: S::Session,
        expires_at: Option<DateTime<Utc>>,
        fields: FieldValues,
    ) -> AuthResult<Option<(S::Session, FieldValues)>> {
        self.store
            .prepare_secondary_update_with_connection(
                self.tx,
                Some(self.tx),
                session,
                expires_at,
                fields,
            )
            .await
    }
    async fn complete_secondary_session_update(
        &self,
        session: S::Session,
        expires_at: Option<DateTime<Utc>>,
        fields: FieldValues,
        persist: bool,
    ) -> AuthResult<Option<S::Session>> {
        use alibi_core::AuthSession;
        let token = session.token().to_owned();
        let result = self
            .store
            .complete_secondary_update_with_connection(
                self.tx,
                Some(self.tx),
                session,
                expires_at,
                fields,
                persist,
            )
            .await?;
        let event = result.as_ref().map_or_else(
            || AfterHook::SessionUpdateMissing(token),
            |model| AfterHook::SessionUpdated(model.clone()),
        );
        self.pending_after.push(event).await;
        Ok(result)
    }
    async fn prepare_secondary_session_creation(
        &self,
        input: CreateSession,
        persist: bool,
    ) -> AuthResult<S::Session> {
        let session = self
            .store
            .prepare_secondary_session_in_tx(self.tx, input, persist)
            .await?;
        self.pending_after
            .push(AfterHook::SessionCreated(session.clone()))
            .await;
        Ok(session)
    }
    async fn create_session(&self, create_session: CreateSession) -> AuthResult<S::Session> {
        let session = self
            .store
            .create_session_in_tx(self.tx, create_session)
            .await?;
        self.pending_after
            .push(AfterHook::SessionCreated(session.clone()))
            .await;
        Ok(session)
    }
    async fn create_verification_record(
        &self,
        data: VerificationCreation,
        publication: VerificationPublication,
    ) -> AuthResult<Option<VerificationSnapshot>> {
        let snapshot = self
            .store
            .create_verification_record_with_connection(self.tx, Some(self.tx), data, publication)
            .await?;
        if let Some(snapshot) = &snapshot {
            self.pending_after
                .push(AfterHook::VerificationRecordCreated(snapshot.clone()))
                .await;
        }
        Ok(snapshot)
    }

    async fn create_verification(
        &self,
        verification: CreateVerification,
    ) -> AuthResult<S::Verification> {
        let verification = self
            .store
            .create_verification_in_tx(self.tx, verification)
            .await?;
        self.pending_after
            .push(AfterHook::VerificationCreated(verification.clone()))
            .await;
        Ok(verification)
    }
    async fn consume_verification_snapshot(
        &self,
        identifier: &str,
    ) -> AuthResult<Option<S::Verification>> {
        let consumed = self
            .store
            .consume_generation_in(self.tx, identifier, None)
            .await?;
        if let Some(verification) = &consumed {
            self.pending_after
                .push(AfterHook::VerificationDeleted(verification.clone()))
                .await;
        }
        Ok(consumed)
    }
}

#[async_trait]
impl<S> TransactionStore<S> for SeaOrmStore<S>
where
    S: AuthSchema,
    S::User: SeaOrmUserModel,
    S::Account: SeaOrmAccountModel,
    S::Session: SeaOrmSessionModel,
    S::Verification: SeaOrmVerificationModel,
{
    async fn transaction_boxed(
        &self,
        work: Box<TransactionWork<S>>,
    ) -> AuthResult<BoxedTransactionValue> {
        self.transaction_boxed_with_policy(
            work,
            alibi_core::store::AfterCommitErrorPolicy::Propagate,
        )
        .await
    }
    async fn transaction_boxed_with_policy(
        &self,
        work: Box<TransactionWork<S>>,
        policy: alibi_core::store::AfterCommitErrorPolicy,
    ) -> AuthResult<BoxedTransactionValue> {
        let tx = self.db.begin().await.map_err(map_db_err)?;
        self.run_transaction(tx, work, policy).await
    }
    async fn write_transaction_boxed(
        &self,
        work: Box<TransactionWork<S>>,
    ) -> AuthResult<BoxedTransactionValue> {
        let tx = self.db.begin_immediate().await?;
        self.run_transaction(
            tx,
            work,
            alibi_core::store::AfterCommitErrorPolicy::Propagate,
        )
        .await
    }
}

impl<S> SeaOrmStore<S>
where
    S: AuthSchema,
    S::User: SeaOrmUserModel,
    S::Account: SeaOrmAccountModel,
    S::Session: SeaOrmSessionModel,
    S::Verification: SeaOrmVerificationModel,
{
    async fn run_transaction(
        &self,
        tx: ScopedTransaction,
        work: Box<TransactionWork<S>>,
        policy: alibi_core::store::AfterCommitErrorPolicy,
    ) -> AuthResult<BoxedTransactionValue> {
        let tx_store = SeaOrmStoreTransaction {
            store: self,
            tx: &tx,
            pending_after: AfterHookQueue::new(),
        };
        let outcome = work(&tx_store).await;
        let pending_after = tx_store.pending_after;
        match outcome {
            Ok(value) => {
                tx.commit().await.map_err(map_db_err)?;
                // After hooks run only once the auth writes are durable, in
                // operation order; rollback drops them.
                pending_after
                    .run_with_policy(self.hooks(), &self.hook_context(None), policy)
                    .await?;
                Ok(value)
            }
            Err(err) => {
                tx.rollback().await.map_err(map_db_err)?;
                Err(err)
            }
        }
    }
}

#[expect(
    clippy::needless_pass_by_value,
    reason = "Result::map_err transfers ownership to this error-boundary adapter"
)]
pub(crate) fn map_db_err(err: DbErr) -> AuthError {
    match err.sql_err() {
        Some(
            SqlErr::UniqueConstraintViolation(message)
            | SqlErr::ForeignKeyConstraintViolation(message),
        ) => AuthError::Database(DatabaseError::Constraint(message)),
        Some(_) | None => AuthError::Database(DatabaseError::Query(err.to_string())),
    }
}

/// Bind raw numeric `LIMIT`/`OFFSET` values; the database validates them.
pub(crate) fn bind_page(
    mut statement: sea_orm::Statement,
    limit: Option<f64>,
    offset: Option<f64>,
) -> AuthResult<sea_orm::Statement> {
    use sea_orm::{DbBackend, Value, Values};
    use std::fmt::Write;
    for (clause, number) in [("LIMIT", limit), ("OFFSET", offset)] {
        if let Some(number) = number {
            let values = statement.values.get_or_insert_with(|| Values(Vec::new()));
            let placeholder = match statement.db_backend {
                DbBackend::Postgres => format!("${}", values.0.len() + 1),
                DbBackend::Sqlite | DbBackend::MySql => "?".into(),
                _ => {
                    return Err(AuthError::not_implemented(
                        "Raw numeric pages are not supported by this database backend",
                    ));
                }
            };
            _ = write!(statement.sql, " {clause} {placeholder}");
            if statement.db_backend == DbBackend::Postgres {
                // node-postgres sends Number parameters as text. Binding float8
                // instead lets PostgreSQL round fractional pages to bigint.
                statement.sql.push_str("::bigint");
                values.0.push(Value::String(Some(
                    ryu_js::Buffer::new().format(number).to_owned(),
                )));
            } else {
                values.0.push(Value::Double(Some(number)));
            }
        }
    }
    Ok(statement)
}

mod sso;

mod scim;
mod scim_decommission;
mod scim_groups;
mod scim_projection;
mod scim_user_create;
mod scim_user_replace;
mod scim_users;