backbone_maintenance/application/service/
maintenance_request_write_service.rs

1//! The hand-authored maintenance request write path (user-owned; survives regen).
2//!
3//! Three verbs own every write a request's lifecycle allows:
4//!   - `create_request`  — file a request (stage defaults to the first visible stage);
5//!   - `update_request`  — change descriptive/recurrence fields (schedule, duration, repeat set);
6//!   - `transition_request` — move a request to another stage. This is the ONE door for stage
7//!     changes: it sets the transaction-local `app.maintenance_managed_transition` marker, so the
8//!     G-MT5 trigger lets a preventive recurring request close — and closing is when the
9//!     clone-on-done engine fires: the successor is cloned INSIDE the same transaction, at most
10//!     once (claim-marker CAS, mirroring the visit engine's completion CAS). ZERO schedulers —
11//!     the recurrence is the transition itself; the successor inherits preventive+recurring+the
12//!     repeat set, so it self-perpetuates by construction.
13//!
14//! The DB backstops raw writers: a raw close of a preventive recurring request RAISES
15//! `maintenance_recurring_close_requires_service_verb` (it would skip the spawn), while a raw stage
16//! change of any non-recurring request still lands the correct kanban reset + close_date via the
17//! trigger. Outbox events (`MaintenanceRequestStageChanged`, `SuccessorSpawned`) are the named seam
18//! for the deferred activity-family feedbacks, consumed through the host relay.
19
20use chrono::{DateTime, Duration, NaiveDate, Utc};
21use rust_decimal::{Decimal, prelude::ToPrimitive};
22use sqlx::PgPool;
23use uuid::Uuid;
24
25use crate::domain::entity::{MaintenanceType, RequestKanbanState, RequestPriority, RepeatType, RepeatUnit};
26use crate::infrastructure::persistence::{
27    MaintenanceRequestRepository, NewRequestRow, RequestFieldUpdates, RequestTransitionRow,
28};
29
30use super::maintenance_events::*;
31use super::maintenance_write_service::{legacy_company_echo, relay_ambient_scope};
32
33#[derive(Debug, thiserror::Error)]
34pub enum RequestWriteError {
35    #[error("db: {0}")]
36    Db(#[from] sqlx::Error),
37    #[error("not found: {0}")]
38    NotFound(&'static str),
39    /// G-MT2 — schedule_end would precede schedule_date (checked on the merged create/update view).
40    #[error("schedule_end must not precede schedule_date")]
41    ScheduleEndBeforeStart,
42    /// G-MT3 — a recurrence step below one unit.
43    #[error("repeat_interval must be at least 1")]
44    RepeatIntervalBelowOne,
45    /// G-MT7 — a recurrence bounded 'until' with no repeat_until date.
46    #[error("repeat_type 'until' requires a repeat_until date")]
47    RepeatUntilMissing,
48    /// G-MT8 — only preventive maintenance may recur.
49    #[error("only preventive maintenance may recur")]
50    CorrectiveCannotRecur,
51    #[error("invalid state: {0}")]
52    InvalidState(&'static str),
53    #[error("invalid input: {0}")]
54    Invalid(String),
55}
56
57/// A request as filed. The stage defaults to the first visible stage when `stage_id` is `None`.
58#[derive(Debug, Clone)]
59pub struct NewMaintenanceRequest {
60    pub name: String,
61    pub description: Option<String>,
62    pub schedule_date: Option<DateTime<Utc>>,
63    pub schedule_end: Option<DateTime<Utc>>,
64    pub duration: Decimal,
65    pub owner_user_id: Option<Uuid>,
66    pub user_id: Option<Uuid>,
67    pub asset_id: Option<Uuid>,
68    pub stage_id: Option<Uuid>,
69    pub kanban_state: RequestKanbanState,
70    pub priority: RequestPriority,
71    pub maintenance_type: MaintenanceType,
72    pub recurring: bool,
73    pub repeat_interval: i32,
74    pub repeat_unit: RepeatUnit,
75    pub repeat_type: RepeatType,
76    pub repeat_until: Option<NaiveDate>,
77}
78
79impl Default for NewMaintenanceRequest {
80    fn default() -> Self {
81        Self {
82            name: String::new(),
83            description: None,
84            schedule_date: None,
85            schedule_end: None,
86            duration: Decimal::ZERO,
87            owner_user_id: None,
88            user_id: None,
89            asset_id: None,
90            stage_id: None,
91            kanban_state: RequestKanbanState::default(),
92            priority: RequestPriority::default(),
93            maintenance_type: MaintenanceType::Corrective,
94            recurring: false,
95            repeat_interval: 1,
96            repeat_unit: RepeatUnit::default(),
97            repeat_type: RepeatType::default(),
98            repeat_until: None,
99        }
100    }
101}
102
103/// A partial request update. `None` leaves the field unchanged. The lifecycle columns (stage,
104/// kanban_state, close_date, the successor markers) are not reachable here — they belong to
105/// [`MaintenanceRequestWriteService::transition_request`] and the spawn engine.
106#[derive(Debug, Clone, Default)]
107pub struct MaintenanceRequestUpdate {
108    pub name: Option<String>,
109    pub description: Option<String>,
110    pub schedule_date: Option<DateTime<Utc>>,
111    pub schedule_end: Option<DateTime<Utc>>,
112    pub duration: Option<Decimal>,
113    pub owner_user_id: Option<Uuid>,
114    pub user_id: Option<Uuid>,
115    pub asset_id: Option<Uuid>,
116    pub priority: Option<RequestPriority>,
117    pub maintenance_type: Option<MaintenanceType>,
118    pub recurring: Option<bool>,
119    pub repeat_interval: Option<i32>,
120    pub repeat_unit: Option<RepeatUnit>,
121    pub repeat_type: Option<RepeatType>,
122    pub repeat_until: Option<NaiveDate>,
123}
124
125/// The outcome of a stage transition. `already: true` means the request was at the target stage (or a
126/// concurrent transition won) — nothing changed and nothing spawned.
127#[derive(Debug, Clone, PartialEq)]
128pub struct TransitionOutcome {
129    pub request_id: Uuid,
130    pub from_stage_id: Option<Uuid>,
131    pub to_stage_id: Uuid,
132    /// The request's close_date after the move (`Some(today)` iff the target stage is done).
133    pub close_date: Option<NaiveDate>,
134    pub spawned_successor_id: Option<Uuid>,
135    pub already: bool,
136}
137
138pub struct MaintenanceRequestWriteService {
139    pool: PgPool,
140    requests: MaintenanceRequestRepository,
141}
142
143impl MaintenanceRequestWriteService {
144    pub fn new(pool: PgPool) -> Self {
145        let requests = MaintenanceRequestRepository::new(pool.clone());
146        Self { pool, requests }
147    }
148
149    /// File a request. The stage defaults to the first visible stage (the shared seed's
150    /// "New Request"); entering a done stage stamps `close_date` via the trigger.
151    pub async fn create_request(&self, r: NewMaintenanceRequest) -> Result<Uuid, RequestWriteError> {
152        Self::guard(
153            r.schedule_date,
154            r.schedule_end,
155            r.repeat_interval,
156            r.maintenance_type,
157            r.recurring,
158            r.repeat_type,
159            r.repeat_until,
160        )?;
161        let stage_id = match r.stage_id {
162            Some(s) => {
163                let stage = self.requests.fetch_stage(&self.pool, s).await?;
164                stage.ok_or(RequestWriteError::NotFound("stage"))?.id
165            }
166            None => self.requests.fetch_first_stage_id(&self.pool).await?
167                .ok_or(RequestWriteError::NotFound("stage"))?,
168        };
169        let id = Uuid::new_v4();
170        self.requests.insert_request(&self.pool, &NewRequestRow {
171                id,
172                name: &r.name,
173                description: r.description.as_deref(),
174                schedule_date: r.schedule_date,
175                schedule_end: r.schedule_end,
176                duration: r.duration,
177                owner_user_id: r.owner_user_id,
178                user_id: r.user_id,
179                asset_id: r.asset_id,
180                stage_id,
181                kanban_state: &r.kanban_state.to_string(),
182                priority: &r.priority.to_string(),
183                maintenance_type: &r.maintenance_type.to_string(),
184                recurring: r.recurring,
185                repeat_interval: r.repeat_interval,
186                repeat_unit: &r.repeat_unit.to_string(),
187                repeat_type: &r.repeat_type.to_string(),
188                repeat_until: r.repeat_until,
189                successor_of_request_id: None,
190            })
191            .await?;
192        Ok(id)
193    }
194
195    /// Change a request's descriptive/recurrence fields. The guards run on the MERGED view (old row
196    /// overlaid with the update), so — the widened G-MT2 fire-rule — moving `schedule_date` past an
197    /// existing `schedule_end` is caught here and by the table CHECK.
198    pub async fn update_request(
199        &self,
200        request_id: Uuid,
201        u: MaintenanceRequestUpdate,
202    ) -> Result<(), RequestWriteError> {
203        let old = self
204            .fetch(&self.pool, request_id)
205            .await?
206            .ok_or(RequestWriteError::NotFound("request"))?;
207
208        let schedule_date = u.schedule_date.or(old.schedule_date);
209        let schedule_end = u.schedule_end.or(old.schedule_end);
210        let repeat_interval = u.repeat_interval.unwrap_or(old.repeat_interval);
211        let maintenance_type = u
212            .maintenance_type
213            .map(|t| t.to_string())
214            .unwrap_or_else(|| old.maintenance_type.clone());
215        let recurring = u.recurring.unwrap_or(old.recurring);
216        let repeat_type = u
217            .repeat_type
218            .map(|t| t.to_string())
219            .unwrap_or_else(|| old.repeat_type.clone());
220        let repeat_until = u.repeat_until.or(old.repeat_until);
221
222        Self::guard_schedule(schedule_date, schedule_end)?;
223        Self::guard_repeat_interval(repeat_interval)?;
224        Self::guard_repeat_until(&repeat_type, repeat_until)?;
225        Self::guard_recurring(&maintenance_type, recurring)?;
226
227        if u.name.is_none()
228            && u.description.is_none()
229            && u.schedule_date.is_none()
230            && u.schedule_end.is_none()
231            && u.duration.is_none()
232            && u.owner_user_id.is_none()
233            && u.user_id.is_none()
234            && u.asset_id.is_none()
235            && u.priority.is_none()
236            && u.maintenance_type.is_none()
237            && u.recurring.is_none()
238            && u.repeat_interval.is_none()
239            && u.repeat_unit.is_none()
240            && u.repeat_type.is_none()
241            && u.repeat_until.is_none()
242        {
243            return Ok(()); // nothing requested — not an error
244        }
245
246        let priority = u.priority.as_ref().map(|p| p.to_string());
247        let maintenance_type_str = u.maintenance_type.as_ref().map(|t| t.to_string());
248        let repeat_unit_str = u.repeat_unit.as_ref().map(|x| x.to_string());
249        let repeat_type_str = u.repeat_type.as_ref().map(|x| x.to_string());
250
251        let moved = self.requests.update_request_fields(&self.pool, request_id, &RequestFieldUpdates {
252                name: u.name.as_deref(),
253                description: u.description.as_deref(),
254                schedule_date: u.schedule_date,
255                schedule_end: u.schedule_end,
256                duration: u.duration,
257                owner_user_id: u.owner_user_id,
258                user_id: u.user_id,
259                asset_id: u.asset_id,
260                priority: priority.as_deref(),
261                maintenance_type: maintenance_type_str.as_deref(),
262                recurring: u.recurring,
263                repeat_interval: u.repeat_interval,
264                repeat_unit: repeat_unit_str.as_deref(),
265                repeat_type: repeat_type_str.as_deref(),
266                repeat_until: u.repeat_until,
267            })
268            .await?;
269        if moved != 1 {
270            return Err(RequestWriteError::NotFound("request"));
271        }
272        Ok(())
273    }
274
275    /// Move a request to another stage — the only sanctioned way a request's stage changes.
276    ///
277    /// Inside one transaction: set the managed-transition marker, CAS the stage move (the G-MT5
278    /// trigger applies the kanban reset + close_date on the same statement), and — when a
279    /// non-done request CLOSES onto a done stage and is preventive+recurring — spawn the successor
280    /// (claim the slot first; at most one successor per source). Stages the outbox events and
281    /// commits; the successor exists iff the close committed.
282    pub async fn transition_request(
283        &self,
284        request_id: Uuid,
285        target_stage_id: Uuid,
286        kanban_state_override: Option<RequestKanbanState>,
287        events: &dyn MaintenanceEventSink,
288    ) -> Result<TransitionOutcome, RequestWriteError> {
289        let old = self
290            .fetch(&self.pool, request_id)
291            .await?
292            .ok_or(RequestWriteError::NotFound("request"))?;
293
294        let target = self.requests.fetch_stage(&self.pool, target_stage_id).await?
295            .ok_or(RequestWriteError::NotFound("stage"))?;
296
297        // Already there — idempotent no-op (a repeat or concurrent call changed nothing).
298        if old.stage_id == target.id {
299            return Ok(TransitionOutcome {
300                request_id,
301                from_stage_id: Some(old.stage_id),
302                to_stage_id: target.id,
303                close_date: old.close_date,
304                spawned_successor_id: None,
305                already: true,
306            });
307        }
308
309        let mut tx = self.pool.begin().await?;
310        // The ambient org scope — the transition tx writes requests/stages/outbox behind the
311        // composed decorator's fence (ADR-0029). Undecorated, the tx stays plain.
312        relay_ambient_scope(&mut tx).await?;
313        // The managed-transition marker: transaction-local (set_config ..., true), so it cannot leak
314        // across pool reuse — the same mechanism as app.company_id. This is what authorizes the G-MT5
315        // recurrence arm to let a preventive recurring request close (the spawn follows below).
316        sqlx::query("SELECT set_config('app.maintenance_managed_transition', '1', true)")
317            .execute(&mut *tx)
318            .await?;
319
320        let from_stage_id = old.stage_id;
321        let moved = self
322            .requests
323            .transition_cas(
324                &mut tx,
325                request_id,
326                target.id,
327                kanban_state_override.as_ref().map(|k| k.to_string()).as_deref(),
328            )
329            .await?;
330        if moved != 1 {
331            tx.rollback().await?;
332            // Raced (or a concurrent call landed the same move first) — re-read for the outcome.
333            let winner = self.fetch(&self.pool, request_id).await?.ok_or(RequestWriteError::NotFound("request"))?;
334            return Ok(TransitionOutcome {
335                request_id,
336                from_stage_id: Some(from_stage_id),
337                to_stage_id: winner.stage_id,
338                close_date: winner.close_date,
339                spawned_successor_id: winner.successor_request_id,
340                already: true,
341            });
342        }
343
344        // ── Clone-on-done: only the CLOSE transition spawns (old stage not done, target done).
345        // A done -> done reclassification (e.g. Repaired -> Scrap) does NOT respawn: Odoo's literal
346        // fire-rule is any done target, but at-most-once-per-close is the deliberate narrowing this
347        // engine commits to — the claim marker makes a second spawn impossible anyway.
348        let mut spawned_successor_id: Option<Uuid> = None;
349        let mut spawn_window: Option<(DateTime<Utc>, DateTime<Utc>)> = None;
350        if !old.stage_done && target.done && old.maintenance_type == "preventive" && old.recurring {
351            let base = old.schedule_date.unwrap_or_else(Utc::now);
352            let next = advance(base, old.repeat_interval, &old.repeat_unit)
353                .ok_or_else(|| RequestWriteError::Invalid("recurrence step overflows the calendar".into()))?;
354            // Termination: forever, or the next date still within repeat_until (inclusive).
355            let continues = old.repeat_type != "until"
356                || old.repeat_until.map(|until| next.date_naive() <= until).unwrap_or(false);
357            if continues {
358                let hours = if old.duration.is_zero() { Decimal::ONE } else { old.duration };
359                let minutes = (hours * Decimal::from(60))
360                    .to_i64()
361                    .ok_or_else(|| RequestWriteError::Invalid("duration out of range".into()))?;
362                let next_end = next + Duration::minutes(minutes);
363
364                let successor_id = Uuid::new_v4();
365                // Claim the slot FIRST: 0 rows means a spawn already happened (repeat/race) — skip.
366                // Deterministic ordering: claim precedes the INSERT, both precede the commit, all
367                // inside the same tx as the stage CAS — the successor exists iff the close committed.
368                let claimed = self
369                    .requests
370                    .claim_successor_slot(&mut tx, request_id, successor_id)
371                    .await?;
372                if claimed == 1 {
373                    let first_stage = self
374                        .requests
375                        .fetch_first_stage_id_on(&mut tx)
376                        .await?
377                        .ok_or(RequestWriteError::Invalid("no visible stage to spawn the successor into".into()))?;
378                    self.requests
379                        .insert_request_on(&mut tx, &NewRequestRow {
380                            id: successor_id,
381                            name: &old.name,
382                            description: old.description.as_deref(),
383                            schedule_date: Some(next),
384                            schedule_end: Some(next_end),
385                            duration: old.duration,
386                            owner_user_id: old.owner_user_id,
387                            user_id: old.user_id,
388                            asset_id: old.asset_id,
389                            stage_id: first_stage,
390                            kanban_state: "normal",
391                            priority: &old.priority,
392                            maintenance_type: &old.maintenance_type,
393                            recurring: old.recurring,
394                            repeat_interval: old.repeat_interval,
395                            repeat_unit: &old.repeat_unit,
396                            repeat_type: &old.repeat_type,
397                            repeat_until: old.repeat_until,
398                            successor_of_request_id: Some(request_id),
399                        })
400                        .await?;
401                    spawned_successor_id = Some(successor_id);
402                    spawn_window = Some((next, next_end));
403                }
404            }
405        }
406
407        // ── Outbox: the stage change (and the spawn) land with the same commit. The payloads and
408        // the outbox fence carry the ambient scope's legacy company echo (ADR-0029 twins).
409        let legacy_company = legacy_company_echo();
410        let close_date = if target.done { Some(Utc::now().date_naive()) } else { None };
411        let stage_event = MaintenanceEvent::MaintenanceRequestStageChanged(MaintenanceRequestStageChanged {
412            request_id,
413            company_id: legacy_company,
414            from_stage_id: Some(from_stage_id),
415            to_stage_id: target.id,
416            close_date,
417            spawned_successor_id,
418        });
419        let stage_record = backbone_outbox::OutboxRecord::new(
420            "MaintenanceRequestStageChanged", "MaintenanceRequest", request_id.to_string(), legacy_company,
421            serde_json::to_value(&stage_event).map_err(|e| RequestWriteError::Invalid(e.to_string()))?,
422            Utc::now(),
423        );
424        backbone_outbox::outbox::stage(&mut *tx, "maintenance", &stage_record)
425            .await
426            .map_err(|e| RequestWriteError::Invalid(format!("outbox stage: {e}")))?;
427
428        let mut spawn_event_value = None;
429        if let Some(successor_id) = spawned_successor_id {
430            let (next, next_end) = spawn_window.expect("spawn window set alongside the successor id");
431            let spawn_event = MaintenanceEvent::SuccessorSpawned(SuccessorSpawned {
432                source_request_id: request_id,
433                successor_request_id: successor_id,
434                company_id: legacy_company,
435                next_schedule_date: Some(next),
436                next_schedule_end: Some(next_end),
437            });
438            let spawn_record = backbone_outbox::OutboxRecord::new(
439                "SuccessorSpawned", "MaintenanceRequest", successor_id.to_string(), legacy_company,
440                serde_json::to_value(&spawn_event).map_err(|e| RequestWriteError::Invalid(e.to_string()))?,
441                Utc::now(),
442            );
443            backbone_outbox::outbox::stage(&mut *tx, "maintenance", &spawn_record)
444                .await
445                .map_err(|e| RequestWriteError::Invalid(format!("outbox stage: {e}")))?;
446            spawn_event_value = Some(spawn_event);
447        }
448
449        tx.commit().await?;
450        events.publish(&stage_event);
451        if let Some(ev) = spawn_event_value {
452            events.publish(&ev);
453        }
454
455        Ok(TransitionOutcome {
456            request_id,
457            from_stage_id: Some(from_stage_id),
458            to_stage_id: target.id,
459            close_date,
460            spawned_successor_id,
461            already: false,
462        })
463    }
464
465    /// Fetch a request's transition view. Caller-scoped (see the repository).
466    async fn fetch(
467        &self,
468        pool: &PgPool,
469        request_id: Uuid,
470    ) -> Result<Option<RequestTransitionRow>, RequestWriteError> {
471        Ok(self.requests.fetch_for_transition(pool, request_id).await?)
472    }
473
474    fn guard_schedule(
475        schedule_date: Option<DateTime<Utc>>,
476        schedule_end: Option<DateTime<Utc>>,
477    ) -> Result<(), RequestWriteError> {
478        if let (Some(start), Some(end)) = (schedule_date, schedule_end) {
479            if end < start {
480                return Err(RequestWriteError::ScheduleEndBeforeStart);
481            }
482        }
483        Ok(())
484    }
485
486    fn guard_repeat_interval(repeat_interval: i32) -> Result<(), RequestWriteError> {
487        if repeat_interval < 1 {
488            return Err(RequestWriteError::RepeatIntervalBelowOne);
489        }
490        Ok(())
491    }
492
493    fn guard_repeat_until(repeat_type: &str, repeat_until: Option<NaiveDate>) -> Result<(), RequestWriteError> {
494        if repeat_type == "until" && repeat_until.is_none() {
495            return Err(RequestWriteError::RepeatUntilMissing);
496        }
497        Ok(())
498    }
499
500    fn guard_recurring(maintenance_type: &str, recurring: bool) -> Result<(), RequestWriteError> {
501        if maintenance_type == "corrective" && recurring {
502            return Err(RequestWriteError::CorrectiveCannotRecur);
503        }
504        Ok(())
505    }
506
507    fn guard(
508        schedule_date: Option<DateTime<Utc>>,
509        schedule_end: Option<DateTime<Utc>>,
510        repeat_interval: i32,
511        maintenance_type: MaintenanceType,
512        recurring: bool,
513        repeat_type: RepeatType,
514        repeat_until: Option<NaiveDate>,
515    ) -> Result<(), RequestWriteError> {
516        Self::guard_schedule(schedule_date, schedule_end)?;
517        Self::guard_repeat_interval(repeat_interval)?;
518        Self::guard_repeat_until(&repeat_type.to_string(), repeat_until)?;
519        Self::guard_recurring(&maintenance_type.to_string(), recurring)?;
520        Ok(())
521    }
522}
523
524/// Advance `base` by one repeat step (interval x unit). Months and years are calendar-exact
525/// (chrono), matching the recurrence intent rather than a fixed 30-day month.
526fn advance(base: DateTime<Utc>, interval: i32, unit: &str) -> Option<DateTime<Utc>> {
527    match unit {
528        "day" => base.checked_add_signed(Duration::days(interval as i64)),
529        "week" => base.checked_add_signed(Duration::weeks(interval as i64)),
530        "month" => base
531            .checked_add_months(chrono::Months::new(u32::try_from(interval).ok()?)),
532        // A year stepped as twelve calendar-exact months (chrono has no Years step).
533        "year" => u32::try_from(interval)
534            .ok()
535            .and_then(|i| i.checked_mul(12))
536            .and_then(|m| base.checked_add_months(chrono::Months::new(m))),
537        _ => None,
538    }
539}
backbone_maintenance/application/service/maintenance_request_write_service.rs

backbone_maintenance/application/service/
maintenance_request_write_service.rs
