alibi_sqlx/store/
members.rs

1use super::entities::member::Model;
2use super::entities::team;
3use super::{SqlxStore, lock_exclusive};
4use crate::model::{self, ActiveRow, SqlxModel};
5use crate::organization_models::{Binding, Row};
6use crate::pool::Exec;
7use crate::schema::AuthSchema;
8use crate::sql::Sql;
9use crate::value::SqlValue;
10use alibi_core::error::{AuthError, AuthResult};
11use alibi_core::store::{ListOrganizationMembersParams, MemberPageQuery, MemberStore};
12use alibi_core::{CreateMember, Member};
13use async_trait::async_trait;
14use chrono::Utc;
15
16impl<S: AuthSchema> SqlxStore<S> {
17    pub(super) async fn create_member_with_connection(
18        &self,
19        exec: Exec<'_>,
20        member: CreateMember,
21    ) -> AuthResult<Member> {
22        let mut active = ActiveRow::new();
23        active.set(
24            "id",
25            self.generated_id(
26                exec,
27                "member",
28                self.organization_models.member.table(),
29                self.organization_models
30                    .member
31                    .physical("id")
32                    .ok_or_else(|| {
33                        alibi_core::AuthError::config("Organization model has no ID column")
34                    })?,
35            )
36            .await?
37            .unwrap_or_else(|| self.organization_models.member.new_id()),
38        );
39        active.set("organization_id", member.organization_id);
40        active.set("user_id", member.user_id);
41        active.set("role", member.role);
42        active.set("created_at", Utc::now());
43        self.organization_models
44            .member
45            .insert(exec, &active)
46            .await
47            .map(|model| Member::from(&model))
48    }
49
50    async fn find_member_by_id(&self, id: &str) -> AuthResult<Option<Row<Model>>> {
51        let mut sql = self.organization_models.member.by_id(self.exec(), id)?;
52        model::limit_one(&mut sql);
53        self.organization_models
54            .member
55            .fetch_optional(self.exec(), sql)
56            .await
57    }
58}
59
60/// A member predicate: one column comparison.
61struct Filter {
62    column: &'static str,
63    operator: &'static str,
64    value: SqlValue,
65}
66
67fn member_column(field: &str) -> Option<&'static str> {
68    match field {
69        "id" => Some("id"),
70        "organizationId" => Some("organization_id"),
71        "userId" => Some("user_id"),
72        "role" => Some("role"),
73        "createdAt" => Some("created_at"),
74        _ => None,
75    }
76}
77
78fn member_filter(params: &ListOrganizationMembersParams) -> Option<Filter> {
79    let field = params.filter_field.as_deref()?;
80    let value = params.filter_value.as_deref()?;
81    let operator = params.filter_operator.as_deref().unwrap_or("eq");
82    match member_column(field)? {
83        "created_at" => {
84            let Ok(parsed) = chrono::DateTime::parse_from_rfc3339(value) else {
85                return Some(Filter {
86                    column: "id",
87                    operator: " = ",
88                    value: "__better_auth_never_matches__".into(),
89                });
90            };
91            let parsed = parsed.with_timezone(&Utc);
92            let operator = match operator {
93                "eq" => " = ",
94                "ne" => " <> ",
95                "gt" => " > ",
96                "gte" => " >= ",
97                "lt" => " < ",
98                "lte" => " <= ",
99                _ => return None,
100            };
101            Some(Filter {
102                column: "created_at",
103                operator,
104                value: parsed.into(),
105            })
106        }
107        column => {
108            let (operator, value) = match operator {
109                "eq" => (" = ", value.to_owned()),
110                "ne" => (" <> ", value.to_owned()),
111                "contains" => (" LIKE ", format!("%{value}%")),
112                "gt" => (" > ", value.to_owned()),
113                "gte" => (" >= ", value.to_owned()),
114                "lt" => (" < ", value.to_owned()),
115                "lte" => (" <= ", value.to_owned()),
116                _ => return None,
117            };
118            Some(Filter {
119                column,
120                operator,
121                value: value.into(),
122            })
123        }
124    }
125}
126
127fn member_where(
128    binding: &Binding<Model>,
129    sql: &mut Sql,
130    organization_id: &str,
131    filter: Option<&Filter>,
132) -> AuthResult<()> {
133    sql.push(" WHERE ");
134    binding.compare(sql, "organization_id", " = ", organization_id)?;
135    if let Some(filter) = filter {
136        sql.push(" AND ");
137        binding.compare(sql, filter.column, filter.operator, filter.value.clone())?;
138    }
139    Ok(())
140}
141
142fn member_sort(
143    binding: &Binding<Model>,
144    sql: &mut Sql,
145    params: &ListOrganizationMembersParams,
146) -> AuthResult<()> {
147    let column = params
148        .sort_by
149        .as_deref()
150        .and_then(member_column)
151        .unwrap_or("created_at");
152    let descending = params.sort_by.as_deref().and_then(member_column).is_some()
153        && matches!(params.sort_direction.as_deref(), Some("desc"));
154    sql.push(" ORDER BY ");
155    binding.column(sql, column)?;
156    sql.push(if descending { " DESC" } else { " ASC" });
157    Ok(())
158}
159
160impl<S: AuthSchema> SqlxStore<S> {
161    async fn count_members(
162        &self,
163        organization_id: &str,
164        filter: Option<&Filter>,
165    ) -> AuthResult<i64> {
166        let mut sql = Sql::with(self.exec().engine(), "SELECT COUNT(*) FROM ");
167        sql.ident(self.organization_models.member.table());
168        member_where(
169            &self.organization_models.member,
170            &mut sql,
171            organization_id,
172            filter,
173        )?;
174        Ok(self
175            .exec()
176            .fetch_scalar::<i64>(sql)
177            .await?
178            .unwrap_or_default())
179    }
180}
181
182#[async_trait]
183impl<S> MemberStore for SqlxStore<S>
184where
185    S: AuthSchema + Send + Sync,
186{
187    async fn create_member(&self, member: CreateMember) -> AuthResult<Member> {
188        self.create_member_with_connection(self.exec(), member)
189            .await
190    }
191
192    async fn get_member(&self, organization_id: &str, user_id: &str) -> AuthResult<Option<Member>> {
193        let mut sql = self.organization_models.member.select(self.exec());
194        sql.push(" WHERE ");
195        self.organization_models.member.compare(
196            &mut sql,
197            "organization_id",
198            " = ",
199            organization_id,
200        )?;
201        sql.push(" AND ");
202        self.organization_models
203            .member
204            .compare(&mut sql, "user_id", " = ", user_id)?;
205        model::limit_one(&mut sql);
206        Ok(self
207            .organization_models
208            .member
209            .fetch_optional(self.exec(), sql)
210            .await?
211            .map(|model| Member::from(&model)))
212    }
213
214    async fn get_member_by_id(&self, id: &str) -> AuthResult<Option<Member>> {
215        Ok(self
216            .find_member_by_id(id)
217            .await?
218            .map(|model| Member::from(&model)))
219    }
220
221    async fn update_member_role(&self, member_id: &str, role: &str) -> AuthResult<Member> {
222        let Some(model) = self.find_member_by_id(member_id).await? else {
223            return Err(AuthError::not_found("Member not found"));
224        };
225        let mut active = model.into_active();
226        active.set("role", role);
227        self.organization_models
228            .member
229            .update(self.exec(), &active)
230            .await?
231            .map(|model| Member::from(&model))
232            .ok_or_else(crate::error::record_not_updated)
233    }
234
235    async fn update_member_role_if_present(
236        &self,
237        member_id: &str,
238        role: &str,
239    ) -> AuthResult<Option<Member>> {
240        let Some(model) = self.find_member_by_id(member_id).await? else {
241            return Ok(None);
242        };
243        let mut active = model.into_active();
244        active.set("role", role);
245        Ok(self
246            .organization_models
247            .member
248            .update(self.exec(), &active)
249            .await?
250            .map(|model| Member::from(&model)))
251    }
252
253    async fn delete_member(&self, member_id: &str) -> AuthResult<()> {
254        let member_id = member_id.to_owned();
255        self.in_transaction(true, async move |tx| {
256            let exec = Exec::tx(tx);
257            let mut select = self
258                .organization_models
259                .member
260                .by_id(exec, member_id.as_str())?;
261            model::limit_one(&mut select);
262            lock_exclusive(&mut select);
263            if let Some(member) = self
264                .organization_models
265                .member
266                .fetch_optional(exec, select)
267                .await?
268            {
269                super::teams::remove_owned_team_members(
270                    tx,
271                    &member.user_id,
272                    Some(&member.organization_id),
273                )
274                .await?;
275                _ = exec
276                    .execute(
277                        self.organization_models
278                            .member
279                            .delete_by_id(exec, member_id)?,
280                    )
281                    .await?;
282            }
283            Ok(())
284        })
285        .await
286    }
287
288    async fn list_organization_members(&self, org_id: &str) -> AuthResult<Vec<Member>> {
289        let mut sql = self.organization_models.member.select(self.exec());
290        member_where(&self.organization_models.member, &mut sql, org_id, None)?;
291        sql.push(" ORDER BY ");
292        self.organization_models
293            .member
294            .column(&mut sql, "created_at")?;
295        sql.push(" ASC");
296        Ok(self
297            .organization_models
298            .member
299            .fetch_all(self.exec(), sql)
300            .await?
301            .iter()
302            .map(Member::from)
303            .collect())
304    }
305
306    async fn delete_member_with_context(
307        &self,
308        member_id: &str,
309        organization_id: &str,
310        user_id: &str,
311        remove_team_members: bool,
312    ) -> AuthResult<()> {
313        let limit = i64::try_from(self.config().advanced.database.default_find_many_limit)
314            .map_err(|_error| AuthError::internal("Member page parameter exceeds u64"))?;
315        let (member_id, organization_id, user_id) = (
316            member_id.to_owned(),
317            organization_id.to_owned(),
318            user_id.to_owned(),
319        );
320        self.in_transaction(true, async move |tx| {
321            let exec = Exec::tx(tx);
322            _ = exec
323                .execute(
324                    self.organization_models
325                        .member
326                        .delete_by_id(exec, member_id)?,
327                )
328                .await?;
329            if remove_team_members {
330                let mut rooms = model::select_model::<team::Model>(exec);
331                rooms.push(" WHERE ");
332                rooms.compare(
333                    team::Model::TABLE,
334                    "organization_id",
335                    " = ",
336                    organization_id,
337                );
338                rooms.push(" LIMIT ");
339                rooms.bind(limit);
340                lock_exclusive(&mut rooms);
341                let rooms = exec.fetch_all::<team::Model>(rooms).await?;
342                super::teams::release_owned_team_members(tx, &user_id, rooms).await?;
343            }
344            Ok(())
345        })
346        .await
347    }
348
349    async fn list_organization_members_page(
350        &self,
351        organization_id: &str,
352        limit: usize,
353    ) -> AuthResult<Vec<Member>> {
354        let limit = i64::try_from(limit)
355            .map_err(|_error| AuthError::internal("Member page parameter exceeds u64"))?;
356        let mut sql = self.organization_models.member.select(self.exec());
357        member_where(
358            &self.organization_models.member,
359            &mut sql,
360            organization_id,
361            None,
362        )?;
363        sql.push(" LIMIT ");
364        sql.bind(limit);
365        Ok(self
366            .organization_models
367            .member
368            .fetch_all(self.exec(), sql)
369            .await?
370            .iter()
371            .map(Member::from)
372            .collect())
373    }
374
375    async fn query_organization_members(
376        &self,
377        params: &ListOrganizationMembersParams,
378    ) -> AuthResult<(Vec<Member>, usize)> {
379        let filter = member_filter(params);
380        let total = usize::try_from(
381            self.count_members(&params.organization_id, filter.as_ref())
382                .await?,
383        )
384        .map_err(|_error| AuthError::internal("Member count exceeds usize"))?;
385
386        let mut sql = self.organization_models.member.select(self.exec());
387        member_where(
388            &self.organization_models.member,
389            &mut sql,
390            &params.organization_id,
391            filter.as_ref(),
392        )?;
393        member_sort(&self.organization_models.member, &mut sql, params)?;
394        if let Some(limit) = params.limit {
395            sql.push(" LIMIT ");
396            sql.bind(
397                i64::try_from(limit)
398                    .map_err(|_error| AuthError::internal("Member page parameter exceeds u64"))?,
399            );
400        }
401        if let Some(offset) = params.offset {
402            sql.push(" OFFSET ");
403            sql.bind(
404                i64::try_from(offset)
405                    .map_err(|_error| AuthError::internal("Member page parameter exceeds u64"))?,
406            );
407        }
408        Ok((
409            self.organization_models
410                .member
411                .fetch_all(self.exec(), sql)
412                .await?
413                .iter()
414                .map(Member::from)
415                .collect(),
416            total,
417        ))
418    }
419
420    async fn query_organization_members_page(
421        &self,
422        params: &MemberPageQuery,
423    ) -> AuthResult<(Vec<Member>, usize)> {
424        let legacy_filter = ListOrganizationMembersParams {
425            organization_id: params.organization_id.clone(),
426            sort_by: params.sort_by.clone(),
427            sort_direction: params.sort_direction.clone(),
428            filter_field: params.filter_field.clone(),
429            filter_value: params.filter_value.clone(),
430            filter_operator: params.filter_operator.clone(),
431            ..Default::default()
432        };
433        let filter = member_filter(&legacy_filter);
434        let total = usize::try_from(
435            self.count_members(&params.organization_id, filter.as_ref())
436                .await?,
437        )
438        .map_err(|error| AuthError::Internal(error.to_string()))?;
439        let mut sql = self.organization_models.member.select(self.exec());
440        member_where(
441            &self.organization_models.member,
442            &mut sql,
443            &params.organization_id,
444            filter.as_ref(),
445        )?;
446        if params.sort_by.as_deref().and_then(member_column).is_some() {
447            member_sort(&self.organization_models.member, &mut sql, &legacy_filter)?;
448        }
449        super::bind_page(&mut sql, params.limit, params.offset);
450        Ok((
451            self.organization_models
452                .member
453                .fetch_all(self.exec(), sql)
454                .await?
455                .iter()
456                .map(Member::from)
457                .collect(),
458            total,
459        ))
460    }
461
462    async fn count_organization_members(&self, org_id: &str) -> AuthResult<i64> {
463        self.count_members(org_id, None).await
464    }
465
466    async fn count_organization_owners(&self, org_id: &str) -> AuthResult<i64> {
467        let filter = Filter {
468            column: "role",
469            operator: " = ",
470            value: "owner".into(),
471        };
472        self.count_members(org_id, Some(&filter)).await
473    }
474}
alibi_sqlx/store/members.rs

alibi_sqlx/store/
members.rs
