alibi_sqlx/store/
organizations.rs

1use super::SqlxStore;
2use super::entities::organization::{JsonMetadata, Model};
3use crate::SqlValue;
4use crate::error::record_not_updated;
5use crate::model::{self, ActiveRow};
6use crate::organization_models::Row;
7use crate::pool::{Engine, Exec};
8use crate::schema::AuthSchema;
9use crate::sql::Sql;
10use crate::value::SqlxValue;
11use alibi_core::error::AuthResult;
12use alibi_core::store::OrganizationStore;
13use alibi_core::{CreateOrganization, Organization, UpdateOrganization};
14use async_trait::async_trait;
15use chrono::Utc;
16use std::collections::HashMap;
17
18#[async_trait]
19impl<S> OrganizationStore for SqlxStore<S>
20where
21    S: AuthSchema + Send + Sync,
22{
23    async fn create_organization(&self, mut org: CreateOrganization) -> AuthResult<Organization> {
24        let now = Utc::now();
25        let metadata = org
26            .metadata
27            .map(|metadata| {
28                JsonMetadata::for_backend(
29                    alibi_core::utils::json::to_value(&metadata)?,
30                    self.exec().engine(),
31                )
32            })
33            .transpose()?;
34        let mut active = ActiveRow::new();
35        active.set(
36            "id",
37            match org.id {
38                Some(id) => id,
39                None => self
40                    .generated_id(
41                        self.exec(),
42                        "organization",
43                        self.organization_models.organization.table(),
44                        self.organization_models
45                            .organization
46                            .physical("id")
47                            .ok_or_else(|| {
48                                alibi_core::AuthError::config("Organization model has no ID column")
49                            })?,
50                    )
51                    .await?
52                    .unwrap_or_else(|| self.organization_models.organization.new_id()),
53            },
54        );
55        active.set("name", org.name);
56        active.set("slug", org.slug);
57        active.set("logo", org.logo);
58        active.set("metadata", metadata.into_sql_value());
59        active.set("created_at", now);
60        active.set("updated_at", now);
61        self.organization_models
62            .organization
63            .set_fields(self.exec(), &mut active, &mut org.additional_fields)
64            .await?;
65        self.organization_models
66            .organization
67            .insert(self.exec(), &active)
68            .await
69            .map(|model| Organization::from(&model))
70    }
71
72    async fn get_organization_by_id(&self, id: &str) -> AuthResult<Option<Organization>> {
73        let mut sql = self
74            .organization_models
75            .organization
76            .by_id(self.exec(), id)?;
77        model::limit_one(&mut sql);
78        Ok(self
79            .organization_models
80            .organization
81            .fetch_optional(self.exec(), sql)
82            .await?
83            .map(|model| Organization::from(&model)))
84    }
85
86    async fn get_organization_by_slug(&self, slug: &str) -> AuthResult<Option<Organization>> {
87        let mut sql = self.organization_models.organization.select(self.exec());
88        sql.push(" WHERE ");
89        self.organization_models
90            .organization
91            .compare(&mut sql, "slug", " = ", slug)?;
92        model::limit_one(&mut sql);
93        Ok(self
94            .organization_models
95            .organization
96            .fetch_optional(self.exec(), sql)
97            .await?
98            .map(|model| Organization::from(&model)))
99    }
100
101    async fn list_organizations_by_ids(&self, ids: &[String]) -> AuthResult<Vec<Organization>> {
102        if ids.is_empty() {
103            return Ok(Vec::new());
104        }
105        let mut sql = self.organization_models.organization.select(self.exec());
106        sql.push(" WHERE ");
107        self.organization_models
108            .organization
109            .column(&mut sql, "id")?;
110        sql.push(" IN ");
111        self.organization_models
112            .organization
113            .bind_list(&mut sql, "id", ids.iter().cloned())?;
114        Ok(self
115            .organization_models
116            .organization
117            .fetch_all(self.exec(), sql)
118            .await?
119            .iter()
120            .map(Organization::from)
121            .collect())
122    }
123
124    async fn update_organization(
125        &self,
126        id: &str,
127        update: UpdateOrganization,
128    ) -> AuthResult<Organization> {
129        let mut sql = self
130            .organization_models
131            .organization
132            .by_id(self.exec(), id)?;
133        model::limit_one(&mut sql);
134        let Some(model) = self
135            .organization_models
136            .organization
137            .fetch_optional(self.exec(), sql)
138            .await?
139        else {
140            return Err(alibi_core::error::AuthError::not_found(
141                "Organization not found",
142            ));
143        };
144        let mut fields = update.additional_fields.clone();
145        let mut active = apply_organization_update(model, update, self.exec().engine())?;
146        self.organization_models
147            .organization
148            .set_fields(self.exec(), &mut active, &mut fields)
149            .await?;
150        self.organization_models
151            .organization
152            .update(self.exec(), &active)
153            .await?
154            .map(|model| Organization::from(&model))
155            .ok_or_else(record_not_updated)
156    }
157
158    async fn patch_organization_if_present(
159        &self,
160        id: &str,
161        mut update: UpdateOrganization,
162    ) -> AuthResult<Option<Organization>> {
163        let backend = self.exec().engine();
164        let mut assignments: Vec<(&str, SqlValue)> = Vec::new();
165        if let Some(name) = update.name {
166            assignments.push(("name", name.into()));
167        }
168        if let Some(slug) = update.slug {
169            assignments.push(("slug", slug.into()));
170        }
171        if let Some(logo) = update.logo {
172            assignments.push(("logo", logo.into()));
173        }
174        if let Some(metadata) = update.metadata {
175            let metadata =
176                JsonMetadata::for_backend(alibi_core::utils::json::to_value(&metadata)?, backend)?;
177            assignments.push(("metadata", metadata.into_sql_value()));
178        }
179        let mut additional = ActiveRow::new();
180        self.organization_models
181            .organization
182            .set_fields(self.exec(), &mut additional, &mut update.additional_fields)
183            .await?;
184        assignments.extend(
185            additional
186                .changed()
187                .map(|(name, value)| (name, value.clone())),
188        );
189        // An empty patch is sent to the database unchanged, which rejects it.
190        let mut sql = Sql::with(backend, "UPDATE ");
191        sql.ident(self.organization_models.organization.table());
192        sql.push(" SET ");
193        for (index, (column, value)) in assignments.into_iter().enumerate() {
194            if index > 0 {
195                sql.push(", ");
196            }
197            self.organization_models
198                .organization
199                .assign(&mut sql, column, value)?;
200        }
201        sql.push(" WHERE ");
202        self.organization_models
203            .organization
204            .assign(&mut sql, "id", id)?;
205        self.organization_models.organization.returning(&mut sql);
206        Ok(self
207            .organization_models
208            .organization
209            .fetch_optional(self.exec(), sql)
210            .await?
211            .as_ref()
212            .map(Organization::from))
213    }
214
215    async fn update_organization_if_present(
216        &self,
217        id: &str,
218        update: UpdateOrganization,
219    ) -> AuthResult<Option<Organization>> {
220        let mut sql = self
221            .organization_models
222            .organization
223            .by_id(self.exec(), id)?;
224        model::limit_one(&mut sql);
225        let Some(model) = self
226            .organization_models
227            .organization
228            .fetch_optional(self.exec(), sql)
229            .await?
230        else {
231            return Ok(None);
232        };
233        let mut fields = update.additional_fields.clone();
234        let mut active = apply_organization_update(model, update, self.exec().engine())?;
235        self.organization_models
236            .organization
237            .set_fields(self.exec(), &mut active, &mut fields)
238            .await?;
239        Ok(self
240            .organization_models
241            .organization
242            .update(self.exec(), &active)
243            .await?
244            .map(|model| Organization::from(&model)))
245    }
246
247    async fn delete_organization(&self, id: &str) -> AuthResult<()> {
248        let id = id.to_owned();
249        self.in_transaction(false, async move |tx| {
250            let exec = Exec::tx(tx);
251            for query in [
252                self.organization_models
253                    .member
254                    .delete_for_organization(exec, &id)?,
255                self.organization_models
256                    .invitation
257                    .delete_for_organization(exec, &id)?,
258            ] {
259                _ = exec.execute(query).await?;
260            }
261            _ = exec
262                .execute(
263                    self.organization_models
264                        .organization
265                        .delete_by_id(exec, id)?,
266                )
267                .await?;
268            Ok(())
269        })
270        .await
271    }
272
273    async fn list_user_organizations(&self, user_id: &str) -> AuthResult<Vec<Organization>> {
274        let mut members = self.organization_models.member.select(self.exec());
275        members.push(" WHERE ");
276        self.organization_models
277            .member
278            .compare(&mut members, "user_id", " = ", user_id)?;
279        members.push(" LIMIT ");
280        members.bind(self.find_many_limit());
281        let member_models = self
282            .organization_models
283            .member
284            .fetch_all(self.exec(), members)
285            .await?;
286
287        if member_models.is_empty() {
288            return Ok(Vec::new());
289        }
290
291        let mut sql = self.organization_models.organization.select(self.exec());
292        sql.push(" WHERE ");
293        self.organization_models
294            .organization
295            .column(&mut sql, "id")?;
296        sql.push(" IN ");
297        self.organization_models.organization.bind_list(
298            &mut sql,
299            "id",
300            member_models
301                .iter()
302                .map(|member| member.organization_id.clone()),
303        )?;
304        let organizations: HashMap<String, Organization> = self
305            .organization_models
306            .organization
307            .fetch_all(self.exec(), sql)
308            .await?
309            .into_iter()
310            .map(|model| (model.id.clone(), Organization::from(&model)))
311            .collect();
312        // Source maps the member page's joined organizations: repeated
313        // memberships repeat the organization, and missing joins are omitted.
314        Ok(member_models
315            .into_iter()
316            .filter_map(|member| organizations.get(&member.organization_id).cloned())
317            .collect())
318    }
319}
320
321fn apply_organization_update(
322    model: Row<Model>,
323    update: UpdateOrganization,
324    backend: Engine,
325) -> AuthResult<ActiveRow> {
326    let mut active = model.into_active();
327    if let Some(name) = update.name {
328        active.set("name", name);
329    }
330    if let Some(slug) = update.slug {
331        active.set("slug", slug);
332    }
333    if let Some(logo) = update.logo {
334        active.set("logo", logo);
335    }
336    if let Some(metadata) = update.metadata {
337        active.set(
338            "metadata",
339            Some(JsonMetadata::for_backend(
340                alibi_core::utils::json::to_value(&metadata)?,
341                backend,
342            )?)
343            .into_sql_value(),
344        );
345    }
346    active.set("updated_at", Utc::now());
347
348    Ok(active)
349}
alibi_sqlx/store/organizations.rs

alibi_sqlx/store/
organizations.rs
