alibi_sqlx/
organization_models.rs

1//! Application-owned organization plugin tables. Models derive [`crate::SqlxModel`]
2//! and use Rust names for plugin fields, with `sqlx(rename)` for physical names.
3
4use crate::model::{self, ActiveRow, SqlxModel};
5use crate::pool::Exec;
6use crate::sql::Sql;
7use crate::store::entities::{invitation, member, organization};
8use crate::value::{ColumnKind, SqlValue};
9use alibi_core::field_policy::FieldValues;
10use alibi_core::{AuthError, AuthResult, Invitation, Member, Organization};
11use async_trait::async_trait;
12use std::{marker::PhantomData, sync::Arc};
13
14/// Bind the three organization entities to application models and ID factories.
15/// This affects every plugin operation, including invitation transactions.
16#[derive(Clone)]
17pub struct OrganizationModels {
18    pub(crate) organization: Binding<organization::Model>,
19    pub(crate) member: Binding<member::Model>,
20    pub(crate) invitation: Binding<invitation::Model>,
21}
22
23impl OrganizationModels {
24    /// Models must declare the plugin's Rust field names. `updated_at` on the
25    /// organization and `team_id` on the invitation may be omitted. Additional
26    /// organization fields are retained and returned by the plugin.
27    #[must_use]
28    pub fn new<O: SqlxModel, M: SqlxModel, I: SqlxModel>(
29        organization_id: fn() -> String,
30        member_id: fn() -> String,
31        invitation_id: fn() -> String,
32    ) -> Self {
33        Self {
34            organization: Binding::new::<O>(organization_id),
35            member: Binding::new::<M>(member_id),
36            invitation: Binding::new::<I>(invitation_id),
37        }
38    }
39}
40
41impl Default for OrganizationModels {
42    fn default() -> Self {
43        fn id() -> String {
44            uuid::Uuid::new_v4().to_string()
45        }
46        Self::new::<organization::Model, member::Model, invitation::Model>(id, id, id)
47    }
48}
49
50#[async_trait]
51trait Backend: Send + Sync {
52    fn table(&self) -> &'static str;
53    fn columns(&self) -> &'static [&'static str];
54    fn fields(&self) -> &'static [(&'static str, &'static str)];
55    fn kind(&self, physical: &str) -> ColumnKind;
56    async fn fetch(&self, exec: Exec<'_>, sql: Sql) -> AuthResult<Vec<ActiveRow>>;
57    async fn insert(&self, exec: Exec<'_>, row: &ActiveRow) -> AuthResult<ActiveRow>;
58    async fn update(&self, exec: Exec<'_>, row: &ActiveRow) -> AuthResult<Option<ActiveRow>>;
59}
60struct TypedBackend<M>(PhantomData<M>);
61#[async_trait]
62impl<M: SqlxModel> Backend for TypedBackend<M> {
63    fn table(&self) -> &'static str {
64        M::TABLE
65    }
66    fn columns(&self) -> &'static [&'static str] {
67        M::COLUMN_NAMES
68    }
69    fn fields(&self) -> &'static [(&'static str, &'static str)] {
70        M::FIELD_COLUMNS
71    }
72    fn kind(&self, physical: &str) -> ColumnKind {
73        M::column_kind(physical)
74    }
75    async fn fetch(&self, exec: Exec<'_>, sql: Sql) -> AuthResult<Vec<ActiveRow>> {
76        Ok(exec
77            .fetch_all::<M>(sql)
78            .await?
79            .into_iter()
80            .map(SqlxModel::into_active)
81            .collect())
82    }
83    async fn insert(&self, exec: Exec<'_>, row: &ActiveRow) -> AuthResult<ActiveRow> {
84        Ok(model::insert::<M>(exec, row).await?.into_active())
85    }
86    async fn update(&self, exec: Exec<'_>, row: &ActiveRow) -> AuthResult<Option<ActiveRow>> {
87        Ok(model::update::<M>(exec, row)
88            .await?
89            .map(SqlxModel::into_active))
90    }
91}
92
93#[derive(Clone)]
94pub(crate) struct Binding<T> {
95    backend: Arc<dyn Backend>,
96    generate_id: fn() -> String,
97    _canonical: PhantomData<T>,
98}
99
100/// Canonical plugin fields and the original loaded write state. Keeping that
101/// state prevents an update from overwriting unrelated application columns.
102pub(crate) struct Row<T> {
103    value: T,
104    active: ActiveRow,
105}
106impl<T> std::ops::Deref for Row<T> {
107    type Target = T;
108    fn deref(&self) -> &T {
109        &self.value
110    }
111}
112impl<T> Row<T> {
113    pub(crate) fn active_role(&self) -> SqlValue {
114        self.active
115            .get("role")
116            .and_then(crate::ActiveValue::value)
117            .cloned()
118            .unwrap_or(SqlValue::Text(None))
119    }
120    pub(crate) fn into_active(self) -> ActiveRow {
121        self.active
122    }
123}
124
125impl<T: SqlxModel> Binding<T> {
126    fn new<M: SqlxModel>(generate_id: fn() -> String) -> Self {
127        Self {
128            backend: Arc::new(TypedBackend::<M>(PhantomData)),
129            generate_id,
130            _canonical: PhantomData,
131        }
132    }
133    pub(crate) async fn set_fields(
134        &self,
135        exec: Exec<'_>,
136        active: &mut ActiveRow,
137        fields: &mut FieldValues,
138    ) -> AuthResult<()> {
139        fields.apply_adapter_transforms_async().await?;
140        for (name, value) in &*fields {
141            let requested = fields.binding_name(name);
142            let column = self.physical(requested).ok_or_else(|| {
143                AuthError::config(format!(
144                    "configured organization field {name} has no model column"
145                ))
146            })?;
147            let field = self.logical(column);
148            if T::COLUMN_NAMES.contains(&field) {
149                return Err(AuthError::config(
150                    "additional organization fields cannot overwrite plugin fields",
151                ));
152            }
153            let value = crate::additional_fields::prepare_value(
154                exec,
155                self.backend.kind(column),
156                crate::additional_fields::raw_value(value)?,
157            )
158            .await?;
159            active.set(field, value);
160        }
161        Ok(())
162    }
163    pub(crate) fn new_id(&self) -> String {
164        (self.generate_id)()
165    }
166    pub(crate) fn table(&self) -> &'static str {
167        self.backend.table()
168    }
169    pub(crate) fn physical(&self, field: &str) -> Option<&'static str> {
170        self.backend
171            .fields()
172            .iter()
173            .find(|(logical, _)| *logical == field)
174            .map(|(_, physical)| *physical)
175            .or_else(|| {
176                self.backend
177                    .columns()
178                    .iter()
179                    .copied()
180                    .find(|column| *column == field)
181            })
182    }
183    fn logical(&self, physical: &'static str) -> &'static str {
184        self.backend
185            .fields()
186            .iter()
187            .find(|(_, column)| *column == physical)
188            .map_or(physical, |(logical, _)| *logical)
189    }
190    fn value(&self, field: &str, value: SqlValue) -> AuthResult<SqlValue> {
191        let column = self.physical(field).ok_or_else(|| {
192            AuthError::internal(format!("organization model has no {field} column"))
193        })?;
194        Ok(match (self.backend.kind(column), value) {
195            (ColumnKind::BpChar, SqlValue::Text(value)) => SqlValue::BpChar(value),
196            (ColumnKind::NaiveTimestamp, SqlValue::Timestamp(value)) => {
197                SqlValue::NaiveTimestamp(value.map(|time| time.naive_utc()))
198            }
199            (ColumnKind::Text, SqlValue::Json(value)) => SqlValue::Text(
200                value
201                    .map(|value| alibi_core::utils::json::to_string(&*value))
202                    .transpose()?,
203            ),
204            (_, value) => value,
205        })
206    }
207    fn stage(&self, active: &ActiveRow) -> AuthResult<ActiveRow> {
208        let mut physical = ActiveRow::new();
209        for (field, value) in active.present() {
210            let Some(column) = self.physical(field) else {
211                if field == "updated_at" || (field == "team_id" && value.is_null()) {
212                    continue;
213                }
214                return Err(AuthError::internal(format!(
215                    "organization model has no {field} column"
216                )));
217            };
218            let value = self.value(field, value.clone())?;
219            if active.get(field).is_some_and(crate::ActiveValue::is_set) {
220                physical.set(column, value);
221            } else {
222                physical.unchanged(column, value);
223            }
224        }
225        Ok(physical)
226    }
227    fn row(&self, physical: &ActiveRow) -> AuthResult<Row<T>> {
228        let mut active = ActiveRow::new();
229        for (column, value) in physical.present() {
230            let field = self.logical(column);
231            let value = match value.clone() {
232                SqlValue::NaiveTimestamp(value) => {
233                    SqlValue::Timestamp(value.map(|time| time.and_utc()))
234                }
235                SqlValue::BpChar(value) => SqlValue::Text(value),
236                value => value,
237            };
238            active.unchanged(field, value);
239        }
240        let mut canonical = active.clone();
241        if T::TABLE == organization::Model::TABLE {
242            if self.physical("updated_at").is_none()
243                && let Some(time) = active.get("created_at").and_then(crate::ActiveValue::value)
244            {
245                canonical.unchanged("updated_at", time.clone());
246            }
247            if canonical
248                .get("metadata")
249                .and_then(crate::ActiveValue::value)
250                .is_some_and(SqlValue::is_null)
251            {
252                canonical.unchanged("metadata", SqlValue::Json(None));
253            }
254        }
255        if T::TABLE == invitation::Model::TABLE && self.physical("team_id").is_none() {
256            canonical.unchanged("team_id", None::<String>);
257        }
258        let mut model = ActiveRow::new();
259        for (logical, physical) in T::FIELD_COLUMNS {
260            if let Some(value) = canonical.get(logical).and_then(crate::ActiveValue::value) {
261                model.unchanged(physical, value.clone());
262            }
263        }
264        Ok(Row {
265            value: T::from_active(model)?,
266            active,
267        })
268    }
269    pub(crate) fn select(&self, exec: Exec<'_>) -> Sql {
270        crate::sql::select(exec.engine(), self.table(), self.backend.columns())
271    }
272    pub(crate) fn by_id(&self, exec: Exec<'_>, id: impl Into<SqlValue>) -> AuthResult<Sql> {
273        let mut sql = self.select(exec);
274        sql.push(" WHERE ");
275        self.compare(&mut sql, "id", " = ", id)?;
276        Ok(sql)
277    }
278    pub(crate) fn delete_for_organization(&self, exec: Exec<'_>, id: &str) -> AuthResult<Sql> {
279        let mut sql = Sql::with(exec.engine(), "DELETE FROM ");
280        sql.ident(self.table());
281        sql.push(" WHERE ");
282        self.compare(&mut sql, "organization_id", " = ", id)?;
283        Ok(sql)
284    }
285    pub(crate) fn delete_by_id(&self, exec: Exec<'_>, id: impl Into<SqlValue>) -> AuthResult<Sql> {
286        let mut sql = Sql::with(exec.engine(), "DELETE FROM ");
287        sql.ident(self.table());
288        sql.push(" WHERE ");
289        self.compare(&mut sql, "id", " = ", id)?;
290        Ok(sql)
291    }
292    pub(crate) fn column(&self, sql: &mut Sql, field: &str) -> AuthResult<()> {
293        let column = self.physical(field).ok_or_else(|| {
294            AuthError::internal(format!("organization model has no {field} column"))
295        })?;
296        sql.column(self.table(), column);
297        Ok(())
298    }
299    pub(crate) fn compare(
300        &self,
301        sql: &mut Sql,
302        field: &str,
303        operator: &str,
304        value: impl Into<SqlValue>,
305    ) -> AuthResult<()> {
306        self.column(sql, field)?;
307        sql.push(operator);
308        sql.bind(self.value(field, value.into())?);
309        Ok(())
310    }
311    pub(crate) fn assign(
312        &self,
313        sql: &mut Sql,
314        field: &str,
315        value: impl Into<SqlValue>,
316    ) -> AuthResult<()> {
317        let column = self.physical(field).ok_or_else(|| {
318            AuthError::internal(format!("organization model has no {field} column"))
319        })?;
320        sql.assign(column, self.value(field, value.into())?);
321        Ok(())
322    }
323    pub(crate) fn bind_list<I: IntoIterator<Item = String>>(
324        &self,
325        sql: &mut Sql,
326        field: &str,
327        values: I,
328    ) -> AuthResult<()> {
329        sql.bind_list(
330            values
331                .into_iter()
332                .map(|value| self.value(field, value.into()))
333                .collect::<AuthResult<Vec<_>>>()?,
334        );
335        Ok(())
336    }
337    pub(crate) fn returning(&self, sql: &mut Sql) {
338        sql.push(" RETURNING ");
339        sql.column_list(self.backend.columns());
340    }
341    pub(crate) async fn fetch_all(&self, exec: Exec<'_>, sql: Sql) -> AuthResult<Vec<Row<T>>> {
342        self.backend
343            .fetch(exec, sql)
344            .await?
345            .into_iter()
346            .map(|active| self.row(&active))
347            .collect()
348    }
349    pub(crate) async fn fetch_optional(
350        &self,
351        exec: Exec<'_>,
352        sql: Sql,
353    ) -> AuthResult<Option<Row<T>>> {
354        Ok(self.fetch_all(exec, sql).await?.into_iter().next())
355    }
356    pub(crate) async fn insert(&self, exec: Exec<'_>, active: &ActiveRow) -> AuthResult<Row<T>> {
357        self.row(&self.backend.insert(exec, &self.stage(active)?).await?)
358    }
359    pub(crate) async fn update(
360        &self,
361        exec: Exec<'_>,
362        active: &ActiveRow,
363    ) -> AuthResult<Option<Row<T>>> {
364        self.backend
365            .update(exec, &self.stage(active)?)
366            .await?
367            .map(|active| self.row(&active))
368            .transpose()
369    }
370}
371
372impl From<&Row<organization::Model>> for Organization {
373    fn from(row: &Row<organization::Model>) -> Self {
374        let mut organization = Self::from(&row.value);
375        for (field, value) in row.active.present() {
376            if organization::Model::COLUMN_NAMES.contains(&field) {
377                continue;
378            }
379            if let Some(value) = output_value(value) {
380                _ = organization
381                    .additional_fields
382                    .insert(field.to_owned(), value);
383            }
384        }
385        organization
386    }
387}
388impl From<&Row<member::Model>> for Member {
389    fn from(row: &Row<member::Model>) -> Self {
390        Self::from(&row.value)
391    }
392}
393impl From<&Row<invitation::Model>> for Invitation {
394    fn from(row: &Row<invitation::Model>) -> Self {
395        let mut invitation = Self::from(&row.value);
396        if row
397            .active
398            .get("role")
399            .and_then(crate::ActiveValue::value)
400            .is_some_and(SqlValue::is_null)
401        {
402            invitation.role = None;
403        }
404        invitation
405    }
406}
407
408fn output_value(value: &SqlValue) -> Option<serde_json::Value> {
409    if value.is_null() {
410        return Some(serde_json::Value::Null);
411    }
412    match value {
413        SqlValue::Text(Some(value)) | SqlValue::BpChar(Some(value)) => Some(value.clone().into()),
414        SqlValue::Bool(Some(value)) => Some((*value).into()),
415        SqlValue::Int(Some(value)) => Some((*value).into()),
416        SqlValue::BigInt(Some(value)) => Some((*value).into()),
417        SqlValue::Float(Some(value)) => {
418            serde_json::Number::from_f64(f64::from(*value)).map(Into::into)
419        }
420        SqlValue::Double(Some(value)) => serde_json::Number::from_f64(*value).map(Into::into),
421        SqlValue::Json(Some(value)) => Some((**value).clone()),
422        _ => None,
423    }
424}
alibi_sqlx/organization_models.rs

alibi_sqlx/
organization_models.rs
