1use async_trait::async_trait;
32use std::collections::HashMap;
33use std::marker::PhantomData;
34use std::sync::Arc;
35
36use backbone_messaging::crud_event::{CrudEvent, CrudEventPublisher, EventMetadata, NoOpCrudEventPublisher};
37
38use crate::persistence::traits::{CrudRepository, PersistentEntity, RepositoryError};
39
40#[derive(Debug, thiserror::Error)]
44pub enum ServiceError {
45 #[error("not found")]
46 NotFound,
47
48 #[error("already exists: {0}")]
49 AlreadyExists(String),
50
51 #[error("validation failed: {0}")]
52 Validation(String),
53
54 #[error("validation failed: {}", .0.iter().map(|v| v.message.as_str()).collect::<Vec<_>>().join("; "))]
58 Violations(Vec<crate::violation::Violation>),
59
60 #[error("repository error: {0}")]
61 Repository(#[from] RepositoryError),
62
63 #[error("internal error: {0}")]
64 Internal(String),
65}
66
67pub type ServiceResult<T> = Result<T, ServiceError>;
68
69pub trait FromCreateDto<DTO>: Sized {
76 fn from_create_dto(dto: DTO) -> ServiceResult<Self>;
77}
78
79pub trait ApplyUpdateDto<DTO> {
81 fn apply_update(self, dto: DTO) -> ServiceResult<Self>
82 where
83 Self: Sized;
84}
85
86#[async_trait]
92pub trait ServiceLifecycle<E: PersistentEntity>: Send + Sync {
93 async fn before_create(&self, entity: &mut E) -> ServiceResult<()> {
94 let _ = entity;
95 Ok(())
96 }
97
98 async fn after_create(&self, entity: &E) -> ServiceResult<()> {
99 let _ = entity;
100 Ok(())
101 }
102
103 async fn before_update(&self, entity: &mut E) -> ServiceResult<()> {
104 let _ = entity;
105 Ok(())
106 }
107
108 async fn after_update(&self, entity: &E) -> ServiceResult<()> {
109 let _ = entity;
110 Ok(())
111 }
112
113 async fn before_delete(&self, entity: &E) -> ServiceResult<()> {
114 let _ = entity;
115 Ok(())
116 }
117
118 async fn after_delete(&self, id: &str) -> ServiceResult<()> {
119 let _ = id;
120 Ok(())
121 }
122
123 async fn before_restore(&self, entity: &E) -> ServiceResult<()> {
125 let _ = entity;
126 Ok(())
127 }
128
129 async fn before_hard_delete(&self, entity: &E) -> ServiceResult<()> {
132 let _ = entity;
133 Ok(())
134 }
135
136 async fn before_restore_all(&self) -> ServiceResult<()> {
139 Ok(())
140 }
141
142 async fn before_empty_trash(&self) -> ServiceResult<()> {
145 Ok(())
146 }
147}
148
149pub struct NoOpLifecycle<E> {
151 _phantom: PhantomData<E>,
152}
153
154impl<E> NoOpLifecycle<E> {
155 pub fn new() -> Self {
156 Self {
157 _phantom: PhantomData,
158 }
159 }
160}
161
162impl<E> Default for NoOpLifecycle<E> {
163 fn default() -> Self {
164 Self::new()
165 }
166}
167
168#[async_trait]
169impl<E: PersistentEntity> ServiceLifecycle<E> for NoOpLifecycle<E> {}
170
171pub struct GenericCrudService<E, C, U, R>
185where
186 E: PersistentEntity + Clone,
187 R: CrudRepository<E>,
188{
189 repository: Arc<R>,
190 lifecycle: Arc<dyn ServiceLifecycle<E>>,
191 event_publisher: Arc<dyn CrudEventPublisher<E>>,
192 guard: Arc<dyn crate::write_guard::WriteGuard<E>>,
193 _phantom: PhantomData<(C, U)>,
194}
195
196impl<E, C, U, R> GenericCrudService<E, C, U, R>
197where
198 E: PersistentEntity + Clone + FromCreateDto<C> + ApplyUpdateDto<U>,
199 C: Send + Sync + 'static,
200 U: Send + Sync + 'static,
201 R: CrudRepository<E>,
202{
203 pub fn new(
205 repository: Arc<R>,
206 lifecycle: Arc<dyn ServiceLifecycle<E>>,
207 event_publisher: Arc<dyn CrudEventPublisher<E>>,
208 ) -> Self {
209 Self {
210 repository,
211 lifecycle,
212 event_publisher,
213 guard: Arc::new(crate::write_guard::AllowAll),
214 _phantom: PhantomData,
215 }
216 }
217
218 pub fn with_guard(mut self, guard: Arc<dyn crate::write_guard::WriteGuard<E>>) -> Self {
221 self.guard = guard;
222 self
223 }
224
225 async fn consult(
227 &self,
228 kind: crate::write_guard::WriteKind,
229 before: Option<&E>,
230 after: Option<&E>,
231 ) -> ServiceResult<()> {
232 self.consult_as(kind, before, after, true).await
233 }
234
235 async fn consult_quietly(
239 &self,
240 kind: crate::write_guard::WriteKind,
241 before: Option<&E>,
242 after: Option<&E>,
243 ) -> ServiceResult<()> {
244 self.consult_as(kind, before, after, false).await
245 }
246
247 async fn consult_as(
248 &self,
249 kind: crate::write_guard::WriteKind,
250 before: Option<&E>,
251 after: Option<&E>,
252 log_shadow: bool,
253 ) -> ServiceResult<()> {
254 let ctx = crate::write_guard::WriteCtx::new(kind, before, after);
255 let outcome = self.guard.check(&ctx).await;
256 for v in outcome.shadow.iter().filter(|_| log_shadow) {
257 tracing::warn!(
258 target: "backbone::write_guard",
259 entity = std::any::type_name::<E>(),
260 kind = ?kind,
261 code = %v.code,
262 path = %v.path,
263 "a rule running in shadow would refuse this write: {}",
264 v.message
265 );
266 }
267 if outcome.refuse.is_empty() {
268 Ok(())
269 } else {
270 Err(ServiceError::Violations(outcome.refuse))
271 }
272 }
273
274 pub fn with_lifecycle(repository: Arc<R>, lifecycle: Arc<dyn ServiceLifecycle<E>>) -> Self {
276 Self::new(repository, lifecycle, NoOpCrudEventPublisher::arc())
277 }
278
279 pub fn with_repository(repository: Arc<R>) -> Self {
281 Self::new(
282 repository,
283 Arc::new(NoOpLifecycle::new()),
284 NoOpCrudEventPublisher::arc(),
285 )
286 }
287
288 pub fn with_event_publisher(mut self, publisher: Arc<dyn CrudEventPublisher<E>>) -> Self {
290 self.event_publisher = publisher;
291 self
292 }
293
294 pub fn repository(&self) -> &R {
296 &self.repository
297 }
298
299 pub async fn list(
302 &self,
303 page: u32,
304 limit: u32,
305 filters: HashMap<String, String>,
306 ) -> ServiceResult<(Vec<E>, u64)> {
307 self.repository
308 .list_filtered(page, limit, filters)
309 .await
310 .map_err(ServiceError::Repository)
311 }
312
313 pub async fn list_with_info(
316 &self,
317 page: u32,
318 limit: u32,
319 filters: HashMap<String, String>,
320 ) -> ServiceResult<(Vec<E>, backbone_orm::repository::PaginationInfo)> {
321 self.repository
322 .list_filtered_with_info(page, limit, filters)
323 .await
324 .map_err(ServiceError::Repository)
325 }
326
327 pub async fn aggregate(
329 &self,
330 spec: &backbone_orm::repository::AggregateSpec,
331 filters: HashMap<String, String>,
332 ) -> ServiceResult<backbone_orm::repository::AggregateResult> {
333 self.repository
334 .aggregate_filtered(spec, filters)
335 .await
336 .map_err(ServiceError::Repository)
337 }
338
339 pub async fn create(&self, dto: C) -> ServiceResult<E> {
340 let mut entity = E::from_create_dto(dto)?;
341 self.consult(crate::write_guard::WriteKind::Create, None, Some(&entity)).await?;
342 self.lifecycle.before_create(&mut entity).await?;
343 let saved = self
344 .repository
345 .create(entity)
346 .await
347 .map_err(ServiceError::Repository)?;
348 self.lifecycle.after_create(&saved).await?;
349
350 let meta = EventMetadata::new(saved.entity_id(), std::any::type_name::<E>());
351 let _ = self
352 .event_publisher
353 .publish(CrudEvent::Created { entity: saved.clone(), metadata: meta })
354 .await;
355
356 Ok(saved)
357 }
358
359 pub async fn get_by_id(&self, id: &str) -> ServiceResult<Option<E>> {
360 self.repository
361 .find_by_id(id)
362 .await
363 .map_err(ServiceError::Repository)
364 }
365
366 pub async fn update(&self, id: &str, dto: U) -> ServiceResult<Option<E>> {
367 let existing = self
368 .repository
369 .find_by_id(id)
370 .await
371 .map_err(ServiceError::Repository)?;
372 let Some(before) = existing else {
373 return Ok(None);
374 };
375 let mut updated = before.clone().apply_update(dto)?;
376 refuse_protected_changes::<E>(&as_object(&before)?, &as_object(&updated)?)?;
377 self.consult(crate::write_guard::WriteKind::Update, Some(&before), Some(&updated)).await?;
378 self.lifecycle.before_update(&mut updated).await?;
379 let Some(saved) = self
381 .repository
382 .update_changed(&before, updated)
383 .await
384 .map_err(ServiceError::Repository)?
385 else {
386 return Ok(None);
387 };
388 self.lifecycle.after_update(&saved).await?;
389
390 let meta = EventMetadata::new(saved.entity_id(), std::any::type_name::<E>());
391 let _ = self
392 .event_publisher
393 .publish(CrudEvent::Updated { before, after: saved.clone(), metadata: meta })
394 .await;
395
396 Ok(Some(saved))
397 }
398
399 pub async fn soft_delete(&self, id: &str) -> ServiceResult<bool> {
400 let existing = self
401 .repository
402 .find_by_id(id)
403 .await
404 .map_err(ServiceError::Repository)?;
405 let Some(entity) = existing else {
406 return Ok(false);
407 };
408 self.consult(crate::write_guard::WriteKind::Delete, Some(&entity), None).await?;
409 self.lifecycle.before_delete(&entity).await?;
410 let deleted = self
411 .repository
412 .soft_delete(id)
413 .await
414 .map_err(ServiceError::Repository)?;
415 self.lifecycle.after_delete(id).await?;
416
417 if deleted {
418 let meta = EventMetadata::new(id, std::any::type_name::<E>());
419 let _ = self
420 .event_publisher
421 .publish(CrudEvent::SoftDeleted { entity, metadata: meta })
422 .await;
423 }
424
425 Ok(deleted)
426 }
427
428 pub async fn restore(&self, id: &str) -> ServiceResult<Option<E>> {
429 let Some(trashed) = self.get_deleted_by_id(id).await? else {
430 return Ok(None);
431 };
432 self.consult(crate::write_guard::WriteKind::Restore, Some(&trashed), None).await?;
433 self.lifecycle.before_restore(&trashed).await?;
434 let restored = self
435 .repository
436 .restore(id)
437 .await
438 .map_err(ServiceError::Repository)?;
439
440 if let Some(ref entity) = restored {
441 let meta = EventMetadata::new(entity.entity_id(), std::any::type_name::<E>());
442 let _ = self
443 .event_publisher
444 .publish(CrudEvent::Restored { entity: entity.clone(), metadata: meta })
445 .await;
446 }
447
448 Ok(restored)
449 }
450
451 pub async fn hard_delete(&self, id: &str) -> ServiceResult<bool> {
452 let Some(entity) = self.get_deleted_by_id(id).await? else {
453 return Ok(false);
454 };
455 self.consult(crate::write_guard::WriteKind::HardDelete, Some(&entity), None).await?;
456 self.lifecycle.before_hard_delete(&entity).await?;
457 let deleted = self
458 .repository
459 .hard_delete(id)
460 .await
461 .map_err(ServiceError::Repository)?;
462
463 if deleted {
464 let meta = EventMetadata::new(id, std::any::type_name::<E>());
465 let _ = self
466 .event_publisher
467 .publish(CrudEvent::HardDeleted {
468 entity_id: id.to_string(),
469 metadata: meta,
470 })
471 .await;
472 }
473
474 Ok(deleted)
475 }
476
477 pub async fn list_deleted(
478 &self,
479 page: u32,
480 limit: u32,
481 ) -> ServiceResult<(Vec<E>, u64)> {
482 self.repository
483 .list_deleted(page, limit)
484 .await
485 .map_err(ServiceError::Repository)
486 }
487
488 pub async fn empty_trash(&self) -> ServiceResult<u64> {
489 self.lifecycle.before_empty_trash().await?;
490 self.repository
491 .empty_trash()
492 .await
493 .map_err(ServiceError::Repository)
494 }
495
496 pub async fn count(&self) -> ServiceResult<u64> {
497 self.repository
498 .count()
499 .await
500 .map_err(ServiceError::Repository)
501 }
502
503 pub async fn count_active(&self) -> ServiceResult<u64> {
504 self.repository
505 .count()
506 .await
507 .map_err(ServiceError::Repository)
508 }
509
510 pub async fn find_by_id(&self, id: &str) -> ServiceResult<Option<E>> {
512 self.get_by_id(id).await
513 }
514
515 pub async fn permanent_delete(&self, id: &str) -> ServiceResult<bool> {
517 self.hard_delete(id).await
518 }
519
520 pub async fn count_deleted(&self) -> ServiceResult<u64> {
522 self.repository
523 .count_deleted()
524 .await
525 .map_err(ServiceError::Repository)
526 }
527
528 pub async fn get_deleted_by_id(&self, id: &str) -> ServiceResult<Option<E>> {
530 self.repository
531 .find_by_id_including_deleted(id)
532 .await
533 .map_err(ServiceError::Repository)
534 }
535
536 pub async fn list_deleted_filtered(
538 &self,
539 page: u32,
540 limit: u32,
541 _filters: HashMap<String, String>,
542 ) -> ServiceResult<(Vec<E>, u64)> {
543 self.list_deleted(page, limit).await
544 }
545
546 pub async fn upsert(&self, dto: C) -> ServiceResult<E> {
548 self.create(dto).await
549 }
550
551 pub async fn partial_update(
556 &self,
557 id: &str,
558 fields: HashMap<String, serde_json::Value>,
559 ) -> ServiceResult<Option<E>>
560 where
561 E: serde::Serialize + serde::de::DeserializeOwned,
562 {
563 let existing = self
564 .repository
565 .find_by_id(id)
566 .await
567 .map_err(ServiceError::Repository)?;
568 let Some(before) = existing else {
569 return Ok(None);
570 };
571 let mut patched = merge_patch(&before, fields)?;
572 self.consult(crate::write_guard::WriteKind::Update, Some(&before), Some(&patched)).await?;
573 self.lifecycle.before_update(&mut patched).await?;
574 let Some(saved) = self
577 .repository
578 .update_changed(&before, patched)
579 .await
580 .map_err(ServiceError::Repository)?
581 else {
582 return Ok(None);
583 };
584 self.lifecycle.after_update(&saved).await?;
585
586 let meta = EventMetadata::new(saved.entity_id(), std::any::type_name::<E>());
587 let _ = self
588 .event_publisher
589 .publish(CrudEvent::Updated { before, after: saved.clone(), metadata: meta })
590 .await;
591
592 Ok(Some(saved))
593 }
594
595 pub async fn bulk_create(&self, dtos: Vec<C>) -> ServiceResult<Vec<E>>
596 where
597 E: FromCreateDto<C>,
598 {
599 let mut results = Vec::with_capacity(dtos.len());
600 for dto in dtos {
601 let entity = self.create(dto).await?;
602 results.push(entity);
603 }
604 Ok(results)
605 }
606
607 pub async fn bulk_soft_delete(&self, ids: Vec<String>) -> ServiceResult<u64> {
616 check_batch_size(ids.len())?;
617 let ids = dedup_ids(ids);
618 let mut entities = Vec::with_capacity(ids.len());
623 for id in &ids {
624 match self
625 .repository
626 .find_by_id(id)
627 .await
628 .map_err(ServiceError::Repository)?
629 {
630 Some(e) => entities.push(e),
631 None => return Err(ServiceError::Validation(format!("id '{id}' not found"))),
632 }
633 }
634 for entity in &entities {
635 self.consult(crate::write_guard::WriteKind::Delete, Some(entity), None).await?;
636 self.lifecycle.before_delete(entity).await?;
637 }
638 let affected = self
639 .repository
640 .bulk_soft_delete(&ids)
641 .await
642 .map_err(ServiceError::Repository)?;
643 for (id, entity) in ids.iter().zip(entities.into_iter()) {
644 self.lifecycle.after_delete(id).await?;
645 let meta = EventMetadata::new(id, std::any::type_name::<E>());
646 let _ = self
647 .event_publisher
648 .publish(CrudEvent::SoftDeleted { entity, metadata: meta })
649 .await;
650 }
651 Ok(affected)
652 }
653
654 pub async fn bulk_restore(&self, ids: Vec<String>) -> ServiceResult<Vec<E>> {
656 check_batch_size(ids.len())?;
657 let ids = dedup_ids(ids);
658 for id in &ids {
661 if let Some(trashed) = self.get_deleted_by_id(id).await? {
662 self.consult(crate::write_guard::WriteKind::Restore, Some(&trashed), None).await?;
663 self.lifecycle.before_restore(&trashed).await?;
664 }
665 }
666 let restored = self
667 .repository
668 .bulk_restore(&ids)
669 .await
670 .map_err(ServiceError::Repository)?;
671 for entity in &restored {
672 let meta = EventMetadata::new(entity.entity_id(), std::any::type_name::<E>());
673 let _ = self
674 .event_publisher
675 .publish(CrudEvent::Restored { entity: entity.clone(), metadata: meta })
676 .await;
677 }
678 Ok(restored)
679 }
680
681 pub async fn restore_all(&self) -> ServiceResult<u64> {
684 self.lifecycle.before_restore_all().await?;
685 let restored = self
686 .repository
687 .restore_all()
688 .await
689 .map_err(ServiceError::Repository)?;
690 for entity in &restored {
691 let meta = EventMetadata::new(entity.entity_id(), std::any::type_name::<E>());
692 let _ = self
693 .event_publisher
694 .publish(CrudEvent::Restored { entity: entity.clone(), metadata: meta })
695 .await;
696 }
697 Ok(restored.len() as u64)
698 }
699
700 pub async fn bulk_permanent_delete(&self, ids: Vec<String>) -> ServiceResult<u64> {
702 check_batch_size(ids.len())?;
703 let ids = dedup_ids(ids);
704 for id in &ids {
710 if let Some(trashed) = self.get_deleted_by_id(id).await? {
711 self.consult(crate::write_guard::WriteKind::HardDelete, Some(&trashed), None).await?;
712 self.lifecycle.before_hard_delete(&trashed).await?;
713 }
714 }
715 let affected = self
716 .repository
717 .bulk_hard_delete(&ids)
718 .await
719 .map_err(ServiceError::Repository)?;
720 for id in &ids {
721 let meta = EventMetadata::new(id, std::any::type_name::<E>());
722 let _ = self
723 .event_publisher
724 .publish(CrudEvent::HardDeleted {
725 entity_id: id.to_string(),
726 metadata: meta,
727 })
728 .await;
729 }
730 Ok(affected)
731 }
732
733 pub async fn bulk_update(&self, items: Vec<(String, U)>) -> ServiceResult<Vec<E>> {
735 check_batch_size(items.len())?;
736 if let Some(dup) = first_duplicate_id(items.iter().map(|(id, _)| id.clone())) {
737 return Err(ServiceError::Validation(format!(
738 "duplicate id '{dup}' in bulk update"
739 )));
740 }
741 let mut befores = Vec::with_capacity(items.len());
742 let mut prepared = Vec::with_capacity(items.len());
743 for (id, dto) in items {
744 let Some(before) = self
745 .repository
746 .find_by_id(&id)
747 .await
748 .map_err(ServiceError::Repository)?
749 else {
750 return Err(ServiceError::Validation(format!("id '{id}' not found")));
751 };
752 let mut updated = before.clone().apply_update(dto)?;
753 refuse_protected_changes::<E>(&as_object(&before)?, &as_object(&updated)?)?;
754 self.consult(crate::write_guard::WriteKind::Update, Some(&before), Some(&updated)).await?;
755 self.lifecycle.before_update(&mut updated).await?;
756 befores.push(before);
757 prepared.push(updated);
758 }
759 let saved = self
760 .repository
761 .bulk_update(prepared)
762 .await
763 .map_err(ServiceError::Repository)?;
764 self.publish_bulk_updates(befores, &saved).await?;
765 Ok(saved)
766 }
767
768 async fn publish_bulk_updates(&self, befores: Vec<E>, saved: &[E]) -> ServiceResult<()> {
772 for (before, after) in befores.into_iter().zip(saved.iter()) {
773 self.lifecycle.after_update(after).await?;
774 let meta = EventMetadata::new(after.entity_id(), std::any::type_name::<E>());
775 let _ = self
776 .event_publisher
777 .publish(CrudEvent::Updated {
778 before,
779 after: after.clone(),
780 metadata: meta,
781 })
782 .await;
783 }
784 Ok(())
785 }
786
787 pub async fn bulk_partial_update(
789 &self,
790 items: Vec<(String, HashMap<String, serde_json::Value>)>,
791 ) -> ServiceResult<Vec<E>>
792 where
793 E: serde::Serialize + serde::de::DeserializeOwned,
794 {
795 check_batch_size(items.len())?;
796 if let Some(dup) = first_duplicate_id(items.iter().map(|(id, _)| id.clone())) {
797 return Err(ServiceError::Validation(format!(
798 "duplicate id '{dup}' in bulk partial update"
799 )));
800 }
801 let mut befores = Vec::with_capacity(items.len());
802 let mut prepared = Vec::with_capacity(items.len());
803 for (id, fields) in items {
804 let Some(before) = self
805 .repository
806 .find_by_id(&id)
807 .await
808 .map_err(ServiceError::Repository)?
809 else {
810 return Err(ServiceError::Validation(format!("id '{id}' not found")));
811 };
812 let mut patched = merge_patch(&before, fields)?;
813 self.consult(crate::write_guard::WriteKind::Update, Some(&before), Some(&patched)).await?;
814 self.lifecycle.before_update(&mut patched).await?;
815 befores.push(before);
816 prepared.push(patched);
817 }
818 let saved = self
819 .repository
820 .bulk_update(prepared)
821 .await
822 .map_err(ServiceError::Repository)?;
823 self.publish_bulk_updates(befores, &saved).await?;
824 Ok(saved)
825 }
826
827 pub async fn preview_bulk_partial_update(
836 &self,
837 items: Vec<(String, HashMap<String, serde_json::Value>)>,
838 ) -> ServiceResult<Vec<crate::http::BulkPreviewRow<E, ServiceError>>>
839 where
840 E: serde::Serialize + serde::de::DeserializeOwned,
841 {
842 check_batch_size(items.len())?;
843 if let Some(dup) = first_duplicate_id(items.iter().map(|(id, _)| id.clone())) {
844 return Err(ServiceError::Validation(format!(
845 "duplicate id '{dup}' in bulk partial update"
846 )));
847 }
848 let mut rows = Vec::with_capacity(items.len());
849 for (id, fields) in items {
850 let outcome = self.preview_one(&id, fields).await;
851 rows.push(crate::http::BulkPreviewRow { id, outcome });
852 }
853 Ok(rows)
854 }
855
856 async fn preview_one(
857 &self,
858 id: &str,
859 fields: HashMap<String, serde_json::Value>,
860 ) -> ServiceResult<(E, E)>
861 where
862 E: serde::Serialize + serde::de::DeserializeOwned,
863 {
864 let Some(before) = self
865 .repository
866 .find_by_id(id)
867 .await
868 .map_err(ServiceError::Repository)?
869 else {
870 return Err(ServiceError::Validation(format!("id '{id}' not found")));
871 };
872 let mut patched = merge_patch(&before, fields)?;
873 self.consult_quietly(crate::write_guard::WriteKind::Update, Some(&before), Some(&patched))
874 .await?;
875 self.lifecycle.before_update(&mut patched).await?;
876 Ok((before, patched))
877 }
878}
879
880pub const ALWAYS_WRITE_PROTECTED: &[&str] = &["id", "metadata"];
887
888fn as_object(entity: &impl serde::Serialize) -> ServiceResult<serde_json::Map<String, serde_json::Value>> {
889 match serde_json::to_value(entity).map_err(|e| ServiceError::Internal(e.to_string()))? {
890 serde_json::Value::Object(map) => Ok(map),
891 _ => Err(ServiceError::Internal("an entity must serialize to a JSON object".into())),
892 }
893}
894
895fn refuse_protected_changes<E: PersistentEntity>(
906 before: &serde_json::Map<String, serde_json::Value>,
907 after: &serde_json::Map<String, serde_json::Value>,
908) -> ServiceResult<()> {
909 let in_transition = crate::write_guard::current_transition().is_some();
910 let refused: Vec<crate::violation::Violation> = ALWAYS_WRITE_PROTECTED
911 .iter()
912 .chain(E::write_protected_fields())
913 .filter(|key| !(in_transition && E::state_fields().contains(*key)))
914 .filter(|key| before.get(**key) != after.get(**key))
915 .map(|key| {
916 crate::violation::Violation::new(
917 *key,
918 "field_not_writable",
919 format!(
920 "field_not_writable: `{key}` cannot be changed by a generic write; \
921 it changes only through the operation that owns it"
922 ),
923 )
924 })
925 .collect();
926 if refused.is_empty() {
927 Ok(())
928 } else {
929 Err(ServiceError::Violations(refused))
930 }
931}
932
933fn merge_patch<E: PersistentEntity>(
941 before: &E,
942 fields: HashMap<String, serde_json::Value>,
943) -> ServiceResult<E> {
944 let before_obj = as_object(before)?;
945 let mut merged = before_obj.clone();
946 let mut given = Vec::new();
947 for (key, value) in fields {
948 if !value.is_null() {
949 given.push(key.clone());
950 }
951 merged.insert(key, value);
952 }
953 let patched: E = serde_json::from_value(serde_json::Value::Object(merged))
954 .map_err(|e| ServiceError::Validation(e.to_string()))?;
955 let after_obj = as_object(&patched)?;
956
957 let mut unknown: Vec<String> = given.into_iter().filter(|k| !after_obj.contains_key(k)).collect();
958 if !unknown.is_empty() {
959 unknown.sort();
960 return Err(ServiceError::Violations(
961 unknown
962 .into_iter()
963 .map(|k| {
964 let message = format!("unknown_field: `{k}` is not a field of this record");
965 crate::violation::Violation::new(k, "unknown_field", message)
966 })
967 .collect(),
968 ));
969 }
970 refuse_protected_changes::<E>(&before_obj, &after_obj)?;
971 Ok(patched)
972}
973
974pub const MAX_BATCH_SIZE: usize = 1000;
978
979fn check_batch_size(count: usize) -> Result<(), ServiceError> {
981 if count > MAX_BATCH_SIZE {
982 return Err(ServiceError::Validation(format!(
983 "Batch too large: {count} items exceeds the maximum of {MAX_BATCH_SIZE}."
984 )));
985 }
986 Ok(())
987}
988
989fn dedup_ids(ids: Vec<String>) -> Vec<String> {
993 let mut seen = std::collections::HashSet::new();
994 ids.into_iter().filter(|id| seen.insert(id.clone())).collect()
995}
996
997fn first_duplicate_id(ids: impl IntoIterator<Item = String>) -> Option<String> {
1000 let mut seen = std::collections::HashSet::new();
1001 for id in ids {
1002 if !seen.insert(id.clone()) {
1003 return Some(id);
1004 }
1005 }
1006 None
1007}
1008
1009#[async_trait::async_trait]
1020impl<E, C, U, R> crate::http::CrudService<E, C, U> for GenericCrudService<E, C, U, R>
1021where
1022 E: PersistentEntity
1023 + Clone
1024 + serde::Serialize
1025 + serde::de::DeserializeOwned
1026 + FromCreateDto<C>
1027 + ApplyUpdateDto<U>
1028 + Send
1029 + Sync
1030 + 'static,
1031 C: Send + Sync + 'static,
1032 U: Send + Sync + 'static,
1033 R: CrudRepository<E> + Send + Sync + 'static,
1034{
1035 type Error = ServiceError;
1036
1037 fn violations_of(err: &ServiceError) -> Option<Vec<crate::violation::Violation>> {
1038 match err {
1039 ServiceError::Violations(v) | ServiceError::Repository(RepositoryError::Violations(v)) => Some(v.clone()),
1040 _ => None,
1041 }
1042 }
1043
1044 fn entity_name() -> &'static str {
1045 std::any::type_name::<E>()
1046 }
1047
1048 async fn fetch_related_json(&self, table: &str, ids: &[String]) -> Vec<serde_json::Value> {
1049 self.repository().fetch_related_json(table, ids).await
1050 }
1051
1052 async fn list(
1053 &self,
1054 page: u32,
1055 limit: u32,
1056 filters: HashMap<String, String>,
1057 ) -> Result<(Vec<E>, u64), ServiceError> {
1058 self.list(page, limit, filters).await
1059 }
1060
1061 async fn list_with_info(
1062 &self,
1063 page: u32,
1064 limit: u32,
1065 filters: HashMap<String, String>,
1066 ) -> Result<(Vec<E>, backbone_orm::repository::PaginationInfo), ServiceError> {
1067 self.list_with_info(page, limit, filters).await
1068 }
1069
1070 async fn aggregate(
1071 &self,
1072 spec: &backbone_orm::repository::AggregateSpec,
1073 filters: HashMap<String, String>,
1074 ) -> Result<backbone_orm::repository::AggregateResult, ServiceError> {
1075 self.aggregate(spec, filters).await
1076 }
1077
1078 fn table_name(&self) -> Option<&str> {
1079 self.repository.table_name()
1080 }
1081
1082 async fn create(&self, dto: C) -> Result<E, ServiceError> {
1083 self.create(dto).await
1084 }
1085
1086 async fn get_by_id(&self, id: &str) -> Result<Option<E>, ServiceError> {
1087 self.get_by_id(id).await
1088 }
1089
1090 async fn update(&self, id: &str, dto: U) -> Result<Option<E>, ServiceError> {
1091 self.update(id, dto).await
1092 }
1093
1094 async fn partial_update(
1095 &self,
1096 id: &str,
1097 fields: HashMap<String, serde_json::Value>,
1098 ) -> Result<Option<E>, ServiceError> {
1099 self.partial_update(id, fields).await
1100 }
1101
1102 async fn soft_delete(&self, id: &str) -> Result<bool, ServiceError> {
1103 self.soft_delete(id).await
1104 }
1105
1106 async fn bulk_create(&self, items: Vec<C>) -> Result<Vec<E>, ServiceError> {
1107 self.bulk_create(items).await
1108 }
1109
1110 async fn upsert(&self, dto: C) -> Result<E, ServiceError> {
1111 self.upsert(dto).await
1112 }
1113
1114 async fn list_deleted(&self, page: u32, limit: u32) -> Result<(Vec<E>, u64), ServiceError> {
1115 self.list_deleted(page, limit).await
1116 }
1117
1118 async fn restore(&self, id: &str) -> Result<Option<E>, ServiceError> {
1119 self.restore(id).await
1120 }
1121
1122 async fn empty_trash(&self) -> Result<u64, ServiceError> {
1123 self.empty_trash().await
1124 }
1125
1126 async fn get_deleted_by_id(&self, id: &str) -> Result<Option<E>, ServiceError> {
1127 self.get_deleted_by_id(id).await
1128 }
1129
1130 async fn permanent_delete(&self, id: &str) -> Result<bool, ServiceError> {
1131 self.permanent_delete(id).await
1132 }
1133
1134 async fn list_deleted_filtered(
1135 &self,
1136 page: u32,
1137 limit: u32,
1138 filters: HashMap<String, String>,
1139 ) -> Result<(Vec<E>, u64), ServiceError> {
1140 self.list_deleted_filtered(page, limit, filters).await
1141 }
1142
1143 async fn count_active(&self) -> Result<u64, ServiceError> {
1144 self.count_active().await
1145 }
1146
1147 async fn count_deleted(&self) -> Result<u64, ServiceError> {
1148 self.count_deleted().await
1149 }
1150
1151 async fn bulk_soft_delete(&self, ids: Vec<String>) -> Result<u64, ServiceError> {
1152 self.bulk_soft_delete(ids).await
1153 }
1154
1155 async fn bulk_restore(&self, ids: Vec<String>) -> Result<Vec<E>, ServiceError> {
1156 self.bulk_restore(ids).await
1157 }
1158
1159 async fn bulk_permanent_delete(&self, ids: Vec<String>) -> Result<u64, ServiceError> {
1160 self.bulk_permanent_delete(ids).await
1161 }
1162
1163 async fn restore_all(&self) -> Result<u64, ServiceError> {
1164 self.restore_all().await
1165 }
1166
1167 async fn bulk_update(&self, items: Vec<(String, U)>) -> Result<Vec<E>, ServiceError> {
1168 self.bulk_update(items).await
1169 }
1170
1171 async fn bulk_partial_update(
1172 &self,
1173 items: Vec<(String, HashMap<String, serde_json::Value>)>,
1174 ) -> Result<Vec<E>, ServiceError> {
1175 self.bulk_partial_update(items).await
1176 }
1177
1178 async fn preview_bulk_partial_update(
1179 &self,
1180 items: Vec<(String, HashMap<String, serde_json::Value>)>,
1181 ) -> Result<Option<Vec<crate::http::BulkPreviewRow<E, ServiceError>>>, ServiceError> {
1182 self.preview_bulk_partial_update(items).await.map(Some)
1183 }
1184}
1185
1186impl<E, C, U, R> Clone for GenericCrudService<E, C, U, R>
1188where
1189 E: PersistentEntity + Clone,
1190 R: CrudRepository<E>,
1191{
1192 fn clone(&self) -> Self {
1193 Self {
1194 repository: self.repository.clone(),
1195 lifecycle: self.lifecycle.clone(),
1196 event_publisher: self.event_publisher.clone(),
1197 guard: self.guard.clone(),
1198 _phantom: PhantomData,
1199 }
1200 }
1201}
1202
1203#[cfg(test)]
1204mod tests {
1205 use super::*;
1206 use chrono::{DateTime, Utc};
1207 use serde::{Deserialize, Serialize};
1208
1209 #[derive(Debug, Clone, PartialEq, Serialize, Deserialize)]
1212 struct Widget {
1213 id: String,
1214 name: String,
1215 #[serde(skip_serializing_if = "Option::is_none")]
1216 note: Option<String>,
1217 secret_hash: String,
1218 #[serde(default)]
1220 state: String,
1221 #[serde(skip_serializing_if = "Option::is_none")]
1222 deleted_at: Option<DateTime<Utc>>,
1223 created_at: DateTime<Utc>,
1224 updated_at: DateTime<Utc>,
1225 }
1226
1227 impl crate::persistence::traits::PersistentEntity for Widget {
1228 fn write_protected_fields() -> &'static [&'static str] {
1229 &["secret_hash", "state"]
1230 }
1231 fn state_fields() -> &'static [&'static str] {
1232 &["state"]
1233 }
1234 fn entity_id(&self) -> String {
1235 self.id.clone()
1236 }
1237 fn set_entity_id(&mut self, id: String) {
1238 self.id = id;
1239 }
1240 fn created_at(&self) -> Option<DateTime<Utc>> {
1241 Some(self.created_at)
1242 }
1243 fn set_created_at(&mut self, ts: DateTime<Utc>) {
1244 self.created_at = ts;
1245 }
1246 fn updated_at(&self) -> Option<DateTime<Utc>> {
1247 Some(self.updated_at)
1248 }
1249 fn set_updated_at(&mut self, ts: DateTime<Utc>) {
1250 self.updated_at = ts;
1251 }
1252 fn deleted_at(&self) -> Option<DateTime<Utc>> {
1253 self.deleted_at
1254 }
1255 fn set_deleted_at(&mut self, ts: Option<DateTime<Utc>>) {
1256 self.deleted_at = ts;
1257 }
1258 }
1259
1260 struct CreateWidgetDto {
1261 name: String,
1262 }
1263
1264 struct UpdateWidgetDto {
1265 name: String,
1266 }
1267
1268 impl FromCreateDto<CreateWidgetDto> for Widget {
1269 fn from_create_dto(dto: CreateWidgetDto) -> ServiceResult<Self> {
1270 Ok(Widget {
1271 id: uuid::Uuid::new_v4().to_string(),
1272 name: dto.name,
1273 note: None,
1274 secret_hash: "h0".into(),
1275 state: "draft".into(),
1276 deleted_at: None,
1277 created_at: Utc::now(),
1278 updated_at: Utc::now(),
1279 })
1280 }
1281 }
1282
1283 impl ApplyUpdateDto<UpdateWidgetDto> for Widget {
1284 fn apply_update(mut self, dto: UpdateWidgetDto) -> ServiceResult<Self> {
1285 self.name = dto.name;
1286 self.updated_at = Utc::now();
1287 Ok(self)
1288 }
1289 }
1290
1291 struct InMemoryWidgetRepo {
1293 data: tokio::sync::Mutex<Vec<Widget>>,
1294 changed_writes: std::sync::Mutex<Vec<(Widget, Widget)>>,
1296 full_writes: std::sync::atomic::AtomicUsize,
1298 vanish_on_write: std::sync::atomic::AtomicBool,
1300 }
1301
1302 impl InMemoryWidgetRepo {
1303 fn new() -> Self {
1304 Self {
1305 data: tokio::sync::Mutex::new(Vec::new()),
1306 changed_writes: std::sync::Mutex::new(Vec::new()),
1307 full_writes: std::sync::atomic::AtomicUsize::new(0),
1308 vanish_on_write: std::sync::atomic::AtomicBool::new(false),
1309 }
1310 }
1311 }
1312
1313 #[async_trait]
1314 impl crate::persistence::traits::CrudRepository<Widget> for InMemoryWidgetRepo {
1315 async fn create(&self, entity: Widget) -> Result<Widget, RepositoryError> {
1316 self.data.lock().await.push(entity.clone());
1317 Ok(entity)
1318 }
1319
1320 async fn find_by_id(&self, id: &str) -> Result<Option<Widget>, RepositoryError> {
1321 Ok(self
1322 .data
1323 .lock()
1324 .await
1325 .iter()
1326 .find(|w| w.id == id && w.deleted_at.is_none())
1327 .cloned())
1328 }
1329
1330 async fn update(&self, entity: Widget) -> Result<Widget, RepositoryError> {
1331 self.full_writes
1332 .fetch_add(1, std::sync::atomic::Ordering::SeqCst);
1333 let mut data = self.data.lock().await;
1334 data.retain(|w| w.id != entity.id);
1335 data.push(entity.clone());
1336 Ok(entity)
1337 }
1338
1339 async fn update_changed(
1340 &self,
1341 before: &Widget,
1342 after: Widget,
1343 ) -> Result<Option<Widget>, RepositoryError> {
1344 self.changed_writes
1345 .lock()
1346 .unwrap()
1347 .push((before.clone(), after.clone()));
1348 if self
1349 .vanish_on_write
1350 .load(std::sync::atomic::Ordering::SeqCst)
1351 {
1352 return Ok(None);
1353 }
1354 let mut data = self.data.lock().await;
1355 data.retain(|w| w.id != after.id);
1356 data.push(after.clone());
1357 Ok(Some(after))
1358 }
1359
1360 async fn soft_delete(&self, id: &str) -> Result<bool, RepositoryError> {
1361 let mut data = self.data.lock().await;
1362 if let Some(w) = data.iter_mut().find(|w| w.id == id) {
1363 w.deleted_at = Some(Utc::now());
1364 return Ok(true);
1365 }
1366 Ok(false)
1367 }
1368
1369 async fn find_by_id_including_deleted(&self, id: &str) -> Result<Option<Widget>, RepositoryError> {
1370 Ok(self.data.lock().await.iter().find(|w| w.id == id).cloned())
1371 }
1372
1373 async fn restore(&self, id: &str) -> Result<Option<Widget>, RepositoryError> {
1374 let mut data = self.data.lock().await;
1375 if let Some(w) = data.iter_mut().find(|w| w.id == id) {
1376 w.deleted_at = None;
1377 return Ok(Some(w.clone()));
1378 }
1379 Ok(None)
1380 }
1381
1382 async fn hard_delete(&self, id: &str) -> Result<bool, RepositoryError> {
1383 let mut data = self.data.lock().await;
1384 let before = data.len();
1385 data.retain(|w| w.id != id);
1386 Ok(data.len() < before)
1387 }
1388
1389 async fn list(
1390 &self,
1391 page: u32,
1392 limit: u32,
1393 ) -> Result<(Vec<Widget>, u64), RepositoryError> {
1394 let data = self.data.lock().await;
1395 let active: Vec<_> = data.iter().filter(|w| w.deleted_at.is_none()).cloned().collect();
1396 let total = active.len() as u64;
1397 let offset = ((page.saturating_sub(1)) as usize) * (limit as usize);
1398 let limit = limit as usize;
1399 let page = active.into_iter().skip(offset).take(limit).collect();
1400 Ok((page, total))
1401 }
1402
1403 async fn list_deleted(
1404 &self,
1405 page: u32,
1406 limit: u32,
1407 ) -> Result<(Vec<Widget>, u64), RepositoryError> {
1408 let data = self.data.lock().await;
1409 let deleted: Vec<_> = data.iter().filter(|w| w.deleted_at.is_some()).cloned().collect();
1410 let total = deleted.len() as u64;
1411 let offset = ((page.saturating_sub(1)) as usize) * (limit as usize);
1412 let limit = limit as usize;
1413 let page = deleted.into_iter().skip(offset).take(limit).collect();
1414 Ok((page, total))
1415 }
1416
1417 async fn count(&self) -> Result<u64, RepositoryError> {
1418 let data = self.data.lock().await;
1419 Ok(data.iter().filter(|w| w.deleted_at.is_none()).count() as u64)
1420 }
1421
1422 async fn count_deleted(&self) -> Result<u64, RepositoryError> {
1423 let data = self.data.lock().await;
1424 Ok(data.iter().filter(|w| w.deleted_at.is_some()).count() as u64)
1425 }
1426
1427 async fn bulk_create(&self, entities: Vec<Widget>) -> Result<Vec<Widget>, RepositoryError> {
1428 let mut data = self.data.lock().await;
1429 data.extend(entities.clone());
1430 Ok(entities)
1431 }
1432
1433 async fn empty_trash(&self) -> Result<u64, RepositoryError> {
1434 let mut data = self.data.lock().await;
1435 let before = data.len();
1436 data.retain(|w| w.deleted_at.is_none());
1437 Ok((before - data.len()) as u64)
1438 }
1439 }
1440
1441 #[tokio::test]
1442 async fn create_and_get_roundtrip() {
1443 let repo = Arc::new(InMemoryWidgetRepo::new());
1444 let service: GenericCrudService<Widget, CreateWidgetDto, UpdateWidgetDto, InMemoryWidgetRepo> =
1445 GenericCrudService::with_repository(repo);
1446
1447 let widget = service
1448 .create(CreateWidgetDto {
1449 name: "sprocket".into(),
1450 })
1451 .await
1452 .unwrap();
1453
1454 let found = service.get_by_id(&widget.id).await.unwrap();
1455 assert!(found.is_some());
1456 assert_eq!(found.unwrap().name, "sprocket");
1457 }
1458
1459 #[tokio::test]
1460 async fn soft_delete_hides_from_list() {
1461 let repo = Arc::new(InMemoryWidgetRepo::new());
1462 let service: GenericCrudService<Widget, CreateWidgetDto, UpdateWidgetDto, InMemoryWidgetRepo> =
1463 GenericCrudService::with_repository(repo);
1464
1465 let w = service.create(CreateWidgetDto { name: "w".into() }).await.unwrap();
1466 service.soft_delete(&w.id).await.unwrap();
1467
1468 let (items, _) = service.list(1, 20, Default::default()).await.unwrap();
1469 assert!(items.is_empty());
1470
1471 let (deleted, _) = service.list_deleted(1, 20).await.unwrap();
1472 assert_eq!(deleted.len(), 1);
1473 }
1474
1475 #[tokio::test]
1476 async fn update_publishes_event_and_returns_new_entity() {
1477 let repo = Arc::new(InMemoryWidgetRepo::new());
1478 let service: GenericCrudService<Widget, CreateWidgetDto, UpdateWidgetDto, InMemoryWidgetRepo> =
1479 GenericCrudService::with_repository(repo);
1480
1481 let w = service.create(CreateWidgetDto { name: "old".into() }).await.unwrap();
1482 let updated = service
1483 .update(&w.id, UpdateWidgetDto { name: "new".into() })
1484 .await
1485 .unwrap();
1486 assert_eq!(updated.unwrap().name, "new");
1487 }
1488
1489 #[tokio::test]
1492 async fn a_patch_writes_only_through_the_changed_columns_path() {
1493 let repo = Arc::new(InMemoryWidgetRepo::new());
1494 let service: GenericCrudService<
1495 Widget,
1496 CreateWidgetDto,
1497 UpdateWidgetDto,
1498 InMemoryWidgetRepo,
1499 > = GenericCrudService::with_repository(repo.clone());
1500 let w = service
1501 .create(CreateWidgetDto { name: "old".into() })
1502 .await
1503 .unwrap();
1504
1505 let mut fields = HashMap::new();
1506 fields.insert("note".to_string(), serde_json::json!("edited"));
1507 let saved = service
1508 .partial_update(&w.id, fields)
1509 .await
1510 .unwrap()
1511 .unwrap();
1512 assert_eq!(saved.note.as_deref(), Some("edited"));
1513
1514 let writes = repo.changed_writes.lock().unwrap().clone();
1515 assert_eq!(writes.len(), 1, "a PATCH must save through update_changed");
1516 assert_eq!(writes[0].0, w, "the write starts from the row as read");
1517 assert_eq!(
1518 repo.full_writes.load(std::sync::atomic::Ordering::SeqCst),
1519 0
1520 );
1521 }
1522
1523 #[tokio::test]
1524 async fn a_put_writes_only_through_the_changed_columns_path() {
1525 let repo = Arc::new(InMemoryWidgetRepo::new());
1526 let service: GenericCrudService<
1527 Widget,
1528 CreateWidgetDto,
1529 UpdateWidgetDto,
1530 InMemoryWidgetRepo,
1531 > = GenericCrudService::with_repository(repo.clone());
1532 let w = service
1533 .create(CreateWidgetDto { name: "old".into() })
1534 .await
1535 .unwrap();
1536 service
1537 .update(&w.id, UpdateWidgetDto { name: "new".into() })
1538 .await
1539 .unwrap()
1540 .unwrap();
1541 assert_eq!(repo.changed_writes.lock().unwrap().len(), 1);
1542 assert_eq!(
1543 repo.full_writes.load(std::sync::atomic::Ordering::SeqCst),
1544 0
1545 );
1546 }
1547
1548 #[tokio::test]
1549 async fn a_write_whose_row_vanished_meanwhile_answers_not_found() {
1550 let repo = Arc::new(InMemoryWidgetRepo::new());
1551 let service: GenericCrudService<
1552 Widget,
1553 CreateWidgetDto,
1554 UpdateWidgetDto,
1555 InMemoryWidgetRepo,
1556 > = GenericCrudService::with_repository(repo.clone());
1557 let w = service
1558 .create(CreateWidgetDto { name: "old".into() })
1559 .await
1560 .unwrap();
1561 repo.vanish_on_write
1562 .store(true, std::sync::atomic::Ordering::SeqCst);
1563
1564 let mut fields = HashMap::new();
1565 fields.insert("note".to_string(), serde_json::json!("edited"));
1566 assert!(service
1567 .partial_update(&w.id, fields)
1568 .await
1569 .unwrap()
1570 .is_none());
1571 assert!(service
1572 .update(&w.id, UpdateWidgetDto { name: "new".into() })
1573 .await
1574 .unwrap()
1575 .is_none());
1576 }
1577
1578 #[tokio::test]
1581 async fn a_transition_moves_its_protected_state_field() {
1582 let service = svc();
1583 let w = service
1584 .create(CreateWidgetDto { name: "w".into() })
1585 .await
1586 .unwrap();
1587 let mut fields = HashMap::new();
1588 fields.insert("state".to_string(), serde_json::json!("active"));
1589 let saved =
1590 crate::write_guard::as_transition("activate", service.partial_update(&w.id, fields))
1591 .await
1592 .expect("a verb moves its own state field")
1593 .unwrap();
1594 assert_eq!(saved.state, "active");
1595 }
1596
1597 #[tokio::test]
1598 async fn a_transition_still_cannot_move_another_protected_field() {
1599 let service = svc();
1600 let w = service
1601 .create(CreateWidgetDto { name: "w".into() })
1602 .await
1603 .unwrap();
1604 let mut fields = HashMap::new();
1605 fields.insert("state".to_string(), serde_json::json!("active"));
1606 fields.insert("secret_hash".to_string(), serde_json::json!("h1"));
1607 let err =
1608 crate::write_guard::as_transition("activate", service.partial_update(&w.id, fields))
1609 .await
1610 .unwrap_err();
1611 match err {
1612 ServiceError::Violations(v) => {
1613 assert_eq!(v.len(), 1, "{v:?}");
1614 assert_eq!(
1615 (v[0].path.as_str(), v[0].code.as_str()),
1616 ("secret_hash", "field_not_writable")
1617 );
1618 }
1619 other => panic!("expected a violation, got {other:?}"),
1620 }
1621 }
1622
1623 #[tokio::test]
1624 async fn a_generic_write_still_cannot_move_the_state_field() {
1625 let service = svc();
1626 let w = service
1627 .create(CreateWidgetDto { name: "w".into() })
1628 .await
1629 .unwrap();
1630 let mut fields = HashMap::new();
1631 fields.insert("state".to_string(), serde_json::json!("active"));
1632 let err = service.partial_update(&w.id, fields).await.unwrap_err();
1633 assert!(
1634 matches!(err, ServiceError::Violations(ref v) if v[0].path == "state"),
1635 "{err:?}"
1636 );
1637 }
1638
1639 fn svc() -> GenericCrudService<Widget, CreateWidgetDto, UpdateWidgetDto, InMemoryWidgetRepo> {
1642 GenericCrudService::with_repository(Arc::new(InMemoryWidgetRepo::new()))
1643 }
1644
1645 #[tokio::test]
1646 async fn bulk_soft_delete_success() {
1647 let service = svc();
1648 let mut ids = Vec::new();
1649 for n in 0..3 {
1650 ids.push(service.create(CreateWidgetDto { name: format!("w{n}") }).await.unwrap().id);
1651 }
1652 let n = service.bulk_soft_delete(ids).await.unwrap();
1653 assert_eq!(n, 3);
1654 let (active, _) = service.list(1, 20, Default::default()).await.unwrap();
1655 assert!(active.is_empty());
1656 }
1657
1658 #[tokio::test]
1659 async fn bulk_soft_delete_is_all_or_nothing_on_missing_id() {
1660 let service = svc();
1661 let a = service.create(CreateWidgetDto { name: "a".into() }).await.unwrap();
1662 let b = service.create(CreateWidgetDto { name: "b".into() }).await.unwrap();
1663
1664 let res = service
1666 .bulk_soft_delete(vec![a.id.clone(), b.id.clone(), "does-not-exist".into()])
1667 .await;
1668 assert!(res.is_err());
1669
1670 let (active, _) = service.list(1, 20, Default::default()).await.unwrap();
1671 assert_eq!(active.len(), 2, "no rows should have been deleted");
1672 }
1673
1674 #[tokio::test]
1675 async fn bulk_restore_and_restore_all() {
1676 let service = svc();
1677 let mut ids = Vec::new();
1678 for n in 0..3 {
1679 ids.push(service.create(CreateWidgetDto { name: format!("w{n}") }).await.unwrap().id);
1680 }
1681 service.bulk_soft_delete(ids.clone()).await.unwrap();
1682
1683 let restored = service.bulk_restore(vec![ids[0].clone(), ids[1].clone()]).await.unwrap();
1685 assert_eq!(restored.len(), 2);
1686
1687 let n = service.restore_all().await.unwrap();
1689 assert_eq!(n, 1);
1690
1691 let (active, _) = service.list(1, 20, Default::default()).await.unwrap();
1692 assert_eq!(active.len(), 3);
1693 }
1694
1695 #[tokio::test]
1696 async fn bulk_update_is_all_or_nothing_on_missing_id() {
1697 let service = svc();
1698 let a = service.create(CreateWidgetDto { name: "a".into() }).await.unwrap();
1699
1700 let res = service
1701 .bulk_update(vec![
1702 (a.id.clone(), UpdateWidgetDto { name: "A2".into() }),
1703 ("missing".into(), UpdateWidgetDto { name: "X".into() }),
1704 ])
1705 .await;
1706 assert!(res.is_err());
1707
1708 let still = service.get_by_id(&a.id).await.unwrap().unwrap();
1710 assert_eq!(still.name, "a");
1711 }
1712
1713 #[tokio::test]
1714 async fn bulk_partial_update_success() {
1715 let service = svc();
1716 let a = service.create(CreateWidgetDto { name: "a".into() }).await.unwrap();
1717 let b = service.create(CreateWidgetDto { name: "b".into() }).await.unwrap();
1718
1719 let mut patch_a = std::collections::HashMap::new();
1720 patch_a.insert("name".to_string(), serde_json::json!("a2"));
1721 let mut patch_b = std::collections::HashMap::new();
1722 patch_b.insert("name".to_string(), serde_json::json!("b2"));
1723
1724 let saved = service
1725 .bulk_partial_update(vec![(a.id.clone(), patch_a), (b.id.clone(), patch_b)])
1726 .await
1727 .unwrap();
1728 assert_eq!(saved.len(), 2);
1729 assert_eq!(service.get_by_id(&a.id).await.unwrap().unwrap().name, "a2");
1730 assert_eq!(service.get_by_id(&b.id).await.unwrap().unwrap().name, "b2");
1731 }
1732
1733 #[tokio::test]
1734 async fn bulk_soft_delete_tolerates_duplicate_ids() {
1735 let service = svc();
1736 let a = service.create(CreateWidgetDto { name: "a".into() }).await.unwrap();
1737 let b = service.create(CreateWidgetDto { name: "b".into() }).await.unwrap();
1738
1739 let n = service
1741 .bulk_soft_delete(vec![a.id.clone(), a.id.clone(), b.id.clone()])
1742 .await
1743 .unwrap();
1744 assert_eq!(n, 2, "two distinct rows soft-deleted");
1745
1746 let (active, _) = service.list(1, 20, Default::default()).await.unwrap();
1747 assert!(active.is_empty());
1748 }
1749
1750 #[tokio::test]
1751 async fn bulk_update_rejects_duplicate_ids() {
1752 let service = svc();
1753 let a = service.create(CreateWidgetDto { name: "a".into() }).await.unwrap();
1754
1755 let res = service
1757 .bulk_update(vec![
1758 (a.id.clone(), UpdateWidgetDto { name: "first".into() }),
1759 (a.id.clone(), UpdateWidgetDto { name: "second".into() }),
1760 ])
1761 .await;
1762 assert!(res.is_err());
1763 assert_eq!(service.get_by_id(&a.id).await.unwrap().unwrap().name, "a");
1764 }
1765
1766 #[tokio::test]
1767 async fn bulk_soft_delete_rejects_oversized_batch() {
1768 let service = svc();
1769 let ids: Vec<String> = (0..MAX_BATCH_SIZE + 1).map(|n| format!("id-{n}")).collect();
1770 assert!(service.bulk_soft_delete(ids).await.is_err());
1772 }
1773
1774 struct RekeyWidgetDto {
1779 name: String,
1780 secret_hash: String,
1781 }
1782
1783 impl ApplyUpdateDto<RekeyWidgetDto> for Widget {
1784 fn apply_update(mut self, dto: RekeyWidgetDto) -> ServiceResult<Self> {
1785 self.name = dto.name;
1786 self.secret_hash = dto.secret_hash;
1787 Ok(self)
1788 }
1789 }
1790
1791 type RekeyService = GenericCrudService<Widget, CreateWidgetDto, RekeyWidgetDto, InMemoryWidgetRepo>;
1792
1793 fn patch(pairs: &[(&str, serde_json::Value)]) -> HashMap<String, serde_json::Value> {
1794 pairs.iter().map(|(k, v)| (k.to_string(), v.clone())).collect()
1795 }
1796
1797 fn validation_message(res: ServiceResult<impl std::fmt::Debug>) -> String {
1800 match res {
1801 Err(ServiceError::Validation(m)) => m,
1802 Err(e @ ServiceError::Violations(_)) => e.to_string(),
1803 other => panic!("expected a validation error, got {other:?}"),
1804 }
1805 }
1806
1807 #[tokio::test]
1808 async fn patch_rejects_a_field_the_entity_does_not_have() {
1809 let service = svc();
1810 let w = service.create(CreateWidgetDto { name: "a".into() }).await.unwrap();
1811
1812 let msg = validation_message(service.partial_update(&w.id, patch(&[("nmae", serde_json::json!("b"))])).await);
1813 assert!(msg.contains("unknown_field") && msg.contains("nmae"), "{msg}");
1814 assert_eq!(service.get_by_id(&w.id).await.unwrap().unwrap().name, "a");
1815 }
1816
1817 #[tokio::test]
1818 async fn patch_rejects_changing_the_id() {
1819 let service = svc();
1820 let w = service.create(CreateWidgetDto { name: "a".into() }).await.unwrap();
1821
1822 let msg = validation_message(service.partial_update(&w.id, patch(&[("id", serde_json::json!("other"))])).await);
1823 assert!(msg.contains("field_not_writable") && msg.contains("id"), "{msg}");
1824 assert!(service.get_by_id(&w.id).await.unwrap().is_some());
1825 }
1826
1827 #[tokio::test]
1828 async fn patch_rejects_changing_a_field_the_entity_protects() {
1829 let service = svc();
1830 let w = service.create(CreateWidgetDto { name: "a".into() }).await.unwrap();
1831
1832 let msg = validation_message(
1833 service.partial_update(&w.id, patch(&[("secret_hash", serde_json::json!("h1"))])).await,
1834 );
1835 assert!(msg.contains("field_not_writable") && msg.contains("secret_hash"), "{msg}");
1836 assert_eq!(service.get_by_id(&w.id).await.unwrap().unwrap().secret_hash, "h0");
1837 }
1838
1839 #[tokio::test]
1840 async fn patch_accepts_a_protected_field_echoed_unchanged() {
1841 let service = svc();
1842 let w = service.create(CreateWidgetDto { name: "a".into() }).await.unwrap();
1843
1844 let saved = service
1845 .partial_update(&w.id, patch(&[("name", serde_json::json!("b")), ("secret_hash", serde_json::json!("h0"))]))
1846 .await
1847 .unwrap()
1848 .unwrap();
1849 assert_eq!(saved.name, "b");
1850 }
1851
1852 #[tokio::test]
1853 async fn patch_sets_and_clears_an_optional_field_that_is_not_serialized_when_empty() {
1854 let service = svc();
1855 let w = service.create(CreateWidgetDto { name: "a".into() }).await.unwrap();
1856
1857 let set = service.partial_update(&w.id, patch(&[("note", serde_json::json!("hi"))])).await.unwrap().unwrap();
1858 assert_eq!(set.note.as_deref(), Some("hi"));
1859 let cleared = service.partial_update(&w.id, patch(&[("note", serde_json::Value::Null)])).await.unwrap().unwrap();
1860 assert_eq!(cleared.note, None);
1861 }
1862
1863 #[tokio::test]
1864 async fn put_rejects_changing_a_field_the_entity_protects() {
1865 let service: RekeyService = GenericCrudService::with_repository(Arc::new(InMemoryWidgetRepo::new()));
1866 let w = service.create(CreateWidgetDto { name: "a".into() }).await.unwrap();
1867
1868 let msg = validation_message(
1869 service.update(&w.id, RekeyWidgetDto { name: "b".into(), secret_hash: "h1".into() }).await,
1870 );
1871 assert!(msg.contains("field_not_writable") && msg.contains("secret_hash"), "{msg}");
1872
1873 let saved = service
1875 .update(&w.id, RekeyWidgetDto { name: "b".into(), secret_hash: "h0".into() })
1876 .await
1877 .unwrap()
1878 .unwrap();
1879 assert_eq!(saved.name, "b");
1880 }
1881
1882 #[tokio::test]
1883 async fn bulk_update_rejects_a_protected_change_before_writing_any_row() {
1884 let service: RekeyService = GenericCrudService::with_repository(Arc::new(InMemoryWidgetRepo::new()));
1885 let a = service.create(CreateWidgetDto { name: "a".into() }).await.unwrap();
1886 let b = service.create(CreateWidgetDto { name: "b".into() }).await.unwrap();
1887
1888 let res = service
1889 .bulk_update(vec![
1890 (a.id.clone(), RekeyWidgetDto { name: "a2".into(), secret_hash: "h0".into() }),
1891 (b.id.clone(), RekeyWidgetDto { name: "b2".into(), secret_hash: "h1".into() }),
1892 ])
1893 .await;
1894 assert!(validation_message(res).contains("field_not_writable"));
1895 assert_eq!(service.get_by_id(&a.id).await.unwrap().unwrap().name, "a");
1896 }
1897
1898 #[tokio::test]
1899 async fn bulk_partial_update_rejects_a_protected_change_before_writing_any_row() {
1900 let service = svc();
1901 let a = service.create(CreateWidgetDto { name: "a".into() }).await.unwrap();
1902 let b = service.create(CreateWidgetDto { name: "b".into() }).await.unwrap();
1903
1904 let res = service
1905 .bulk_partial_update(vec![
1906 (a.id.clone(), patch(&[("name", serde_json::json!("a2"))])),
1907 (b.id.clone(), patch(&[("secret_hash", serde_json::json!("h1"))])),
1908 ])
1909 .await;
1910 assert!(validation_message(res).contains("field_not_writable"));
1911 assert_eq!(service.get_by_id(&a.id).await.unwrap().unwrap().name, "a");
1912 }
1913
1914 #[tokio::test]
1915 async fn a_bulk_preview_says_what_each_row_would_do_and_writes_nothing() {
1916 let service = svc();
1917 let a = service.create(CreateWidgetDto { name: "a".into() }).await.unwrap();
1918 let b = service.create(CreateWidgetDto { name: "b".into() }).await.unwrap();
1919
1920 let rows = service
1921 .preview_bulk_partial_update(vec![
1922 (a.id.clone(), patch(&[("name", serde_json::json!("a2"))])),
1923 (b.id.clone(), patch(&[("secret_hash", serde_json::json!("h1"))])),
1924 ("missing".into(), patch(&[("name", serde_json::json!("x"))])),
1925 ])
1926 .await
1927 .unwrap();
1928 assert_eq!(rows.len(), 3);
1929 let (before, after) = rows[0].outcome.as_ref().unwrap();
1930 assert_eq!((before.name.as_str(), after.name.as_str()), ("a", "a2"));
1931 let refused = rows[1].outcome.as_ref().err().unwrap().to_string();
1932 assert!(refused.contains("field_not_writable"), "{refused}");
1933 assert!(rows[2].outcome.is_err(), "a row that is not there is refused, not dropped");
1934 assert_eq!(service.get_by_id(&a.id).await.unwrap().unwrap().name, "a");
1936 }
1937
1938 #[tokio::test]
1939 async fn a_bulk_preview_refuses_a_batch_the_write_would_refuse_whole() {
1940 let service = svc();
1941 let a = service.create(CreateWidgetDto { name: "a".into() }).await.unwrap();
1942 let res = service
1943 .preview_bulk_partial_update(vec![
1944 (a.id.clone(), patch(&[("name", serde_json::json!("a2"))])),
1945 (a.id.clone(), patch(&[("name", serde_json::json!("a3"))])),
1946 ])
1947 .await;
1948 assert!(validation_message(res.map(|r| r.len())).contains("duplicate id"));
1949 }
1950
1951 #[tokio::test]
1952 async fn bulk_partial_update_rejects_an_unknown_field() {
1953 let service = svc();
1954 let a = service.create(CreateWidgetDto { name: "a".into() }).await.unwrap();
1955
1956 let res = service.bulk_partial_update(vec![(a.id.clone(), patch(&[("colour", serde_json::json!("red"))]))]).await;
1957 assert!(validation_message(res).contains("unknown_field"));
1958 }
1959
1960 #[derive(Default)]
1963 struct RecordingPublisher {
1964 kinds: tokio::sync::Mutex<Vec<&'static str>>,
1965 }
1966
1967 #[async_trait]
1968 impl CrudEventPublisher<Widget> for RecordingPublisher {
1969 async fn publish(&self, event: CrudEvent<Widget>) -> Result<(), backbone_messaging::EventError> {
1970 let kind = match event {
1971 CrudEvent::Updated { .. } => "updated",
1972 CrudEvent::Patched { .. } => "patched",
1973 _ => "other",
1974 };
1975 self.kinds.lock().await.push(kind);
1976 Ok(())
1977 }
1978 }
1979
1980 #[tokio::test]
1981 async fn patch_publishes_an_updated_event() {
1982 let publisher = Arc::new(RecordingPublisher::default());
1983 let service = svc().with_event_publisher(publisher.clone());
1984 let w = service.create(CreateWidgetDto { name: "a".into() }).await.unwrap();
1985 publisher.kinds.lock().await.clear();
1986
1987 service.partial_update(&w.id, patch(&[("name", serde_json::json!("b"))])).await.unwrap();
1988 assert_eq!(*publisher.kinds.lock().await, vec!["updated"]);
1989 }
1990
1991 struct RefuseTrash;
1994
1995 #[async_trait]
1996 impl ServiceLifecycle<Widget> for RefuseTrash {
1997 async fn before_restore(&self, _entity: &Widget) -> ServiceResult<()> {
1998 Err(ServiceError::Validation("restore refused".into()))
1999 }
2000 async fn before_hard_delete(&self, _entity: &Widget) -> ServiceResult<()> {
2001 Err(ServiceError::Validation("hard delete refused".into()))
2002 }
2003 async fn before_restore_all(&self) -> ServiceResult<()> {
2004 Err(ServiceError::Validation("restore all refused".into()))
2005 }
2006 async fn before_empty_trash(&self) -> ServiceResult<()> {
2007 Err(ServiceError::Validation("empty trash refused".into()))
2008 }
2009 }
2010
2011 #[tokio::test]
2012 async fn every_trash_operation_consults_the_lifecycle() {
2013 let repo = Arc::new(InMemoryWidgetRepo::new());
2014 let service: GenericCrudService<Widget, CreateWidgetDto, UpdateWidgetDto, InMemoryWidgetRepo> =
2015 GenericCrudService::with_lifecycle(repo, Arc::new(RefuseTrash));
2016 let w = service.create(CreateWidgetDto { name: "a".into() }).await.unwrap();
2017 service.soft_delete(&w.id).await.unwrap();
2018
2019 assert!(service.restore(&w.id).await.is_err(), "restore");
2020 assert!(service.bulk_restore(vec![w.id.clone()]).await.is_err(), "bulk restore");
2021 assert!(service.restore_all().await.is_err(), "restore all");
2022 assert!(service.hard_delete(&w.id).await.is_err(), "hard delete");
2023 assert!(service.bulk_permanent_delete(vec![w.id.clone()]).await.is_err(), "bulk permanent delete");
2024 assert!(service.empty_trash().await.is_err(), "empty trash");
2025
2026 let still = service.get_deleted_by_id(&w.id).await.unwrap().unwrap();
2028 assert!(still.deleted_at.is_some());
2029 }
2030
2031 #[tokio::test]
2034 async fn a_refused_field_is_a_violation_naming_its_path_and_code() {
2035 let service = svc();
2036 let w = service.create(CreateWidgetDto { name: "a".into() }).await.unwrap();
2037
2038 let err = service
2039 .partial_update(&w.id, patch(&[("secret_hash", serde_json::json!("h1")), ("colour", serde_json::json!("red"))]))
2040 .await
2041 .unwrap_err();
2042 let ServiceError::Violations(v) = &err else { panic!("expected violations, got {err:?}") };
2043 let pairs: Vec<(&str, &str)> = v.iter().map(|x| (x.path.as_str(), x.code.as_str())).collect();
2044 assert_eq!(pairs, vec![("colour", "unknown_field")]);
2045 assert!(err.to_string().starts_with("validation failed: unknown_field: `colour`"), "{err}");
2047
2048 let err = service.partial_update(&w.id, patch(&[("secret_hash", serde_json::json!("h1"))])).await.unwrap_err();
2049 let ServiceError::Violations(v) = &err else { panic!("expected violations, got {err:?}") };
2050 assert_eq!((v[0].path.as_str(), v[0].code.as_str()), ("secret_hash", "field_not_writable"));
2051 }
2052
2053 #[derive(Default)]
2056 struct NameGuard {
2057 seen: std::sync::Mutex<Vec<crate::write_guard::WriteKind>>,
2058 }
2059
2060 #[async_trait]
2061 impl crate::write_guard::WriteGuard<Widget> for NameGuard {
2062 async fn check(&self, ctx: &crate::write_guard::WriteCtx<'_, Widget>) -> crate::write_guard::GuardOutcome {
2063 self.seen.lock().unwrap().push(ctx.kind);
2064 let row = ctx.after.or(ctx.before).expect("a write concerns a row");
2065 let mut out = crate::write_guard::GuardOutcome::default();
2066 match row.name.as_str() {
2067 "forbidden" => out.refuse.push(crate::violation::Violation::new("name", "name_forbidden", "this name is refused")),
2068 "watched" => out.shadow.push(crate::violation::Violation::new("name", "name_watched", "would be refused")),
2069 _ => {}
2070 }
2071 out
2072 }
2073 }
2074
2075 fn guarded(guard: Arc<NameGuard>) -> GenericCrudService<Widget, CreateWidgetDto, UpdateWidgetDto, InMemoryWidgetRepo> {
2076 svc().with_guard(guard)
2077 }
2078
2079 #[tokio::test]
2080 async fn a_guard_refusal_stops_create_update_and_patch_with_nothing_written() {
2081 let guard = Arc::new(NameGuard::default());
2082 let service = guarded(guard.clone());
2083
2084 let err = service.create(CreateWidgetDto { name: "forbidden".into() }).await.unwrap_err();
2085 assert!(matches!(&err, ServiceError::Violations(v) if v[0].code == "name_forbidden"), "{err:?}");
2086 assert!(service.list(1, 20, Default::default()).await.unwrap().0.is_empty(), "nothing created");
2087
2088 let w = service.create(CreateWidgetDto { name: "ok".into() }).await.unwrap();
2089 assert!(service.update(&w.id, UpdateWidgetDto { name: "forbidden".into() }).await.is_err());
2090 assert!(service.partial_update(&w.id, patch(&[("name", serde_json::json!("forbidden"))])).await.is_err());
2091 assert!(service
2092 .bulk_partial_update(vec![(w.id.clone(), patch(&[("name", serde_json::json!("forbidden"))]))])
2093 .await
2094 .is_err());
2095 assert_eq!(service.get_by_id(&w.id).await.unwrap().unwrap().name, "ok");
2096 }
2097
2098 #[tokio::test]
2099 async fn a_shadowed_rule_lets_the_write_through() {
2100 let service = guarded(Arc::new(NameGuard::default()));
2101 let w = service.create(CreateWidgetDto { name: "watched".into() }).await.unwrap();
2102 assert_eq!(service.get_by_id(&w.id).await.unwrap().unwrap().name, "watched");
2103 }
2104
2105 #[tokio::test]
2106 async fn the_guard_is_asked_about_every_write_kind() {
2107 use crate::write_guard::WriteKind::*;
2108 let guard = Arc::new(NameGuard::default());
2109 let service = guarded(guard.clone());
2110 let w = service.create(CreateWidgetDto { name: "a".into() }).await.unwrap();
2111 service.update(&w.id, UpdateWidgetDto { name: "b".into() }).await.unwrap();
2112 service.partial_update(&w.id, patch(&[("name", serde_json::json!("c"))])).await.unwrap();
2113 service.soft_delete(&w.id).await.unwrap();
2114 service.restore(&w.id).await.unwrap();
2115 service.soft_delete(&w.id).await.unwrap();
2116 service.hard_delete(&w.id).await.unwrap();
2117 assert_eq!(*guard.seen.lock().unwrap(), vec![Create, Update, Update, Delete, Restore, Delete, HardDelete]);
2118 }
2119
2120 #[test]
2121 fn the_generic_service_hands_its_violations_to_the_http_layer() {
2122 use crate::http::CrudService;
2123 type S = GenericCrudService<Widget, CreateWidgetDto, UpdateWidgetDto, InMemoryWidgetRepo>;
2124 let v = vec![crate::violation::Violation::new("name", "x", "y")];
2125 assert_eq!(<S as CrudService<Widget, CreateWidgetDto, UpdateWidgetDto>>::violations_of(&ServiceError::Violations(v.clone())), Some(v.clone()));
2126 assert_eq!(<S as CrudService<Widget, CreateWidgetDto, UpdateWidgetDto>>::violations_of(&ServiceError::NotFound), None);
2127 let db = ServiceError::Repository(RepositoryError::Violations(v.clone()));
2129 assert_eq!(<S as CrudService<Widget, CreateWidgetDto, UpdateWidgetDto>>::violations_of(&db), Some(v));
2130 }
2131}