shared/models/
mod.rs

1use crate::database::DatabaseError;
2use compact_str::CompactStringExt;
3use futures_util::{StreamExt, TryStreamExt};
4use garde::Validate;
5use parking_lot::RwLock;
6use serde::{Deserialize, Serialize, de::DeserializeOwned};
7use sqlx::{
8    Arguments, Postgres, QueryBuilder, Row,
9    encode::IsNull,
10    error::BoxDynError,
11    postgres::{PgArgumentBuffer, PgArguments, PgRow, PgTypeInfo},
12};
13use std::{
14    collections::{BTreeMap, HashSet},
15    marker::PhantomData,
16    pin::Pin,
17    sync::{Arc, LazyLock},
18};
19use utoipa::ToSchema;
20
21pub mod admin_activity;
22pub mod announcement;
23pub mod backup_configuration;
24pub mod database_agent_host;
25pub mod database_agent_template;
26pub mod database_host;
27pub mod egg_configuration;
28pub mod egg_repository;
29pub mod egg_repository_egg;
30pub mod location;
31pub mod location_database_agent_host;
32pub mod location_database_host;
33pub mod mount;
34pub mod nest;
35pub mod nest_egg;
36pub mod nest_egg_mount;
37pub mod nest_egg_variable;
38pub mod node;
39pub mod node_allocation;
40pub mod node_database_agent_host;
41pub mod node_database_host;
42pub mod node_mount;
43pub mod oauth_provider;
44pub mod oauth_provider_mapping;
45pub mod role;
46pub mod server;
47pub mod server_activity;
48pub mod server_allocation;
49pub mod server_backup;
50pub mod server_backup_group;
51pub mod server_database;
52pub mod server_database_instance;
53pub mod server_mount;
54pub mod server_schedule;
55pub mod server_schedule_step;
56pub mod server_subuser;
57pub mod server_variable;
58pub mod system_backup_policy;
59pub mod system_backup_policy_location;
60pub mod system_backup_policy_node;
61pub mod system_backup_policy_server;
62pub mod user;
63pub mod user_activity;
64pub mod user_api_key;
65pub mod user_command_snippet;
66pub mod user_email_verification;
67pub mod user_oauth_link;
68pub mod user_password_reset;
69pub mod user_recovery_code;
70pub mod user_security_key;
71pub mod user_server_group;
72pub mod user_session;
73pub mod user_ssh_key;
74pub mod user_two_factor_code;
75
76#[derive(ToSchema, Validate, Deserialize, Serialize)]
77pub struct PaginationParams {
78    #[garde(range(min = 1))]
79    #[schema(minimum = 1)]
80    #[serde(default = "Pagination::default_page")]
81    pub page: i64,
82    #[garde(range(min = 1, max = 100))]
83    #[schema(minimum = 1, maximum = 100)]
84    #[serde(default = "Pagination::default_per_page")]
85    pub per_page: i64,
86}
87
88#[derive(ToSchema, Validate, Deserialize, Serialize)]
89pub struct PaginationParamsWithSearch {
90    #[garde(range(min = 1))]
91    #[schema(minimum = 1)]
92    #[serde(default = "Pagination::default_page")]
93    pub page: i64,
94    #[garde(range(min = 1, max = 100))]
95    #[schema(minimum = 1, maximum = 100)]
96    #[serde(default = "Pagination::default_per_page")]
97    pub per_page: i64,
98    #[garde(length(chars, min = 1, max = 128))]
99    #[schema(min_length = 1, max_length = 128)]
100    #[serde(
101        default,
102        deserialize_with = "crate::deserialize::deserialize_string_option"
103    )]
104    pub search: Option<compact_str::CompactString>,
105}
106
107#[derive(ToSchema, Deserialize, Serialize)]
108pub struct Pagination<T: Serialize = serde_json::Value> {
109    pub total: i64,
110    pub per_page: i64,
111    pub page: i64,
112
113    pub data: Vec<T>,
114}
115
116impl Pagination {
117    #[inline]
118    pub const fn default_page() -> i64 {
119        1
120    }
121
122    #[inline]
123    pub const fn default_per_page() -> i64 {
124        25
125    }
126}
127
128impl<T: Serialize> Pagination<T> {
129    pub async fn async_map<R: serde::Serialize, Fut: Future<Output = R>>(
130        self,
131        mapper: impl Fn(T) -> Fut,
132    ) -> Pagination<R> {
133        let mut results = Vec::new();
134        results.reserve_exact(self.data.len());
135        let mut result_stream =
136            futures_util::stream::iter(self.data.into_iter().map(mapper)).buffered(25);
137
138        while let Some(result) = result_stream.next().await {
139            results.push(result);
140        }
141
142        Pagination {
143            total: self.total,
144            per_page: self.per_page,
145            page: self.page,
146            data: results,
147        }
148    }
149
150    pub async fn try_async_map<R: serde::Serialize, E, Fut: Future<Output = Result<R, E>>>(
151        self,
152        mapper: impl Fn(T) -> Fut,
153    ) -> Result<Pagination<R>, E> {
154        let mut results = Vec::new();
155        results.reserve_exact(self.data.len());
156        let mut result_stream =
157            futures_util::stream::iter(self.data.into_iter().map(mapper)).buffered(25);
158
159        while let Some(result) = result_stream.try_next().await? {
160            results.push(result);
161        }
162
163        Ok(Pagination {
164            total: self.total,
165            per_page: self.per_page,
166            page: self.page,
167            data: results,
168        })
169    }
170}
171
172pub type ModelExtensionList = parking_lot::RwLock<Vec<Box<dyn ModelExtension + Send + Sync>>>;
173pub type ModelExtensionData = Vec<(compact_str::CompactString, Vec<u8>)>;
174pub type ModelExtensionMapType = Box<dyn erased_serde::Serialize>;
175
176pub trait ModelExtension {
177    fn extension_name(&self) -> &'static str;
178
179    fn extended_columns(&self, prefix: &str) -> BTreeMap<&'static str, compact_str::CompactString>;
180
181    fn map_extended(
182        &self,
183        prefix: &str,
184        row: &PgRow,
185    ) -> Result<ModelExtensionMapType, crate::database::DatabaseError>;
186}
187
188pub trait SafeModelExtension: ModelExtension {
189    type Value: Serialize + DeserializeOwned;
190
191    fn name() -> &'static str;
192}
193
194pub trait BaseModel: Serialize + DeserializeOwned {
195    const NAME: &'static str;
196
197    fn get_extension_list() -> &'static ModelExtensionList;
198    fn get_extension_data(&self) -> &ModelExtensionData;
199
200    /// Registers a model extension. If an extension with the same name is already registered, this function will do nothing.
201    fn register_model_extension(extension: impl ModelExtension + Send + Sync + 'static) {
202        let mut extensions = Self::get_extension_list().write();
203
204        if extensions
205            .iter()
206            .any(|e| e.extension_name() == extension.extension_name())
207        {
208            return;
209        }
210
211        extensions.push(Box::new(extension));
212    }
213
214    /// Parses a model extension from the model's extension data. If the extension is not found, or if the data cannot be deserialized, an error is returned.
215    ///
216    /// This can be costly depending on what is stored, so use sparingly.
217    fn parse_model_extension<Extension: SafeModelExtension>(
218        &self,
219    ) -> Result<Extension::Value, crate::database::DatabaseError>
220    where
221        Extension::Value: Serialize + DeserializeOwned,
222    {
223        let data = self.get_extension_data();
224
225        for (name, value) in data.iter() {
226            if name.as_str() == Extension::name() {
227                let deserialized =
228                    rmp_serde::from_slice::<Extension::Value>(value).map_err(anyhow::Error::new)?;
229
230                return Ok(deserialized);
231            }
232        }
233
234        Err(crate::database::DatabaseError::Any(anyhow::anyhow!(
235            "model extension not found"
236        )))
237    }
238
239    fn base_columns(prefix: Option<&str>) -> BTreeMap<&'static str, compact_str::CompactString>;
240    fn columns(prefix: Option<&str>) -> BTreeMap<&'static str, compact_str::CompactString> {
241        let extensions = Self::get_extension_list().read();
242
243        let mut columns = Self::base_columns(prefix);
244
245        for extension in extensions.iter() {
246            columns.extend(extension.extended_columns(prefix.unwrap_or_default()));
247        }
248
249        columns
250    }
251
252    #[inline]
253    fn columns_sql(prefix: Option<&str>) -> compact_str::CompactString {
254        Self::columns(prefix)
255            .iter()
256            .map(|(key, value)| compact_str::format_compact!("{key} as {value}"))
257            .join_compact(", ")
258    }
259
260    fn map_extensions(
261        prefix: &str,
262        row: &PgRow,
263    ) -> Result<ModelExtensionData, crate::database::DatabaseError> {
264        let mut data = Vec::new();
265
266        let extensions = Self::get_extension_list().read();
267        for extension in extensions.iter() {
268            let value = extension.map_extended(prefix, row)?;
269            let serialized = rmp_serde::to_vec(&value).map_err(anyhow::Error::new)?;
270
271            data.push((
272                compact_str::CompactString::const_new(extension.extension_name()),
273                serialized,
274            ));
275        }
276
277        Ok(data)
278    }
279
280    fn map(prefix: Option<&str>, row: &PgRow) -> Result<Self, crate::database::DatabaseError>;
281}
282
283pub trait EventEmittingModel: BaseModel {
284    type Event: Send + Sync + 'static;
285
286    fn get_event_emitter() -> &'static crate::events::EventEmitter<Self::Event>;
287
288    fn register_event_handler<
289        F: Fn(crate::State, Arc<Self::Event>) -> Fut + Send + Sync + 'static,
290        Fut: Future<Output = Result<(), anyhow::Error>> + Send + 'static,
291    >(
292        listener: F,
293    ) -> crate::events::EventHandlerHandle {
294        Self::get_event_emitter().register_event_handler(listener)
295    }
296}
297
298type CreateHandlerResult<'a> =
299    Pin<Box<dyn Future<Output = Result<(), crate::database::DatabaseError>> + Send + 'a>>;
300type CreateHandler<M> = dyn for<'a> Fn(
301        &'a mut <M as CreatableModel>::CreateOptions<'_>,
302        &'a mut InsertQueryBuilder,
303        &'a crate::State,
304        &'a mut sqlx::Transaction<'_, sqlx::Postgres>,
305    ) -> CreateHandlerResult<'a>
306    + Send
307    + Sync;
308type CreateAfterHandler<M> = dyn for<'a> Fn(
309        &'a mut <M as CreatableModel>::CreateResult,
310        &'a <M as CreatableModel>::CreateOptions<'_>,
311        &'a crate::State,
312        &'a mut sqlx::Transaction<'_, sqlx::Postgres>,
313    ) -> CreateHandlerResult<'a>
314    + Send
315    + Sync;
316pub type CreateListenerList<M> =
317    Arc<ModelHandlerList<Arc<CreateHandler<M>>, Arc<CreateAfterHandler<M>>>>;
318
319#[async_trait::async_trait]
320pub trait CreatableModel: BaseModel + Send + Sync + 'static {
321    type CreateOptions<'a>: Send + Sync + Validate;
322    type CreateResult: Send;
323
324    fn get_create_handlers() -> &'static LazyLock<CreateListenerList<Self>>;
325
326    fn register_create_handler<
327        F: for<'a> Fn(
328                &'a mut Self::CreateOptions<'_>,
329                &'a mut InsertQueryBuilder,
330                &'a crate::State,
331                &'a mut sqlx::Transaction<'_, sqlx::Postgres>,
332            ) -> Pin<
333                Box<dyn Future<Output = Result<(), crate::database::DatabaseError>> + Send + 'a>,
334            > + Send
335            + Sync
336            + 'static,
337    >(
338        priority: ListenerPriority,
339        callback: F,
340    ) {
341        let erased = Arc::new(callback) as Arc<CreateHandler<Self>>;
342
343        Self::get_create_handlers().register_handler(priority, erased);
344    }
345
346    fn register_after_create_handler<
347        F: for<'a> Fn(
348                &'a mut Self::CreateResult,
349                &'a Self::CreateOptions<'_>,
350                &'a crate::State,
351                &'a mut sqlx::Transaction<'_, sqlx::Postgres>,
352            ) -> Pin<
353                Box<dyn Future<Output = Result<(), crate::database::DatabaseError>> + Send + 'a>,
354            > + Send
355            + Sync
356            + 'static,
357    >(
358        priority: ListenerPriority,
359        callback: F,
360    ) {
361        let erased = Arc::new(callback) as Arc<CreateAfterHandler<Self>>;
362
363        Self::get_create_handlers().register_after_handler(priority, erased);
364    }
365
366    async fn run_create_handlers(
367        options: &mut Self::CreateOptions<'_>,
368        query_builder: &mut InsertQueryBuilder,
369        state: &crate::State,
370        transaction: &mut sqlx::Transaction<'_, sqlx::Postgres>,
371    ) -> Result<(), crate::database::DatabaseError> {
372        let callbacks = Self::get_create_handlers()
373            .before_handlers
374            .read()
375            .iter()
376            .map(|l| l.callback.clone())
377            .collect::<Vec<_>>();
378
379        for callback in callbacks.iter() {
380            (*callback)(options, query_builder, state, transaction).await?;
381        }
382
383        Ok(())
384    }
385
386    async fn run_after_create_handlers(
387        result: &mut Self::CreateResult,
388        options: &Self::CreateOptions<'_>,
389        state: &crate::State,
390        transaction: &mut sqlx::Transaction<'_, sqlx::Postgres>,
391    ) -> Result<(), crate::database::DatabaseError> {
392        let callbacks = Self::get_create_handlers()
393            .after_handlers
394            .read()
395            .iter()
396            .map(|l| l.callback.clone())
397            .collect::<Vec<_>>();
398
399        for callback in callbacks.iter() {
400            (*callback)(result, options, state, transaction).await?;
401        }
402
403        Ok(())
404    }
405
406    async fn create_with_transaction(
407        state: &crate::State,
408        options: Self::CreateOptions<'_>,
409        transaction: &mut sqlx::Transaction<'_, sqlx::Postgres>,
410    ) -> Result<Self::CreateResult, crate::database::DatabaseError>;
411
412    async fn create(
413        state: &crate::State,
414        options: Self::CreateOptions<'_>,
415    ) -> Result<Self::CreateResult, crate::database::DatabaseError> {
416        let mut transaction = state.database.write().begin().await?;
417
418        let result = match Self::create_with_transaction(state, options, &mut transaction).await {
419            Ok(result) => result,
420            Err(err) => {
421                transaction.rollback().await?;
422                return Err(err);
423            }
424        };
425
426        transaction.commit().await?;
427
428        Ok(result)
429    }
430}
431
432type UpdateHandlerResult<'a> =
433    Pin<Box<dyn Future<Output = Result<(), crate::database::DatabaseError>> + Send + 'a>>;
434type UpdateHandler<M> = dyn for<'a> Fn(
435        &'a mut M,
436        &'a mut <M as UpdatableModel>::UpdateOptions,
437        &'a mut UpdateQueryBuilder,
438        &'a crate::State,
439        &'a mut sqlx::Transaction<'_, sqlx::Postgres>,
440    ) -> UpdateHandlerResult<'a>
441    + Send
442    + Sync;
443type UpdateAfterHandler<M> = dyn for<'a> Fn(
444        &'a mut M,
445        &'a crate::State,
446        &'a mut sqlx::Transaction<'_, sqlx::Postgres>,
447    ) -> UpdateHandlerResult<'a>
448    + Send
449    + Sync;
450pub type UpdateHandlerList<M> =
451    Arc<ModelHandlerList<Arc<UpdateHandler<M>>, Arc<UpdateAfterHandler<M>>>>;
452
453#[async_trait::async_trait]
454pub trait UpdatableModel: BaseModel + Send + Sync + 'static {
455    type UpdateOptions: Send + Sync + Default + ToSchema + DeserializeOwned + Serialize + Validate;
456
457    fn get_update_handlers() -> &'static LazyLock<UpdateHandlerList<Self>>;
458
459    fn register_update_handler<
460        F: for<'a> Fn(
461                &'a mut Self,
462                &'a mut Self::UpdateOptions,
463                &'a mut UpdateQueryBuilder,
464                &'a crate::State,
465                &'a mut sqlx::Transaction<'_, sqlx::Postgres>,
466            ) -> Pin<
467                Box<dyn Future<Output = Result<(), crate::database::DatabaseError>> + Send + 'a>,
468            > + Send
469            + Sync
470            + 'static,
471    >(
472        priority: ListenerPriority,
473        callback: F,
474    ) {
475        let erased = Arc::new(callback) as Arc<UpdateHandler<Self>>;
476
477        Self::get_update_handlers().register_handler(priority, erased);
478    }
479
480    fn register_after_update_handler<
481        F: for<'a> Fn(
482                &'a mut Self,
483                &'a crate::State,
484                &'a mut sqlx::Transaction<'_, sqlx::Postgres>,
485            ) -> Pin<
486                Box<dyn Future<Output = Result<(), crate::database::DatabaseError>> + Send + 'a>,
487            > + Send
488            + Sync
489            + 'static,
490    >(
491        priority: ListenerPriority,
492        callback: F,
493    ) {
494        let erased = Arc::new(callback) as Arc<UpdateAfterHandler<Self>>;
495
496        Self::get_update_handlers().register_after_handler(priority, erased);
497    }
498
499    async fn run_update_handlers(
500        &mut self,
501        options: &mut Self::UpdateOptions,
502        query_builder: &mut UpdateQueryBuilder,
503        state: &crate::State,
504        transaction: &mut sqlx::Transaction<'_, sqlx::Postgres>,
505    ) -> Result<(), crate::database::DatabaseError> {
506        let callbacks = Self::get_update_handlers()
507            .before_handlers
508            .read()
509            .iter()
510            .map(|l| l.callback.clone())
511            .collect::<Vec<_>>();
512
513        for callback in callbacks.iter() {
514            (*callback)(self, options, query_builder, state, transaction).await?;
515        }
516
517        Ok(())
518    }
519
520    async fn run_after_update_handlers(
521        &mut self,
522        state: &crate::State,
523        transaction: &mut sqlx::Transaction<'_, sqlx::Postgres>,
524    ) -> Result<(), crate::database::DatabaseError> {
525        let callbacks = Self::get_update_handlers()
526            .after_handlers
527            .read()
528            .iter()
529            .map(|l| l.callback.clone())
530            .collect::<Vec<_>>();
531
532        for callback in callbacks.iter() {
533            (*callback)(self, state, transaction).await?;
534        }
535
536        Ok(())
537    }
538
539    async fn update_with_transaction(
540        &mut self,
541        state: &crate::State,
542        options: Self::UpdateOptions,
543        transaction: &mut sqlx::Transaction<'_, sqlx::Postgres>,
544    ) -> Result<(), crate::database::DatabaseError>;
545
546    async fn update(
547        &mut self,
548        state: &crate::State,
549        options: Self::UpdateOptions,
550    ) -> Result<(), crate::database::DatabaseError> {
551        let mut transaction = state.database.write().begin().await?;
552
553        if let Err(err) = self
554            .update_with_transaction(state, options, &mut transaction)
555            .await
556        {
557            transaction.rollback().await?;
558            return Err(err);
559        }
560
561        transaction.commit().await?;
562
563        Ok(())
564    }
565}
566
567type DeleteHandlerResult<'a> = Pin<Box<dyn Future<Output = Result<(), anyhow::Error>> + Send + 'a>>;
568type DeleteHandler<M> = dyn for<'a> Fn(
569        &'a M,
570        &'a <M as DeletableModel>::DeleteOptions,
571        &'a crate::State,
572        &'a mut sqlx::Transaction<'_, sqlx::Postgres>,
573    ) -> DeleteHandlerResult<'a>
574    + Send
575    + Sync;
576type DeleteAfterHandler<M> = dyn for<'a> Fn(
577        &'a M,
578        &'a <M as DeletableModel>::DeleteOptions,
579        &'a crate::State,
580        &'a mut sqlx::Transaction<'_, sqlx::Postgres>,
581    ) -> DeleteHandlerResult<'a>
582    + Send
583    + Sync;
584pub type DeleteHandlerList<M> =
585    Arc<ModelHandlerList<Arc<DeleteHandler<M>>, Arc<DeleteAfterHandler<M>>>>;
586
587#[async_trait::async_trait]
588pub trait DeletableModel: BaseModel + Send + Sync + 'static {
589    type DeleteOptions: Send + Sync + Default + Clone;
590
591    fn get_delete_handlers() -> &'static LazyLock<DeleteHandlerList<Self>>;
592
593    fn register_delete_handler<
594        F: for<'a> Fn(
595                &'a Self,
596                &'a Self::DeleteOptions,
597                &'a crate::State,
598                &'a mut sqlx::Transaction<'_, sqlx::Postgres>,
599            )
600                -> Pin<Box<dyn Future<Output = Result<(), anyhow::Error>> + Send + 'a>>
601            + Send
602            + Sync
603            + 'static,
604    >(
605        priority: ListenerPriority,
606        callback: F,
607    ) {
608        let erased = Arc::new(callback) as Arc<DeleteHandler<Self>>;
609
610        Self::get_delete_handlers().register_handler(priority, erased);
611    }
612
613    fn register_after_delete_handler<
614        F: for<'a> Fn(
615                &'a Self,
616                &'a Self::DeleteOptions,
617                &'a crate::State,
618                &'a mut sqlx::Transaction<'_, sqlx::Postgres>,
619            )
620                -> Pin<Box<dyn Future<Output = Result<(), anyhow::Error>> + Send + 'a>>
621            + Send
622            + Sync
623            + 'static,
624    >(
625        priority: ListenerPriority,
626        callback: F,
627    ) {
628        let erased = Arc::new(callback) as Arc<DeleteAfterHandler<Self>>;
629
630        Self::get_delete_handlers().register_after_handler(priority, erased);
631    }
632
633    async fn run_delete_handlers(
634        &self,
635        options: &Self::DeleteOptions,
636        state: &crate::State,
637        transaction: &mut sqlx::Transaction<'_, sqlx::Postgres>,
638    ) -> Result<(), anyhow::Error> {
639        let callbacks = Self::get_delete_handlers()
640            .before_handlers
641            .read()
642            .iter()
643            .map(|l| l.callback.clone())
644            .collect::<Vec<_>>();
645
646        for callback in callbacks.iter() {
647            (*callback)(self, options, state, transaction).await?;
648        }
649
650        Ok(())
651    }
652
653    async fn run_after_delete_handlers(
654        &self,
655        options: &Self::DeleteOptions,
656        state: &crate::State,
657        transaction: &mut sqlx::Transaction<'_, sqlx::Postgres>,
658    ) -> Result<(), anyhow::Error> {
659        let callbacks = Self::get_delete_handlers()
660            .after_handlers
661            .read()
662            .iter()
663            .map(|l| l.callback.clone())
664            .collect::<Vec<_>>();
665
666        for callback in callbacks.iter() {
667            (*callback)(self, options, state, transaction).await?;
668        }
669
670        Ok(())
671    }
672
673    async fn delete_with_transaction(
674        &self,
675        state: &crate::State,
676        options: Self::DeleteOptions,
677        transaction: &mut sqlx::Transaction<'_, sqlx::Postgres>,
678    ) -> Result<(), anyhow::Error>;
679
680    async fn delete(
681        &self,
682        state: &crate::State,
683        options: Self::DeleteOptions,
684    ) -> Result<(), anyhow::Error> {
685        let mut transaction = state.database.write().begin().await?;
686
687        if let Err(err) = self
688            .delete_with_transaction(state, options, &mut transaction)
689            .await
690        {
691            transaction.rollback().await?;
692            return Err(err);
693        }
694
695        transaction.commit().await?;
696
697        Ok(())
698    }
699}
700
701type DuplicateHandlerResult<'a> =
702    Pin<Box<dyn Future<Output = Result<(), crate::database::DatabaseError>> + Send + 'a>>;
703type DuplicateHandler<M> = dyn for<'a> Fn(
704        &'a M,
705        &'a <M as DuplicableModel>::DuplicateOptions<'_>,
706        &'a crate::State,
707        &'a mut sqlx::Transaction<'_, sqlx::Postgres>,
708    ) -> DuplicateHandlerResult<'a>
709    + Send
710    + Sync;
711type DuplicateAfterHandler<M> = dyn for<'a> Fn(
712        &'a M,
713        &'a mut M,
714        &'a <M as DuplicableModel>::DuplicateOptions<'_>,
715        &'a crate::State,
716        &'a mut sqlx::Transaction<'_, sqlx::Postgres>,
717    ) -> DuplicateHandlerResult<'a>
718    + Send
719    + Sync;
720pub type DuplicateHandlerList<M> =
721    Arc<ModelHandlerList<Arc<DuplicateHandler<M>>, Arc<DuplicateAfterHandler<M>>>>;
722
723#[async_trait::async_trait]
724pub trait DuplicableModel: BaseModel + Send + Sync + 'static {
725    type DuplicateOptions<'a>: Send + Sync + Validate;
726
727    fn get_duplicate_handlers() -> &'static LazyLock<DuplicateHandlerList<Self>>;
728
729    fn register_duplicate_handler<
730        F: for<'a> Fn(
731                &'a Self,
732                &'a Self::DuplicateOptions<'_>,
733                &'a crate::State,
734                &'a mut sqlx::Transaction<'_, sqlx::Postgres>,
735            ) -> DuplicateHandlerResult<'a>
736            + Send
737            + Sync
738            + 'static,
739    >(
740        priority: ListenerPriority,
741        callback: F,
742    ) {
743        let erased = Arc::new(callback) as Arc<DuplicateHandler<Self>>;
744
745        Self::get_duplicate_handlers().register_handler(priority, erased);
746    }
747
748    fn register_after_duplicate_handler<
749        F: for<'a> Fn(
750                &'a Self,
751                &'a mut Self,
752                &'a Self::DuplicateOptions<'_>,
753                &'a crate::State,
754                &'a mut sqlx::Transaction<'_, sqlx::Postgres>,
755            ) -> DuplicateHandlerResult<'a>
756            + Send
757            + Sync
758            + 'static,
759    >(
760        priority: ListenerPriority,
761        callback: F,
762    ) {
763        let erased = Arc::new(callback) as Arc<DuplicateAfterHandler<Self>>;
764
765        Self::get_duplicate_handlers().register_after_handler(priority, erased);
766    }
767
768    async fn run_duplicate_handlers(
769        &self,
770        options: &Self::DuplicateOptions<'_>,
771        state: &crate::State,
772        transaction: &mut sqlx::Transaction<'_, sqlx::Postgres>,
773    ) -> Result<(), crate::database::DatabaseError> {
774        let callbacks = Self::get_duplicate_handlers()
775            .before_handlers
776            .read()
777            .iter()
778            .map(|l| l.callback.clone())
779            .collect::<Vec<_>>();
780
781        for callback in callbacks.iter() {
782            (*callback)(self, options, state, transaction).await?;
783        }
784
785        Ok(())
786    }
787
788    async fn run_after_duplicate_handlers(
789        &self,
790        duplicated: &mut Self,
791        options: &Self::DuplicateOptions<'_>,
792        state: &crate::State,
793        transaction: &mut sqlx::Transaction<'_, sqlx::Postgres>,
794    ) -> Result<(), crate::database::DatabaseError> {
795        let callbacks = Self::get_duplicate_handlers()
796            .after_handlers
797            .read()
798            .iter()
799            .map(|l| l.callback.clone())
800            .collect::<Vec<_>>();
801
802        for callback in callbacks.iter() {
803            (*callback)(self, duplicated, options, state, transaction).await?;
804        }
805
806        Ok(())
807    }
808
809    async fn duplicate_with_transaction(
810        &self,
811        state: &crate::State,
812        options: Self::DuplicateOptions<'_>,
813        transaction: &mut sqlx::Transaction<'_, sqlx::Postgres>,
814    ) -> Result<Self, crate::database::DatabaseError>;
815
816    async fn duplicate(
817        &self,
818        state: &crate::State,
819        options: Self::DuplicateOptions<'_>,
820    ) -> Result<Self, crate::database::DatabaseError> {
821        let mut transaction = state.database.write().begin().await?;
822
823        let duplicated = match self
824            .duplicate_with_transaction(state, options, &mut transaction)
825            .await
826        {
827            Ok(duplicated) => duplicated,
828            Err(err) => {
829                transaction.rollback().await?;
830                return Err(err);
831            }
832        };
833
834        transaction.commit().await?;
835
836        Ok(duplicated)
837    }
838}
839
840#[async_trait::async_trait]
841pub trait ByUuid: BaseModel {
842    async fn by_uuid(
843        database: &crate::database::Database,
844        uuid: uuid::Uuid,
845    ) -> Result<Self, DatabaseError>;
846
847    async fn by_uuid_with_transaction(
848        transaction: &mut sqlx::Transaction<'_, sqlx::Postgres>,
849        uuid: uuid::Uuid,
850    ) -> Result<Self, DatabaseError>;
851
852    async fn by_uuid_cached(
853        database: &crate::database::Database,
854        uuid: uuid::Uuid,
855    ) -> Result<Self, anyhow::Error> {
856        database
857            .cache
858            .cached(&format!("{}::{uuid}", Self::NAME), 10, || {
859                Self::by_uuid(database, uuid)
860            })
861            .await
862    }
863
864    async fn by_uuid_optional(
865        database: &crate::database::Database,
866        uuid: uuid::Uuid,
867    ) -> Result<Option<Self>, DatabaseError> {
868        match Self::by_uuid(database, uuid).await {
869            Ok(res) => Ok(Some(res)),
870            Err(DatabaseError::Sqlx(sqlx::Error::RowNotFound)) => Ok(None),
871            Err(err) => Err(err),
872        }
873    }
874
875    async fn by_uuid_optional_with_transaction(
876        transaction: &mut sqlx::Transaction<'_, sqlx::Postgres>,
877        uuid: uuid::Uuid,
878    ) -> Result<Option<Self>, DatabaseError> {
879        match Self::by_uuid_with_transaction(transaction, uuid).await {
880            Ok(res) => Ok(Some(res)),
881            Err(DatabaseError::Sqlx(sqlx::Error::RowNotFound)) => Ok(None),
882            Err(err) => Err(err),
883        }
884    }
885
886    async fn by_uuid_optional_cached(
887        database: &crate::database::Database,
888        uuid: uuid::Uuid,
889    ) -> Result<Option<Self>, anyhow::Error> {
890        match Self::by_uuid_cached(database, uuid).await {
891            Ok(res) => Ok(Some(res)),
892            Err(err) => {
893                if let Some(DatabaseError::Sqlx(sqlx::Error::RowNotFound)) =
894                    err.downcast_ref::<DatabaseError>()
895                {
896                    Ok(None)
897                } else {
898                    Err(err)
899                }
900            }
901        }
902    }
903
904    #[inline]
905    fn get_fetchable(uuid: uuid::Uuid) -> Fetchable<Self> {
906        Fetchable {
907            uuid,
908            _model: PhantomData,
909        }
910    }
911
912    #[inline]
913    fn get_fetchable_from_row(row: &PgRow, column: impl AsRef<str>) -> Option<Fetchable<Self>> {
914        match row.try_get(column.as_ref()) {
915            Ok(uuid) => Some(Fetchable {
916                uuid,
917                _model: PhantomData,
918            }),
919            Err(_) => None,
920        }
921    }
922}
923
924#[derive(Default, Debug, Clone, Copy, PartialEq, Eq)]
925pub enum ListenerPriority {
926    Highest,
927    High,
928    #[default]
929    Normal,
930    Low,
931    Lowest,
932}
933
934impl ListenerPriority {
935    #[inline]
936    fn rank(self) -> u8 {
937        match self {
938            Self::Highest => 5,
939            Self::High => 4,
940            Self::Normal => 3,
941            Self::Low => 2,
942            Self::Lowest => 1,
943        }
944    }
945}
946
947impl PartialOrd for ListenerPriority {
948    fn partial_cmp(&self, other: &Self) -> Option<std::cmp::Ordering> {
949        Some(self.cmp(other))
950    }
951}
952
953impl Ord for ListenerPriority {
954    fn cmp(&self, other: &Self) -> std::cmp::Ordering {
955        let self_rank = self.rank();
956        let other_rank = other.rank();
957
958        other_rank.cmp(&self_rank)
959    }
960}
961
962impl<F: Send + Sync, AfterF: Send + Sync> crate::events::DisconnectEventHandler
963    for ModelHandlerList<F, AfterF>
964{
965    #[inline]
966    fn disconnect(&self, id: uuid::Uuid) {
967        self.before_handlers.write().retain(|l| l.uuid != id);
968        self.after_handlers.write().retain(|l| l.uuid != id);
969    }
970}
971
972pub struct ModelHandlerList<F: Send + Sync + 'static, AfterF: Send + Sync + 'static> {
973    before_handlers: RwLock<Vec<ModelHandler<F>>>,
974    after_handlers: RwLock<Vec<ModelHandler<AfterF>>>,
975}
976
977impl<F: Send + Sync + 'static, AfterF: Send + Sync + 'static> Default
978    for ModelHandlerList<F, AfterF>
979{
980    fn default() -> Self {
981        Self {
982            before_handlers: RwLock::new(Vec::new()),
983            after_handlers: RwLock::new(Vec::new()),
984        }
985    }
986}
987
988impl<F: Send + Sync + 'static, AfterF: Send + Sync + 'static> ModelHandlerList<F, AfterF> {
989    pub fn register_handler(
990        self: &Arc<Self>,
991        priority: ListenerPriority,
992        callback: F,
993    ) -> ModelHandlerHandle {
994        let (listener, aborter) = ModelHandler::new(callback, priority, self.clone());
995
996        let mut self_listeners = self.before_handlers.write();
997        self_listeners.push(listener);
998        self_listeners.sort_by_key(|a| a.priority);
999
1000        aborter
1001    }
1002
1003    pub fn register_after_handler(
1004        self: &Arc<Self>,
1005        priority: ListenerPriority,
1006        callback: AfterF,
1007    ) -> ModelHandlerHandle {
1008        let (listener, aborter) = ModelHandler::new(callback, priority, self.clone());
1009
1010        let mut self_listeners = self.after_handlers.write();
1011        self_listeners.push(listener);
1012        self_listeners.sort_by_key(|a| a.priority);
1013
1014        aborter
1015    }
1016}
1017
1018pub struct ModelHandler<F: Send + Sync + 'static> {
1019    uuid: uuid::Uuid,
1020    priority: ListenerPriority,
1021
1022    pub callback: F,
1023}
1024
1025impl<F: Send + Sync + 'static> ModelHandler<F> {
1026    pub(crate) fn new(
1027        callback: F,
1028        priority: ListenerPriority,
1029        list: Arc<dyn crate::events::DisconnectEventHandler + Send + Sync>,
1030    ) -> (Self, ModelHandlerHandle) {
1031        let handler = Self {
1032            uuid: uuid::Uuid::new_v4(),
1033            priority,
1034            callback,
1035        };
1036        let handle = ModelHandlerHandle {
1037            list_ref: list,
1038            id: handler.uuid,
1039        };
1040        (handler, handle)
1041    }
1042}
1043
1044pub struct ModelHandlerHandle {
1045    list_ref: Arc<dyn crate::events::DisconnectEventHandler + Send + Sync>,
1046    id: uuid::Uuid,
1047}
1048
1049impl ModelHandlerHandle {
1050    pub fn disconnect(&self) {
1051        self.list_ref.disconnect(self.id);
1052    }
1053}
1054
1055#[derive(Serialize, Deserialize, Clone, Copy)]
1056pub struct Fetchable<M: ByUuid> {
1057    pub uuid: uuid::Uuid,
1058    #[serde(skip)]
1059    _model: PhantomData<M>,
1060}
1061
1062impl<M: ByUuid + Send> Fetchable<M> {
1063    #[inline]
1064    pub async fn fetch(&self, database: &crate::database::Database) -> Result<M, DatabaseError> {
1065        M::by_uuid(database, self.uuid).await
1066    }
1067
1068    #[inline]
1069    pub async fn fetch_cached(
1070        &self,
1071        database: &crate::database::Database,
1072    ) -> Result<M, anyhow::Error> {
1073        M::by_uuid_cached(database, self.uuid).await
1074    }
1075
1076    #[inline]
1077    pub async fn fetch_optional(
1078        &self,
1079        database: &crate::database::Database,
1080    ) -> Result<Option<M>, DatabaseError> {
1081        M::by_uuid_optional(database, self.uuid).await
1082    }
1083
1084    #[inline]
1085    pub async fn fetch_optional_cached(
1086        &self,
1087        database: &crate::database::Database,
1088    ) -> Result<Option<M>, anyhow::Error> {
1089        M::by_uuid_optional_cached(database, self.uuid).await
1090    }
1091}
1092
1093pub struct InsertQueryBuilder<'a> {
1094    table: &'a str,
1095    columns: Vec<&'a str>,
1096    expressions: Vec<String>,
1097    arguments: PgArguments,
1098    returning_clause: Option<&'a str>,
1099}
1100
1101impl<'a> InsertQueryBuilder<'a> {
1102    pub fn new(table: &'a str) -> Self {
1103        Self {
1104            table,
1105            columns: Vec::new(),
1106            expressions: Vec::new(),
1107            arguments: PgArguments::default(),
1108            returning_clause: None,
1109        }
1110    }
1111
1112    pub fn set<T: 'a + sqlx::Encode<'a, Postgres> + sqlx::Type<Postgres> + Send>(
1113        &mut self,
1114        column: &'a str,
1115        value: T,
1116    ) -> &mut Self {
1117        if self.columns.contains(&column) {
1118            return self;
1119        }
1120
1121        if self.arguments.add(value).is_ok() {
1122            self.columns.push(column);
1123            let idx = self.arguments.len();
1124            self.expressions.push(format!("${}", idx));
1125        }
1126
1127        self
1128    }
1129
1130    pub fn set_expr<T: 'a + sqlx::Encode<'a, Postgres> + sqlx::Type<Postgres> + Send>(
1131        &mut self,
1132        column: &'a str,
1133        expression: &str,
1134        values: Vec<T>,
1135    ) -> &mut Self {
1136        if self.columns.contains(&column) {
1137            return self;
1138        }
1139
1140        let start_len = self.arguments.len();
1141
1142        for value in values {
1143            if self.arguments.add(value).is_err() {
1144                return self;
1145            }
1146        }
1147
1148        let mut expr = expression.to_string();
1149        let added_count = self.arguments.len() - start_len;
1150
1151        for i in (1..=added_count).rev() {
1152            let global_idx = start_len + i;
1153            expr = expr.replace(&format!("${}", i), &format!("${}", global_idx));
1154        }
1155
1156        self.columns.push(column);
1157        self.expressions.push(expr);
1158
1159        self
1160    }
1161
1162    pub fn returning(mut self, clause: &'a str) -> Self {
1163        self.returning_clause = Some(clause);
1164        self
1165    }
1166
1167    fn build_sql(&self) -> String {
1168        let columns_sql = self.columns.join(", ");
1169        let values_sql = self.expressions.join(", ");
1170
1171        let mut sql = format!(
1172            "INSERT INTO {} ({}) VALUES ({})",
1173            self.table, columns_sql, values_sql
1174        );
1175
1176        if let Some(clause) = self.returning_clause {
1177            sql.push_str(" RETURNING ");
1178            sql.push_str(clause);
1179        }
1180
1181        sql
1182    }
1183
1184    pub async fn execute(
1185        self,
1186        executor: impl sqlx::Executor<'a, Database = Postgres>,
1187    ) -> Result<sqlx::postgres::PgQueryResult, sqlx::Error> {
1188        let sql = self.build_sql();
1189        sqlx::query_with(sqlx::AssertSqlSafe(sql), self.arguments)
1190            .execute(executor)
1191            .await
1192    }
1193
1194    pub async fn fetch_one(
1195        self,
1196        executor: impl sqlx::Executor<'a, Database = Postgres>,
1197    ) -> Result<sqlx::postgres::PgRow, sqlx::Error> {
1198        let sql = self.build_sql();
1199        sqlx::query_with(sqlx::AssertSqlSafe(sql), self.arguments)
1200            .fetch_one(executor)
1201            .await
1202    }
1203}
1204
1205pub struct UpdateQueryBuilder<'a> {
1206    builder: QueryBuilder<Postgres>,
1207    updated_fields: HashSet<&'a str>,
1208    has_set_fields: bool,
1209    has_where: bool,
1210}
1211
1212impl<'a> UpdateQueryBuilder<'a> {
1213    pub fn new(table: &'a str) -> Self {
1214        let mut builder = QueryBuilder::new("UPDATE ");
1215        builder.push(table);
1216        builder.push(" SET ");
1217
1218        Self {
1219            builder,
1220            updated_fields: HashSet::new(),
1221            has_set_fields: false,
1222            has_where: false,
1223        }
1224    }
1225
1226    /// Adds a field to be updated, if `None`, will not add the field
1227    /// To set a field to null (`None`), you need a `Some(None)`
1228    pub fn set<T: 'a + sqlx::Encode<'a, Postgres> + sqlx::Type<Postgres> + Send>(
1229        &mut self,
1230        column: &'a str,
1231        value: Option<T>,
1232    ) -> &mut Self {
1233        let Some(value) = value else {
1234            return self;
1235        };
1236
1237        if !self.updated_fields.insert(column) {
1238            return self;
1239        }
1240
1241        if self.has_set_fields {
1242            self.builder.push(", ");
1243        }
1244
1245        self.builder.push(column);
1246        self.builder.push(" = ");
1247        self.builder.push_bind(value);
1248
1249        self.has_set_fields = true;
1250        self
1251    }
1252
1253    pub fn where_eq<T: 'a + sqlx::Encode<'a, Postgres> + sqlx::Type<Postgres> + Send>(
1254        &mut self,
1255        column: &'a str,
1256        value: T,
1257    ) -> &mut Self {
1258        if self.has_where {
1259            self.builder.push(" AND ");
1260        } else {
1261            self.builder.push(" WHERE ");
1262            self.has_where = true;
1263        }
1264
1265        self.builder.push(column);
1266        self.builder.push(" = ");
1267        self.builder.push_bind(value);
1268        self
1269    }
1270
1271    pub async fn execute(
1272        mut self,
1273        executor: impl sqlx::Executor<'a, Database = Postgres>,
1274    ) -> Result<sqlx::any::AnyQueryResult, sqlx::Error> {
1275        if !self.has_set_fields {
1276            return Ok(sqlx::any::AnyQueryResult::default());
1277        }
1278
1279        let query = self.builder.build();
1280        query.execute(executor).await.map(|r| r.into())
1281    }
1282}
1283
1284/// SQLx helper type to preserve order of keys when encoding JSON. By default, SQLx encodes JSON using `serde_json::Value`, which does not preserve order of keys. This type allows you to encode any serializable type as JSON while preserving the order of keys.
1285pub struct OrderedJson<T>(pub T);
1286
1287impl<T: Serialize> sqlx::Encode<'_, sqlx::Postgres> for OrderedJson<T> {
1288    fn encode_by_ref(&self, buf: &mut PgArgumentBuffer) -> Result<IsNull, BoxDynError> {
1289        serde_json::to_writer(&mut **buf, &self.0)?;
1290        Ok(IsNull::No)
1291    }
1292}
1293
1294impl<T> sqlx::Type<sqlx::Postgres> for OrderedJson<T> {
1295    fn type_info() -> PgTypeInfo {
1296        // JSON, not JSONB, to preserve order of keys
1297        PgTypeInfo::with_oid(sqlx::postgres::types::Oid(114))
1298    }
1299}
1300
1301#[async_trait::async_trait]
1302pub trait IntoApiObject {
1303    type ApiObject: Send;
1304    type ExtraArgs<'a>: Send;
1305
1306    async fn into_api_object<'a>(
1307        self,
1308        state: &crate::State,
1309        args: Self::ExtraArgs<'a>,
1310    ) -> Result<Self::ApiObject, DatabaseError>;
1311}
1312
1313#[async_trait::async_trait]
1314pub trait IntoAdminApiObject {
1315    type AdminApiObject: Send;
1316    type ExtraArgs<'a>: Send;
1317
1318    async fn into_admin_api_object<'a>(
1319        self,
1320        state: &crate::State,
1321        args: Self::ExtraArgs<'a>,
1322    ) -> Result<Self::AdminApiObject, DatabaseError>;
1323}
shared/models/mod.rs

shared/models/
mod.rs
