shared/models/
mod.rs

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

shared/models/
mod.rs
