Skip to main content

backbone_core/
service.rs

1//! Generic CRUD service — one implementation used by all entities.
2//!
3//! Generated code emits only a type alias:
4//!
5//! ```rust,ignore
6//! // Generated (was ~250 lines, now 1 line):
7//! pub type StoredFileService = GenericCrudService<
8//!     StoredFile,
9//!     CreateStoredFileDto,
10//!     UpdateStoredFileDto,
11//!     PostgresStoredFileRepository,
12//! >;
13//! ```
14//!
15//! Custom services wrap the alias with a decorator:
16//!
17//! ```rust,ignore
18//! pub struct StoredFileServiceCustom {
19//!     inner: Arc<StoredFileService>,
20//!     notifications: Arc<NotificationService>,
21//! }
22//!
23//! impl StoredFileServiceCustom {
24//!     pub async fn archive(&self, id: &str, reason: String) -> ServiceResult<StoredFile> {
25//!         let entity = self.inner.get_by_id(id).await?.ok_or(ServiceError::NotFound)?;
26//!         // custom business logic
27//!     }
28//! }
29//! ```
30
31use 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// ─── Service error ───────────────────────────────────────────────────────────
41
42/// Standard service-layer error type.
43#[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    /// The write broke named rules: one violation per rule, each with the
55    /// field it concerns and a stable code. Displays as the same
56    /// `validation failed: …` sentence a plain validation error does.
57    #[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
69// ─── DTO conversion traits ────────────────────────────────────────────────────
70
71/// Convert a Create DTO into a new entity instance.
72///
73/// Implemented once per entity by generated code, may be overridden in custom
74/// decorators via `UseCaseHooks`.
75pub trait FromCreateDto<DTO>: Sized {
76    fn from_create_dto(dto: DTO) -> ServiceResult<Self>;
77}
78
79/// Apply an Update DTO's fields to an existing entity (full update).
80pub trait ApplyUpdateDto<DTO> {
81    fn apply_update(self, dto: DTO) -> ServiceResult<Self>
82    where
83        Self: Sized;
84}
85
86// ─── ServiceLifecycle hooks ──────────────────────────────────────────────────
87
88/// Optional lifecycle callbacks injected into `GenericCrudService`.
89///
90/// Override individual methods — all default to no-op.
91#[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    /// Called before a soft-deleted entity is restored, singly or in a batch.
124    async fn before_restore(&self, entity: &E) -> ServiceResult<()> {
125        let _ = entity;
126        Ok(())
127    }
128
129    /// Called before a soft-deleted entity is removed for good, singly or in
130    /// a batch.
131    async fn before_hard_delete(&self, entity: &E) -> ServiceResult<()> {
132        let _ = entity;
133        Ok(())
134    }
135
136    /// Called once before the whole trash is restored. The rows are not
137    /// loaded for this; refuse here to keep the bulk path closed.
138    async fn before_restore_all(&self) -> ServiceResult<()> {
139        Ok(())
140    }
141
142    /// Called once before the whole trash is removed for good. The rows are
143    /// not loaded for this; refuse here to keep the bulk path closed.
144    async fn before_empty_trash(&self) -> ServiceResult<()> {
145        Ok(())
146    }
147}
148
149/// No-op lifecycle — the default used by generated type aliases.
150pub 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
171// ─── GenericCrudService ──────────────────────────────────────────────────────
172
173/// A fully generic CRUD service.
174///
175/// Type parameters:
176/// - `E`  — entity type (must implement `PersistentEntity`)
177/// - `C`  — create DTO (entity implements `FromCreateDto<C>`)
178/// - `U`  — update DTO (entity implements `ApplyUpdateDto<U>`)
179/// - `R`  — repository (implements `CrudRepository<E>`)
180///
181/// The service always holds an `Arc<dyn CrudEventPublisher<E>>`.  When no
182/// publisher is needed, `NoOpCrudEventPublisher` is used (zero overhead).
183/// Event-publish errors are fire-and-forget — they never fail the operation.
184pub 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    /// Full constructor — supply all three dependencies.
204    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    /// Check every write against `guard`: its refusals refuse the write, its
219    /// shadowed violations are logged and the write proceeds.
220    pub fn with_guard(mut self, guard: Arc<dyn crate::write_guard::WriteGuard<E>>) -> Self {
221        self.guard = guard;
222        self
223    }
224
225    /// Ask the guard about one write.
226    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    /// [`consult`](Self::consult) for a dry run: the same refusals, without
236    /// logging what a shadow rule would refuse — the real write logs that, and a
237    /// preview before it would count every shadow violation twice.
238    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    /// Create with a custom lifecycle and the no-op event publisher.
275    pub fn with_lifecycle(repository: Arc<R>, lifecycle: Arc<dyn ServiceLifecycle<E>>) -> Self {
276        Self::new(repository, lifecycle, NoOpCrudEventPublisher::arc())
277    }
278
279    /// Create with the no-op lifecycle and no-op event publisher (generated default).
280    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    /// Replace the event publisher on an existing service (builder pattern).
289    pub fn with_event_publisher(mut self, publisher: Arc<dyn CrudEventPublisher<E>>) -> Self {
290        self.event_publisher = publisher;
291        self
292    }
293
294    /// Access the underlying repository for custom query extensions.
295    pub fn repository(&self) -> &R {
296        &self.repository
297    }
298
299    // ─── Individual operations ────────────────────────────────────────────
300
301    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    /// `list`, carrying the pagination info (cursor positions included) so
314    /// the HTTP layer can surface keyset paging on every generic route.
315    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    /// Group and reduce the rows `list` would return, under the same filters.
328    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        // The same changed-fields, live-row-only save as `partial_update`.
380        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    /// Alias for `get_by_id` — used by generated handlers.
511    pub async fn find_by_id(&self, id: &str) -> ServiceResult<Option<E>> {
512        self.get_by_id(id).await
513    }
514
515    /// Alias for `hard_delete` — used by generated handlers.
516    pub async fn permanent_delete(&self, id: &str) -> ServiceResult<bool> {
517        self.hard_delete(id).await
518    }
519
520    /// Count soft-deleted entities.
521    pub async fn count_deleted(&self) -> ServiceResult<u64> {
522        self.repository
523            .count_deleted()
524            .await
525            .map_err(ServiceError::Repository)
526    }
527
528    /// Retrieve a soft-deleted entity by ID (includes deleted records).
529    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    /// List deleted entities with filter support (filters currently ignored, delegates to list_deleted).
537    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    /// Upsert: create the entity if it doesn't exist (no ID-based lookup — always creates).
547    pub async fn upsert(&self, dto: C) -> ServiceResult<E> {
548        self.create(dto).await
549    }
550
551    /// Apply a partial update from a JSON map of field values.
552    ///
553    /// Fields not present in the map are left unchanged. A key the entity does
554    /// not have, or a new value for a protected field, refuses the whole write.
555    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        // Only the fields this write changed are saved, and only while the row is still live:
575        // see `CrudRepository::update_changed`. A row deleted since the read answers not found.
576        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    // ── Atomic batch operations ───────────────────────────────────────────────
608    //
609    // All-or-nothing: ids are validated up-front (so a missing id fails the whole
610    // request before any write), then the repository applies the change inside a
611    // single transaction. Lifecycle hooks and CRUD events fire per affected
612    // entity, mirroring the single-row methods. Events are best-effort.
613
614    /// Soft-delete many entities by id (atomic).
615    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        // Load the entities up-front: they are the payloads for the lifecycle
619        // `before_delete` hook and the `SoftDeleted` event below. The repository's
620        // atomic affected-row check (not this loop) is what guarantees the write
621        // is all-or-nothing.
622        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    /// Restore many soft-deleted entities by id (atomic).
655    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        // Ask the lifecycle about every row it can see; an id that is not in
659        // the trash is left to the repository's atomic check below.
660        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    /// Restore every soft-deleted entity (atomic). Returns the number restored.
682    /// Emits a `Restored` event per entity, mirroring [`bulk_restore`](Self::bulk_restore).
683    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    /// Permanently delete many soft-deleted entities by id (atomic).
701    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        // The lifecycle is asked about every row it can see. Whether every id
705        // is in the trash stays with `bulk_hard_delete`, which deletes only
706        // trashed rows and rolls the whole batch back unless every id matched,
707        // so a row that moves between this loop and the write cannot slip
708        // through.
709        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    /// Full-update many entities (atomic). Each item is `(id, UpdateDto)`.
734    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    /// Fire `after_update` and a `Updated` event for each `(before, after)` pair.
769    /// Shared by [`bulk_update`](Self::bulk_update) and
770    /// [`bulk_partial_update`](Self::bulk_partial_update); events are best-effort.
771    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    /// Partial-update many entities (atomic). Each item is `(id, field_map)`.
788    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    /// What [`bulk_partial_update`](Self::bulk_partial_update) would do, with
828    /// nothing written. Each row goes through the same steps the write takes —
829    /// it is found, the patch is merged and checked against the protected fields,
830    /// the write rules are consulted (quietly: a shadow rule logs on the real
831    /// write, not here) and the update hook runs — and its outcome is
832    /// kept instead of failing the batch, so one dialog can say which rows would
833    /// change and which would be refused. Only a batch that is too large or names
834    /// an id twice fails as a whole, as the write does.
835    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
880// ─── Generic write protection ────────────────────────────────────────────────
881
882/// Serialized fields no generic write may change on any entity: the row's
883/// identity, and the audit block that records who created, changed and
884/// deleted it. Entities add their own through
885/// [`PersistentEntity::write_protected_fields`].
886pub 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
895/// Refuse a generic write that gives a protected field a new value. Carrying
896/// the stored value unchanged is allowed, so a form that sends the whole
897/// record is not refused for fields it did not touch.
898///
899/// A state machine's verb writes through the same path, inside
900/// [`as_transition`](crate::write_guard::as_transition), and its whole point is
901/// to move the state field the entity protects from generic writes. Inside a
902/// transition the entity's [`PersistentEntity::state_fields`] are therefore let
903/// through — those and nothing else; the write guard still decides whether the
904/// move itself is allowed.
905fn 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
933/// Merge a partial update onto an entity.
934///
935/// The merge goes through the entity's JSON form, so a key the entity has no
936/// field for would otherwise be dropped without a word and the edit lost. It
937/// is refused instead: a key given a value that does not survive the round
938/// trip is not a field of this record. A key set to `null` is exempt, since
939/// an empty optional field may be left out of the JSON form entirely.
940fn 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
974/// Maximum number of ids / items a single batch operation may contain. Enforced
975/// here at the service layer (not only in the HTTP handlers) so every caller —
976/// gRPC, background jobs, internal code — gets the same bound.
977pub const MAX_BATCH_SIZE: usize = 1000;
978
979/// Reject an over-sized batch before any work is done.
980fn 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
989/// Drop duplicate ids while preserving first-seen order. Id-list batch ops issue
990/// a SQL `IN (...)` whose distinct-row count would never match a list that
991/// repeats an id, so a stray duplicate would otherwise fail the whole batch.
992fn 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
997/// Return the first id that appears more than once, if any. Used to reject
998/// ambiguous bulk updates (the same id mapped to two different payloads).
999fn 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// ─── Blanket CrudService impl ────────────────────────────────────────────────
1010
1011/// Blanket impl so that any `GenericCrudService<E, C, U, R>` automatically
1012/// satisfies `CrudService<E, C, U>` without generated adapter structs.
1013///
1014/// Generated code emits only a type alias:
1015///   `pub type UserService = GenericCrudService<User, CreateUserDto, UpdateUserDto, UserRepository>;`
1016///
1017/// That alias now directly implements `CrudService` and can be passed to
1018/// `BackboneCrudHandler` without any additional boilerplate.
1019#[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
1186// Make the service cloneable (needed when Arc-ing it and sharing across handlers).
1187impl<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    // ── Minimal in-memory repository for testing ──────────────────────────
1210
1211    #[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        /// A state-machine field: protected from generic writes, moved by verbs.
1219        #[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    // Minimal in-memory CrudRepository
1292    struct InMemoryWidgetRepo {
1293        data: tokio::sync::Mutex<Vec<Widget>>,
1294        /// Every `update_changed` call, as (before, after).
1295        changed_writes: std::sync::Mutex<Vec<(Widget, Widget)>>,
1296        /// Every full `update` call.
1297        full_writes: std::sync::atomic::AtomicUsize,
1298        /// Make the next changed-columns write find the row gone (a concurrent delete).
1299        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    // ── Generic writes go through the changed-columns path ───────────────────
1490
1491    #[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    // ── A transition may move the state field, and only that ─────────────────
1579
1580    #[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    // ── Batch operations ──────────────────────────────────────────────────────
1640
1641    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        // One id is missing → whole batch must be rejected with nothing deleted.
1665        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        // Restore two explicitly.
1684        let restored = service.bulk_restore(vec![ids[0].clone(), ids[1].clone()]).await.unwrap();
1685        assert_eq!(restored.len(), 2);
1686
1687        // Restore the remaining one via restore_all.
1688        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        // The valid row must be untouched because the batch aborted pre-write.
1709        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        // A repeated id must not fail the batch: ids are de-duplicated first.
1740        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        // The same id mapped to two payloads is ambiguous → rejected, nothing written.
1756        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        // Enforced at the service layer, independent of any HTTP handler.
1771        assert!(service.bulk_soft_delete(ids).await.is_err());
1772    }
1773
1774    // ── Generic writes may not change protected or unknown fields ─────────
1775
1776    /// A full update that carries the protected field, the way a form that
1777    /// replaces the whole record would.
1778    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    /// The refusal's message, whether the service refused with named
1798    /// violations or with a plain validation sentence.
1799    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        // The same form echoing the stored value is an ordinary edit.
1874        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        // One refused row does not refuse the others, and nothing was written.
1935        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    // ── Events and trash lifecycle ────────────────────────────────────────
1961
1962    #[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    /// A lifecycle that refuses every trash operation, so a test can tell
1992    /// whether the operation consulted it.
1993    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        // Nothing moved: the row is still in the trash.
2027        let still = service.get_deleted_by_id(&w.id).await.unwrap().unwrap();
2028        assert!(still.deleted_at.is_some());
2029    }
2030
2031    // ── Named violations and the write guard ──────────────────────────────
2032
2033    #[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        // The sentence clients already read is unchanged.
2046        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    /// Refuses any write whose resulting widget is named `forbidden`, and
2054    /// shadow-reports any named `watched`. Records every kind it was asked about.
2055    #[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        // A constraint the database refused with reaches the HTTP layer as the same 422.
2128        let db = ServiceError::Repository(RepositoryError::Violations(v.clone()));
2129        assert_eq!(<S as CrudService<Widget, CreateWidgetDto, UpdateWidgetDto>>::violations_of(&db), Some(v));
2130    }
2131}