alibi_sqlx/store/

users.rs

use super::{SqlxStore, lock_exclusive, lock_shared};
use crate::error::record_not_updated;
use crate::model::{self, ActiveRow, SqlxModel};
use crate::pool::{Engine, Exec, SqlxTransaction};
use crate::schema::{AuthSchema, SqlxUserModel};
use crate::sql::Sql;
use crate::value::{ColumnKind, SqlValue};
use alibi_core::error::{AuthError, AuthResult};
use alibi_core::store::adapter::cancelled_by_hook;
use alibi_core::store::{NumericTextInput, UserCreationDefaults, UserStore};
use alibi_core::types::{CreateUser, ListUsersParams, UpdateUser};
use alibi_core::user_validation::PreparedUserCreation;
use alibi_core::{AuthUser, UserFilterValue};
use async_trait::async_trait;
use chrono::Utc;

/// Row lock applied to a user lookup inside a transaction.
#[derive(Clone, Copy)]
pub(super) enum Lock {
    None,
    Shared,
    Exclusive,
}

pub(super) async fn find_user_by_id<M: SqlxUserModel>(
    exec: Exec<'_>,
    id: &str,
    lock: Lock,
) -> AuthResult<Option<M>> {
    let id = M::parse_id(id)?;
    let mut sql = model::select_model::<M>(exec);
    sql.push(" WHERE ");
    sql.compare_model::<M>(M::TABLE, M::id_column(), " = ", id);
    model::limit_one(&mut sql);
    match lock {
        Lock::None => {}
        Lock::Shared => lock_shared(&mut sql),
        Lock::Exclusive => lock_exclusive(&mut sql),
    }
    exec.fetch_optional(sql).await
}

pub(super) async fn provider_verification_output<M: SqlxUserModel>(
    exec: Exec<'_>,
    id: &str,
) -> AuthResult<Option<serde_json::Value>> {
    if exec.engine() != Engine::Sqlite {
        return Ok(None);
    }
    let Some(column) = M::PROVIDER_VERIFICATION_COLUMN else {
        return Ok(None);
    };
    let mut sql = Sql::new(exec.engine());
    sql.push("SELECT CASE typeof(");
    sql.ident(column);
    sql.push(") WHEN 'text' THEN json_quote(");
    sql.ident(column);
    sql.push(") WHEN 'null' THEN 'null' ELSE CASE WHEN ");
    sql.ident(column);
    sql.push(" = 1 THEN 'true' ELSE 'false' END END FROM ");
    sql.ident(M::TABLE);
    sql.push(" WHERE ");
    sql.compare(M::TABLE, M::id_column(), " = ", M::parse_id(id)?);
    sql.push(" LIMIT 1");
    exec.fetch_scalar::<String>(sql)
        .await?
        .map(|value| {
            alibi_core::utils::json::from_slice(value.as_bytes())
                .map_err(|error| AuthError::internal(error.to_string()))
        })
        .transpose()
}

fn stage_provider_verification<M: SqlxUserModel>(
    active: &mut ActiveRow,
    value: Option<serde_json::Value>,
    engine: Engine,
) -> AuthResult<()> {
    let Some(value) = value else {
        return Ok(());
    };
    // Boolean inputs retain the ordinary typed-model path on every engine.
    if value.is_boolean() {
        return Ok(());
    }
    let column = M::PROVIDER_VERIFICATION_COLUMN.ok_or_else(|| {
        AuthError::internal("The user model does not expose its provider verification column")
    })?;
    let value = match value {
        serde_json::Value::Null => SqlValue::Bool(None),
        serde_json::Value::String(value) => SqlValue::Text(Some(value)),
        serde_json::Value::Number(value) => {
            if engine == Engine::Postgres {
                SqlValue::Text(Some(
                    alibi_core::utils::json::number_to_string(&value)
                        .map_err(|error| AuthError::internal(error.to_string()))?,
                ))
            } else if let Some(value) = value.as_i64() {
                SqlValue::BigInt(Some(value))
            } else {
                SqlValue::Double(value.as_f64())
            }
        }
        _ => {
            return Err(AuthError::internal(
                "Unsupported provider verification SQL parameter",
            ));
        }
    };
    active.set(column, value);
    Ok(())
}

