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 owner_field() -> Option<&'static str> {
82 None
83 }
84
85 fn company_field() -> Option<&'static str> {
101 None
102 }
103
104 fn relations() -> &'static [(&'static str, &'static str, &'static str)] {
109 &[]
110 }
111}
112
113#[derive(Debug, thiserror::Error)]
123#[error("entity is company-scoped ({column}) but the request carries no company scope")]
124pub struct MissingCompanyScope {
125 pub column: &'static str,
127}
128
129pub fn company_fence<T: EntityRepoMeta>(company: Option<Uuid>) -> Result<Option<String>, MissingCompanyScope> {
141 match (T::company_field(), company) {
142 (None, _) => Ok(None),
143 (Some(column), None) => Err(MissingCompanyScope { column }),
144 (Some(column), Some(id)) => Ok(Some(format!("{column} = '{id}'"))),
145 }
146}
147
148pub fn strip_client_company_filters<T: EntityRepoMeta>(filters: &mut HashMap<String, String>) {
160 let Some(column) = T::company_field() else {
161 return;
162 };
163 let camel = snake_to_camel(column);
164 filters.retain(|key, _| {
165 let base = key.split('[').next().unwrap_or(key);
166 !base.eq_ignore_ascii_case(column) && !base.eq_ignore_ascii_case(&camel)
167 });
168}
169
170pub fn and_conditions(a: Option<&str>, b: Option<String>) -> Option<String> {
175 match (a, b) {
176 (None, None) => None,
177 (Some(a), None) => Some(a.to_string()),
178 (None, Some(b)) => Some(b),
179 (Some(a), Some(b)) => Some(format!("{a} AND {b}")),
180 }
181}
182
183fn snake_to_camel(s: &str) -> String {
184 let mut out = String::with_capacity(s.len());
185 let mut upper = false;
186 for c in s.chars() {
187 if c == '_' {
188 upper = true;
189 } else if upper {
190 out.push(c.to_ascii_uppercase());
191 upper = false;
192 } else {
193 out.push(c);
194 }
195 }
196 out
197}
198
199pub struct SoftDelete;
203
204pub struct HardDelete;
206
207pub struct GenericCrudRepository<T, D = SoftDelete>
216where
217 T: for<'r> FromRow<'r, PgRow> + Send + Unpin,
218{
219 inner: PostgresRepository<T>,
220 _mode: PhantomData<D>,
221}
222
223impl<T, D> GenericCrudRepository<T, D>
226where
227 T: for<'r> FromRow<'r, PgRow> + Send + Unpin,
228{
229 pub fn new(pool: PgPool, table_name: &str) -> Self {
230 Self {
231 inner: PostgresRepository::new(pool, table_name),
232 _mode: PhantomData,
233 }
234 }
235
236 pub fn pool(&self) -> &PgPool {
237 self.inner.pool()
238 }
239
240 pub fn table_name(&self) -> &str {
241 self.inner.table_name()
242 }
243
244 pub async fn create(&self, entity: &T) -> Result<T>
246 where
247 T: Serialize + Send + Sync,
248 {
249 self.inner.create(entity).await
250 }
251
252 pub async fn bulk_create(&self, entities: &[T]) -> Result<Vec<T>>
254 where
255 T: Serialize + Send + Sync,
256 {
257 let tx = sqlx::pool::Pool::begin(self.pool()).await?;
258 let mut results = Vec::with_capacity(entities.len());
259 for entity in entities {
260 results.push(self.create(entity).await?);
261 }
262 tx.commit().await?;
263 Ok(results)
264 }
265
266 async fn find_by_text_field_with_cond(
273 &self,
274 field: &str,
275 value: &str,
276 extra: &str,
277 ) -> Result<Option<T>> {
278 let query = format!(
279 "SELECT * FROM {} WHERE {} = $1{}",
280 self.table_name(), field, extra
281 );
282 let result = crate::company_scope::fetch_optional_scoped(
283 self.pool(),
284 sqlx::query_as::<_, T>(&query).bind(value),
285 )
286 .await?;
287 Ok(result)
288 }
289
290 async fn exists_by_text_field_with_cond(
292 &self,
293 field: &str,
294 value: &str,
295 extra: &str,
296 ) -> Result<bool> {
297 let query = format!(
298 "SELECT 1 FROM {} WHERE {} = $1{} LIMIT 1",
299 self.table_name(), field, extra
300 );
301 let result = crate::company_scope::fetch_optional_scalar_scoped(
302 self.pool(),
303 sqlx::query_scalar::<_, i32>(&query).bind(value),
304 )
305 .await?;
306 Ok(result.is_some())
307 }
308
309 async fn find_by_uuid_field_with_cond(
311 &self,
312 field: &str,
313 value: Uuid,
314 extra: &str,
315 ) -> Result<Option<T>> {
316 let query = format!(
317 "SELECT * FROM {} WHERE {} = $1{}",
318 self.table_name(), field, extra
319 );
320 let result = crate::company_scope::fetch_optional_scoped(
321 self.pool(),
322 sqlx::query_as::<_, T>(&query).bind(value),
323 )
324 .await?;
325 Ok(result)
326 }
327
328 async fn exists_by_uuid_field_with_cond(
330 &self,
331 field: &str,
332 value: Uuid,
333 extra: &str,
334 ) -> Result<bool> {
335 let query = format!(
336 "SELECT 1 FROM {} WHERE {} = $1{} LIMIT 1",
337 self.table_name(), field, extra
338 );
339 let result = crate::company_scope::fetch_optional_scalar_scoped(
340 self.pool(),
341 sqlx::query_scalar::<_, i32>(&query).bind(value),
342 )
343 .await?;
344 Ok(result.is_some())
345 }
346
347 pub async fn run_filtered_query(
358 &self,
359 pagination: PaginationParams,
360 base_condition: Option<&str>,
361 filters: &HashMap<String, String>,
362 column_types: &HashMap<String, String>,
363 search_fields: &[&str],
364 ) -> Result<PaginatedResult<T>>
365 where
366 T: Send + Sync,
367 {
368 let mut filters_map = filters.clone();
369 if let Some(cond) = base_condition {
370 filters_map.insert("__base_condition".to_string(), cond.to_string());
371 }
372 self.inner
373 .list_with_filters(pagination, &filters_map, column_types, search_fields)
374 .await
375 }
376
377 pub async fn run_aggregate_query(
384 &self,
385 spec: &crate::repository::AggregateSpec,
386 base_condition: Option<&str>,
387 filters: &HashMap<String, String>,
388 column_types: &HashMap<String, String>,
389 search_fields: &[&str],
390 ) -> Result<crate::repository::AggregateResult>
391 where
392 T: crate::EntityRepoMeta + Send + Sync,
393 {
394 let mut spec = spec.clone();
398 if let (Some(group), Some(label)) = (&spec.group_by, &spec.label_field) {
399 let _ = label;
400 let camel = snake_to_camel(group);
401 if let Some((_, table, _)) = T::relations().iter().find(|(_, _, fk)| *fk == camel) {
402 let base_fk = camel_to_snake(&camel);
403 spec.label_relation = Some((table.to_string(), base_fk));
404 }
405 }
406 let mut filters_map = filters.clone();
407 if let Some(cond) = base_condition {
408 filters_map.insert("__base_condition".to_string(), cond.to_string());
409 }
410 self.inner
411 .aggregate_with_filters(&spec, &filters_map, column_types, search_fields)
412 .await
413 }
414}
415
416impl<T> GenericCrudRepository<T, SoftDelete>
419where
420 T: for<'r> FromRow<'r, PgRow> + Send + Unpin,
421{
422 pub async fn find_by_text_field(&self, field: &str, value: &str) -> Result<Option<T>> {
428 self.find_by_text_field_with_cond(field, value, " AND metadata->>'deleted_at' IS NULL").await
429 }
430
431 pub async fn exists_by_text_field(&self, field: &str, value: &str) -> Result<bool> {
433 self.exists_by_text_field_with_cond(field, value, " AND metadata->>'deleted_at' IS NULL").await
434 }
435
436 pub async fn find_by_uuid_field(&self, field: &str, value: Uuid) -> Result<Option<T>> {
438 self.find_by_uuid_field_with_cond(field, value, " AND metadata->>'deleted_at' IS NULL").await
439 }
440
441 pub async fn exists_by_uuid_field(&self, field: &str, value: Uuid) -> Result<bool> {
443 self.exists_by_uuid_field_with_cond(field, value, " AND metadata->>'deleted_at' IS NULL").await
444 }
445
446 pub async fn list_paginated_filtered(
452 &self,
453 pagination: PaginationParams,
454 filters: Option<&HashMap<String, String>>,
455 ) -> Result<PaginatedResult<T>>
456 where
457 T: EntityRepoMeta + Send + Sync,
458 {
459 let filters_map = filters.cloned().unwrap_or_default();
460 let column_types = T::column_types();
461 let search_fields_owned: Vec<&'static str> = T::search_fields().iter().copied().collect();
462 self.run_filtered_query(
463 pagination,
464 Some("metadata->>'deleted_at' IS NULL"),
465 &filters_map,
466 &column_types,
467 &search_fields_owned,
468 ).await
469 }
470
471 pub async fn aggregate_filtered(
473 &self,
474 spec: &crate::repository::AggregateSpec,
475 filters: Option<&HashMap<String, String>>,
476 ) -> Result<crate::repository::AggregateResult>
477 where
478 T: EntityRepoMeta + Send + Sync,
479 {
480 let filters_map = filters.cloned().unwrap_or_default();
481 let column_types = T::column_types();
482 let search_fields_owned: Vec<&'static str> = T::search_fields().iter().copied().collect();
483 self.run_aggregate_query(spec, Some("metadata->>'deleted_at' IS NULL"), &filters_map, &column_types, &search_fields_owned).await
484 }
485
486 pub async fn list_paginated_filtered_scoped(
497 &self,
498 pagination: PaginationParams,
499 filters: Option<&HashMap<String, String>>,
500 company: Option<Uuid>,
501 ) -> Result<PaginatedResult<T>>
502 where
503 T: EntityRepoMeta + Send + Sync,
504 {
505 let fence = company_fence::<T>(company)?;
506 let mut filters_map = filters.cloned().unwrap_or_default();
507 strip_client_company_filters::<T>(&mut filters_map);
508 let column_types = T::column_types();
509 let search_fields_owned: Vec<&'static str> = T::search_fields().iter().copied().collect();
510 self.run_filtered_query(
511 pagination,
512 and_conditions(Some("metadata->>'deleted_at' IS NULL"), fence).as_deref(),
513 &filters_map,
514 &column_types,
515 &search_fields_owned,
516 )
517 .await
518 }
519
520 pub async fn list_deleted_filtered_scoped(
525 &self,
526 pagination: PaginationParams,
527 filters: Option<&HashMap<String, String>>,
528 company: Option<Uuid>,
529 ) -> Result<PaginatedResult<T>>
530 where
531 T: EntityRepoMeta + Send + Sync,
532 {
533 let fence = company_fence::<T>(company)?;
534 let mut filters_map = filters.cloned().unwrap_or_default();
535 strip_client_company_filters::<T>(&mut filters_map);
536 let column_types = T::column_types();
537 let empty: &[&str] = &[];
538 self.run_filtered_query(
539 pagination,
540 and_conditions(Some("metadata->>'deleted_at' IS NOT NULL"), fence).as_deref(),
541 &filters_map,
542 &column_types,
543 empty,
544 )
545 .await
546 }
547
548 pub async fn list_deleted_filtered(
550 &self,
551 pagination: PaginationParams,
552 filters: Option<&HashMap<String, String>>,
553 ) -> Result<PaginatedResult<T>>
554 where
555 T: EntityRepoMeta + Send + Sync,
556 {
557 let filters_map = filters.cloned().unwrap_or_default();
558 let column_types = T::column_types();
559 let empty: &[&str] = &[];
560 self.run_filtered_query(
561 pagination,
562 Some("metadata->>'deleted_at' IS NOT NULL"),
563 &filters_map,
564 &column_types,
565 empty,
566 ).await
567 }
568
569 pub async fn find_by_id(&self, id: &str) -> Result<Option<T>> {
571 let query = format!(
572 "SELECT * FROM {} WHERE id = $1::uuid AND metadata->>'deleted_at' IS NULL",
573 self.table_name()
574 );
575 let result = crate::company_scope::fetch_optional_scoped(
576 self.pool(),
577 sqlx::query_as::<_, T>(&query).bind(id),
578 )
579 .await?;
580 Ok(result)
581 }
582
583 pub async fn find_all(&self) -> Result<Vec<T>> {
585 let query = format!(
586 "SELECT * FROM {} WHERE metadata->>'deleted_at' IS NULL",
587 self.table_name()
588 );
589 let results = crate::company_scope::fetch_all_scoped(
590 self.pool(),
591 sqlx::query_as::<_, T>(&query),
592 )
593 .await?;
594 Ok(results)
595 }
596
597 pub async fn update(&self, id: &str, entity: &T) -> Result<Option<T>>
599 where
600 T: Serialize + Send + Sync,
601 {
602 if self.find_by_id(id).await?.is_none() {
603 return Ok(None);
604 }
605 self.inner.update(id, entity).await
606 }
607
608 pub async fn delete(&self, id: &str) -> Result<bool> {
610 self.soft_delete(id).await
611 }
612
613 pub async fn count(&self) -> Result<u64> {
615 self.count_active().await
616 }
617
618 pub async fn exists(&self, id: &str) -> Result<bool> {
620 let query = format!(
621 "SELECT 1 FROM {} WHERE id = $1::uuid AND metadata->>'deleted_at' IS NULL LIMIT 1",
622 self.table_name()
623 );
624 let result = crate::company_scope::fetch_optional_scalar_scoped(
625 self.pool(),
626 sqlx::query_scalar::<_, i32>(&query).bind(id),
627 )
628 .await?;
629 Ok(result.is_some())
630 }
631
632 pub async fn list_paginated(&self, pagination: PaginationParams) -> Result<PaginatedResult<T>> {
634 let offset = pagination.offset();
635 let limit = pagination.limit();
636 let query = format!(
637 "SELECT * FROM {} WHERE metadata->>'deleted_at' IS NULL \
638 ORDER BY id DESC LIMIT $1 OFFSET $2",
639 self.table_name()
640 );
641 let data = crate::company_scope::fetch_all_scoped(
642 self.pool(),
643 sqlx::query_as::<_, T>(&query).bind(limit as i64).bind(offset as i64),
644 )
645 .await?;
646 let total = self.count_active().await?;
647 Ok(PaginatedResult {
648 data,
649 pagination: PaginationInfo::new(pagination.page, pagination.per_page, total),
650 })
651 }
652
653 pub async fn soft_delete(&self, id: &str) -> Result<bool> {
657 let query = format!(
658 "UPDATE {} SET metadata = jsonb_set(\
659 COALESCE(metadata, '{{}}'), \
660 '{{deleted_at}}', \
661 to_jsonb(NOW())\
662 ) WHERE id = $1::uuid AND (metadata->>'deleted_at') IS NULL",
663 self.table_name()
664 );
665 let result = crate::company_scope::execute_scoped(
666 self.pool(),
667 sqlx::query(&query).bind(id),
668 )
669 .await?;
670 Ok(result.rows_affected() > 0)
671 }
672
673 pub async fn restore(&self, id: &str) -> Result<Option<T>> {
675 let query = format!(
676 "UPDATE {} SET metadata = metadata - 'deleted_at' \
677 WHERE id = $1::uuid AND (metadata->>'deleted_at') IS NOT NULL \
678 RETURNING *",
679 self.table_name()
680 );
681 let result = crate::company_scope::fetch_optional_scoped(
682 self.pool(),
683 sqlx::query_as::<_, T>(&query).bind(id),
684 )
685 .await?;
686 Ok(result)
687 }
688
689 pub async fn list_deleted(&self, pagination: PaginationParams) -> Result<PaginatedResult<T>> {
691 let offset = pagination.offset();
692 let limit = pagination.limit();
693 let query = format!(
694 "SELECT * FROM {} WHERE (metadata->>'deleted_at') IS NOT NULL \
695 ORDER BY (metadata->>'deleted_at') DESC LIMIT $1 OFFSET $2",
696 self.table_name()
697 );
698 let data = crate::company_scope::fetch_all_scoped(
699 self.pool(),
700 sqlx::query_as::<_, T>(&query).bind(limit as i64).bind(offset as i64),
701 )
702 .await?;
703 let count_query = format!(
704 "SELECT COUNT(*) FROM {} WHERE (metadata->>'deleted_at') IS NOT NULL",
705 self.table_name()
706 );
707 let total = crate::company_scope::fetch_one_scalar_scoped(
708 self.pool(),
709 sqlx::query_scalar::<_, i64>(&count_query),
710 )
711 .await? as u64;
712 Ok(PaginatedResult {
713 data,
714 pagination: PaginationInfo::new(pagination.page, pagination.per_page, total),
715 })
716 }
717
718 pub async fn empty_trash(&self) -> Result<u64> {
720 let query = format!(
721 "DELETE FROM {} WHERE (metadata->>'deleted_at') IS NOT NULL",
722 self.table_name()
723 );
724 let result = crate::company_scope::execute_scoped(self.pool(), sqlx::query(&query)).await?;
725 Ok(result.rows_affected())
726 }
727
728 pub async fn find_deleted_by_id(&self, id: &str) -> Result<Option<T>> {
730 let query = format!(
731 "SELECT * FROM {} WHERE id = $1::uuid AND (metadata->>'deleted_at') IS NOT NULL",
732 self.table_name()
733 );
734 let result = crate::company_scope::fetch_optional_scoped(
735 self.pool(),
736 sqlx::query_as::<_, T>(&query).bind(id),
737 )
738 .await?;
739 Ok(result)
740 }
741
742 pub async fn permanent_delete(&self, id: &str) -> Result<bool> {
744 let query = format!(
745 "DELETE FROM {} WHERE id = $1::uuid AND (metadata->>'deleted_at') IS NOT NULL",
746 self.table_name()
747 );
748 let result = crate::company_scope::execute_scoped(
749 self.pool(),
750 sqlx::query(&query).bind(id),
751 )
752 .await?;
753 Ok(result.rows_affected() > 0)
754 }
755
756 pub async fn count_active(&self) -> Result<u64> {
758 let query = format!(
759 "SELECT COUNT(*) FROM {} WHERE (metadata->>'deleted_at') IS NULL",
760 self.table_name()
761 );
762 let count = crate::company_scope::fetch_one_scalar_scoped(
763 self.pool(),
764 sqlx::query_scalar::<_, i64>(&query),
765 )
766 .await? as u64;
767 Ok(count)
768 }
769
770 pub async fn count_deleted(&self) -> Result<u64> {
772 let query = format!(
773 "SELECT COUNT(*) FROM {} WHERE (metadata->>'deleted_at') IS NOT NULL",
774 self.table_name()
775 );
776 let count = crate::company_scope::fetch_one_scalar_scoped(
777 self.pool(),
778 sqlx::query_scalar::<_, i64>(&query),
779 )
780 .await? as u64;
781 Ok(count)
782 }
783
784 pub async fn bulk_soft_delete(&self, ids: &[String]) -> Result<u64> {
793 if ids.is_empty() {
794 return Ok(0);
795 }
796 let placeholders = id_in_placeholders(ids.len());
797 let query = format!(
798 "UPDATE {} SET metadata = jsonb_set(\
799 COALESCE(metadata, '{{}}'), '{{deleted_at}}', to_jsonb(NOW())\
800 ) WHERE id IN ({placeholders}) AND (metadata->>'deleted_at') IS NULL",
801 self.table_name()
802 );
803 let mut tx = self.pool().begin().await?;
804 crate::company_scope::bind_current_company(&mut tx).await?;
805 let mut q = sqlx::query(&query);
806 for id in ids {
807 q = q.bind(id);
808 }
809 let affected = q.execute(&mut *tx).await?.rows_affected();
810 if affected != ids.len() as u64 {
811 return Err(anyhow::anyhow!(
813 "bulk_soft_delete: {} of {} ids were not active/deletable; rolled back",
814 ids.len() as u64 - affected,
815 ids.len()
816 ));
817 }
818 tx.commit().await?;
819 Ok(affected)
820 }
821
822 pub async fn bulk_restore(&self, ids: &[String]) -> Result<Vec<T>> {
824 if ids.is_empty() {
825 return Ok(Vec::new());
826 }
827 let placeholders = id_in_placeholders(ids.len());
828 let query = format!(
829 "UPDATE {} SET metadata = metadata - 'deleted_at' \
830 WHERE id IN ({placeholders}) AND (metadata->>'deleted_at') IS NOT NULL \
831 RETURNING *",
832 self.table_name()
833 );
834 let mut tx = self.pool().begin().await?;
835 crate::company_scope::bind_current_company(&mut tx).await?;
836 let mut q = sqlx::query_as::<_, T>(&query);
837 for id in ids {
838 q = q.bind(id);
839 }
840 let rows = q.fetch_all(&mut *tx).await?;
841 if rows.len() != ids.len() {
842 return Err(anyhow::anyhow!(
843 "bulk_restore: {} of {} ids were not in trash; rolled back",
844 ids.len() - rows.len(),
845 ids.len()
846 ));
847 }
848 tx.commit().await?;
849 Ok(rows)
850 }
851
852 pub async fn bulk_permanent_delete(&self, ids: &[String]) -> Result<u64> {
854 if ids.is_empty() {
855 return Ok(0);
856 }
857 let placeholders = id_in_placeholders(ids.len());
858 let query = format!(
859 "DELETE FROM {} WHERE id IN ({placeholders}) AND (metadata->>'deleted_at') IS NOT NULL",
860 self.table_name()
861 );
862 let mut tx = self.pool().begin().await?;
863 crate::company_scope::bind_current_company(&mut tx).await?;
864 let mut q = sqlx::query(&query);
865 for id in ids {
866 q = q.bind(id);
867 }
868 let affected = q.execute(&mut *tx).await?.rows_affected();
869 if affected != ids.len() as u64 {
870 return Err(anyhow::anyhow!(
871 "bulk_permanent_delete: {} of {} ids were not in trash; rolled back",
872 ids.len() as u64 - affected,
873 ids.len()
874 ));
875 }
876 tx.commit().await?;
877 Ok(affected)
878 }
879
880 pub async fn restore_all(&self) -> Result<Vec<T>> {
884 let query = format!(
885 "UPDATE {} SET metadata = metadata - 'deleted_at' \
886 WHERE (metadata->>'deleted_at') IS NOT NULL \
887 RETURNING *",
888 self.table_name()
889 );
890 let rows = crate::company_scope::fetch_all_scoped(
891 self.pool(),
892 sqlx::query_as::<_, T>(&query),
893 )
894 .await?;
895 Ok(rows)
896 }
897
898 pub async fn bulk_update(&self, entities: &[T]) -> Result<Vec<T>>
901 where
902 T: Serialize + Send + Sync,
903 {
904 bulk_update_rows(
905 self.pool(),
906 self.table_name(),
907 " AND t.metadata->>'deleted_at' IS NULL",
908 entities,
909 )
910 .await
911 }
912}
913
914fn id_in_placeholders(n: usize) -> String {
916 (1..=n)
917 .map(|i| format!("${i}::uuid"))
918 .collect::<Vec<_>>()
919 .join(", ")
920}
921
922fn build_update_parts<T: Serialize>(entity: &T) -> Result<(String, String, String)> {
925 let json_value = serde_json::to_value(entity)?;
926 let json_obj = match json_value {
927 serde_json::Value::Object(obj) => obj,
928 _ => return Err(anyhow::anyhow!("entity must serialize to a JSON object")),
929 };
930 let id = json_obj
931 .get("id")
932 .and_then(|v| v.as_str())
933 .ok_or_else(|| anyhow::anyhow!("entity missing string 'id' field"))?
934 .to_string();
935 let column_names = json_obj
936 .keys()
937 .filter(|k| *k != "id")
938 .map(|k| format!("\"{k}\""))
939 .collect::<Vec<_>>()
940 .join(", ");
941 let json_str = serde_json::to_string(&json_obj)?;
942 Ok((id, json_str, column_names))
943}
944
945impl<T> GenericCrudRepository<T, HardDelete>
948where
949 T: for<'r> FromRow<'r, PgRow> + Send + Sync + Unpin + Serialize,
950{
951 pub async fn find_by_text_field(&self, field: &str, value: &str) -> Result<Option<T>> {
955 self.find_by_text_field_with_cond(field, value, "").await
956 }
957
958 pub async fn exists_by_text_field(&self, field: &str, value: &str) -> Result<bool> {
960 self.exists_by_text_field_with_cond(field, value, "").await
961 }
962
963 pub async fn find_by_uuid_field(&self, field: &str, value: Uuid) -> Result<Option<T>> {
965 self.find_by_uuid_field_with_cond(field, value, "").await
966 }
967
968 pub async fn exists_by_uuid_field(&self, field: &str, value: Uuid) -> Result<bool> {
970 self.exists_by_uuid_field_with_cond(field, value, "").await
971 }
972
973 pub async fn list_paginated_filtered(
977 &self,
978 pagination: PaginationParams,
979 filters: Option<&HashMap<String, String>>,
980 ) -> Result<PaginatedResult<T>>
981 where
982 T: EntityRepoMeta + Send + Sync,
983 {
984 let filters_map = filters.cloned().unwrap_or_default();
985 let column_types = T::column_types();
986 let search_fields_owned: Vec<&'static str> = T::search_fields().iter().copied().collect();
987 self.run_filtered_query(pagination, None, &filters_map, &column_types, &search_fields_owned).await
988 }
989
990 pub async fn aggregate_filtered(
992 &self,
993 spec: &crate::repository::AggregateSpec,
994 filters: Option<&HashMap<String, String>>,
995 ) -> Result<crate::repository::AggregateResult>
996 where
997 T: EntityRepoMeta + Send + Sync,
998 {
999 let filters_map = filters.cloned().unwrap_or_default();
1000 let column_types = T::column_types();
1001 let search_fields_owned: Vec<&'static str> = T::search_fields().iter().copied().collect();
1002 self.run_aggregate_query(spec, None, &filters_map, &column_types, &search_fields_owned).await
1003 }
1004
1005 pub async fn find_by_id(&self, id: &str) -> Result<Option<T>> {
1007 self.inner.find_by_id(id).await
1008 }
1009
1010 pub async fn find_all(&self) -> Result<Vec<T>> {
1012 let query = format!("SELECT * FROM {}", self.table_name());
1013 let results = crate::company_scope::fetch_all_scoped(
1014 self.pool(),
1015 sqlx::query_as::<_, T>(&query),
1016 )
1017 .await?;
1018 Ok(results)
1019 }
1020
1021 pub async fn update(&self, id: &str, entity: &T) -> Result<Option<T>> {
1023 self.inner.update(id, entity).await
1024 }
1025
1026 pub async fn delete(&self, id: &str) -> Result<bool> {
1028 self.inner.delete(id).await
1029 }
1030
1031 pub async fn count(&self) -> Result<u64> {
1033 let query = format!("SELECT COUNT(*) FROM {}", self.table_name());
1034 let count = crate::company_scope::fetch_one_scalar_scoped(
1035 self.pool(),
1036 sqlx::query_scalar::<_, i64>(&query),
1037 )
1038 .await? as u64;
1039 Ok(count)
1040 }
1041
1042 pub async fn exists(&self, id: &str) -> Result<bool> {
1044 let query = format!(
1045 "SELECT 1 FROM {} WHERE id = $1::uuid LIMIT 1",
1046 self.table_name()
1047 );
1048 let result = crate::company_scope::fetch_optional_scalar_scoped(
1049 self.pool(),
1050 sqlx::query_scalar::<_, i32>(&query).bind(id),
1051 )
1052 .await?;
1053 Ok(result.is_some())
1054 }
1055
1056 pub async fn list_paginated(&self, pagination: PaginationParams) -> Result<PaginatedResult<T>> {
1058 let offset = pagination.offset();
1059 let limit = pagination.limit();
1060 let query = format!(
1061 "SELECT * FROM {} ORDER BY id DESC LIMIT $1 OFFSET $2",
1062 self.table_name()
1063 );
1064 let data = crate::company_scope::fetch_all_scoped(
1065 self.pool(),
1066 sqlx::query_as::<_, T>(&query).bind(limit as i64).bind(offset as i64),
1067 )
1068 .await?;
1069 let total = self.count().await?;
1070 Ok(PaginatedResult {
1071 data,
1072 pagination: PaginationInfo::new(pagination.page, pagination.per_page, total),
1073 })
1074 }
1075
1076 pub async fn bulk_delete(&self, ids: &[String]) -> Result<u64> {
1080 if ids.is_empty() {
1081 return Ok(0);
1082 }
1083 let placeholders = id_in_placeholders(ids.len());
1084 let query = format!(
1085 "DELETE FROM {} WHERE id IN ({placeholders})",
1086 self.table_name()
1087 );
1088 let mut tx = self.pool().begin().await?;
1089 crate::company_scope::bind_current_company(&mut tx).await?;
1090 let mut q = sqlx::query(&query);
1091 for id in ids {
1092 q = q.bind(id);
1093 }
1094 let affected = q.execute(&mut *tx).await?.rows_affected();
1095 if affected != ids.len() as u64 {
1096 return Err(anyhow::anyhow!(
1097 "bulk_delete: {} of {} ids not found; rolled back",
1098 ids.len() as u64 - affected,
1099 ids.len()
1100 ));
1101 }
1102 tx.commit().await?;
1103 Ok(affected)
1104 }
1105
1106 pub async fn bulk_update(&self, entities: &[T]) -> Result<Vec<T>> {
1108 bulk_update_rows(self.pool(), self.table_name(), "", entities).await
1109 }
1110}
1111
1112pub fn qualify_relation_table(caller_table: &str, target_table: &str) -> String {
1122 if target_table.contains('.') || !caller_table.contains('.') {
1123 target_table.to_string()
1124 } else {
1125 let schema = caller_table.split('.').next().unwrap_or(caller_table);
1126 format!("{schema}.{target_table}")
1127 }
1128}
1129
1130fn is_undefined_table(err: &anyhow::Error) -> bool {
1135 err.chain()
1136 .filter_map(|cause| cause.downcast_ref::<sqlx::Error>())
1137 .any(|sqlx_err| {
1138 sqlx_err
1139 .as_database_error()
1140 .map(|db| db.code().as_deref() == Some("42P01"))
1141 .unwrap_or(false)
1142 })
1143}
1144
1145pub async fn fetch_by_ids_as_json(
1160 pool: &PgPool,
1161 caller_table: &str,
1162 table: &str,
1163 ids: &[String],
1164) -> Result<Vec<serde_json::Value>> {
1165 if ids.is_empty() {
1166 return Ok(Vec::new());
1167 }
1168 let qualified = qualify_relation_table(caller_table, table);
1169 match fetch_rows_as_json(pool, &qualified, ids).await {
1170 Ok(rows) => Ok(rows),
1171 Err(err) if qualified != table && is_undefined_table(&err) => {
1172 fetch_rows_as_json(pool, table, ids).await
1173 }
1174 Err(err) => Err(err),
1175 }
1176}
1177
1178async fn fetch_rows_as_json(
1183 pool: &PgPool,
1184 table: &str,
1185 ids: &[String],
1186) -> Result<Vec<serde_json::Value>> {
1187 let query = format!("SELECT row_to_json(t) AS j FROM {table} t WHERE t.id = ANY($1::uuid[])");
1188 let rows: Vec<(serde_json::Value,)> =
1189 crate::company_scope::fetch_all_scoped(pool, sqlx::query_as(&query).bind(ids)).await?;
1190 Ok(rows.into_iter().map(|(j,)| j).collect())
1191}
1192
1193async fn bulk_update_rows<T>(
1198 pool: &PgPool,
1199 table: &str,
1200 active_guard: &str,
1201 entities: &[T],
1202) -> Result<Vec<T>>
1203where
1204 T: for<'r> FromRow<'r, PgRow> + Send + Sync + Unpin + Serialize,
1205{
1206 if entities.is_empty() {
1207 return Ok(Vec::new());
1208 }
1209 let mut tx = pool.begin().await?;
1210 crate::company_scope::bind_current_company(&mut tx).await?;
1211 let mut out = Vec::with_capacity(entities.len());
1212 for entity in entities {
1213 let (id, json_str, column_names) = build_update_parts(entity)?;
1214 let query = format!(
1215 "WITH new_row AS (\
1216 SELECT (jsonb_populate_record(NULL::{table}, $1::jsonb)).*\
1217 ) UPDATE {table} AS t \
1218 SET ({columns}) = (SELECT {columns} FROM new_row) \
1219 WHERE t.id = $2::uuid{guard} \
1220 RETURNING t.*",
1221 table = table,
1222 columns = column_names,
1223 guard = active_guard,
1224 );
1225 let updated = sqlx::query_as::<_, T>(&query)
1226 .bind(&json_str)
1227 .bind(&id)
1228 .fetch_optional(&mut *tx)
1229 .await?;
1230 match updated {
1231 Some(e) => out.push(e),
1232 None => {
1233 return Err(anyhow::anyhow!(
1234 "bulk_update: id '{id}' not found or already deleted; rolled back"
1235 ));
1236 }
1237 }
1238 }
1239 tx.commit().await?;
1240 Ok(out)
1241}
1242
1243fn camel_to_snake(s: &str) -> String {
1245 let mut out = String::with_capacity(s.len() + 4);
1246 for ch in s.chars() {
1247 if ch.is_uppercase() {
1248 out.push('_');
1249 out.extend(ch.to_lowercase());
1250 } else {
1251 out.push(ch);
1252 }
1253 }
1254 out
1255}