alibi_sqlx/store/
organizations.rs

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

alibi_sqlx/store/
organizations.rs
