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

alibi_sqlx/store/
members.rs