async fn stage_provider_text<M: SqlxUserModel>(
    exec: Exec<'_>,
    active: &mut ActiveRow,
    name: Option<serde_json::Value>,
    image: Option<serde_json::Value>,
) -> AuthResult<()> {
    for (field, raw) in [("name", name), ("image", image)] {
        let Some(raw) = raw else {
            continue;
        };
        if raw.is_array() || raw.is_object() {
            return Err(AuthError::internal(
                "Unsupported provider text SQL parameter",
            ));
        }
        let column = if field == "name" {
            Some(M::name_column())
        } else {
            M::list_users_column("image")
        }
        .ok_or_else(|| {
            AuthError::internal("The user model does not expose its provider text column")
        })?;
        let value =
            crate::additional_fields::raw_value(&alibi_core::utils::json::JsValue::from(raw))?;
        let value =
            crate::additional_fields::prepare_value(exec, M::column_kind(column), value).await?;
        active.set(column, value);
    }
    Ok(())
}

impl<S> SqlxStore<S>
where
    S: AuthSchema,
    S::User: SqlxUserModel,
{
    async fn create_user_with_connection(
        &self,
        exec: Exec<'_>,
        tx: Option<&SqlxTransaction>,
        mut create_user: CreateUser,
        defaults: UserCreationDefaults,
    ) -> AuthResult<S::User> {
        let hook_context = self.hook_context(tx);
        for hook in self.hooks() {
            if hook
                .before_create_user(&mut create_user, &hook_context)
                .await?
                .is_cancelled()
            {
                return Err(AuthError::UserCreationCancelled);
            }
        }
        let mut create_user = defaults.apply(create_user)?;
        if let Some(username) = create_user.username.as_mut() {
            *username = username.to_lowercase();
        }
        let now = Utc::now();
        if create_user.id.is_none() {
            create_user.id = self
                .generated_id(
                    exec,
                    "user",
                    <S::User as SqlxModel>::TABLE,
                    S::User::id_column(),
                )
                .await?;
        }
        let user_id = create_user
            .id
            .as_deref()
            .map(S::User::parse_id)
            .transpose()?;
        let mut fields = std::mem::take(&mut create_user.additional_fields);
        fields.apply_adapter_transforms_async().await?;
        let provider_name = create_user.provider_name.take();
        let provider_image = create_user.provider_image.take();
        let provider_verification = create_user.provider_email_verified.take();
        let mut active = S::User::new_active(user_id, create_user, now);
        stage_provider_verification::<S::User>(&mut active, provider_verification, exec.engine())?;
        stage_provider_text::<S::User>(exec, &mut active, provider_name, provider_image).await?;
        let backend = exec.engine();
        for (column, value) in S::User::additional_field_bindings(&fields, backend)? {
            let value = crate::additional_fields::prepare_value(
                exec,
                <S::User as SqlxModel>::column_kind(column),
                value,
            )
            .await?;
            S::User::set_additional_field(&mut active, column, value, backend)?;
        }
        S::User::prepare_json_metadata(&mut active, backend)?;

        let user = model::insert::<S::User>(exec, &active).await?;
        if tx.is_none() {
            for hook in self.hooks() {
                hook.after_create_user(&user, &hook_context).await?;
            }
        }
        Ok(user)
    }

    pub(crate) async fn create_user_in_tx(
        &self,
        tx: &SqlxTransaction,
        mut create_user: CreateUser,
    ) -> AuthResult<S::User> {
        create_user.email = create_user.email.map(|email| normalize_user_email(&email));
        self.create_user_with_connection(
            Exec::tx(tx),
            Some(tx),
            create_user,
            alibi_core::store::UserCreationDefaults::default(),
        )
        .await
    }

    pub(crate) async fn create_user_prepared_in_tx(
        &self,
        tx: &SqlxTransaction,
        prepared: PreparedUserCreation,
    ) -> AuthResult<S::User> {
        let (data, defaults) = prepared.into_parts();
        self.create_user_with_connection(Exec::tx(tx), Some(tx), data, defaults)
            .await
    }

    async fn users_by_ids(&self, ids: &[String], limit: Option<f64>) -> AuthResult<Vec<S::User>> {
        if ids.is_empty() {
            return Ok(Vec::new());
        }
        let user_ids = ids
            .iter()
            .map(|id| S::User::parse_id(id))
            .collect::<AuthResult<Vec<_>>>()?;
        let mut sql = model::select_model::<S::User>(self.exec());
        sql.push(" WHERE ");
        sql.column(<S::User as SqlxModel>::TABLE, S::User::id_column());
        sql.push(" IN ");
        sql.bind_list(user_ids);
        super::bind_page(&mut sql, limit, None);
        self.exec().fetch_all(sql).await
    }

    /// Bind an admin user-list predicate with the configured SQL semantics.
    fn push_user_filter(
        &self,
        sql: &mut Sql,
        filter_field: Option<&str>,
        operator: &str,
        value: &UserFilterValue,
    ) -> AuthResult<()> {
        let table = <S::User as SqlxModel>::TABLE;
        let field = filter_field
            .filter(|field| !field.is_empty())
            .unwrap_or("email");
        let column = S::User::list_users_column(field)
            .ok_or_else(|| AuthError::bad_request("User filter field has no configured column"))?;
        let operands = match value {
            UserFilterValue::Multiple(values) => values.as_slice(),
            UserFilterValue::Scalar(value) if operator != "in" => std::slice::from_ref(value),
            UserFilterValue::Scalar(_) => {
                return Err(AuthError::bad_request("Value must be an array"));
            }
        };
        // Upstream coerces a scalar string on a boolean field before binding;
        // array operands keep their original strings.
        let numeric_cast = if self.exec().engine() == Engine::Postgres {
            match <S::User as SqlxModel>::column_kind(column) {
                ColumnKind::Int => Some("int4"),
                ColumnKind::BigInt => Some("int8"),
                ColumnKind::Float => Some("float4"),
                ColumnKind::Double => Some("float8"),
                _ => None,
            }
        } else {
            None
        };
        let bindings: Vec<SqlValue> = if numeric_cast.is_some() {
            numeric_filter_text(operands)
                .into_iter()
                .map(Into::into)
                .collect()
        } else if matches!(value, UserFilterValue::Scalar(_))
            && <S::User as SqlxModel>::column_kind(column) == ColumnKind::Boolean
        {
            operands
                .iter()
                .map(|value_2| (value_2 == "true").into())
                .collect()
        } else {
            operands
                .iter()
                .cloned()
                .map(|value| S::User::column_value(column, value.into()))
                .collect()
        };
        match operator {
            "in" | "not_in" if bindings.is_empty() && numeric_cast.is_none() => {
                sql.push(if operator == "in" { "1 = 2" } else { "1 = 1" });
            }
            "in" => {
                sql.column(table, column);
                sql.push(" IN ");
                bind_filter_values(sql, bindings, numeric_cast);
            }
            "not_in" => {
                sql.column(table, column);
                sql.push(" NOT IN ");
                bind_filter_values(sql, bindings, numeric_cast);
            }
            // The pinned adapter interpolates the complete array's
            // comma-joined value into a bound LIKE pattern. Actual SQL
            // retains backend case, wildcard and NULL semantics.
            "contains" => {
                sql.compare_model::<S::User>(
                    table,
                    column,
                    " LIKE ",
                    format!("%{}%", operands.join(",")),
                );
            }
            "starts_with" => {
                sql.compare_model::<S::User>(
                    table,
                    column,
                    " LIKE ",
                    format!("{}%", operands.join(",")),
                );
            }
            "ends_with" => {
                sql.compare_model::<S::User>(
                    table,
                    column,
                    " LIKE ",
                    format!("%{}", operands.join(",")),
                );
            }
            "eq" | "ne" | "lt" | "lte" | "gt" | "gte" => {
                let comparison = match operator {
                    "eq" => " = ",
                    "ne" => " <> ",
                    "lt" => " < ",
                    "lte" => " <= ",
                    "gt" => " > ",
                    _ => " >= ",
                };
                sql.ident(column);
                sql.push(comparison);
                bind_filter_values(sql, bindings, numeric_cast);
            }
            _ => return Err(AuthError::bad_request("Unsupported user filter operator")),
        }
        Ok(())
    }

    fn user_list_predicates(&self, sql: &mut Sql, params: &ListUsersParams) -> AuthResult<()> {
        let filtered = if let Some(search) = params
            .search_value
            .as_ref()
            .filter(|search| !search.is_empty())
        {
            sql.push(" WHERE ");
            self.push_user_filter(
                sql,
                params.search_field.as_deref(),
                params.search_operator.as_deref().unwrap_or("contains"),
                &UserFilterValue::Scalar(search.clone()),
            )?;
            true
        } else {
            false
        };
        if let Some(value) = &params.filter_value {
            sql.push(if filtered { " AND " } else { " WHERE " });
            self.push_user_filter(
                sql,
                params.filter_field.as_deref(),
                params.filter_operator.as_deref().unwrap_or("eq"),
                value,
            )?;
        }
        Ok(())
    }

    fn user_lookup(&self, column: &str, value: impl Into<SqlValue>) -> Sql {
        let mut sql = model::select_model::<S::User>(self.exec());
        sql.push(" WHERE ");
        sql.compare_model::<S::User>(<S::User as SqlxModel>::TABLE, column, " = ", value);
        sql.push(" LIMIT 1");
        sql
    }
}

