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

alibi_sqlx/store/
organizations.rs
