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 if let Some(conn) = crate::company_scope::current_request_conn() {
280 let mut held = conn.lock().await;
281 let mut tx = sqlx::Connection::begin(&mut **held).await?;
282 let rows = insert_each(&mut tx, table, entities).await?;
283 tx.commit().await?;
284 return Ok(rows);
285 }
286 let mut tx = self.pool().begin().await?;
287 crate::company_scope::bind_current_company(&mut tx).await?;
288 let rows = insert_each(&mut tx, table, entities).await?;
289 tx.commit().await?;
290 Ok(rows)
291 }
292
293 async fn bulk_update_changed_guarded(&self, pairs: &[(T, T)], guard: &str) -> Result<Vec<T>>
300 where
301 T: Serialize + Send + Sync,
302 {
303 if pairs.is_empty() {
304 return Ok(Vec::new());
305 }
306 let table = self.table_name();
307 if let Some(conn) = crate::company_scope::current_request_conn() {
308 let mut held = conn.lock().await;
309 let mut tx = sqlx::Connection::begin(&mut **held).await?;
310 let rows = update_each_changed(&mut tx, table, guard, pairs).await?;
311 tx.commit().await?;
312 return Ok(rows);
313 }
314 let mut tx = self.pool().begin().await?;
315 crate::company_scope::bind_current_company(&mut tx).await?;
316 let rows = update_each_changed(&mut tx, table, guard, pairs).await?;
317 tx.commit().await?;
318 Ok(rows)
319 }
320
321 async fn find_by_text_field_with_cond(
328 &self,
329 field: &str,
330 value: &str,
331 extra: &str,
332 ) -> Result<Option<T>> {
333 let query = format!(
334 "SELECT * FROM {} WHERE {} = $1{}",
335 self.table_name(), field, extra
336 );
337 let result = crate::company_scope::fetch_optional_scoped(
338 self.pool(),
339 sqlx::query_as::<_, T>(&query).bind(value),
340 )
341 .await?;
342 Ok(result)
343 }
344
345 async fn exists_by_text_field_with_cond(
347 &self,
348 field: &str,
349 value: &str,
350 extra: &str,
351 ) -> Result<bool> {
352 let query = format!(
353 "SELECT 1 FROM {} WHERE {} = $1{} LIMIT 1",
354 self.table_name(), field, extra
355 );
356 let result = crate::company_scope::fetch_optional_scalar_scoped(
357 self.pool(),
358 sqlx::query_scalar::<_, i32>(&query).bind(value),
359 )
360 .await?;
361 Ok(result.is_some())
362 }
363
364 async fn find_by_uuid_field_with_cond(
366 &self,
367 field: &str,
368 value: Uuid,
369 extra: &str,
370 ) -> Result<Option<T>> {
371 let query = format!(
372 "SELECT * FROM {} WHERE {} = $1{}",
373 self.table_name(), field, extra
374 );
375 let result = crate::company_scope::fetch_optional_scoped(
376 self.pool(),
377 sqlx::query_as::<_, T>(&query).bind(value),
378 )
379 .await?;
380 Ok(result)
381 }
382
383 async fn exists_by_uuid_field_with_cond(
385 &self,
386 field: &str,
387 value: Uuid,
388 extra: &str,
389 ) -> Result<bool> {
390 let query = format!(
391 "SELECT 1 FROM {} WHERE {} = $1{} LIMIT 1",
392 self.table_name(), field, extra
393 );
394 let result = crate::company_scope::fetch_optional_scalar_scoped(
395 self.pool(),
396 sqlx::query_scalar::<_, i32>(&query).bind(value),
397 )
398 .await?;
399 Ok(result.is_some())
400 }
401
402 pub async fn run_filtered_query(
413 &self,
414 pagination: PaginationParams,
415 base_condition: Option<&str>,
416 filters: &HashMap<String, String>,
417 column_types: &HashMap<String, String>,
418 search_fields: &[&str],
419 ) -> Result<PaginatedResult<T>>
420 where
421 T: Send + Sync,
422 {
423 self.inner
424 .list_with_filters_and_base(
425 pagination,
426 base_condition,
427 filters,
428 column_types,
429 search_fields,
430 )
431 .await
432 }
433
434 pub async fn run_aggregate_query(
441 &self,
442 spec: &crate::repository::AggregateSpec,
443 base_condition: Option<&str>,
444 filters: &HashMap<String, String>,
445 column_types: &HashMap<String, String>,
446 search_fields: &[&str],
447 ) -> Result<crate::repository::AggregateResult>
448 where
449 T: crate::EntityRepoMeta + Send + Sync,
450 {
451 let mut spec = spec.clone();
455 if let (Some(group), Some(label)) = (&spec.group_by, &spec.label_field) {
456 let _ = label;
457 let camel = snake_to_camel(group);
458 if let Some((_, table, _)) = T::relations().iter().find(|(_, _, fk)| *fk == camel) {
459 let base_fk = camel_to_snake(&camel);
460 spec.label_relation = Some((table.to_string(), base_fk));
461 }
462 }
463 self.inner
464 .aggregate_with_filters_and_base(
465 &spec,
466 base_condition,
467 filters,
468 column_types,
469 search_fields,
470 )
471 .await
472 }
473}
474
475impl<T> GenericCrudRepository<T, SoftDelete>
478where
479 T: for<'r> FromRow<'r, PgRow> + Send + Unpin,
480{
481 pub async fn find_by_text_field(&self, field: &str, value: &str) -> Result<Option<T>> {
487 self.find_by_text_field_with_cond(field, value, " AND metadata->>'deleted_at' IS NULL").await
488 }
489
490 pub async fn exists_by_text_field(&self, field: &str, value: &str) -> Result<bool> {
492 self.exists_by_text_field_with_cond(field, value, " AND metadata->>'deleted_at' IS NULL").await
493 }
494
495 pub async fn find_by_uuid_field(&self, field: &str, value: Uuid) -> Result<Option<T>> {
497 self.find_by_uuid_field_with_cond(field, value, " AND metadata->>'deleted_at' IS NULL").await
498 }
499
500 pub async fn exists_by_uuid_field(&self, field: &str, value: Uuid) -> Result<bool> {
502 self.exists_by_uuid_field_with_cond(field, value, " AND metadata->>'deleted_at' IS NULL").await
503 }
504
505 pub async fn list_paginated_filtered(
511 &self,
512 pagination: PaginationParams,
513 filters: Option<&HashMap<String, String>>,
514 ) -> Result<PaginatedResult<T>>
515 where
516 T: EntityRepoMeta + Send + Sync,
517 {
518 let filters_map = filters.cloned().unwrap_or_default();
519 let column_types = T::column_types();
520 let search_fields_owned: Vec<&'static str> = T::search_fields().iter().copied().collect();
521 self.run_filtered_query(
522 pagination,
523 Some("metadata->>'deleted_at' IS NULL"),
524 &filters_map,
525 &column_types,
526 &search_fields_owned,
527 ).await
528 }
529
530 pub async fn aggregate_filtered(
532 &self,
533 spec: &crate::repository::AggregateSpec,
534 filters: Option<&HashMap<String, String>>,
535 ) -> Result<crate::repository::AggregateResult>
536 where
537 T: EntityRepoMeta + Send + Sync,
538 {
539 let filters_map = filters.cloned().unwrap_or_default();
540 let column_types = T::column_types();
541 let search_fields_owned: Vec<&'static str> = T::search_fields().iter().copied().collect();
542 self.run_aggregate_query(spec, Some("metadata->>'deleted_at' IS NULL"), &filters_map, &column_types, &search_fields_owned).await
543 }
544
545 pub async fn list_paginated_filtered_scoped(
556 &self,
557 pagination: PaginationParams,
558 filters: Option<&HashMap<String, String>>,
559 company: Option<Uuid>,
560 ) -> Result<PaginatedResult<T>>
561 where
562 T: EntityRepoMeta + Send + Sync,
563 {
564 let fence = company_fence::<T>(company)?;
565 let mut filters_map = filters.cloned().unwrap_or_default();
566 strip_client_company_filters::<T>(&mut filters_map);
567 let column_types = T::column_types();
568 let search_fields_owned: Vec<&'static str> = T::search_fields().iter().copied().collect();
569 self.run_filtered_query(
570 pagination,
571 and_conditions(Some("metadata->>'deleted_at' IS NULL"), fence).as_deref(),
572 &filters_map,
573 &column_types,
574 &search_fields_owned,
575 )
576 .await
577 }
578
579 pub async fn list_deleted_filtered_scoped(
584 &self,
585 pagination: PaginationParams,
586 filters: Option<&HashMap<String, String>>,
587 company: Option<Uuid>,
588 ) -> Result<PaginatedResult<T>>
589 where
590 T: EntityRepoMeta + Send + Sync,
591 {
592 let fence = company_fence::<T>(company)?;
593 let mut filters_map = filters.cloned().unwrap_or_default();
594 strip_client_company_filters::<T>(&mut filters_map);
595 let column_types = T::column_types();
596 let search_fields: Vec<&'static str> = T::search_fields().to_vec();
597 self.run_filtered_query(
598 pagination,
599 and_conditions(Some("metadata->>'deleted_at' IS NOT NULL"), fence).as_deref(),
600 &filters_map,
601 &column_types,
602 &search_fields,
603 )
604 .await
605 }
606
607 pub async fn list_deleted_filtered(
610 &self,
611 pagination: PaginationParams,
612 filters: Option<&HashMap<String, String>>,
613 ) -> Result<PaginatedResult<T>>
614 where
615 T: EntityRepoMeta + Send + Sync,
616 {
617 let filters_map = filters.cloned().unwrap_or_default();
618 let column_types = T::column_types();
619 let search_fields: Vec<&'static str> = T::search_fields().to_vec();
620 self.run_filtered_query(
621 pagination,
622 Some("metadata->>'deleted_at' IS NOT NULL"),
623 &filters_map,
624 &column_types,
625 &search_fields,
626 ).await
627 }
628
629 pub async fn find_by_id(&self, id: &str) -> Result<Option<T>> {
631 let query = format!(
632 "SELECT * FROM {} WHERE id = $1::uuid AND metadata->>'deleted_at' IS NULL",
633 self.table_name()
634 );
635 let result = crate::company_scope::fetch_optional_scoped(
636 self.pool(),
637 sqlx::query_as::<_, T>(&query).bind(id),
638 )
639 .await?;
640 Ok(result)
641 }
642
643 pub async fn find_all(&self) -> Result<Vec<T>> {
645 let query = format!(
646 "SELECT * FROM {} WHERE metadata->>'deleted_at' IS NULL",
647 self.table_name()
648 );
649 let results = crate::company_scope::fetch_all_scoped(
650 self.pool(),
651 sqlx::query_as::<_, T>(&query),
652 )
653 .await?;
654 Ok(results)
655 }
656
657 pub async fn update(&self, id: &str, entity: &T) -> Result<Option<T>>
662 where
663 T: Serialize + Send + Sync,
664 {
665 self.inner
666 .update_guarded(id, entity, None, ACTIVE_ROW_GUARD)
667 .await
668 }
669
670 pub async fn update_changed(&self, id: &str, before: &T, after: &T) -> Result<Option<T>>
678 where
679 T: Serialize + Send + Sync,
680 {
681 let changed = crate::repository::changed_columns(before, after)?;
682 self.inner
683 .update_guarded(id, after, Some(&changed), ACTIVE_ROW_GUARD)
684 .await
685 }
686
687 pub async fn delete(&self, id: &str) -> Result<bool> {
689 self.soft_delete(id).await
690 }
691
692 pub async fn count(&self) -> Result<u64> {
694 self.count_active().await
695 }
696
697 pub async fn exists(&self, id: &str) -> Result<bool> {
699 let query = format!(
700 "SELECT 1 FROM {} WHERE id = $1::uuid AND metadata->>'deleted_at' IS NULL LIMIT 1",
701 self.table_name()
702 );
703 let result = crate::company_scope::fetch_optional_scalar_scoped(
704 self.pool(),
705 sqlx::query_scalar::<_, i32>(&query).bind(id),
706 )
707 .await?;
708 Ok(result.is_some())
709 }
710
711 pub async fn list_paginated(&self, pagination: PaginationParams) -> Result<PaginatedResult<T>> {
713 let offset = pagination.offset();
714 let limit = pagination.limit();
715 let query = format!(
716 "SELECT * FROM {} WHERE metadata->>'deleted_at' IS NULL \
717 ORDER BY id DESC LIMIT $1 OFFSET $2",
718 self.table_name()
719 );
720 let data = crate::company_scope::fetch_all_scoped(
721 self.pool(),
722 sqlx::query_as::<_, T>(&query).bind(limit as i64).bind(offset as i64),
723 )
724 .await?;
725 let total = self.count_active().await?;
726 Ok(PaginatedResult {
727 data,
728 pagination: PaginationInfo::new(pagination.page, pagination.per_page, total),
729 })
730 }
731
732 pub async fn soft_delete(&self, id: &str) -> Result<bool> {
736 let query = format!(
737 "UPDATE {} SET metadata = jsonb_set(\
738 COALESCE(metadata, '{{}}'), \
739 '{{deleted_at}}', \
740 to_jsonb(NOW())\
741 ) WHERE id = $1::uuid AND (metadata->>'deleted_at') IS NULL",
742 self.table_name()
743 );
744 let result = crate::company_scope::execute_scoped(
745 self.pool(),
746 sqlx::query(&query).bind(id),
747 )
748 .await?;
749 Ok(result.rows_affected() > 0)
750 }
751
752 pub async fn restore(&self, id: &str) -> Result<Option<T>> {
754 let query = format!(
755 "UPDATE {} SET metadata = metadata - 'deleted_at' \
756 WHERE id = $1::uuid AND (metadata->>'deleted_at') IS NOT NULL \
757 RETURNING *",
758 self.table_name()
759 );
760 let result = crate::company_scope::fetch_optional_scoped(
761 self.pool(),
762 sqlx::query_as::<_, T>(&query).bind(id),
763 )
764 .await?;
765 Ok(result)
766 }
767
768 pub async fn list_deleted(&self, pagination: PaginationParams) -> Result<PaginatedResult<T>> {
770 let offset = pagination.offset();
771 let limit = pagination.limit();
772 let query = format!(
773 "SELECT * FROM {} WHERE (metadata->>'deleted_at') IS NOT NULL \
774 ORDER BY (metadata->>'deleted_at') DESC LIMIT $1 OFFSET $2",
775 self.table_name()
776 );
777 let data = crate::company_scope::fetch_all_scoped(
778 self.pool(),
779 sqlx::query_as::<_, T>(&query).bind(limit as i64).bind(offset as i64),
780 )
781 .await?;
782 let count_query = format!(
783 "SELECT COUNT(*) FROM {} WHERE (metadata->>'deleted_at') IS NOT NULL",
784 self.table_name()
785 );
786 let total = crate::company_scope::fetch_one_scalar_scoped(
787 self.pool(),
788 sqlx::query_scalar::<_, i64>(&count_query),
789 )
790 .await? as u64;
791 Ok(PaginatedResult {
792 data,
793 pagination: PaginationInfo::new(pagination.page, pagination.per_page, total),
794 })
795 }
796
797 pub async fn empty_trash(&self) -> Result<u64> {
799 let query = format!(
800 "DELETE FROM {} WHERE (metadata->>'deleted_at') IS NOT NULL",
801 self.table_name()
802 );
803 let result = crate::company_scope::execute_scoped(self.pool(), sqlx::query(&query)).await?;
804 Ok(result.rows_affected())
805 }
806
807 pub async fn find_deleted_by_id(&self, id: &str) -> Result<Option<T>> {
809 let query = format!(
810 "SELECT * FROM {} WHERE id = $1::uuid AND (metadata->>'deleted_at') IS NOT NULL",
811 self.table_name()
812 );
813 let result = crate::company_scope::fetch_optional_scoped(
814 self.pool(),
815 sqlx::query_as::<_, T>(&query).bind(id),
816 )
817 .await?;
818 Ok(result)
819 }
820
821 pub async fn permanent_delete(&self, id: &str) -> Result<bool> {
823 let query = format!(
824 "DELETE FROM {} WHERE id = $1::uuid AND (metadata->>'deleted_at') IS NOT NULL",
825 self.table_name()
826 );
827 let result = crate::company_scope::execute_scoped(
828 self.pool(),
829 sqlx::query(&query).bind(id),
830 )
831 .await?;
832 Ok(result.rows_affected() > 0)
833 }
834
835 pub async fn count_active(&self) -> Result<u64> {
837 let query = format!(
838 "SELECT COUNT(*) FROM {} WHERE (metadata->>'deleted_at') IS NULL",
839 self.table_name()
840 );
841 let count = crate::company_scope::fetch_one_scalar_scoped(
842 self.pool(),
843 sqlx::query_scalar::<_, i64>(&query),
844 )
845 .await? as u64;
846 Ok(count)
847 }
848
849 pub async fn count_deleted(&self) -> Result<u64> {
851 let query = format!(
852 "SELECT COUNT(*) FROM {} WHERE (metadata->>'deleted_at') IS NOT NULL",
853 self.table_name()
854 );
855 let count = crate::company_scope::fetch_one_scalar_scoped(
856 self.pool(),
857 sqlx::query_scalar::<_, i64>(&query),
858 )
859 .await? as u64;
860 Ok(count)
861 }
862
863 pub async fn bulk_soft_delete(&self, ids: &[String]) -> Result<u64> {
872 if ids.is_empty() {
873 return Ok(0);
874 }
875 let placeholders = id_in_placeholders(ids.len());
876 let query = format!(
877 "UPDATE {} SET metadata = jsonb_set(\
878 COALESCE(metadata, '{{}}'), '{{deleted_at}}', to_jsonb(NOW())\
879 ) WHERE id IN ({placeholders}) AND (metadata->>'deleted_at') IS NULL",
880 self.table_name()
881 );
882 let mut tx = self.pool().begin().await?;
883 crate::company_scope::bind_current_company(&mut tx).await?;
884 let mut q = sqlx::query(&query);
885 for id in ids {
886 q = q.bind(id);
887 }
888 let affected = q.execute(&mut *tx).await?.rows_affected();
889 if affected != ids.len() as u64 {
890 return Err(anyhow::anyhow!(
892 "bulk_soft_delete: {} of {} ids were not active/deletable; rolled back",
893 ids.len() as u64 - affected,
894 ids.len()
895 ));
896 }
897 tx.commit().await?;
898 Ok(affected)
899 }
900
901 pub async fn bulk_restore(&self, ids: &[String]) -> Result<Vec<T>> {
903 if ids.is_empty() {
904 return Ok(Vec::new());
905 }
906 let placeholders = id_in_placeholders(ids.len());
907 let query = format!(
908 "UPDATE {} SET metadata = metadata - 'deleted_at' \
909 WHERE id IN ({placeholders}) AND (metadata->>'deleted_at') IS NOT NULL \
910 RETURNING *",
911 self.table_name()
912 );
913 let mut tx = self.pool().begin().await?;
914 crate::company_scope::bind_current_company(&mut tx).await?;
915 let mut q = sqlx::query_as::<_, T>(&query);
916 for id in ids {
917 q = q.bind(id);
918 }
919 let rows = q.fetch_all(&mut *tx).await?;
920 if rows.len() != ids.len() {
921 return Err(anyhow::anyhow!(
922 "bulk_restore: {} of {} ids were not in trash; rolled back",
923 ids.len() - rows.len(),
924 ids.len()
925 ));
926 }
927 tx.commit().await?;
928 Ok(rows)
929 }
930
931 pub async fn bulk_permanent_delete(&self, ids: &[String]) -> Result<u64> {
933 if ids.is_empty() {
934 return Ok(0);
935 }
936 let placeholders = id_in_placeholders(ids.len());
937 let query = format!(
938 "DELETE FROM {} WHERE id IN ({placeholders}) AND (metadata->>'deleted_at') IS NOT NULL",
939 self.table_name()
940 );
941 let mut tx = self.pool().begin().await?;
942 crate::company_scope::bind_current_company(&mut tx).await?;
943 let mut q = sqlx::query(&query);
944 for id in ids {
945 q = q.bind(id);
946 }
947 let affected = q.execute(&mut *tx).await?.rows_affected();
948 if affected != ids.len() as u64 {
949 return Err(anyhow::anyhow!(
950 "bulk_permanent_delete: {} of {} ids were not in trash; rolled back",
951 ids.len() as u64 - affected,
952 ids.len()
953 ));
954 }
955 tx.commit().await?;
956 Ok(affected)
957 }
958
959 pub async fn restore_all(&self) -> Result<Vec<T>> {
963 let query = format!(
964 "UPDATE {} SET metadata = metadata - 'deleted_at' \
965 WHERE (metadata->>'deleted_at') IS NOT NULL \
966 RETURNING *",
967 self.table_name()
968 );
969 let rows = crate::company_scope::fetch_all_scoped(
970 self.pool(),
971 sqlx::query_as::<_, T>(&query),
972 )
973 .await?;
974 Ok(rows)
975 }
976
977 pub async fn bulk_update(&self, entities: &[T]) -> Result<Vec<T>>
980 where
981 T: Serialize + Send + Sync,
982 {
983 bulk_update_rows(
984 self.pool(),
985 self.table_name(),
986 ACTIVE_ROW_GUARD,
987 entities,
988 )
989 .await
990 }
991
992 pub async fn bulk_update_changed(&self, pairs: &[(T, T)]) -> Result<Vec<T>>
996 where
997 T: Serialize + Send + Sync,
998 {
999 self.bulk_update_changed_guarded(pairs, ACTIVE_ROW_GUARD).await
1000 }
1001}
1002
1003fn id_in_placeholders(n: usize) -> String {
1005 (1..=n)
1006 .map(|i| format!("${i}::uuid"))
1007 .collect::<Vec<_>>()
1008 .join(", ")
1009}
1010
1011#[derive(Debug)]
1017pub struct BatchItemError {
1018 pub index: usize,
1020 pub error: anyhow::Error,
1022}
1023
1024impl std::fmt::Display for BatchItemError {
1025 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
1026 write!(f, "item {} of the batch: {}; rolled back", self.index, self.error)
1027 }
1028}
1029
1030impl std::error::Error for BatchItemError {
1031 fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
1032 Some(self.error.as_ref())
1033 }
1034}
1035
1036async fn insert_each<T>(conn: &mut sqlx::PgConnection, table: &str, entities: &[T]) -> Result<Vec<T>>
1038where
1039 T: for<'r> FromRow<'r, PgRow> + Serialize + Send + Sync + Unpin,
1040{
1041 let mut out = Vec::with_capacity(entities.len());
1042 for (index, entity) in entities.iter().enumerate() {
1043 let (query, json) = crate::repository::insert_statement(table, entity)
1044 .map_err(|error| BatchItemError { index, error })?;
1045 let statement = match json {
1046 Some(json) => sqlx::query_as::<_, T>(&query).bind(json),
1047 None => sqlx::query_as::<_, T>(&query),
1048 };
1049 let row = statement.fetch_one(&mut *conn).await.map_err(|e| BatchItemError {
1050 index,
1051 error: crate::repository::explain_unknown_column(e, table),
1052 })?;
1053 out.push(row);
1054 }
1055 Ok(out)
1056}
1057
1058async fn update_each_changed<T>(
1061 conn: &mut sqlx::PgConnection,
1062 table: &str,
1063 guard: &str,
1064 pairs: &[(T, T)],
1065) -> Result<Vec<T>>
1066where
1067 T: for<'r> FromRow<'r, PgRow> + Serialize + Send + Sync + Unpin,
1068{
1069 use crate::repository::UpdateStatement;
1070 let mut out = Vec::with_capacity(pairs.len());
1071 for (index, (before, after)) in pairs.iter().enumerate() {
1072 let item = |error: anyhow::Error| BatchItemError { index, error };
1073 let id = match serde_json::to_value(after).map_err(|e| item(e.into()))? {
1074 serde_json::Value::Object(obj) => match obj.get("id") {
1075 Some(serde_json::Value::String(id)) => id.clone(),
1076 _ => return Err(item(anyhow::anyhow!("entity missing string 'id' field")).into()),
1077 },
1078 _ => return Err(item(anyhow::anyhow!("entity must serialize to a JSON object")).into()),
1079 };
1080 let changed = crate::repository::changed_columns(before, after).map_err(item)?;
1081 let updated = match crate::repository::update_statement(table, after, Some(&changed), guard)
1082 .map_err(item)?
1083 {
1084 UpdateStatement::Unchanged(query) => {
1085 sqlx::query_as::<_, T>(&query).bind(&id).fetch_optional(&mut *conn).await
1086 }
1087 UpdateStatement::Update { query, json } => {
1088 sqlx::query_as::<_, T>(&query)
1089 .bind(json)
1090 .bind(&id)
1091 .fetch_optional(&mut *conn)
1092 .await
1093 }
1094 }
1095 .map_err(|e| item(e.into()))?;
1096 match updated {
1097 Some(row) => out.push(row),
1098 None => {
1099 return Err(item(anyhow::anyhow!("id '{id}' not found or already deleted")).into())
1100 }
1101 }
1102 }
1103 Ok(out)
1104}
1105
1106fn build_update_parts<T: Serialize>(entity: &T) -> Result<(String, String, String)> {
1109 let json_value = serde_json::to_value(entity)?;
1110 let json_obj = match json_value {
1111 serde_json::Value::Object(obj) => obj,
1112 _ => return Err(anyhow::anyhow!("entity must serialize to a JSON object")),
1113 };
1114 let id = json_obj
1115 .get("id")
1116 .and_then(|v| v.as_str())
1117 .ok_or_else(|| anyhow::anyhow!("entity missing string 'id' field"))?
1118 .to_string();
1119 let column_names = json_obj
1120 .keys()
1121 .filter(|k| *k != "id")
1122 .map(|k| format!("\"{k}\""))
1123 .collect::<Vec<_>>()
1124 .join(", ");
1125 let json_str = serde_json::to_string(&json_obj)?;
1126 Ok((id, json_str, column_names))
1127}
1128
1129impl<T> GenericCrudRepository<T, HardDelete>
1132where
1133 T: for<'r> FromRow<'r, PgRow> + Send + Sync + Unpin + Serialize,
1134{
1135 pub async fn find_by_text_field(&self, field: &str, value: &str) -> Result<Option<T>> {
1139 self.find_by_text_field_with_cond(field, value, "").await
1140 }
1141
1142 pub async fn exists_by_text_field(&self, field: &str, value: &str) -> Result<bool> {
1144 self.exists_by_text_field_with_cond(field, value, "").await
1145 }
1146
1147 pub async fn find_by_uuid_field(&self, field: &str, value: Uuid) -> Result<Option<T>> {
1149 self.find_by_uuid_field_with_cond(field, value, "").await
1150 }
1151
1152 pub async fn exists_by_uuid_field(&self, field: &str, value: Uuid) -> Result<bool> {
1154 self.exists_by_uuid_field_with_cond(field, value, "").await
1155 }
1156
1157 pub async fn list_paginated_filtered(
1161 &self,
1162 pagination: PaginationParams,
1163 filters: Option<&HashMap<String, String>>,
1164 ) -> Result<PaginatedResult<T>>
1165 where
1166 T: EntityRepoMeta + Send + Sync,
1167 {
1168 let filters_map = filters.cloned().unwrap_or_default();
1169 let column_types = T::column_types();
1170 let search_fields_owned: Vec<&'static str> = T::search_fields().iter().copied().collect();
1171 self.run_filtered_query(pagination, None, &filters_map, &column_types, &search_fields_owned).await
1172 }
1173
1174 pub async fn aggregate_filtered(
1176 &self,
1177 spec: &crate::repository::AggregateSpec,
1178 filters: Option<&HashMap<String, String>>,
1179 ) -> Result<crate::repository::AggregateResult>
1180 where
1181 T: EntityRepoMeta + Send + Sync,
1182 {
1183 let filters_map = filters.cloned().unwrap_or_default();
1184 let column_types = T::column_types();
1185 let search_fields_owned: Vec<&'static str> = T::search_fields().iter().copied().collect();
1186 self.run_aggregate_query(spec, None, &filters_map, &column_types, &search_fields_owned).await
1187 }
1188
1189 pub async fn find_by_id(&self, id: &str) -> Result<Option<T>> {
1191 self.inner.find_by_id(id).await
1192 }
1193
1194 pub async fn find_all(&self) -> Result<Vec<T>> {
1196 let query = format!("SELECT * FROM {}", self.table_name());
1197 let results = crate::company_scope::fetch_all_scoped(
1198 self.pool(),
1199 sqlx::query_as::<_, T>(&query),
1200 )
1201 .await?;
1202 Ok(results)
1203 }
1204
1205 pub async fn update(&self, id: &str, entity: &T) -> Result<Option<T>> {
1207 self.inner.update(id, entity).await
1208 }
1209
1210 pub async fn update_changed(&self, id: &str, before: &T, after: &T) -> Result<Option<T>> {
1214 let changed = crate::repository::changed_columns(before, after)?;
1215 self.inner
1216 .update_guarded(id, after, Some(&changed), "")
1217 .await
1218 }
1219
1220 pub async fn delete(&self, id: &str) -> Result<bool> {
1222 self.inner.delete(id).await
1223 }
1224
1225 pub async fn count(&self) -> Result<u64> {
1227 let query = format!("SELECT COUNT(*) FROM {}", self.table_name());
1228 let count = crate::company_scope::fetch_one_scalar_scoped(
1229 self.pool(),
1230 sqlx::query_scalar::<_, i64>(&query),
1231 )
1232 .await? as u64;
1233 Ok(count)
1234 }
1235
1236 pub async fn exists(&self, id: &str) -> Result<bool> {
1238 let query = format!(
1239 "SELECT 1 FROM {} WHERE id = $1::uuid LIMIT 1",
1240 self.table_name()
1241 );
1242 let result = crate::company_scope::fetch_optional_scalar_scoped(
1243 self.pool(),
1244 sqlx::query_scalar::<_, i32>(&query).bind(id),
1245 )
1246 .await?;
1247 Ok(result.is_some())
1248 }
1249
1250 pub async fn list_paginated(&self, pagination: PaginationParams) -> Result<PaginatedResult<T>> {
1252 let offset = pagination.offset();
1253 let limit = pagination.limit();
1254 let query = format!(
1255 "SELECT * FROM {} ORDER BY id DESC LIMIT $1 OFFSET $2",
1256 self.table_name()
1257 );
1258 let data = crate::company_scope::fetch_all_scoped(
1259 self.pool(),
1260 sqlx::query_as::<_, T>(&query).bind(limit as i64).bind(offset as i64),
1261 )
1262 .await?;
1263 let total = self.count().await?;
1264 Ok(PaginatedResult {
1265 data,
1266 pagination: PaginationInfo::new(pagination.page, pagination.per_page, total),
1267 })
1268 }
1269
1270 pub async fn bulk_delete(&self, ids: &[String]) -> Result<u64> {
1274 if ids.is_empty() {
1275 return Ok(0);
1276 }
1277 let placeholders = id_in_placeholders(ids.len());
1278 let query = format!(
1279 "DELETE FROM {} WHERE id IN ({placeholders})",
1280 self.table_name()
1281 );
1282 let mut tx = self.pool().begin().await?;
1283 crate::company_scope::bind_current_company(&mut tx).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 *tx).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 tx.commit().await?;
1297 Ok(affected)
1298 }
1299
1300 pub async fn bulk_update(&self, entities: &[T]) -> Result<Vec<T>> {
1302 bulk_update_rows(self.pool(), self.table_name(), "", entities).await
1303 }
1304
1305 pub async fn bulk_update_changed(&self, pairs: &[(T, T)]) -> Result<Vec<T>> {
1308 self.bulk_update_changed_guarded(pairs, "").await
1309 }
1310}
1311
1312pub fn qualify_relation_table(caller_table: &str, target_table: &str) -> String {
1322 if target_table.contains('.') || !caller_table.contains('.') {
1323 target_table.to_string()
1324 } else {
1325 let schema = caller_table.split('.').next().unwrap_or(caller_table);
1326 format!("{schema}.{target_table}")
1327 }
1328}
1329
1330fn is_undefined_table(err: &anyhow::Error) -> bool {
1335 err.chain()
1336 .filter_map(|cause| cause.downcast_ref::<sqlx::Error>())
1337 .any(|sqlx_err| {
1338 sqlx_err
1339 .as_database_error()
1340 .map(|db| db.code().as_deref() == Some("42P01"))
1341 .unwrap_or(false)
1342 })
1343}
1344
1345pub async fn fetch_by_ids_as_json(
1360 pool: &PgPool,
1361 caller_table: &str,
1362 table: &str,
1363 ids: &[String],
1364) -> Result<Vec<serde_json::Value>> {
1365 if ids.is_empty() {
1366 return Ok(Vec::new());
1367 }
1368 let qualified = qualify_relation_table(caller_table, table);
1369 match fetch_rows_as_json(pool, &qualified, ids).await {
1370 Ok(rows) => Ok(rows),
1371 Err(err) if qualified != table && is_undefined_table(&err) => {
1372 fetch_rows_as_json(pool, table, ids).await
1373 }
1374 Err(err) => Err(err),
1375 }
1376}
1377
1378async fn fetch_rows_as_json(
1383 pool: &PgPool,
1384 table: &str,
1385 ids: &[String],
1386) -> Result<Vec<serde_json::Value>> {
1387 let query = format!("SELECT row_to_json(t) AS j FROM {table} t WHERE t.id = ANY($1::uuid[])");
1388 let rows: Vec<(serde_json::Value,)> =
1389 crate::company_scope::fetch_all_scoped(pool, sqlx::query_as(&query).bind(ids)).await?;
1390 Ok(rows.into_iter().map(|(j,)| j).collect())
1391}
1392
1393async fn bulk_update_rows<T>(
1398 pool: &PgPool,
1399 table: &str,
1400 active_guard: &str,
1401 entities: &[T],
1402) -> Result<Vec<T>>
1403where
1404 T: for<'r> FromRow<'r, PgRow> + Send + Sync + Unpin + Serialize,
1405{
1406 if entities.is_empty() {
1407 return Ok(Vec::new());
1408 }
1409 let mut tx = pool.begin().await?;
1410 crate::company_scope::bind_current_company(&mut tx).await?;
1411 let mut out = Vec::with_capacity(entities.len());
1412 for entity in entities {
1413 let (id, json_str, column_names) = build_update_parts(entity)?;
1414 let query = format!(
1415 "WITH new_row AS (\
1416 SELECT (jsonb_populate_record(NULL::{table}, $1::jsonb)).*\
1417 ) UPDATE {table} AS t \
1418 SET ({columns}) = (SELECT {columns} FROM new_row) \
1419 WHERE t.id = $2::uuid{guard} \
1420 RETURNING t.*",
1421 table = table,
1422 columns = column_names,
1423 guard = active_guard,
1424 );
1425 let updated = sqlx::query_as::<_, T>(&query)
1426 .bind(&json_str)
1427 .bind(&id)
1428 .fetch_optional(&mut *tx)
1429 .await?;
1430 match updated {
1431 Some(e) => out.push(e),
1432 None => {
1433 return Err(anyhow::anyhow!(
1434 "bulk_update: id '{id}' not found or already deleted; rolled back"
1435 ));
1436 }
1437 }
1438 }
1439 tx.commit().await?;
1440 Ok(out)
1441}
1442
1443fn camel_to_snake(s: &str) -> String {
1445 let mut out = String::with_capacity(s.len() + 4);
1446 for ch in s.chars() {
1447 if ch.is_uppercase() {
1448 out.push('_');
1449 out.extend(ch.to_lowercase());
1450 } else {
1451 out.push(ch);
1452 }
1453 }
1454 out
1455}