#[async_trait]
impl<S> UserStore<S> for SqlxStore<S>
where
    S: AuthSchema + Send + Sync,
    S::User: SqlxUserModel,
{
    async fn provider_verification_output(
        &self,
        id: &str,
    ) -> AuthResult<Option<serde_json::Value>> {
        provider_verification_output::<S::User>(self.exec(), id).await
    }

    async fn create_user(&self, mut create_user: CreateUser) -> AuthResult<S::User> {
        create_user.email = create_user.email.map(|email| normalize_user_email(&email));
        self.create_user_with_connection(
            self.exec(),
            None,
            create_user,
            alibi_core::store::UserCreationDefaults::default(),
        )
        .await
    }

    async fn create_user_prepared(&self, prepared: PreparedUserCreation) -> AuthResult<S::User> {
        let (data, defaults) = prepared.into_parts();
        self.create_user_with_connection(self.exec(), None, data, defaults)
            .await
    }

    async fn coerce_user_text_number(&self, input: NumericTextInput) -> AuthResult<String> {
        let value: SqlValue = match input {
            NumericTextInput::Integer(value) => value.into(),
            NumericTextInput::Real(value) if !value.is_nan() => value.into(),
            NumericTextInput::Real(_) => {
                return Err(AuthError::bad_request("Numeric text input must not be NaN"));
            }
        };
        let backend = self.exec().engine();
        let mut sql = Sql::with(backend, "SELECT CAST(");
        sql.bind(value);
        sql.push(" AS TEXT) AS value");
        // The database's own CAST decides the stored text.
        self.exec()
            .fetch_scalar::<String>(sql)
            .await?
            .ok_or_else(|| AuthError::internal("Numeric text coercion returned no value"))
    }

    async fn get_user_by_id(&self, id: &str) -> AuthResult<Option<S::User>> {
        find_user_by_id::<S::User>(self.exec(), id, Lock::None).await
    }

    async fn list_users_by_ids(&self, ids: &[String]) -> AuthResult<Vec<S::User>> {
        self.users_by_ids(ids, None).await
    }

    async fn list_users_by_ids_page(&self, ids: &[String], limit: f64) -> AuthResult<Vec<S::User>> {
        self.users_by_ids(ids, Some(limit)).await
    }

    async fn get_user_by_email(&self, email: &str) -> AuthResult<Option<S::User>> {
        self.get_user_by_email_with_connection(self.exec(), email)
            .await
    }

    async fn get_user_by_username(&self, username: &str) -> AuthResult<Option<S::User>> {
        let Some(column) = S::User::username_column() else {
            return Ok(None);
        };
        let sql = self.user_lookup(column, username);
        self.exec().fetch_optional(sql).await
    }

    async fn get_user_by_phone_number(&self, phone_number: &str) -> AuthResult<Option<S::User>> {
        let column = S::User::phone_number_column()
            .ok_or_else(|| AuthError::internal("the user schema has no phone-number field"))?;
        let sql = self.user_lookup(column, phone_number);
        self.exec().fetch_optional(sql).await
    }

    async fn update_user(&self, id: &str, update: UpdateUser) -> AuthResult<S::User> {
        self.update_user_with_connection(self.exec(), None, id, update)
            .await
    }

    async fn delete_user(&self, id: &str) -> AuthResult<()> {
        _ = S::User::parse_id(id)?;
        let Some(user) = self.get_user_by_id(id).await? else {
            return Err(AuthError::UserNotFound);
        };
        let hook_context = self.hook_context(None);
        for hook in self.hooks() {
            if hook
                .before_delete_user(&user, &hook_context)
                .await?
                .is_cancelled()
            {
                return Ok(());
            }
        }
        // API keys retain their polymorphic reference. Redemption resolves the
        // owner and rejects a missing user with the public identity error.
        let owner = user.id().into_owned();
        let id = id.to_owned();
        self.in_transaction(true, async move |tx| {
            let exec = Exec::tx(tx);
            _ = find_user_by_id::<S::User>(exec, &id, Lock::Exclusive).await?;
            super::teams::remove_owned_team_members(tx, &owner, None).await?;
            super::wallets::remove_owned_wallets(tx, &owner).await?;
            let mut users = Sql::with(exec.engine(), "DELETE FROM ");
            users.ident(<S::User as SqlxModel>::TABLE);
            users.push(" WHERE ");
            users.compare_model::<S::User>(
                <S::User as SqlxModel>::TABLE,
                S::User::id_column(),
                " = ",
                S::User::parse_id(&id)?,
            );
            _ = exec.execute(users).await?;
            Ok(())
        })
        .await?;
        for hook in self.hooks() {
            hook.after_delete_user(&user, &hook_context).await?;
        }
        Ok(())
    }

    async fn list_users(&self, params: ListUsersParams) -> AuthResult<(Vec<S::User>, usize)> {
        let table = <S::User as SqlxModel>::TABLE;
        let mut sql = model::select_model::<S::User>(self.exec());
        self.user_list_predicates(&mut sql, &params)?;
        let field = params
            .sort_by
            .as_deref()
            .filter(|field| !field.is_empty())
            .unwrap_or("createdAt");
        let column = S::User::list_users_column(field)
            .ok_or_else(|| AuthError::bad_request("User sort field has no configured column"))?;
        sql.push(" ORDER BY ");
        sql.column(table, column);
        let direction = params.sort_direction.as_deref().unwrap_or_else(|| {
            if params.sort_by.is_some() {
                "asc"
            } else {
                "desc"
            }
        });
        sql.push(if direction == "desc" { " DESC" } else { " ASC" });
        let number = |value: usize| value.to_string().parse::<f64>().ok();
        let limit = params
            .source_limit
            .or_else(|| params.limit.and_then(number))
            .or_else(|| self.find_many_limit().to_string().parse().ok());
        let offset = params
            .source_offset
            .or_else(|| params.offset.and_then(number));
        super::bind_page(&mut sql, limit, offset);
        let users = self.exec().fetch_all(sql).await?;
        let mut count = Sql::with(self.exec().engine(), "SELECT COUNT(*) FROM ");
        count.ident(table);
        self.user_list_predicates(&mut count, &params)?;
        let total = usize::try_from(
            self.exec()
                .fetch_scalar::<i64>(count)
                .await?
                .unwrap_or_default(),
        )
        .map_err(|_| AuthError::internal("User count exceeds usize"))?;
        Ok((users, total))
    }
}

