1use std::collections::HashMap;
39use std::marker::PhantomData;
40
41use anyhow::Result;
42use serde::Serialize;
43use sqlx::postgres::PgRow;
44use sqlx::{FromRow, PgPool};
45use uuid::Uuid;
46
47use crate::repository::{
48 DatabaseOperations, PaginatedResult, PaginationInfo, PaginationParams, PostgresRepository,
49};
50
51pub trait EntityRepoMeta {
63 fn column_types() -> HashMap<String, String>;
65 fn search_fields() -> &'static [&'static str];
67
68 fn private_fields() -> &'static [&'static str] {
75 &[]
76 }
77
78 fn secret_fields() -> &'static [&'static str] {
81 &[]
82 }
83
84 fn relation_secret_fields(_relation: &str) -> &'static [&'static str] {
88 &[]
89 }
90
91 fn owner_field() -> Option<&'static str> {
95 None
96 }
97
98 fn company_field() -> Option<&'static str> {
114 None
115 }
116
117 fn relations() -> &'static [(&'static str, &'static str, &'static str)] {
122 &[]
123 }
124}
125
126#[derive(Debug, thiserror::Error)]
136#[error("entity is company-scoped ({column}) but the request carries no company scope")]
137pub struct MissingCompanyScope {
138 pub column: &'static str,
140}
141
142pub fn company_fence<T: EntityRepoMeta>(company: Option<Uuid>) -> Result<Option<String>, MissingCompanyScope> {
153 match (T::company_field(), company) {
154 (None, _) => Ok(None),
155 (Some(column), None) => Err(MissingCompanyScope { column }),
156 (Some(column), Some(id)) => Ok(Some(format!("{column} = '{id}'"))),
157 }
158}
159
160pub fn strip_client_company_filters<T: EntityRepoMeta>(filters: &mut HashMap<String, String>) {
172 let Some(column) = T::company_field() else {
173 return;
174 };
175 let camel = snake_to_camel(column);
176 filters.retain(|key, _| {
177 let base = key.split('[').next().unwrap_or(key);
178 !base.eq_ignore_ascii_case(column) && !base.eq_ignore_ascii_case(&camel)
179 });
180}
181
182pub fn and_conditions(a: Option<&str>, b: Option<String>) -> Option<String> {
185 match (a, b) {
186 (None, None) => None,
187 (Some(a), None) => Some(a.to_string()),
188 (None, Some(b)) => Some(b),
189 (Some(a), Some(b)) => Some(format!("{a} AND {b}")),
190 }
191}
192
193fn snake_to_camel(s: &str) -> String {
194 let mut out = String::with_capacity(s.len());
195 let mut upper = false;
196 for c in s.chars() {
197 if c == '_' {
198 upper = true;
199 } else if upper {
200 out.push(c.to_ascii_uppercase());
201 upper = false;
202 } else {
203 out.push(c);
204 }
205 }
206 out
207}
208
209const ACTIVE_ROW_GUARD: &str = " AND t.metadata->>'deleted_at' IS NULL";
211
212pub struct SoftDelete;
216
217pub struct HardDelete;
219
220pub struct GenericCrudRepository<T, D = SoftDelete>
229where
230 T: for<'r> FromRow<'r, PgRow> + Send + Unpin,
231{
232 inner: PostgresRepository<T>,
233 _mode: PhantomData<D>,
234}
235
236impl<T, D> GenericCrudRepository<T, D>
239where
240 T: for<'r> FromRow<'r, PgRow> + Send + Unpin,
241{
242 pub fn new(pool: PgPool, table_name: &str) -> Self {
243 Self {
244 inner: PostgresRepository::new(pool, table_name),
245 _mode: PhantomData,
246 }
247 }
248
249 pub fn pool(&self) -> &PgPool {
250 self.inner.pool()
251 }
252
253 pub fn table_name(&self) -> &str {
254 self.inner.table_name()
255 }
256
257 pub async fn create(&self, entity: &T) -> Result<T>
259 where
260 T: Serialize + Send + Sync,
261 {
262 self.inner.create(entity).await
263 }
264
265 pub async fn bulk_create(&self, entities: &[T]) -> Result<Vec<T>>
272 where
273 T: Serialize + Send + Sync,
274 {
275 if entities.is_empty() {
276 return Ok(Vec::new());
277 }
278 let table = self.table_name();
279 let unit = crate::write_tx::begin_unit(self.pool(), true).await?;
280 let rows = async {
281 let mut conn = unit.conn().await;
282 insert_each(&mut conn, table, entities).await
283 }
284 .await;
285 unit.finish(rows).await
286 }
287
288 async fn bulk_update_changed_guarded(&self, pairs: &[(T, T)], guard: &str) -> Result<Vec<T>>
295 where
296 T: Serialize + Send + Sync,
297 {
298 if pairs.is_empty() {
299 return Ok(Vec::new());
300 }
301 let table = self.table_name();
302 let unit = crate::write_tx::begin_unit(self.pool(), true).await?;
303 let rows = async {
304 let mut conn = unit.conn().await;
305 update_each_changed(&mut conn, table, guard, pairs).await
306 }
307 .await;
308 unit.finish(rows).await
309 }
310
311 async fn find_by_text_field_with_cond(
318 &self,
319 field: &str,
320 value: &str,
321 extra: &str,
322 ) -> Result<Option<T>> {
323 let query = format!(
324 "SELECT * FROM {} WHERE {} = $1{}",
325 self.table_name(), field, extra
326 );
327 let result = crate::company_scope::fetch_optional_scoped(
328 self.pool(),
329 sqlx::query_as::<_, T>(&query).bind(value),
330 )
331 .await?;
332 Ok(result)
333 }
334
335 async fn exists_by_text_field_with_cond(
337 &self,
338 field: &str,
339 value: &str,
340 extra: &str,
341 ) -> Result<bool> {
342 let query = format!(
343 "SELECT 1 FROM {} WHERE {} = $1{} LIMIT 1",
344 self.table_name(), field, extra
345 );
346 let result = crate::company_scope::fetch_optional_scalar_scoped(
347 self.pool(),
348 sqlx::query_scalar::<_, i32>(&query).bind(value),
349 )
350 .await?;
351 Ok(result.is_some())
352 }
353
354 async fn find_by_uuid_field_with_cond(
356 &self,
357 field: &str,
358 value: Uuid,
359 extra: &str,
360 ) -> Result<Option<T>> {
361 let query = format!(
362 "SELECT * FROM {} WHERE {} = $1{}",
363 self.table_name(), field, extra
364 );
365 let result = crate::company_scope::fetch_optional_scoped(
366 self.pool(),
367 sqlx::query_as::<_, T>(&query).bind(value),
368 )
369 .await?;
370 Ok(result)
371 }
372
373 async fn exists_by_uuid_field_with_cond(
375 &self,
376 field: &str,
377 value: Uuid,
378 extra: &str,
379 ) -> Result<bool> {
380 let query = format!(
381 "SELECT 1 FROM {} WHERE {} = $1{} LIMIT 1",
382 self.table_name(), field, extra
383 );
384 let result = crate::company_scope::fetch_optional_scalar_scoped(
385 self.pool(),
386 sqlx::query_scalar::<_, i32>(&query).bind(value),
387 )
388 .await?;
389 Ok(result.is_some())
390 }
391
392 pub async fn run_filtered_query(
403 &self,
404 pagination: PaginationParams,
405 base_condition: Option<&str>,
406 filters: &HashMap<String, String>,
407 column_types: &HashMap<String, String>,
408 search_fields: &[&str],
409 ) -> Result<PaginatedResult<T>>
410 where
411 T: Send + Sync,
412 {
413 self.inner
414 .list_with_filters_and_base(
415 pagination,
416 base_condition,
417 filters,
418 column_types,
419 search_fields,
420 )
421 .await
422 }
423
424 pub async fn run_aggregate_query(
431 &self,
432 spec: &crate::repository::AggregateSpec,
433 base_condition: Option<&str>,
434 filters: &HashMap<String, String>,
435 column_types: &HashMap<String, String>,
436 search_fields: &[&str],
437 ) -> Result<crate::repository::AggregateResult>
438 where
439 T: crate::EntityRepoMeta + Send + Sync,
440 {
441 let mut spec = spec.clone();
445 if let (Some(group), Some(label)) = (&spec.group_by, &spec.label_field) {
446 let _ = label;
447 let camel = snake_to_camel(group);
448 if let Some((_, table, _)) = T::relations().iter().find(|(_, _, fk)| *fk == camel) {
449 let base_fk = camel_to_snake(&camel);
450 spec.label_relation = Some((table.to_string(), base_fk));
451 }
452 }
453 self.inner
454 .aggregate_with_filters_and_base(
455 &spec,
456 base_condition,
457 filters,
458 column_types,
459 search_fields,
460 )
461 .await
462 }
463}
464
465impl<T> GenericCrudRepository<T, SoftDelete>
468where
469 T: for<'r> FromRow<'r, PgRow> + Send + Unpin,
470{
471 pub async fn find_by_text_field(&self, field: &str, value: &str) -> Result<Option<T>> {
477 self.find_by_text_field_with_cond(field, value, " AND metadata->>'deleted_at' IS NULL").await
478 }
479
480 pub async fn exists_by_text_field(&self, field: &str, value: &str) -> Result<bool> {
482 self.exists_by_text_field_with_cond(field, value, " AND metadata->>'deleted_at' IS NULL").await
483 }
484
485 pub async fn find_by_uuid_field(&self, field: &str, value: Uuid) -> Result<Option<T>> {
487 self.find_by_uuid_field_with_cond(field, value, " AND metadata->>'deleted_at' IS NULL").await
488 }
489
490 pub async fn exists_by_uuid_field(&self, field: &str, value: Uuid) -> Result<bool> {
492 self.exists_by_uuid_field_with_cond(field, value, " AND metadata->>'deleted_at' IS NULL").await
493 }
494
495 pub async fn list_paginated_filtered(
501 &self,
502 pagination: PaginationParams,
503 filters: Option<&HashMap<String, String>>,
504 ) -> Result<PaginatedResult<T>>
505 where
506 T: EntityRepoMeta + Send + Sync,
507 {
508 let filters_map = filters.cloned().unwrap_or_default();
509 let column_types = T::column_types();
510 let search_fields_owned: Vec<&'static str> = T::search_fields().iter().copied().collect();
511 self.run_filtered_query(
512 pagination,
513 Some("metadata->>'deleted_at' IS NULL"),
514 &filters_map,
515 &column_types,
516 &search_fields_owned,
517 ).await
518 }
519
520 pub async fn aggregate_filtered(
522 &self,
523 spec: &crate::repository::AggregateSpec,
524 filters: Option<&HashMap<String, String>>,
525 ) -> Result<crate::repository::AggregateResult>
526 where
527 T: EntityRepoMeta + Send + Sync,
528 {
529 let filters_map = filters.cloned().unwrap_or_default();
530 let column_types = T::column_types();
531 let search_fields_owned: Vec<&'static str> = T::search_fields().iter().copied().collect();
532 self.run_aggregate_query(spec, Some("metadata->>'deleted_at' IS NULL"), &filters_map, &column_types, &search_fields_owned).await
533 }
534
535 pub async fn list_paginated_filtered_scoped(
546 &self,
547 pagination: PaginationParams,
548 filters: Option<&HashMap<String, String>>,
549 company: Option<Uuid>,
550 ) -> Result<PaginatedResult<T>>
551 where
552 T: EntityRepoMeta + Send + Sync,
553 {
554 let fence = company_fence::<T>(company)?;
555 let mut filters_map = filters.cloned().unwrap_or_default();
556 strip_client_company_filters::<T>(&mut filters_map);
557 let column_types = T::column_types();
558 let search_fields_owned: Vec<&'static str> = T::search_fields().iter().copied().collect();
559 self.run_filtered_query(
560 pagination,
561 and_conditions(Some("metadata->>'deleted_at' IS NULL"), fence).as_deref(),
562 &filters_map,
563 &column_types,
564 &search_fields_owned,
565 )
566 .await
567 }
568
569 pub async fn list_deleted_filtered_scoped(
574 &self,
575 pagination: PaginationParams,
576 filters: Option<&HashMap<String, String>>,
577 company: Option<Uuid>,
578 ) -> Result<PaginatedResult<T>>
579 where
580 T: EntityRepoMeta + Send + Sync,
581 {
582 let fence = company_fence::<T>(company)?;
583 let mut filters_map = filters.cloned().unwrap_or_default();
584 strip_client_company_filters::<T>(&mut filters_map);
585 let column_types = T::column_types();
586 let search_fields: Vec<&'static str> = T::search_fields().to_vec();
587 self.run_filtered_query(
588 pagination,
589 and_conditions(Some("metadata->>'deleted_at' IS NOT NULL"), fence).as_deref(),
590 &filters_map,
591 &column_types,
592 &search_fields,
593 )
594 .await
595 }
596
597 pub async fn list_deleted_filtered(
600 &self,
601 pagination: PaginationParams,
602 filters: Option<&HashMap<String, String>>,
603 ) -> Result<PaginatedResult<T>>
604 where
605 T: EntityRepoMeta + Send + Sync,
606 {
607 let filters_map = filters.cloned().unwrap_or_default();
608 let column_types = T::column_types();
609 let search_fields: Vec<&'static str> = T::search_fields().to_vec();
610 self.run_filtered_query(
611 pagination,
612 Some("metadata->>'deleted_at' IS NOT NULL"),
613 &filters_map,
614 &column_types,
615 &search_fields,
616 ).await
617 }
618
619 pub async fn find_by_id(&self, id: &str) -> Result<Option<T>> {
621 let query = format!(
622 "SELECT * FROM {} WHERE id = $1::uuid AND metadata->>'deleted_at' IS NULL",
623 self.table_name()
624 );
625 let result = crate::company_scope::fetch_optional_scoped(
626 self.pool(),
627 sqlx::query_as::<_, T>(&query).bind(id),
628 )
629 .await?;
630 Ok(result)
631 }
632
633 pub async fn find_all(&self) -> Result<Vec<T>> {
635 let query = format!(
636 "SELECT * FROM {} WHERE metadata->>'deleted_at' IS NULL",
637 self.table_name()
638 );
639 let results = crate::company_scope::fetch_all_scoped(
640 self.pool(),
641 sqlx::query_as::<_, T>(&query),
642 )
643 .await?;
644 Ok(results)
645 }
646
647 pub async fn update(&self, id: &str, entity: &T) -> Result<Option<T>>
652 where
653 T: Serialize + Send + Sync,
654 {
655 self.inner
656 .update_guarded(id, entity, None, ACTIVE_ROW_GUARD)
657 .await
658 }
659
660 pub async fn update_changed(&self, id: &str, before: &T, after: &T) -> Result<Option<T>>
668 where
669 T: Serialize + Send + Sync,
670 {
671 let changed = crate::repository::changed_columns(before, after)?;
672 self.inner
673 .update_guarded(id, after, Some(&changed), ACTIVE_ROW_GUARD)
674 .await
675 }
676
677 pub async fn delete(&self, id: &str) -> Result<bool> {
679 self.soft_delete(id).await
680 }
681
682 pub async fn count(&self) -> Result<u64> {
684 self.count_active().await
685 }
686
687 pub async fn exists(&self, id: &str) -> Result<bool> {
689 let query = format!(
690 "SELECT 1 FROM {} WHERE id = $1::uuid AND metadata->>'deleted_at' IS NULL LIMIT 1",
691 self.table_name()
692 );
693 let result = crate::company_scope::fetch_optional_scalar_scoped(
694 self.pool(),
695 sqlx::query_scalar::<_, i32>(&query).bind(id),
696 )
697 .await?;
698 Ok(result.is_some())
699 }
700
701 pub async fn list_paginated(&self, pagination: PaginationParams) -> Result<PaginatedResult<T>> {
703 let offset = pagination.offset();
704 let limit = pagination.limit();
705 let query = format!(
706 "SELECT * FROM {} WHERE metadata->>'deleted_at' IS NULL \
707 ORDER BY id DESC LIMIT $1 OFFSET $2",
708 self.table_name()
709 );
710 let data = crate::company_scope::fetch_all_scoped(
711 self.pool(),
712 sqlx::query_as::<_, T>(&query).bind(limit as i64).bind(offset as i64),
713 )
714 .await?;
715 let total = self.count_active().await?;
716 Ok(PaginatedResult {
717 data,
718 pagination: PaginationInfo::new(pagination.page, pagination.per_page, total),
719 })
720 }
721
722 pub async fn soft_delete(&self, id: &str) -> Result<bool> {
726 let query = format!(
727 "UPDATE {} SET metadata = jsonb_set(\
728 COALESCE(metadata, '{{}}'), \
729 '{{deleted_at}}', \
730 to_jsonb(NOW())\
731 ) WHERE id = $1::uuid AND (metadata->>'deleted_at') IS NULL",
732 self.table_name()
733 );
734 let result = crate::company_scope::execute_scoped(
735 self.pool(),
736 sqlx::query(&query).bind(id),
737 )
738 .await?;
739 Ok(result.rows_affected() > 0)
740 }
741
742 pub async fn restore(&self, id: &str) -> Result<Option<T>> {
744 let query = format!(
745 "UPDATE {} SET metadata = metadata - 'deleted_at' \
746 WHERE id = $1::uuid AND (metadata->>'deleted_at') IS NOT NULL \
747 RETURNING *",
748 self.table_name()
749 );
750 let result = crate::company_scope::fetch_optional_scoped(
751 self.pool(),
752 sqlx::query_as::<_, T>(&query).bind(id),
753 )
754 .await?;
755 Ok(result)
756 }
757
758 pub async fn list_deleted(&self, pagination: PaginationParams) -> Result<PaginatedResult<T>> {
760 let offset = pagination.offset();
761 let limit = pagination.limit();
762 let query = format!(
763 "SELECT * FROM {} WHERE (metadata->>'deleted_at') IS NOT NULL \
764 ORDER BY (metadata->>'deleted_at') DESC LIMIT $1 OFFSET $2",
765 self.table_name()
766 );
767 let data = crate::company_scope::fetch_all_scoped(
768 self.pool(),
769 sqlx::query_as::<_, T>(&query).bind(limit as i64).bind(offset as i64),
770 )
771 .await?;
772 let count_query = format!(
773 "SELECT COUNT(*) FROM {} WHERE (metadata->>'deleted_at') IS NOT NULL",
774 self.table_name()
775 );
776 let total = crate::company_scope::fetch_one_scalar_scoped(
777 self.pool(),
778 sqlx::query_scalar::<_, i64>(&count_query),
779 )
780 .await? as u64;
781 Ok(PaginatedResult {
782 data,
783 pagination: PaginationInfo::new(pagination.page, pagination.per_page, total),
784 })
785 }
786
787 pub async fn empty_trash(&self) -> Result<u64> {
789 let query = format!(
790 "DELETE FROM {} WHERE (metadata->>'deleted_at') IS NOT NULL",
791 self.table_name()
792 );
793 let result = crate::company_scope::execute_scoped(self.pool(), sqlx::query(&query)).await?;
794 Ok(result.rows_affected())
795 }
796
797 pub async fn find_deleted_by_id(&self, id: &str) -> Result<Option<T>> {
799 let query = format!(
800 "SELECT * FROM {} WHERE id = $1::uuid AND (metadata->>'deleted_at') IS NOT NULL",
801 self.table_name()
802 );
803 let result = crate::company_scope::fetch_optional_scoped(
804 self.pool(),
805 sqlx::query_as::<_, T>(&query).bind(id),
806 )
807 .await?;
808 Ok(result)
809 }
810
811 pub async fn permanent_delete(&self, id: &str) -> Result<bool> {
813 let query = format!(
814 "DELETE FROM {} WHERE id = $1::uuid AND (metadata->>'deleted_at') IS NOT NULL",
815 self.table_name()
816 );
817 let result = crate::company_scope::execute_scoped(
818 self.pool(),
819 sqlx::query(&query).bind(id),
820 )
821 .await?;
822 Ok(result.rows_affected() > 0)
823 }
824
825 pub async fn count_active(&self) -> Result<u64> {
827 let query = format!(
828 "SELECT COUNT(*) FROM {} WHERE (metadata->>'deleted_at') IS NULL",
829 self.table_name()
830 );
831 let count = crate::company_scope::fetch_one_scalar_scoped(
832 self.pool(),
833 sqlx::query_scalar::<_, i64>(&query),
834 )
835 .await? as u64;
836 Ok(count)
837 }
838
839 pub async fn count_deleted(&self) -> Result<u64> {
841 let query = format!(
842 "SELECT COUNT(*) FROM {} WHERE (metadata->>'deleted_at') IS NOT NULL",
843 self.table_name()
844 );
845 let count = crate::company_scope::fetch_one_scalar_scoped(
846 self.pool(),
847 sqlx::query_scalar::<_, i64>(&query),
848 )
849 .await? as u64;
850 Ok(count)
851 }
852
853 pub async fn bulk_soft_delete(&self, ids: &[String]) -> Result<u64> {
862 if ids.is_empty() {
863 return Ok(0);
864 }
865 let placeholders = id_in_placeholders(ids.len());
866 let query = format!(
867 "UPDATE {} SET metadata = jsonb_set(\
868 COALESCE(metadata, '{{}}'), '{{deleted_at}}', to_jsonb(NOW())\
869 ) WHERE id IN ({placeholders}) AND (metadata->>'deleted_at') IS NULL",
870 self.table_name()
871 );
872 let unit = crate::write_tx::begin_unit(self.pool(), false).await?;
873 let outcome: Result<u64> = async {
874 let mut conn = unit.conn().await;
875 let mut q = sqlx::query(&query);
876 for id in ids {
877 q = q.bind(id);
878 }
879 let affected = q.execute(&mut *conn).await?.rows_affected();
880 if affected != ids.len() as u64 {
881 return Err(anyhow::anyhow!(
883 "bulk_soft_delete: {} of {} ids were not active/deletable; rolled back",
884 ids.len() as u64 - affected,
885 ids.len()
886 ));
887 }
888 Ok(affected)
889 }
890 .await;
891 unit.finish(outcome).await
892 }
893
894 pub async fn bulk_restore(&self, ids: &[String]) -> Result<Vec<T>> {
896 if ids.is_empty() {
897 return Ok(Vec::new());
898 }
899 let placeholders = id_in_placeholders(ids.len());
900 let query = format!(
901 "UPDATE {} SET metadata = metadata - 'deleted_at' \
902 WHERE id IN ({placeholders}) AND (metadata->>'deleted_at') IS NOT NULL \
903 RETURNING *",
904 self.table_name()
905 );
906 let unit = crate::write_tx::begin_unit(self.pool(), false).await?;
907 let outcome: Result<Vec<T>> = async {
908 let mut conn = unit.conn().await;
909 let mut q = sqlx::query_as::<_, T>(&query);
910 for id in ids {
911 q = q.bind(id);
912 }
913 let rows = q.fetch_all(&mut *conn).await?;
914 if rows.len() != ids.len() {
915 return Err(anyhow::anyhow!(
916 "bulk_restore: {} of {} ids were not in trash; rolled back",
917 ids.len() - rows.len(),
918 ids.len()
919 ));
920 }
921 Ok(rows)
922 }
923 .await;
924 unit.finish(outcome).await
925 }
926
927 pub async fn bulk_permanent_delete(&self, ids: &[String]) -> Result<u64> {
929 if ids.is_empty() {
930 return Ok(0);
931 }
932 let placeholders = id_in_placeholders(ids.len());
933 let query = format!(
934 "DELETE FROM {} WHERE id IN ({placeholders}) AND (metadata->>'deleted_at') IS NOT NULL",
935 self.table_name()
936 );
937 let unit = crate::write_tx::begin_unit(self.pool(), false).await?;
938 let outcome: Result<u64> = async {
939 let mut conn = unit.conn().await;
940 let mut q = sqlx::query(&query);
941 for id in ids {
942 q = q.bind(id);
943 }
944 let affected = q.execute(&mut *conn).await?.rows_affected();
945 if affected != ids.len() as u64 {
946 return Err(anyhow::anyhow!(
947 "bulk_permanent_delete: {} of {} ids were not in trash; rolled back",
948 ids.len() as u64 - affected,
949 ids.len()
950 ));
951 }
952 Ok(affected)
953 }
954 .await;
955 unit.finish(outcome).await
956 }
957
958 pub async fn restore_all(&self) -> Result<Vec<T>> {
962 let query = format!(
963 "UPDATE {} SET metadata = metadata - 'deleted_at' \
964 WHERE (metadata->>'deleted_at') IS NOT NULL \
965 RETURNING *",
966 self.table_name()
967 );
968 let rows = crate::company_scope::fetch_all_scoped(
969 self.pool(),
970 sqlx::query_as::<_, T>(&query),
971 )
972 .await?;
973 Ok(rows)
974 }
975
976 pub async fn bulk_update(&self, entities: &[T]) -> Result<Vec<T>>
979 where
980 T: Serialize + Send + Sync,
981 {
982 bulk_update_rows(
983 self.pool(),
984 self.table_name(),
985 ACTIVE_ROW_GUARD,
986 entities,
987 )
988 .await
989 }
990
991 pub async fn bulk_update_changed(&self, pairs: &[(T, T)]) -> Result<Vec<T>>
995 where
996 T: Serialize + Send + Sync,
997 {
998 self.bulk_update_changed_guarded(pairs, ACTIVE_ROW_GUARD).await
999 }
1000}
1001
1002fn id_in_placeholders(n: usize) -> String {
1004 (1..=n)
1005 .map(|i| format!("${i}::uuid"))
1006 .collect::<Vec<_>>()
1007 .join(", ")
1008}
1009
1010#[derive(Debug)]
1016pub struct BatchItemError {
1017 pub index: usize,
1019 pub error: anyhow::Error,
1021}
1022
1023impl std::fmt::Display for BatchItemError {
1024 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1025 write!(f, "item {} of the batch: {}; rolled back", self.index, self.error)
1026 }
1027}
1028
1029impl std::error::Error for BatchItemError {
1030 fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
1031 Some(self.error.as_ref())
1032 }
1033}
1034
1035async fn insert_each<T>(conn: &mut sqlx::PgConnection, table: &str, entities: &[T]) -> Result<Vec<T>>
1037where
1038 T: for<'r> FromRow<'r, PgRow> + Serialize + Send + Sync + Unpin,
1039{
1040 let mut out = Vec::with_capacity(entities.len());
1041 for (index, entity) in entities.iter().enumerate() {
1042 let (query, json) = crate::repository::insert_statement(table, entity)
1043 .map_err(|error| BatchItemError { index, error })?;
1044 let statement = match json {
1045 Some(json) => sqlx::query_as::<_, T>(&query).bind(json),
1046 None => sqlx::query_as::<_, T>(&query),
1047 };
1048 let row = statement.fetch_one(&mut *conn).await.map_err(|e| BatchItemError {
1049 index,
1050 error: crate::repository::explain_unknown_column(e, table),
1051 })?;
1052 out.push(row);
1053 }
1054 Ok(out)
1055}
1056
1057async fn update_each_changed<T>(
1060 conn: &mut sqlx::PgConnection,
1061 table: &str,
1062 guard: &str,
1063 pairs: &[(T, T)],
1064) -> Result<Vec<T>>
1065where
1066 T: for<'r> FromRow<'r, PgRow> + Serialize + Send + Sync + Unpin,
1067{
1068 use crate::repository::UpdateStatement;
1069 let mut out = Vec::with_capacity(pairs.len());
1070 for (index, (before, after)) in pairs.iter().enumerate() {
1071 let item = |error: anyhow::Error| BatchItemError { index, error };
1072 let id = match serde_json::to_value(after).map_err(|e| item(e.into()))? {
1073 serde_json::Value::Object(obj) => match obj.get("id") {
1074 Some(serde_json::Value::String(id)) => id.clone(),
1075 _ => return Err(item(anyhow::anyhow!("entity missing string 'id' field")).into()),
1076 },
1077 _ => return Err(item(anyhow::anyhow!("entity must serialize to a JSON object")).into()),
1078 };
1079 let changed = crate::repository::changed_columns(before, after).map_err(item)?;
1080 let updated = match crate::repository::update_statement(table, after, Some(&changed), guard)
1081 .map_err(item)?
1082 {
1083 UpdateStatement::Unchanged(query) => {
1084 sqlx::query_as::<_, T>(&query).bind(&id).fetch_optional(&mut *conn).await
1085 }
1086 UpdateStatement::Update { query, json } => {
1087 sqlx::query_as::<_, T>(&query)
1088 .bind(json)
1089 .bind(&id)
1090 .fetch_optional(&mut *conn)
1091 .await
1092 }
1093 }
1094 .map_err(|e| item(e.into()))?;
1095 match updated {
1096 Some(row) => out.push(row),
1097 None => {
1098 return Err(item(anyhow::anyhow!("id '{id}' not found or already deleted")).into())
1099 }
1100 }
1101 }
1102 Ok(out)
1103}
1104
1105fn build_update_parts<T: Serialize>(entity: &T) -> Result<(String, String, String)> {
1108 let json_value = serde_json::to_value(entity)?;
1109 let json_obj = match json_value {
1110 serde_json::Value::Object(obj) => obj,
1111 _ => return Err(anyhow::anyhow!("entity must serialize to a JSON object")),
1112 };
1113 let id = json_obj
1114 .get("id")
1115 .and_then(|v| v.as_str())
1116 .ok_or_else(|| anyhow::anyhow!("entity missing string 'id' field"))?
1117 .to_string();
1118 let column_names = json_obj
1119 .keys()
1120 .filter(|k| *k != "id")
1121 .map(|k| format!("\"{k}\""))
1122 .collect::<Vec<_>>()
1123 .join(", ");
1124 let json_str = serde_json::to_string(&json_obj)?;
1125 Ok((id, json_str, column_names))
1126}
1127
1128impl<T> GenericCrudRepository<T, HardDelete>
1131where
1132 T: for<'r> FromRow<'r, PgRow> + Send + Sync + Unpin + Serialize,
1133{
1134 pub async fn find_by_text_field(&self, field: &str, value: &str) -> Result<Option<T>> {
1138 self.find_by_text_field_with_cond(field, value, "").await
1139 }
1140
1141 pub async fn exists_by_text_field(&self, field: &str, value: &str) -> Result<bool> {
1143 self.exists_by_text_field_with_cond(field, value, "").await
1144 }
1145
1146 pub async fn find_by_uuid_field(&self, field: &str, value: Uuid) -> Result<Option<T>> {
1148 self.find_by_uuid_field_with_cond(field, value, "").await
1149 }
1150
1151 pub async fn exists_by_uuid_field(&self, field: &str, value: Uuid) -> Result<bool> {
1153 self.exists_by_uuid_field_with_cond(field, value, "").await
1154 }
1155
1156 pub async fn list_paginated_filtered(
1160 &self,
1161 pagination: PaginationParams,
1162 filters: Option<&HashMap<String, String>>,
1163 ) -> Result<PaginatedResult<T>>
1164 where
1165 T: EntityRepoMeta + Send + Sync,
1166 {
1167 let filters_map = filters.cloned().unwrap_or_default();
1168 let column_types = T::column_types();
1169 let search_fields_owned: Vec<&'static str> = T::search_fields().iter().copied().collect();
1170 self.run_filtered_query(pagination, None, &filters_map, &column_types, &search_fields_owned).await
1171 }
1172
1173 pub async fn aggregate_filtered(
1175 &self,
1176 spec: &crate::repository::AggregateSpec,
1177 filters: Option<&HashMap<String, String>>,
1178 ) -> Result<crate::repository::AggregateResult>
1179 where
1180 T: EntityRepoMeta + Send + Sync,
1181 {
1182 let filters_map = filters.cloned().unwrap_or_default();
1183 let column_types = T::column_types();
1184 let search_fields_owned: Vec<&'static str> = T::search_fields().iter().copied().collect();
1185 self.run_aggregate_query(spec, None, &filters_map, &column_types, &search_fields_owned).await
1186 }
1187
1188 pub async fn find_by_id(&self, id: &str) -> Result<Option<T>> {
1190 self.inner.find_by_id(id).await
1191 }
1192
1193 pub async fn find_all(&self) -> Result<Vec<T>> {
1195 let query = format!("SELECT * FROM {}", self.table_name());
1196 let results = crate::company_scope::fetch_all_scoped(
1197 self.pool(),
1198 sqlx::query_as::<_, T>(&query),
1199 )
1200 .await?;
1201 Ok(results)
1202 }
1203
1204 pub async fn update(&self, id: &str, entity: &T) -> Result<Option<T>> {
1206 self.inner.update(id, entity).await
1207 }
1208
1209 pub async fn update_changed(&self, id: &str, before: &T, after: &T) -> Result<Option<T>> {
1213 let changed = crate::repository::changed_columns(before, after)?;
1214 self.inner
1215 .update_guarded(id, after, Some(&changed), "")
1216 .await
1217 }
1218
1219 pub async fn delete(&self, id: &str) -> Result<bool> {
1221 self.inner.delete(id).await
1222 }
1223
1224 pub async fn count(&self) -> Result<u64> {
1226 let query = format!("SELECT COUNT(*) FROM {}", self.table_name());
1227 let count = crate::company_scope::fetch_one_scalar_scoped(
1228 self.pool(),
1229 sqlx::query_scalar::<_, i64>(&query),
1230 )
1231 .await? as u64;
1232 Ok(count)
1233 }
1234
1235 pub async fn exists(&self, id: &str) -> Result<bool> {
1237 let query = format!(
1238 "SELECT 1 FROM {} WHERE id = $1::uuid LIMIT 1",
1239 self.table_name()
1240 );
1241 let result = crate::company_scope::fetch_optional_scalar_scoped(
1242 self.pool(),
1243 sqlx::query_scalar::<_, i32>(&query).bind(id),
1244 )
1245 .await?;
1246 Ok(result.is_some())
1247 }
1248
1249 pub async fn list_paginated(&self, pagination: PaginationParams) -> Result<PaginatedResult<T>> {
1251 let offset = pagination.offset();
1252 let limit = pagination.limit();
1253 let query = format!(
1254 "SELECT * FROM {} ORDER BY id DESC LIMIT $1 OFFSET $2",
1255 self.table_name()
1256 );
1257 let data = crate::company_scope::fetch_all_scoped(
1258 self.pool(),
1259 sqlx::query_as::<_, T>(&query).bind(limit as i64).bind(offset as i64),
1260 )
1261 .await?;
1262 let total = self.count().await?;
1263 Ok(PaginatedResult {
1264 data,
1265 pagination: PaginationInfo::new(pagination.page, pagination.per_page, total),
1266 })
1267 }
1268
1269 pub async fn bulk_delete(&self, ids: &[String]) -> Result<u64> {
1273 if ids.is_empty() {
1274 return Ok(0);
1275 }
1276 let placeholders = id_in_placeholders(ids.len());
1277 let query = format!(
1278 "DELETE FROM {} WHERE id IN ({placeholders})",
1279 self.table_name()
1280 );
1281 let unit = crate::write_tx::begin_unit(self.pool(), false).await?;
1282 let outcome: Result<u64> = async {
1283 let mut conn = unit.conn().await;
1284 let mut q = sqlx::query(&query);
1285 for id in ids {
1286 q = q.bind(id);
1287 }
1288 let affected = q.execute(&mut *conn).await?.rows_affected();
1289 if affected != ids.len() as u64 {
1290 return Err(anyhow::anyhow!(
1291 "bulk_delete: {} of {} ids not found; rolled back",
1292 ids.len() as u64 - affected,
1293 ids.len()
1294 ));
1295 }
1296 Ok(affected)
1297 }
1298 .await;
1299 unit.finish(outcome).await
1300 }
1301
1302 pub async fn bulk_update(&self, entities: &[T]) -> Result<Vec<T>> {
1304 bulk_update_rows(self.pool(), self.table_name(), "", entities).await
1305 }
1306
1307 pub async fn bulk_update_changed(&self, pairs: &[(T, T)]) -> Result<Vec<T>> {
1310 self.bulk_update_changed_guarded(pairs, "").await
1311 }
1312}
1313
1314pub fn qualify_relation_table(caller_table: &str, target_table: &str) -> String {
1324 if target_table.contains('.') || !caller_table.contains('.') {
1325 target_table.to_string()
1326 } else {
1327 let schema = caller_table.split('.').next().unwrap_or(caller_table);
1328 format!("{schema}.{target_table}")
1329 }
1330}
1331
1332fn is_undefined_table(err: &anyhow::Error) -> bool {
1337 err.chain()
1338 .filter_map(|cause| cause.downcast_ref::<sqlx::Error>())
1339 .any(|sqlx_err| {
1340 sqlx_err
1341 .as_database_error()
1342 .map(|db| db.code().as_deref() == Some("42P01"))
1343 .unwrap_or(false)
1344 })
1345}
1346
1347pub async fn fetch_by_ids_as_json(
1362 pool: &PgPool,
1363 caller_table: &str,
1364 table: &str,
1365 ids: &[String],
1366) -> Result<Vec<serde_json::Value>> {
1367 if ids.is_empty() {
1368 return Ok(Vec::new());
1369 }
1370 let qualified = qualify_relation_table(caller_table, table);
1371 match fetch_rows_as_json(pool, &qualified, ids).await {
1372 Ok(rows) => Ok(rows),
1373 Err(err) if qualified != table && is_undefined_table(&err) => {
1374 fetch_rows_as_json(pool, table, ids).await
1375 }
1376 Err(err) => Err(err),
1377 }
1378}
1379
1380pub async fn reference_visible(
1394 pool: &PgPool,
1395 caller_table: &str,
1396 table: &str,
1397 column: &str,
1398 value: &serde_json::Value,
1399) -> Result<Option<bool>> {
1400 let plain = |s: &str, dot: bool| {
1401 !s.is_empty() && s.chars().all(|c| c.is_ascii_alphanumeric() || c == '_' || (dot && c == '.'))
1402 };
1403 if !plain(table, true) || !plain(column, false) {
1404 anyhow::bail!("reference target `{table}`.`{column}` is not a plain identifier");
1405 }
1406 let qualified = qualify_relation_table(caller_table, table);
1407 match row_visible(pool, &qualified, column, value).await {
1408 Ok(seen) => Ok(Some(seen)),
1409 Err(err) if is_undefined_table(&err) => {
1410 if qualified == table {
1411 return Ok(None);
1412 }
1413 match row_visible(pool, table, column, value).await {
1414 Ok(seen) => Ok(Some(seen)),
1415 Err(err) if is_undefined_table(&err) => Ok(None),
1416 Err(err) => Err(err),
1417 }
1418 }
1419 Err(err) => Err(err),
1420 }
1421}
1422
1423async fn row_visible(
1424 pool: &PgPool,
1425 table: &str,
1426 column: &str,
1427 value: &serde_json::Value,
1428) -> Result<bool> {
1429 let query = format!(
1430 "SELECT EXISTS (SELECT 1 FROM {table} t \
1431 WHERE t.{column} = (jsonb_populate_record(NULL::{table}, $1::jsonb)).{column})"
1432 );
1433 let probe = serde_json::json!({ column: value });
1434 Ok(crate::write_tx::within_savepoint(crate::company_scope::fetch_one_scalar_scoped(
1437 pool,
1438 sqlx::query_scalar::<_, bool>(&query).bind(probe),
1439 ))
1440 .await?)
1441}
1442
1443async fn fetch_rows_as_json(
1448 pool: &PgPool,
1449 table: &str,
1450 ids: &[String],
1451) -> Result<Vec<serde_json::Value>> {
1452 let query = format!("SELECT row_to_json(t) AS j FROM {table} t WHERE t.id = ANY($1::uuid[])");
1453 let rows: Vec<(serde_json::Value,)> =
1454 crate::company_scope::fetch_all_scoped(pool, sqlx::query_as(&query).bind(ids)).await?;
1455 Ok(rows.into_iter().map(|(j,)| j).collect())
1456}
1457
1458async fn bulk_update_rows<T>(
1463 pool: &PgPool,
1464 table: &str,
1465 active_guard: &str,
1466 entities: &[T],
1467) -> Result<Vec<T>>
1468where
1469 T: for<'r> FromRow<'r, PgRow> + Send + Sync + Unpin + Serialize,
1470{
1471 if entities.is_empty() {
1472 return Ok(Vec::new());
1473 }
1474 let unit = crate::write_tx::begin_unit(pool, false).await?;
1475 let outcome: Result<Vec<T>> = async {
1476 let mut conn = unit.conn().await;
1477 let mut out = Vec::with_capacity(entities.len());
1478 for entity in entities {
1479 let (id, json_str, column_names) = build_update_parts(entity)?;
1480 let query = format!(
1481 "WITH new_row AS (\
1482 SELECT (jsonb_populate_record(NULL::{table}, $1::jsonb)).*\
1483 ) UPDATE {table} AS t \
1484 SET ({columns}) = (SELECT {columns} FROM new_row) \
1485 WHERE t.id = $2::uuid{guard} \
1486 RETURNING t.*",
1487 table = table,
1488 columns = column_names,
1489 guard = active_guard,
1490 );
1491 let updated = sqlx::query_as::<_, T>(&query)
1492 .bind(&json_str)
1493 .bind(&id)
1494 .fetch_optional(&mut *conn)
1495 .await?;
1496 match updated {
1497 Some(e) => out.push(e),
1498 None => {
1499 return Err(anyhow::anyhow!(
1500 "bulk_update: id '{id}' not found or already deleted; rolled back"
1501 ));
1502 }
1503 }
1504 }
1505 Ok(out)
1506 }
1507 .await;
1508 unit.finish(outcome).await
1509}
1510
1511fn camel_to_snake(s: &str) -> String {
1513 let mut out = String::with_capacity(s.len() + 4);
1514 for ch in s.chars() {
1515 if ch.is_uppercase() {
1516 out.push('_');
1517 out.extend(ch.to_lowercase());
1518 } else {
1519 out.push(ch);
1520 }
1521 }
1522 out
1523}