backbone_inventory/application/service/
inventory_reservation.rs

1//! The store-level reservation surface (hand-written; user-owned) — the
2//! checkout-intent ledger the register row asked for: intent + release
3//! verbs on the stock authority, checkout consumes them.
4//!
5//! Posture (the design recorded on the issue):
6//!
7//! - Holds NEVER write the quant counters. `stock_quants.reserved_quantity`
8//!   mirrors open stock-move lines and is recomputed from them; a hold
9//!   written there would be wiped by the next mirror pass. Holds live in
10//!   `inventory.stock_reservations` and the AVAILABILITY READ subtracts
11//!   them at the warehouse grain.
12//! - Hold is idempotent per (ref, item): a re-hold REPLACES the line
13//!   (release-then-insert in one transaction) so the cart's freeze step can
14//!   run freely.
15//! - Hold refuses typed when the fresh availability triangle minus held
16//!   reservations cannot cover the ask — the advisory check-then-act race
17//!   window stays (two carts can still both pass the read), but a passed
18//!   hold is bounded by the ledger's own arithmetic.
19//! - Consume is the bridge at order confirm: the hold's quantity is
20//!   superseded by the real reservation the delivery door's move lines
21//!   create (the mirror owns reserved_quantity from there).
22//! - Expiry is the abandonment policy: holds past their TTL expire in
23//!   bounded SKIP LOCKED batches (the same idiom the promo claim sweep
24//!   uses).
25
26use chrono::{DateTime, Utc};
27use rust_decimal::Decimal;
28use uuid::Uuid;
29
30use crate::infrastructure::persistence::stock_reservation_repository::{
31    ReservationRepoError, StockReservationRepository as Repo,
32};
33
34#[derive(Debug, thiserror::Error)]
35pub enum ReservationError {
36    #[error("db: {0}")]
37    Db(#[from] sqlx::Error),
38    #[error("cannot hold {asked} of item {item}: availability after existing holds is {available}")]
39    InsufficientForHold {
40        item: Uuid,
41        asked: Decimal,
42        available: Decimal,
43    },
44    #[error("hold lines carry no quantities")]
45    EmptyHold,
46}
47
48/// One line of a hold request.
49#[derive(Debug, Clone)]
50pub struct HoldLine {
51    pub item_id: Uuid,
52    pub qty: Decimal,
53}
54
55/// The hold outcome, per line: the reservation id and the availability
56/// that remained after the hold landed.
57#[derive(Debug, Clone)]
58pub struct HoldAckLine {
59    pub item_id: Uuid,
60    pub reservation_id: Uuid,
61    pub remaining_after_hold: Decimal,
62}
63
64/// The verbs. Stateless over a pool; each verb runs in one transaction with
65/// the ambient org scope relayed (the module's uniform posture).
66#[derive(Clone)]
67pub struct ReservationService {
68    pool: sqlx::PgPool,
69}
70
71impl ReservationService {
72    pub fn new(pool: sqlx::PgPool) -> Self {
73        Self { pool }
74    }
75
76    async fn scoped_tx(
77        &self,
78    ) -> Result<sqlx::Transaction<'_, sqlx::Postgres>, sqlx::Error> {
79        let mut tx = self.pool.begin().await?;
80        if let Some(scope) = backbone_orm::org_scope::current_org_scope() {
81            backbone_orm::org_scope::bind_org_scope_on(&mut tx, &scope).await?;
82        }
83        Ok(tx)
84    }
85
86    /// Hold `lines` for `ref` at `warehouse` until `expires_at`. Idempotent
87    /// per (ref, item): existing held lines for the ref are replaced. Every
88    /// line must fit the fresh availability triangle minus held
89    /// reservations, or the whole hold refuses (all-or-nothing).
90    pub async fn hold(
91        &self,
92        ref_type: &str,
93        ref_id: Uuid,
94        warehouse_id: Uuid,
95        lines: &[HoldLine],
96        expires_at: DateTime<Utc>,
97    ) -> Result<Vec<HoldAckLine>, ReservationError> {
98        if lines.is_empty() || lines.iter().any(|l| l.qty <= Decimal::ZERO) {
99            return Err(ReservationError::EmptyHold);
100        }
101        // The availability triangle this hold must fit inside: fresh
102        // per call, held subtracted, at the warehouse grain.
103        let availability = self
104            .availability_after_holds(warehouse_id)
105            .await?;
106        let mut tx = self.scoped_tx().await?;
107        // Replace semantics: drop the ref's held lines first (a no-op on
108        // the first hold).
109        Repo::release_for_ref(&mut tx, ref_type, ref_id, Utc::now()).await?;
110        let mut acks = Vec::with_capacity(lines.len());
111        for line in lines {
112            let available = availability
113                .iter()
114                .find(|(item, _)| *item == line.item_id)
115                .map(|(_, qty)| *qty)
116                .unwrap_or(Decimal::ZERO);
117            if line.qty > available {
118                return Err(ReservationError::InsufficientForHold {
119                    item: line.item_id,
120                    asked: line.qty,
121                    available,
122                });
123            }
124            // The insert's own id is generated; read it back for the ack.
125            let id: Uuid = sqlx::query_scalar(
126                r#"INSERT INTO inventory.stock_reservations
127                     (ref_type, ref_id, warehouse_id, item_id, qty, status, expires_at)
128                   VALUES ($1, $2, $3, $4, $5, 'held', $6)
129                   RETURNING id"#,
130            )
131            .bind(ref_type)
132            .bind(ref_id)
133            .bind(warehouse_id)
134            .bind(line.item_id)
135            .bind(line.qty)
136            .bind(expires_at)
137            .fetch_one(&mut *tx)
138            .await?;
139            acks.push(HoldAckLine {
140                item_id: line.item_id,
141                reservation_id: id,
142                remaining_after_hold: available - line.qty,
143            });
144        }
145        tx.commit().await?;
146        Ok(acks)
147    }
148
149    /// Release the ref's held lines (the cart-close hook). Idempotent.
150    pub async fn release(&self, ref_type: &str, ref_id: Uuid) -> Result<usize, ReservationError> {
151        let mut tx = self.scoped_tx().await?;
152        let released = Repo::release_for_ref(&mut tx, ref_type, ref_id, Utc::now()).await?;
153        tx.commit().await?;
154        Ok(released.len())
155    }
156
157    /// Consume the ref's held lines (the order-confirm bridge: the real
158    /// move's reservation takes over). Idempotent.
159    pub async fn consume(&self, ref_type: &str, ref_id: Uuid) -> Result<usize, ReservationError> {
160        let mut tx = self.scoped_tx().await?;
161        let consumed = Repo::consume_for_ref(&mut tx, ref_type, ref_id, Utc::now()).await?;
162        tx.commit().await?;
163        Ok(consumed.len())
164    }
165
166    /// The ref's held lines (inspect).
167    pub async fn held_for_ref(
168        &self,
169        ref_type: &str,
170        ref_id: Uuid,
171    ) -> Result<
172        Vec<crate::infrastructure::persistence::stock_reservation_repository::HeldReservation>,
173        ReservationError,
174    > {
175        let mut conn = self.pool.acquire().await?;
176        if let Some(scope) = backbone_orm::org_scope::current_org_scope() {
177            // A short-lived read connection with the fence bound, the
178            // module's uniform read posture.
179            let _ = backbone_orm::org_scope::bind_org_scope_on(&mut conn, &scope).await;
180        }
181        Ok(Repo::held_for_ref(&mut conn, ref_type, ref_id).await?)
182    }
183
184    /// The expiry sweep — the abandonment policy. Bounded batches; the
185    /// caller (a host job pass) loops until it answers zero.
186    pub async fn sweep_expired(&self, limit: i64) -> Result<usize, ReservationError> {
187        let mut tx = self.scoped_tx().await?;
188        let expired = Repo::expire_stale(&mut tx, Utc::now(), limit).await?;
189        tx.commit().await?;
190        Ok(expired.len())
191    }
192
193    /// The fresh triangle minus held reservations per item at one warehouse:
194    /// `(quantity - reserved_quantity)` summed over the warehouse's internal
195    /// locations, then the held ledger subtracted at the warehouse grain.
196    /// The availability read's own arithmetic, re-expressed here so the
197    /// hold check and the read can never diverge.
198    async fn availability_after_holds(
199        &self,
200        warehouse_id: Uuid,
201    ) -> Result<Vec<(Uuid, Decimal)>, ReservationError> {
202        let mut conn = self.pool.acquire().await?;
203        if let Some(scope) = backbone_orm::org_scope::current_org_scope() {
204            let _ = backbone_orm::org_scope::bind_org_scope_on(&mut conn, &scope).await;
205        }
206        let rows: Vec<(Uuid, Decimal)> = sqlx::query_as(
207            r#"WITH triangle AS (
208                   SELECT q.item_id,
209                          COALESCE(SUM(q.quantity - q.reserved_quantity), 0) AS avail
210                     FROM inventory.stock_quants q
211                     JOIN inventory.locations l ON l.id = q.location_id
212                    WHERE l.warehouse_id = $1
213                      AND l.usage::text = 'internal'
214                      AND (q.metadata->>'deleted_at') IS NULL
215                      AND (l.metadata->>'deleted_at') IS NULL
216                    GROUP BY q.item_id
217               ),
218               held AS (
219                   SELECT item_id, COALESCE(SUM(qty), 0) AS held
220                     FROM inventory.stock_reservations
221                    WHERE warehouse_id = $1 AND status = 'held' AND expires_at > now()
222                    GROUP BY item_id
223               )
224               SELECT t.item_id, t.avail - COALESCE(h.held, 0)
225                 FROM triangle t LEFT JOIN held h ON h.item_id = t.item_id"#,
226        )
227        .bind(warehouse_id)
228        .fetch_all(&mut *conn)
229        .await?;
230        Ok(rows)
231    }
232}
233
234impl From<ReservationRepoError> for ReservationError {
235    fn from(e: ReservationRepoError) -> Self {
236        match e {
237            ReservationRepoError::Db(db) => ReservationError::Db(db),
238        }
239    }
240}
backbone_inventory/application/service/inventory_reservation.rs

backbone_inventory/application/service/
inventory_reservation.rs