fn normalize_user_email(email: &str) -> String {
    email.to_lowercase()
}

// Source converts a nonempty numeric scalar, or every numeric array operand,
// to Number before node-postgres serializes the parameter. An invalid array
// keeps all its original strings. PostgreSQL validates the resulting text.
fn numeric_filter_text(values: &[String]) -> Vec<String> {
    let numbers = values
        .iter()
        .map(|value| {
            (!alibi_core::utils::javascript::trim(value).is_empty())
                .then(|| alibi_core::utils::javascript::string_to_number(value))
                .flatten()
                .filter(|number| !number.is_nan())
        })
        .collect::<Option<Vec<_>>>();
    numbers.map_or_else(
        || values.to_vec(),
        |numbers| {
            numbers
                .into_iter()
                .map(|number| ryu_js::Buffer::new().format(number).to_owned())
                .collect()
        },
    )
}

fn bind_filter_values(sql: &mut Sql, values: Vec<SqlValue>, cast: Option<&str>) {
    sql.push("(");
    for (index, value) in values.into_iter().enumerate() {
        if index > 0 {
            sql.push(", ");
        }
        sql.bind(value);
        if let Some(cast) = cast {
            sql.push("::");
            sql.push(cast);
        }
    }
    sql.push(")");
}

impl<S> SqlxStore<S>
where
    S: AuthSchema,
    S::User: crate::schema::SqlxUserModel,
{
    pub(super) async fn update_user_with_connection(
        &self,
        exec: Exec<'_>,
        tx: Option<&SqlxTransaction>,
        id: &str,
        mut update: UpdateUser,
    ) -> AuthResult<S::User> {
        update.email = update.email.map(|email| normalize_user_email(&email));
        _ = S::User::parse_id(id)?;
        let hook_context = self.hook_context(tx);
        for hook in self.hooks() {
            if hook
                .before_update_user(id, &mut update, &hook_context)
                .await
                .map_err(alibi_core::store::adapter::callback_error)?
                .is_cancelled()
            {
                return Err(cancelled_by_hook("user update"));
            }
        }
        if let Some(username) = update.username.as_mut() {
            *username = username.to_lowercase();
        }
        let Some(model) = find_user_by_id::<S::User>(exec, id, Lock::None).await? else {
            return Err(AuthError::UserNotFound);
        };

        let mut active = model.into_active();
        let mut fields = std::mem::take(&mut update.additional_fields);
        fields.apply_adapter_transforms_async().await?;
        let provider_name = update.provider_name.take();
        let provider_image = update.provider_image.take();
        let provider_verification = update.provider_email_verified.take();
        S::User::apply_update(&mut active, update, Utc::now());
        stage_provider_verification::<S::User>(&mut active, provider_verification, exec.engine())?;
        stage_provider_text::<S::User>(exec, &mut active, provider_name, provider_image).await?;
        let backend = exec.engine();
        for (column, value) in S::User::additional_field_bindings(&fields, backend)? {
            let value = crate::additional_fields::prepare_value(
                exec,
                <S::User as SqlxModel>::column_kind(column),
                value,
            )
            .await?;
            S::User::set_additional_field(&mut active, column, value, backend)?;
        }
        S::User::prepare_json_metadata(&mut active, backend)?;

        let user = model::update::<S::User>(exec, &active)
            .await?
            .ok_or_else(record_not_updated)?;
        if tx.is_none() {
            for hook in self.hooks() {
                hook.after_update_user(&user, &hook_context)
                    .await
                    .map_err(alibi_core::store::adapter::callback_error)?;
            }
        }
        Ok(user)
    }
}

impl<S> SqlxStore<S>
where
    S: AuthSchema,
    S::User: crate::schema::SqlxUserModel,
{
    pub(super) async fn get_user_by_email_with_connection(
        &self,
        exec: Exec<'_>,
        email: &str,
    ) -> AuthResult<Option<S::User>> {
        let email = normalize_user_email(email);
        let sql = self.user_lookup(S::User::email_column(), email);
        exec.fetch_optional(sql).await
    }
}