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()).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()).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>> {
704 let offset = pagination.offset();
705 let limit = pagination.limit();
706 let query = format!(
707 "SELECT * FROM {} WHERE metadata->>'deleted_at' IS NULL \
708 ORDER BY id LIMIT $1 OFFSET $2",
709 self.table_name()
710 );
711 let data = crate::company_scope::fetch_all_scoped(
712 self.pool(),
713 sqlx::query_as::<_, T>(&query).bind(limit as i64).bind(offset as i64),
714 )
715 .await?;
716 let total = self.count_active().await?;
717 Ok(PaginatedResult {
718 data,
719 pagination: PaginationInfo::new(pagination.page, pagination.per_page, total),
720 })
721 }
722
723 pub async fn soft_delete(&self, id: &str) -> Result<bool> {
727 let query = format!(
728 "UPDATE {} SET metadata = jsonb_set(\
729 COALESCE(metadata, '{{}}'), \
730 '{{deleted_at}}', \
731 to_jsonb(NOW())\
732 ) WHERE id = $1::uuid AND (metadata->>'deleted_at') IS NULL",
733 self.table_name()
734 );
735 let result = crate::company_scope::execute_scoped(
736 self.pool(),
737 sqlx::query(&query).bind(id),
738 )
739 .await?;
740 Ok(result.rows_affected() > 0)
741 }
742
743 pub async fn restore(&self, id: &str) -> Result<Option<T>> {
745 let query = format!(
746 "UPDATE {} SET metadata = metadata - 'deleted_at' \
747 WHERE id = $1::uuid AND (metadata->>'deleted_at') IS NOT NULL \
748 RETURNING *",
749 self.table_name()
750 );
751 let result = crate::company_scope::fetch_optional_scoped(
752 self.pool(),
753 sqlx::query_as::<_, T>(&query).bind(id),
754 )
755 .await?;
756 Ok(result)
757 }
758
759 pub async fn list_deleted(&self, pagination: PaginationParams) -> Result<PaginatedResult<T>> {
762 let offset = pagination.offset();
763 let limit = pagination.limit();
764 let query = format!(
765 "SELECT * FROM {} WHERE (metadata->>'deleted_at') IS NOT NULL \
766 ORDER BY (metadata->>'deleted_at') DESC, id LIMIT $1 OFFSET $2",
767 self.table_name()
768 );
769 let data = crate::company_scope::fetch_all_scoped(
770 self.pool(),
771 sqlx::query_as::<_, T>(&query).bind(limit as i64).bind(offset as i64),
772 )
773 .await?;
774 let count_query = format!(
775 "SELECT COUNT(*) FROM {} WHERE (metadata->>'deleted_at') IS NOT NULL",
776 self.table_name()
777 );
778 let total = crate::company_scope::fetch_one_scalar_scoped(
779 self.pool(),
780 sqlx::query_scalar::<_, i64>(&count_query),
781 )
782 .await? as u64;
783 Ok(PaginatedResult {
784 data,
785 pagination: PaginationInfo::new(pagination.page, pagination.per_page, total),
786 })
787 }
788
789 pub async fn empty_trash(&self) -> Result<u64> {
791 let query = format!(
792 "DELETE FROM {} WHERE (metadata->>'deleted_at') IS NOT NULL",
793 self.table_name()
794 );
795 let result = crate::company_scope::execute_scoped(self.pool(), sqlx::query(&query)).await?;
796 Ok(result.rows_affected())
797 }
798
799 pub async fn find_deleted_by_id(&self, id: &str) -> Result<Option<T>> {
801 let query = format!(
802 "SELECT * FROM {} WHERE id = $1::uuid AND (metadata->>'deleted_at') IS NOT NULL",
803 self.table_name()
804 );
805 let result = crate::company_scope::fetch_optional_scoped(
806 self.pool(),
807 sqlx::query_as::<_, T>(&query).bind(id),
808 )
809 .await?;
810 Ok(result)
811 }
812
813 pub async fn permanent_delete(&self, id: &str) -> Result<bool> {
815 let query = format!(
816 "DELETE FROM {} WHERE id = $1::uuid AND (metadata->>'deleted_at') IS NOT NULL",
817 self.table_name()
818 );
819 let result = crate::company_scope::execute_scoped(
820 self.pool(),
821 sqlx::query(&query).bind(id),
822 )
823 .await?;
824 Ok(result.rows_affected() > 0)
825 }
826
827 pub async fn count_active(&self) -> Result<u64> {
829 let query = format!(
830 "SELECT COUNT(*) FROM {} WHERE (metadata->>'deleted_at') IS NULL",
831 self.table_name()
832 );
833 let count = crate::company_scope::fetch_one_scalar_scoped(
834 self.pool(),
835 sqlx::query_scalar::<_, i64>(&query),
836 )
837 .await? as u64;
838 Ok(count)
839 }
840
841 pub async fn count_deleted(&self) -> Result<u64> {
843 let query = format!(
844 "SELECT COUNT(*) FROM {} WHERE (metadata->>'deleted_at') IS NOT NULL",
845 self.table_name()
846 );
847 let count = crate::company_scope::fetch_one_scalar_scoped(
848 self.pool(),
849 sqlx::query_scalar::<_, i64>(&query),
850 )
851 .await? as u64;
852 Ok(count)
853 }
854
855 pub async fn bulk_soft_delete(&self, ids: &[String]) -> Result<u64> {
864 if ids.is_empty() {
865 return Ok(0);
866 }
867 let placeholders = id_in_placeholders(ids.len());
868 let query = format!(
869 "UPDATE {} SET metadata = jsonb_set(\
870 COALESCE(metadata, '{{}}'), '{{deleted_at}}', to_jsonb(NOW())\
871 ) WHERE id IN ({placeholders}) AND (metadata->>'deleted_at') IS NULL",
872 self.table_name()
873 );
874 let unit = crate::write_tx::begin_unit(self.pool()).await?;
875 let outcome: Result<u64> = async {
876 let mut conn = unit.conn().await;
877 let mut q = sqlx::query(&query);
878 for id in ids {
879 q = q.bind(id);
880 }
881 let affected = q.execute(&mut *conn).await?.rows_affected();
882 if affected != ids.len() as u64 {
883 return Err(anyhow::anyhow!(
885 "bulk_soft_delete: {} of {} ids were not active/deletable; rolled back",
886 ids.len() as u64 - affected,
887 ids.len()
888 ));
889 }
890 Ok(affected)
891 }
892 .await;
893 unit.finish(outcome).await
894 }
895
896 pub async fn bulk_restore(&self, ids: &[String]) -> Result<Vec<T>> {
898 if ids.is_empty() {
899 return Ok(Vec::new());
900 }
901 let placeholders = id_in_placeholders(ids.len());
902 let query = format!(
903 "UPDATE {} SET metadata = metadata - 'deleted_at' \
904 WHERE id IN ({placeholders}) AND (metadata->>'deleted_at') IS NOT NULL \
905 RETURNING *",
906 self.table_name()
907 );
908 let unit = crate::write_tx::begin_unit(self.pool()).await?;
909 let outcome: Result<Vec<T>> = async {
910 let mut conn = unit.conn().await;
911 let mut q = sqlx::query_as::<_, T>(&query);
912 for id in ids {
913 q = q.bind(id);
914 }
915 let rows = q.fetch_all(&mut *conn).await?;
916 if rows.len() != ids.len() {
917 return Err(anyhow::anyhow!(
918 "bulk_restore: {} of {} ids were not in trash; rolled back",
919 ids.len() - rows.len(),
920 ids.len()
921 ));
922 }
923 Ok(rows)
924 }
925 .await;
926 unit.finish(outcome).await
927 }
928
929 pub async fn bulk_permanent_delete(&self, ids: &[String]) -> Result<u64> {
931 if ids.is_empty() {
932 return Ok(0);
933 }
934 let placeholders = id_in_placeholders(ids.len());
935 let query = format!(
936 "DELETE FROM {} WHERE id IN ({placeholders}) AND (metadata->>'deleted_at') IS NOT NULL",
937 self.table_name()
938 );
939 let unit = crate::write_tx::begin_unit(self.pool()).await?;
940 let outcome: Result<u64> = async {
941 let mut conn = unit.conn().await;
942 let mut q = sqlx::query(&query);
943 for id in ids {
944 q = q.bind(id);
945 }
946 let affected = q.execute(&mut *conn).await?.rows_affected();
947 if affected != ids.len() as u64 {
948 return Err(anyhow::anyhow!(
949 "bulk_permanent_delete: {} of {} ids were not in trash; rolled back",
950 ids.len() as u64 - affected,
951 ids.len()
952 ));
953 }
954 Ok(affected)
955 }
956 .await;
957 unit.finish(outcome).await
958 }
959
960 pub async fn restore_all(&self) -> Result<Vec<T>> {
964 let query = format!(
965 "UPDATE {} SET metadata = metadata - 'deleted_at' \
966 WHERE (metadata->>'deleted_at') IS NOT NULL \
967 RETURNING *",
968 self.table_name()
969 );
970 let rows = crate::company_scope::fetch_all_scoped(
971 self.pool(),
972 sqlx::query_as::<_, T>(&query),
973 )
974 .await?;
975 Ok(rows)
976 }
977
978 pub async fn bulk_update(&self, entities: &[T]) -> Result<Vec<T>>
981 where
982 T: Serialize + Send + Sync,
983 {
984 bulk_update_rows(
985 self.pool(),
986 self.table_name(),
987 ACTIVE_ROW_GUARD,
988 entities,
989 )
990 .await
991 }
992
993 pub async fn bulk_update_changed(&self, pairs: &[(T, T)]) -> Result<Vec<T>>
997 where
998 T: Serialize + Send + Sync,
999 {
1000 self.bulk_update_changed_guarded(pairs, ACTIVE_ROW_GUARD).await
1001 }
1002}
1003
1004fn id_in_placeholders(n: usize) -> String {
1006 (1..=n)
1007 .map(|i| format!("${i}::uuid"))
1008 .collect::<Vec<_>>()
1009 .join(", ")
1010}
1011
1012#[derive(Debug)]
1018pub struct BatchItemError {
1019 pub index: usize,
1021 pub error: anyhow::Error,
1023}
1024
1025impl std::fmt::Display for BatchItemError {
1026 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1027 write!(f, "item {} of the batch: {}; rolled back", self.index, self.error)
1028 }
1029}
1030
1031impl std::error::Error for BatchItemError {
1032 fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
1033 Some(self.error.as_ref())
1034 }
1035}
1036
1037async fn insert_each<T>(conn: &mut sqlx::PgConnection, table: &str, entities: &[T]) -> Result<Vec<T>>
1039where
1040 T: for<'r> FromRow<'r, PgRow> + Serialize + Send + Sync + Unpin,
1041{
1042 let mut out = Vec::with_capacity(entities.len());
1043 for (index, entity) in entities.iter().enumerate() {
1044 let (query, json) = crate::repository::insert_statement(table, entity)
1045 .map_err(|error| BatchItemError { index, error })?;
1046 let statement = match json {
1047 Some(json) => sqlx::query_as::<_, T>(&query).bind(json),
1048 None => sqlx::query_as::<_, T>(&query),
1049 };
1050 let row = statement.fetch_one(&mut *conn).await.map_err(|e| BatchItemError {
1051 index,
1052 error: crate::repository::explain_unknown_column(e, table),
1053 })?;
1054 out.push(row);
1055 }
1056 Ok(out)
1057}
1058
1059async fn update_each_changed<T>(
1062 conn: &mut sqlx::PgConnection,
1063 table: &str,
1064 guard: &str,
1065 pairs: &[(T, T)],
1066) -> Result<Vec<T>>
1067where
1068 T: for<'r> FromRow<'r, PgRow> + Serialize + Send + Sync + Unpin,
1069{
1070 use crate::repository::UpdateStatement;
1071 let mut out = Vec::with_capacity(pairs.len());
1072 for (index, (before, after)) in pairs.iter().enumerate() {
1073 let item = |error: anyhow::Error| BatchItemError { index, error };
1074 let id = match serde_json::to_value(after).map_err(|e| item(e.into()))? {
1075 serde_json::Value::Object(obj) => match obj.get("id") {
1076 Some(serde_json::Value::String(id)) => id.clone(),
1077 _ => return Err(item(anyhow::anyhow!("entity missing string 'id' field")).into()),
1078 },
1079 _ => return Err(item(anyhow::anyhow!("entity must serialize to a JSON object")).into()),
1080 };
1081 let changed = crate::repository::changed_columns(before, after).map_err(item)?;
1082 let updated = match crate::repository::update_statement(table, after, Some(&changed), guard)
1083 .map_err(item)?
1084 {
1085 UpdateStatement::Unchanged(query) => {
1086 sqlx::query_as::<_, T>(&query).bind(&id).fetch_optional(&mut *conn).await
1087 }
1088 UpdateStatement::Update { query, json } => {
1089 sqlx::query_as::<_, T>(&query)
1090 .bind(json)
1091 .bind(&id)
1092 .fetch_optional(&mut *conn)
1093 .await
1094 }
1095 }
1096 .map_err(|e| item(e.into()))?;
1097 match updated {
1098 Some(row) => out.push(row),
1099 None => {
1100 return Err(item(anyhow::anyhow!("id '{id}' not found or already deleted")).into())
1101 }
1102 }
1103 }
1104 Ok(out)
1105}
1106
1107fn build_update_parts<T: Serialize>(entity: &T) -> Result<(String, String, String)> {
1110 let json_value = serde_json::to_value(entity)?;
1111 let json_obj = match json_value {
1112 serde_json::Value::Object(obj) => obj,
1113 _ => return Err(anyhow::anyhow!("entity must serialize to a JSON object")),
1114 };
1115 let id = json_obj
1116 .get("id")
1117 .and_then(|v| v.as_str())
1118 .ok_or_else(|| anyhow::anyhow!("entity missing string 'id' field"))?
1119 .to_string();
1120 let column_names = json_obj
1121 .keys()
1122 .filter(|k| *k != "id")
1123 .map(|k| format!("\"{k}\""))
1124 .collect::<Vec<_>>()
1125 .join(", ");
1126 let json_str = serde_json::to_string(&json_obj)?;
1127 Ok((id, json_str, column_names))
1128}
1129
1130impl<T> GenericCrudRepository<T, HardDelete>
1133where
1134 T: for<'r> FromRow<'r, PgRow> + Send + Sync + Unpin + Serialize,
1135{
1136 pub async fn find_by_text_field(&self, field: &str, value: &str) -> Result<Option<T>> {
1140 self.find_by_text_field_with_cond(field, value, "").await
1141 }
1142
1143 pub async fn exists_by_text_field(&self, field: &str, value: &str) -> Result<bool> {
1145 self.exists_by_text_field_with_cond(field, value, "").await
1146 }
1147
1148 pub async fn find_by_uuid_field(&self, field: &str, value: Uuid) -> Result<Option<T>> {
1150 self.find_by_uuid_field_with_cond(field, value, "").await
1151 }
1152
1153 pub async fn exists_by_uuid_field(&self, field: &str, value: Uuid) -> Result<bool> {
1155 self.exists_by_uuid_field_with_cond(field, value, "").await
1156 }
1157
1158 pub async fn list_paginated_filtered(
1162 &self,
1163 pagination: PaginationParams,
1164 filters: Option<&HashMap<String, String>>,
1165 ) -> Result<PaginatedResult<T>>
1166 where
1167 T: EntityRepoMeta + Send + Sync,
1168 {
1169 let filters_map = filters.cloned().unwrap_or_default();
1170 let column_types = T::column_types();
1171 let search_fields_owned: Vec<&'static str> = T::search_fields().iter().copied().collect();
1172 self.run_filtered_query(pagination, None, &filters_map, &column_types, &search_fields_owned).await
1173 }
1174
1175 pub async fn aggregate_filtered(
1177 &self,
1178 spec: &crate::repository::AggregateSpec,
1179 filters: Option<&HashMap<String, String>>,
1180 ) -> Result<crate::repository::AggregateResult>
1181 where
1182 T: EntityRepoMeta + Send + Sync,
1183 {
1184 let filters_map = filters.cloned().unwrap_or_default();
1185 let column_types = T::column_types();
1186 let search_fields_owned: Vec<&'static str> = T::search_fields().iter().copied().collect();
1187 self.run_aggregate_query(spec, None, &filters_map, &column_types, &search_fields_owned).await
1188 }
1189
1190 pub async fn find_by_id(&self, id: &str) -> Result<Option<T>> {
1192 self.inner.find_by_id(id).await
1193 }
1194
1195 pub async fn find_all(&self) -> Result<Vec<T>> {
1197 let query = format!("SELECT * FROM {}", self.table_name());
1198 let results = crate::company_scope::fetch_all_scoped(
1199 self.pool(),
1200 sqlx::query_as::<_, T>(&query),
1201 )
1202 .await?;
1203 Ok(results)
1204 }
1205
1206 pub async fn update(&self, id: &str, entity: &T) -> Result<Option<T>> {
1208 self.inner.update(id, entity).await
1209 }
1210
1211 pub async fn update_changed(&self, id: &str, before: &T, after: &T) -> Result<Option<T>> {
1215 let changed = crate::repository::changed_columns(before, after)?;
1216 self.inner
1217 .update_guarded(id, after, Some(&changed), "")
1218 .await
1219 }
1220
1221 pub async fn delete(&self, id: &str) -> Result<bool> {
1223 self.inner.delete(id).await
1224 }
1225
1226 pub async fn count(&self) -> Result<u64> {
1228 let query = format!("SELECT COUNT(*) FROM {}", self.table_name());
1229 let count = crate::company_scope::fetch_one_scalar_scoped(
1230 self.pool(),
1231 sqlx::query_scalar::<_, i64>(&query),
1232 )
1233 .await? as u64;
1234 Ok(count)
1235 }
1236
1237 pub async fn exists(&self, id: &str) -> Result<bool> {
1239 let query = format!(
1240 "SELECT 1 FROM {} WHERE id = $1::uuid LIMIT 1",
1241 self.table_name()
1242 );
1243 let result = crate::company_scope::fetch_optional_scalar_scoped(
1244 self.pool(),
1245 sqlx::query_scalar::<_, i32>(&query).bind(id),
1246 )
1247 .await?;
1248 Ok(result.is_some())
1249 }
1250
1251 pub async fn list_paginated(&self, pagination: PaginationParams) -> Result<PaginatedResult<T>> {
1254 let offset = pagination.offset();
1255 let limit = pagination.limit();
1256 let query = format!(
1257 "SELECT * FROM {} ORDER BY id LIMIT $1 OFFSET $2",
1258 self.table_name()
1259 );
1260 let data = crate::company_scope::fetch_all_scoped(
1261 self.pool(),
1262 sqlx::query_as::<_, T>(&query).bind(limit as i64).bind(offset as i64),
1263 )
1264 .await?;
1265 let total = self.count().await?;
1266 Ok(PaginatedResult {
1267 data,
1268 pagination: PaginationInfo::new(pagination.page, pagination.per_page, total),
1269 })
1270 }
1271
1272 pub async fn bulk_delete(&self, ids: &[String]) -> Result<u64> {
1276 if ids.is_empty() {
1277 return Ok(0);
1278 }
1279 let placeholders = id_in_placeholders(ids.len());
1280 let query = format!(
1281 "DELETE FROM {} WHERE id IN ({placeholders})",
1282 self.table_name()
1283 );
1284 let unit = crate::write_tx::begin_unit(self.pool()).await?;
1285 let outcome: Result<u64> = async {
1286 let mut conn = unit.conn().await;
1287 let mut q = sqlx::query(&query);
1288 for id in ids {
1289 q = q.bind(id);
1290 }
1291 let affected = q.execute(&mut *conn).await?.rows_affected();
1292 if affected != ids.len() as u64 {
1293 return Err(anyhow::anyhow!(
1294 "bulk_delete: {} of {} ids not found; rolled back",
1295 ids.len() as u64 - affected,
1296 ids.len()
1297 ));
1298 }
1299 Ok(affected)
1300 }
1301 .await;
1302 unit.finish(outcome).await
1303 }
1304
1305 pub async fn bulk_update(&self, entities: &[T]) -> Result<Vec<T>> {
1307 bulk_update_rows(self.pool(), self.table_name(), "", entities).await
1308 }
1309
1310 pub async fn bulk_update_changed(&self, pairs: &[(T, T)]) -> Result<Vec<T>> {
1313 self.bulk_update_changed_guarded(pairs, "").await
1314 }
1315}
1316
1317pub fn qualify_relation_table(caller_table: &str, target_table: &str) -> String {
1327 if target_table.contains('.') || !caller_table.contains('.') {
1328 target_table.to_string()
1329 } else {
1330 let schema = caller_table.split('.').next().unwrap_or(caller_table);
1331 format!("{schema}.{target_table}")
1332 }
1333}
1334
1335fn is_undefined_table(err: &anyhow::Error) -> bool {
1340 err.chain()
1341 .filter_map(|cause| cause.downcast_ref::<sqlx::Error>())
1342 .any(|sqlx_err| {
1343 sqlx_err
1344 .as_database_error()
1345 .map(|db| db.code().as_deref() == Some("42P01"))
1346 .unwrap_or(false)
1347 })
1348}
1349
1350pub async fn fetch_by_ids_as_json(
1365 pool: &PgPool,
1366 caller_table: &str,
1367 table: &str,
1368 ids: &[String],
1369) -> Result<Vec<serde_json::Value>> {
1370 if ids.is_empty() {
1371 return Ok(Vec::new());
1372 }
1373 let qualified = qualify_relation_table(caller_table, table);
1374 match fetch_rows_as_json(pool, &qualified, ids).await {
1375 Ok(rows) => Ok(rows),
1376 Err(err) if qualified != table && is_undefined_table(&err) => {
1377 fetch_rows_as_json(pool, table, ids).await
1378 }
1379 Err(err) => Err(err),
1380 }
1381}
1382
1383pub async fn reference_visible(
1397 pool: &PgPool,
1398 caller_table: &str,
1399 table: &str,
1400 column: &str,
1401 value: &serde_json::Value,
1402) -> Result<Option<bool>> {
1403 let plain = |s: &str, dot: bool| {
1404 !s.is_empty() && s.chars().all(|c| c.is_ascii_alphanumeric() || c == '_' || (dot && c == '.'))
1405 };
1406 if !plain(table, true) || !plain(column, false) {
1407 anyhow::bail!("reference target `{table}`.`{column}` is not a plain identifier");
1408 }
1409 let qualified = qualify_relation_table(caller_table, table);
1410 match row_visible(pool, &qualified, column, value).await {
1411 Ok(seen) => Ok(Some(seen)),
1412 Err(err) if is_undefined_table(&err) => {
1413 if qualified == table {
1414 return Ok(None);
1415 }
1416 match row_visible(pool, table, column, value).await {
1417 Ok(seen) => Ok(Some(seen)),
1418 Err(err) if is_undefined_table(&err) => Ok(None),
1419 Err(err) => Err(err),
1420 }
1421 }
1422 Err(err) => Err(err),
1423 }
1424}
1425
1426async fn row_visible(
1427 pool: &PgPool,
1428 table: &str,
1429 column: &str,
1430 value: &serde_json::Value,
1431) -> Result<bool> {
1432 let query = format!(
1433 "SELECT EXISTS (SELECT 1 FROM {table} t \
1434 WHERE t.{column} = (jsonb_populate_record(NULL::{table}, $1::jsonb)).{column})"
1435 );
1436 let probe = serde_json::json!({ column: value });
1437 Ok(crate::write_tx::within_savepoint(crate::company_scope::fetch_one_scalar_scoped(
1440 pool,
1441 sqlx::query_scalar::<_, bool>(&query).bind(probe),
1442 ))
1443 .await?)
1444}
1445
1446async fn fetch_rows_as_json(
1451 pool: &PgPool,
1452 table: &str,
1453 ids: &[String],
1454) -> Result<Vec<serde_json::Value>> {
1455 let query = format!("SELECT row_to_json(t) AS j FROM {table} t WHERE t.id = ANY($1::uuid[])");
1456 let rows: Vec<(serde_json::Value,)> =
1457 crate::company_scope::fetch_all_scoped(pool, sqlx::query_as(&query).bind(ids)).await?;
1458 Ok(rows.into_iter().map(|(j,)| j).collect())
1459}
1460
1461async fn bulk_update_rows<T>(
1466 pool: &PgPool,
1467 table: &str,
1468 active_guard: &str,
1469 entities: &[T],
1470) -> Result<Vec<T>>
1471where
1472 T: for<'r> FromRow<'r, PgRow> + Send + Sync + Unpin + Serialize,
1473{
1474 if entities.is_empty() {
1475 return Ok(Vec::new());
1476 }
1477 let unit = crate::write_tx::begin_unit(pool).await?;
1478 let outcome: Result<Vec<T>> = async {
1479 let mut conn = unit.conn().await;
1480 let mut out = Vec::with_capacity(entities.len());
1481 for entity in entities {
1482 let (id, json_str, column_names) = build_update_parts(entity)?;
1483 let query = format!(
1484 "WITH new_row AS (\
1485 SELECT (jsonb_populate_record(NULL::{table}, $1::jsonb)).*\
1486 ) UPDATE {table} AS t \
1487 SET ({columns}) = (SELECT {columns} FROM new_row) \
1488 WHERE t.id = $2::uuid{guard} \
1489 RETURNING t.*",
1490 table = table,
1491 columns = column_names,
1492 guard = active_guard,
1493 );
1494 let updated = sqlx::query_as::<_, T>(&query)
1495 .bind(&json_str)
1496 .bind(&id)
1497 .fetch_optional(&mut *conn)
1498 .await?;
1499 match updated {
1500 Some(e) => out.push(e),
1501 None => {
1502 return Err(anyhow::anyhow!(
1503 "bulk_update: id '{id}' not found or already deleted; rolled back"
1504 ));
1505 }
1506 }
1507 }
1508 Ok(out)
1509 }
1510 .await;
1511 unit.finish(outcome).await
1512}
1513
1514fn camel_to_snake(s: &str) -> String {
1516 let mut out = String::with_capacity(s.len() + 4);
1517 for ch in s.chars() {
1518 if ch.is_uppercase() {
1519 out.push('_');
1520 out.extend(ch.to_lowercase());
1521 } else {
1522 out.push(ch);
1523 }
1524 }
1525 out
1526}