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>>
267 where
268 T: Serialize + Send + Sync,
269 {
270 let tx = sqlx::pool::Pool::begin(self.pool()).await?;
271 let mut results = Vec::with_capacity(entities.len());
272 for entity in entities {
273 results.push(self.create(entity).await?);
274 }
275 tx.commit().await?;
276 Ok(results)
277 }
278
279 async fn find_by_text_field_with_cond(
286 &self,
287 field: &str,
288 value: &str,
289 extra: &str,
290 ) -> Result<Option<T>> {
291 let query = format!(
292 "SELECT * FROM {} WHERE {} = $1{}",
293 self.table_name(), field, extra
294 );
295 let result = crate::company_scope::fetch_optional_scoped(
296 self.pool(),
297 sqlx::query_as::<_, T>(&query).bind(value),
298 )
299 .await?;
300 Ok(result)
301 }
302
303 async fn exists_by_text_field_with_cond(
305 &self,
306 field: &str,
307 value: &str,
308 extra: &str,
309 ) -> Result<bool> {
310 let query = format!(
311 "SELECT 1 FROM {} WHERE {} = $1{} LIMIT 1",
312 self.table_name(), field, extra
313 );
314 let result = crate::company_scope::fetch_optional_scalar_scoped(
315 self.pool(),
316 sqlx::query_scalar::<_, i32>(&query).bind(value),
317 )
318 .await?;
319 Ok(result.is_some())
320 }
321
322 async fn find_by_uuid_field_with_cond(
324 &self,
325 field: &str,
326 value: Uuid,
327 extra: &str,
328 ) -> Result<Option<T>> {
329 let query = format!(
330 "SELECT * FROM {} WHERE {} = $1{}",
331 self.table_name(), field, extra
332 );
333 let result = crate::company_scope::fetch_optional_scoped(
334 self.pool(),
335 sqlx::query_as::<_, T>(&query).bind(value),
336 )
337 .await?;
338 Ok(result)
339 }
340
341 async fn exists_by_uuid_field_with_cond(
343 &self,
344 field: &str,
345 value: Uuid,
346 extra: &str,
347 ) -> Result<bool> {
348 let query = format!(
349 "SELECT 1 FROM {} WHERE {} = $1{} LIMIT 1",
350 self.table_name(), field, extra
351 );
352 let result = crate::company_scope::fetch_optional_scalar_scoped(
353 self.pool(),
354 sqlx::query_scalar::<_, i32>(&query).bind(value),
355 )
356 .await?;
357 Ok(result.is_some())
358 }
359
360 pub async fn run_filtered_query(
371 &self,
372 pagination: PaginationParams,
373 base_condition: Option<&str>,
374 filters: &HashMap<String, String>,
375 column_types: &HashMap<String, String>,
376 search_fields: &[&str],
377 ) -> Result<PaginatedResult<T>>
378 where
379 T: Send + Sync,
380 {
381 self.inner
382 .list_with_filters_and_base(
383 pagination,
384 base_condition,
385 filters,
386 column_types,
387 search_fields,
388 )
389 .await
390 }
391
392 pub async fn run_aggregate_query(
399 &self,
400 spec: &crate::repository::AggregateSpec,
401 base_condition: Option<&str>,
402 filters: &HashMap<String, String>,
403 column_types: &HashMap<String, String>,
404 search_fields: &[&str],
405 ) -> Result<crate::repository::AggregateResult>
406 where
407 T: crate::EntityRepoMeta + Send + Sync,
408 {
409 let mut spec = spec.clone();
413 if let (Some(group), Some(label)) = (&spec.group_by, &spec.label_field) {
414 let _ = label;
415 let camel = snake_to_camel(group);
416 if let Some((_, table, _)) = T::relations().iter().find(|(_, _, fk)| *fk == camel) {
417 let base_fk = camel_to_snake(&camel);
418 spec.label_relation = Some((table.to_string(), base_fk));
419 }
420 }
421 self.inner
422 .aggregate_with_filters_and_base(
423 &spec,
424 base_condition,
425 filters,
426 column_types,
427 search_fields,
428 )
429 .await
430 }
431}
432
433impl<T> GenericCrudRepository<T, SoftDelete>
436where
437 T: for<'r> FromRow<'r, PgRow> + Send + Unpin,
438{
439 pub async fn find_by_text_field(&self, field: &str, value: &str) -> Result<Option<T>> {
445 self.find_by_text_field_with_cond(field, value, " AND metadata->>'deleted_at' IS NULL").await
446 }
447
448 pub async fn exists_by_text_field(&self, field: &str, value: &str) -> Result<bool> {
450 self.exists_by_text_field_with_cond(field, value, " AND metadata->>'deleted_at' IS NULL").await
451 }
452
453 pub async fn find_by_uuid_field(&self, field: &str, value: Uuid) -> Result<Option<T>> {
455 self.find_by_uuid_field_with_cond(field, value, " AND metadata->>'deleted_at' IS NULL").await
456 }
457
458 pub async fn exists_by_uuid_field(&self, field: &str, value: Uuid) -> Result<bool> {
460 self.exists_by_uuid_field_with_cond(field, value, " AND metadata->>'deleted_at' IS NULL").await
461 }
462
463 pub async fn list_paginated_filtered(
469 &self,
470 pagination: PaginationParams,
471 filters: Option<&HashMap<String, String>>,
472 ) -> Result<PaginatedResult<T>>
473 where
474 T: EntityRepoMeta + Send + Sync,
475 {
476 let filters_map = filters.cloned().unwrap_or_default();
477 let column_types = T::column_types();
478 let search_fields_owned: Vec<&'static str> = T::search_fields().iter().copied().collect();
479 self.run_filtered_query(
480 pagination,
481 Some("metadata->>'deleted_at' IS NULL"),
482 &filters_map,
483 &column_types,
484 &search_fields_owned,
485 ).await
486 }
487
488 pub async fn aggregate_filtered(
490 &self,
491 spec: &crate::repository::AggregateSpec,
492 filters: Option<&HashMap<String, String>>,
493 ) -> Result<crate::repository::AggregateResult>
494 where
495 T: EntityRepoMeta + Send + Sync,
496 {
497 let filters_map = filters.cloned().unwrap_or_default();
498 let column_types = T::column_types();
499 let search_fields_owned: Vec<&'static str> = T::search_fields().iter().copied().collect();
500 self.run_aggregate_query(spec, Some("metadata->>'deleted_at' IS NULL"), &filters_map, &column_types, &search_fields_owned).await
501 }
502
503 pub async fn list_paginated_filtered_scoped(
514 &self,
515 pagination: PaginationParams,
516 filters: Option<&HashMap<String, String>>,
517 company: Option<Uuid>,
518 ) -> Result<PaginatedResult<T>>
519 where
520 T: EntityRepoMeta + Send + Sync,
521 {
522 let fence = company_fence::<T>(company)?;
523 let mut filters_map = filters.cloned().unwrap_or_default();
524 strip_client_company_filters::<T>(&mut filters_map);
525 let column_types = T::column_types();
526 let search_fields_owned: Vec<&'static str> = T::search_fields().iter().copied().collect();
527 self.run_filtered_query(
528 pagination,
529 and_conditions(Some("metadata->>'deleted_at' IS NULL"), fence).as_deref(),
530 &filters_map,
531 &column_types,
532 &search_fields_owned,
533 )
534 .await
535 }
536
537 pub async fn list_deleted_filtered_scoped(
542 &self,
543 pagination: PaginationParams,
544 filters: Option<&HashMap<String, String>>,
545 company: Option<Uuid>,
546 ) -> Result<PaginatedResult<T>>
547 where
548 T: EntityRepoMeta + Send + Sync,
549 {
550 let fence = company_fence::<T>(company)?;
551 let mut filters_map = filters.cloned().unwrap_or_default();
552 strip_client_company_filters::<T>(&mut filters_map);
553 let column_types = T::column_types();
554 let empty: &[&str] = &[];
555 self.run_filtered_query(
556 pagination,
557 and_conditions(Some("metadata->>'deleted_at' IS NOT NULL"), fence).as_deref(),
558 &filters_map,
559 &column_types,
560 empty,
561 )
562 .await
563 }
564
565 pub async fn list_deleted_filtered(
567 &self,
568 pagination: PaginationParams,
569 filters: Option<&HashMap<String, String>>,
570 ) -> Result<PaginatedResult<T>>
571 where
572 T: EntityRepoMeta + Send + Sync,
573 {
574 let filters_map = filters.cloned().unwrap_or_default();
575 let column_types = T::column_types();
576 let empty: &[&str] = &[];
577 self.run_filtered_query(
578 pagination,
579 Some("metadata->>'deleted_at' IS NOT NULL"),
580 &filters_map,
581 &column_types,
582 empty,
583 ).await
584 }
585
586 pub async fn find_by_id(&self, id: &str) -> Result<Option<T>> {
588 let query = format!(
589 "SELECT * FROM {} WHERE id = $1::uuid AND metadata->>'deleted_at' IS NULL",
590 self.table_name()
591 );
592 let result = crate::company_scope::fetch_optional_scoped(
593 self.pool(),
594 sqlx::query_as::<_, T>(&query).bind(id),
595 )
596 .await?;
597 Ok(result)
598 }
599
600 pub async fn find_all(&self) -> Result<Vec<T>> {
602 let query = format!(
603 "SELECT * FROM {} WHERE metadata->>'deleted_at' IS NULL",
604 self.table_name()
605 );
606 let results = crate::company_scope::fetch_all_scoped(
607 self.pool(),
608 sqlx::query_as::<_, T>(&query),
609 )
610 .await?;
611 Ok(results)
612 }
613
614 pub async fn update(&self, id: &str, entity: &T) -> Result<Option<T>>
619 where
620 T: Serialize + Send + Sync,
621 {
622 self.inner
623 .update_guarded(id, entity, None, ACTIVE_ROW_GUARD)
624 .await
625 }
626
627 pub async fn update_changed(&self, id: &str, before: &T, after: &T) -> Result<Option<T>>
635 where
636 T: Serialize + Send + Sync,
637 {
638 let changed = crate::repository::changed_columns(before, after)?;
639 self.inner
640 .update_guarded(id, after, Some(&changed), ACTIVE_ROW_GUARD)
641 .await
642 }
643
644 pub async fn delete(&self, id: &str) -> Result<bool> {
646 self.soft_delete(id).await
647 }
648
649 pub async fn count(&self) -> Result<u64> {
651 self.count_active().await
652 }
653
654 pub async fn exists(&self, id: &str) -> Result<bool> {
656 let query = format!(
657 "SELECT 1 FROM {} WHERE id = $1::uuid AND metadata->>'deleted_at' IS NULL LIMIT 1",
658 self.table_name()
659 );
660 let result = crate::company_scope::fetch_optional_scalar_scoped(
661 self.pool(),
662 sqlx::query_scalar::<_, i32>(&query).bind(id),
663 )
664 .await?;
665 Ok(result.is_some())
666 }
667
668 pub async fn list_paginated(&self, pagination: PaginationParams) -> Result<PaginatedResult<T>> {
670 let offset = pagination.offset();
671 let limit = pagination.limit();
672 let query = format!(
673 "SELECT * FROM {} WHERE metadata->>'deleted_at' IS NULL \
674 ORDER BY id DESC LIMIT $1 OFFSET $2",
675 self.table_name()
676 );
677 let data = crate::company_scope::fetch_all_scoped(
678 self.pool(),
679 sqlx::query_as::<_, T>(&query).bind(limit as i64).bind(offset as i64),
680 )
681 .await?;
682 let total = self.count_active().await?;
683 Ok(PaginatedResult {
684 data,
685 pagination: PaginationInfo::new(pagination.page, pagination.per_page, total),
686 })
687 }
688
689 pub async fn soft_delete(&self, id: &str) -> Result<bool> {
693 let query = format!(
694 "UPDATE {} SET metadata = jsonb_set(\
695 COALESCE(metadata, '{{}}'), \
696 '{{deleted_at}}', \
697 to_jsonb(NOW())\
698 ) WHERE id = $1::uuid AND (metadata->>'deleted_at') IS NULL",
699 self.table_name()
700 );
701 let result = crate::company_scope::execute_scoped(
702 self.pool(),
703 sqlx::query(&query).bind(id),
704 )
705 .await?;
706 Ok(result.rows_affected() > 0)
707 }
708
709 pub async fn restore(&self, id: &str) -> Result<Option<T>> {
711 let query = format!(
712 "UPDATE {} SET metadata = metadata - 'deleted_at' \
713 WHERE id = $1::uuid AND (metadata->>'deleted_at') IS NOT NULL \
714 RETURNING *",
715 self.table_name()
716 );
717 let result = crate::company_scope::fetch_optional_scoped(
718 self.pool(),
719 sqlx::query_as::<_, T>(&query).bind(id),
720 )
721 .await?;
722 Ok(result)
723 }
724
725 pub async fn list_deleted(&self, pagination: PaginationParams) -> Result<PaginatedResult<T>> {
727 let offset = pagination.offset();
728 let limit = pagination.limit();
729 let query = format!(
730 "SELECT * FROM {} WHERE (metadata->>'deleted_at') IS NOT NULL \
731 ORDER BY (metadata->>'deleted_at') DESC LIMIT $1 OFFSET $2",
732 self.table_name()
733 );
734 let data = crate::company_scope::fetch_all_scoped(
735 self.pool(),
736 sqlx::query_as::<_, T>(&query).bind(limit as i64).bind(offset as i64),
737 )
738 .await?;
739 let count_query = format!(
740 "SELECT COUNT(*) FROM {} WHERE (metadata->>'deleted_at') IS NOT NULL",
741 self.table_name()
742 );
743 let total = crate::company_scope::fetch_one_scalar_scoped(
744 self.pool(),
745 sqlx::query_scalar::<_, i64>(&count_query),
746 )
747 .await? as u64;
748 Ok(PaginatedResult {
749 data,
750 pagination: PaginationInfo::new(pagination.page, pagination.per_page, total),
751 })
752 }
753
754 pub async fn empty_trash(&self) -> Result<u64> {
756 let query = format!(
757 "DELETE FROM {} WHERE (metadata->>'deleted_at') IS NOT NULL",
758 self.table_name()
759 );
760 let result = crate::company_scope::execute_scoped(self.pool(), sqlx::query(&query)).await?;
761 Ok(result.rows_affected())
762 }
763
764 pub async fn find_deleted_by_id(&self, id: &str) -> Result<Option<T>> {
766 let query = format!(
767 "SELECT * FROM {} WHERE id = $1::uuid AND (metadata->>'deleted_at') IS NOT NULL",
768 self.table_name()
769 );
770 let result = crate::company_scope::fetch_optional_scoped(
771 self.pool(),
772 sqlx::query_as::<_, T>(&query).bind(id),
773 )
774 .await?;
775 Ok(result)
776 }
777
778 pub async fn permanent_delete(&self, id: &str) -> Result<bool> {
780 let query = format!(
781 "DELETE FROM {} WHERE id = $1::uuid AND (metadata->>'deleted_at') IS NOT NULL",
782 self.table_name()
783 );
784 let result = crate::company_scope::execute_scoped(
785 self.pool(),
786 sqlx::query(&query).bind(id),
787 )
788 .await?;
789 Ok(result.rows_affected() > 0)
790 }
791
792 pub async fn count_active(&self) -> Result<u64> {
794 let query = format!(
795 "SELECT COUNT(*) FROM {} WHERE (metadata->>'deleted_at') IS NULL",
796 self.table_name()
797 );
798 let count = crate::company_scope::fetch_one_scalar_scoped(
799 self.pool(),
800 sqlx::query_scalar::<_, i64>(&query),
801 )
802 .await? as u64;
803 Ok(count)
804 }
805
806 pub async fn count_deleted(&self) -> Result<u64> {
808 let query = format!(
809 "SELECT COUNT(*) FROM {} WHERE (metadata->>'deleted_at') IS NOT NULL",
810 self.table_name()
811 );
812 let count = crate::company_scope::fetch_one_scalar_scoped(
813 self.pool(),
814 sqlx::query_scalar::<_, i64>(&query),
815 )
816 .await? as u64;
817 Ok(count)
818 }
819
820 pub async fn bulk_soft_delete(&self, ids: &[String]) -> Result<u64> {
829 if ids.is_empty() {
830 return Ok(0);
831 }
832 let placeholders = id_in_placeholders(ids.len());
833 let query = format!(
834 "UPDATE {} SET metadata = jsonb_set(\
835 COALESCE(metadata, '{{}}'), '{{deleted_at}}', to_jsonb(NOW())\
836 ) WHERE id IN ({placeholders}) AND (metadata->>'deleted_at') IS NULL",
837 self.table_name()
838 );
839 let mut tx = self.pool().begin().await?;
840 crate::company_scope::bind_current_company(&mut tx).await?;
841 let mut q = sqlx::query(&query);
842 for id in ids {
843 q = q.bind(id);
844 }
845 let affected = q.execute(&mut *tx).await?.rows_affected();
846 if affected != ids.len() as u64 {
847 return Err(anyhow::anyhow!(
849 "bulk_soft_delete: {} of {} ids were not active/deletable; rolled back",
850 ids.len() as u64 - affected,
851 ids.len()
852 ));
853 }
854 tx.commit().await?;
855 Ok(affected)
856 }
857
858 pub async fn bulk_restore(&self, ids: &[String]) -> Result<Vec<T>> {
860 if ids.is_empty() {
861 return Ok(Vec::new());
862 }
863 let placeholders = id_in_placeholders(ids.len());
864 let query = format!(
865 "UPDATE {} SET metadata = metadata - 'deleted_at' \
866 WHERE id IN ({placeholders}) AND (metadata->>'deleted_at') IS NOT NULL \
867 RETURNING *",
868 self.table_name()
869 );
870 let mut tx = self.pool().begin().await?;
871 crate::company_scope::bind_current_company(&mut tx).await?;
872 let mut q = sqlx::query_as::<_, T>(&query);
873 for id in ids {
874 q = q.bind(id);
875 }
876 let rows = q.fetch_all(&mut *tx).await?;
877 if rows.len() != ids.len() {
878 return Err(anyhow::anyhow!(
879 "bulk_restore: {} of {} ids were not in trash; rolled back",
880 ids.len() - rows.len(),
881 ids.len()
882 ));
883 }
884 tx.commit().await?;
885 Ok(rows)
886 }
887
888 pub async fn bulk_permanent_delete(&self, ids: &[String]) -> Result<u64> {
890 if ids.is_empty() {
891 return Ok(0);
892 }
893 let placeholders = id_in_placeholders(ids.len());
894 let query = format!(
895 "DELETE FROM {} WHERE id IN ({placeholders}) AND (metadata->>'deleted_at') IS NOT NULL",
896 self.table_name()
897 );
898 let mut tx = self.pool().begin().await?;
899 crate::company_scope::bind_current_company(&mut tx).await?;
900 let mut q = sqlx::query(&query);
901 for id in ids {
902 q = q.bind(id);
903 }
904 let affected = q.execute(&mut *tx).await?.rows_affected();
905 if affected != ids.len() as u64 {
906 return Err(anyhow::anyhow!(
907 "bulk_permanent_delete: {} of {} ids were not in trash; rolled back",
908 ids.len() as u64 - affected,
909 ids.len()
910 ));
911 }
912 tx.commit().await?;
913 Ok(affected)
914 }
915
916 pub async fn restore_all(&self) -> Result<Vec<T>> {
920 let query = format!(
921 "UPDATE {} SET metadata = metadata - 'deleted_at' \
922 WHERE (metadata->>'deleted_at') IS NOT NULL \
923 RETURNING *",
924 self.table_name()
925 );
926 let rows = crate::company_scope::fetch_all_scoped(
927 self.pool(),
928 sqlx::query_as::<_, T>(&query),
929 )
930 .await?;
931 Ok(rows)
932 }
933
934 pub async fn bulk_update(&self, entities: &[T]) -> Result<Vec<T>>
937 where
938 T: Serialize + Send + Sync,
939 {
940 bulk_update_rows(
941 self.pool(),
942 self.table_name(),
943 ACTIVE_ROW_GUARD,
944 entities,
945 )
946 .await
947 }
948}
949
950fn id_in_placeholders(n: usize) -> String {
952 (1..=n)
953 .map(|i| format!("${i}::uuid"))
954 .collect::<Vec<_>>()
955 .join(", ")
956}
957
958fn build_update_parts<T: Serialize>(entity: &T) -> Result<(String, String, String)> {
961 let json_value = serde_json::to_value(entity)?;
962 let json_obj = match json_value {
963 serde_json::Value::Object(obj) => obj,
964 _ => return Err(anyhow::anyhow!("entity must serialize to a JSON object")),
965 };
966 let id = json_obj
967 .get("id")
968 .and_then(|v| v.as_str())
969 .ok_or_else(|| anyhow::anyhow!("entity missing string 'id' field"))?
970 .to_string();
971 let column_names = json_obj
972 .keys()
973 .filter(|k| *k != "id")
974 .map(|k| format!("\"{k}\""))
975 .collect::<Vec<_>>()
976 .join(", ");
977 let json_str = serde_json::to_string(&json_obj)?;
978 Ok((id, json_str, column_names))
979}
980
981impl<T> GenericCrudRepository<T, HardDelete>
984where
985 T: for<'r> FromRow<'r, PgRow> + Send + Sync + Unpin + Serialize,
986{
987 pub async fn find_by_text_field(&self, field: &str, value: &str) -> Result<Option<T>> {
991 self.find_by_text_field_with_cond(field, value, "").await
992 }
993
994 pub async fn exists_by_text_field(&self, field: &str, value: &str) -> Result<bool> {
996 self.exists_by_text_field_with_cond(field, value, "").await
997 }
998
999 pub async fn find_by_uuid_field(&self, field: &str, value: Uuid) -> Result<Option<T>> {
1001 self.find_by_uuid_field_with_cond(field, value, "").await
1002 }
1003
1004 pub async fn exists_by_uuid_field(&self, field: &str, value: Uuid) -> Result<bool> {
1006 self.exists_by_uuid_field_with_cond(field, value, "").await
1007 }
1008
1009 pub async fn list_paginated_filtered(
1013 &self,
1014 pagination: PaginationParams,
1015 filters: Option<&HashMap<String, String>>,
1016 ) -> Result<PaginatedResult<T>>
1017 where
1018 T: EntityRepoMeta + Send + Sync,
1019 {
1020 let filters_map = filters.cloned().unwrap_or_default();
1021 let column_types = T::column_types();
1022 let search_fields_owned: Vec<&'static str> = T::search_fields().iter().copied().collect();
1023 self.run_filtered_query(pagination, None, &filters_map, &column_types, &search_fields_owned).await
1024 }
1025
1026 pub async fn aggregate_filtered(
1028 &self,
1029 spec: &crate::repository::AggregateSpec,
1030 filters: Option<&HashMap<String, String>>,
1031 ) -> Result<crate::repository::AggregateResult>
1032 where
1033 T: EntityRepoMeta + Send + Sync,
1034 {
1035 let filters_map = filters.cloned().unwrap_or_default();
1036 let column_types = T::column_types();
1037 let search_fields_owned: Vec<&'static str> = T::search_fields().iter().copied().collect();
1038 self.run_aggregate_query(spec, None, &filters_map, &column_types, &search_fields_owned).await
1039 }
1040
1041 pub async fn find_by_id(&self, id: &str) -> Result<Option<T>> {
1043 self.inner.find_by_id(id).await
1044 }
1045
1046 pub async fn find_all(&self) -> Result<Vec<T>> {
1048 let query = format!("SELECT * FROM {}", self.table_name());
1049 let results = crate::company_scope::fetch_all_scoped(
1050 self.pool(),
1051 sqlx::query_as::<_, T>(&query),
1052 )
1053 .await?;
1054 Ok(results)
1055 }
1056
1057 pub async fn update(&self, id: &str, entity: &T) -> Result<Option<T>> {
1059 self.inner.update(id, entity).await
1060 }
1061
1062 pub async fn update_changed(&self, id: &str, before: &T, after: &T) -> Result<Option<T>> {
1066 let changed = crate::repository::changed_columns(before, after)?;
1067 self.inner
1068 .update_guarded(id, after, Some(&changed), "")
1069 .await
1070 }
1071
1072 pub async fn delete(&self, id: &str) -> Result<bool> {
1074 self.inner.delete(id).await
1075 }
1076
1077 pub async fn count(&self) -> Result<u64> {
1079 let query = format!("SELECT COUNT(*) FROM {}", self.table_name());
1080 let count = crate::company_scope::fetch_one_scalar_scoped(
1081 self.pool(),
1082 sqlx::query_scalar::<_, i64>(&query),
1083 )
1084 .await? as u64;
1085 Ok(count)
1086 }
1087
1088 pub async fn exists(&self, id: &str) -> Result<bool> {
1090 let query = format!(
1091 "SELECT 1 FROM {} WHERE id = $1::uuid LIMIT 1",
1092 self.table_name()
1093 );
1094 let result = crate::company_scope::fetch_optional_scalar_scoped(
1095 self.pool(),
1096 sqlx::query_scalar::<_, i32>(&query).bind(id),
1097 )
1098 .await?;
1099 Ok(result.is_some())
1100 }
1101
1102 pub async fn list_paginated(&self, pagination: PaginationParams) -> Result<PaginatedResult<T>> {
1104 let offset = pagination.offset();
1105 let limit = pagination.limit();
1106 let query = format!(
1107 "SELECT * FROM {} ORDER BY id DESC LIMIT $1 OFFSET $2",
1108 self.table_name()
1109 );
1110 let data = crate::company_scope::fetch_all_scoped(
1111 self.pool(),
1112 sqlx::query_as::<_, T>(&query).bind(limit as i64).bind(offset as i64),
1113 )
1114 .await?;
1115 let total = self.count().await?;
1116 Ok(PaginatedResult {
1117 data,
1118 pagination: PaginationInfo::new(pagination.page, pagination.per_page, total),
1119 })
1120 }
1121
1122 pub async fn bulk_delete(&self, ids: &[String]) -> Result<u64> {
1126 if ids.is_empty() {
1127 return Ok(0);
1128 }
1129 let placeholders = id_in_placeholders(ids.len());
1130 let query = format!(
1131 "DELETE FROM {} WHERE id IN ({placeholders})",
1132 self.table_name()
1133 );
1134 let mut tx = self.pool().begin().await?;
1135 crate::company_scope::bind_current_company(&mut tx).await?;
1136 let mut q = sqlx::query(&query);
1137 for id in ids {
1138 q = q.bind(id);
1139 }
1140 let affected = q.execute(&mut *tx).await?.rows_affected();
1141 if affected != ids.len() as u64 {
1142 return Err(anyhow::anyhow!(
1143 "bulk_delete: {} of {} ids not found; rolled back",
1144 ids.len() as u64 - affected,
1145 ids.len()
1146 ));
1147 }
1148 tx.commit().await?;
1149 Ok(affected)
1150 }
1151
1152 pub async fn bulk_update(&self, entities: &[T]) -> Result<Vec<T>> {
1154 bulk_update_rows(self.pool(), self.table_name(), "", entities).await
1155 }
1156}
1157
1158pub fn qualify_relation_table(caller_table: &str, target_table: &str) -> String {
1168 if target_table.contains('.') || !caller_table.contains('.') {
1169 target_table.to_string()
1170 } else {
1171 let schema = caller_table.split('.').next().unwrap_or(caller_table);
1172 format!("{schema}.{target_table}")
1173 }
1174}
1175
1176fn is_undefined_table(err: &anyhow::Error) -> bool {
1181 err.chain()
1182 .filter_map(|cause| cause.downcast_ref::<sqlx::Error>())
1183 .any(|sqlx_err| {
1184 sqlx_err
1185 .as_database_error()
1186 .map(|db| db.code().as_deref() == Some("42P01"))
1187 .unwrap_or(false)
1188 })
1189}
1190
1191pub async fn fetch_by_ids_as_json(
1206 pool: &PgPool,
1207 caller_table: &str,
1208 table: &str,
1209 ids: &[String],
1210) -> Result<Vec<serde_json::Value>> {
1211 if ids.is_empty() {
1212 return Ok(Vec::new());
1213 }
1214 let qualified = qualify_relation_table(caller_table, table);
1215 match fetch_rows_as_json(pool, &qualified, ids).await {
1216 Ok(rows) => Ok(rows),
1217 Err(err) if qualified != table && is_undefined_table(&err) => {
1218 fetch_rows_as_json(pool, table, ids).await
1219 }
1220 Err(err) => Err(err),
1221 }
1222}
1223
1224async fn fetch_rows_as_json(
1229 pool: &PgPool,
1230 table: &str,
1231 ids: &[String],
1232) -> Result<Vec<serde_json::Value>> {
1233 let query = format!("SELECT row_to_json(t) AS j FROM {table} t WHERE t.id = ANY($1::uuid[])");
1234 let rows: Vec<(serde_json::Value,)> =
1235 crate::company_scope::fetch_all_scoped(pool, sqlx::query_as(&query).bind(ids)).await?;
1236 Ok(rows.into_iter().map(|(j,)| j).collect())
1237}
1238
1239async fn bulk_update_rows<T>(
1244 pool: &PgPool,
1245 table: &str,
1246 active_guard: &str,
1247 entities: &[T],
1248) -> Result<Vec<T>>
1249where
1250 T: for<'r> FromRow<'r, PgRow> + Send + Sync + Unpin + Serialize,
1251{
1252 if entities.is_empty() {
1253 return Ok(Vec::new());
1254 }
1255 let mut tx = pool.begin().await?;
1256 crate::company_scope::bind_current_company(&mut tx).await?;
1257 let mut out = Vec::with_capacity(entities.len());
1258 for entity in entities {
1259 let (id, json_str, column_names) = build_update_parts(entity)?;
1260 let query = format!(
1261 "WITH new_row AS (\
1262 SELECT (jsonb_populate_record(NULL::{table}, $1::jsonb)).*\
1263 ) UPDATE {table} AS t \
1264 SET ({columns}) = (SELECT {columns} FROM new_row) \
1265 WHERE t.id = $2::uuid{guard} \
1266 RETURNING t.*",
1267 table = table,
1268 columns = column_names,
1269 guard = active_guard,
1270 );
1271 let updated = sqlx::query_as::<_, T>(&query)
1272 .bind(&json_str)
1273 .bind(&id)
1274 .fetch_optional(&mut *tx)
1275 .await?;
1276 match updated {
1277 Some(e) => out.push(e),
1278 None => {
1279 return Err(anyhow::anyhow!(
1280 "bulk_update: id '{id}' not found or already deleted; rolled back"
1281 ));
1282 }
1283 }
1284 }
1285 tx.commit().await?;
1286 Ok(out)
1287}
1288
1289fn camel_to_snake(s: &str) -> String {
1291 let mut out = String::with_capacity(s.len() + 4);
1292 for ch in s.chars() {
1293 if ch.is_uppercase() {
1294 out.push('_');
1295 out.extend(ch.to_lowercase());
1296 } else {
1297 out.push(ch);
1298 }
1299 }
1300 out
1301}