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    /// The database this call runs on: the composer's request pool when one
77    /// is bound (a tenant mount, or a relay consumer wrapped by the host),
78    /// else the composed pool (ADR-0029 pool law).
79    fn rpool(&self) -> sqlx::PgPool {
80        crate::request_pool::current().unwrap_or_else(|| self.pool.clone())
81    }
82
83    async fn scoped_tx(
84        &self,
85    ) -> Result<sqlx::Transaction<'_, sqlx::Postgres>, sqlx::Error> {
86        let mut tx = self.rpool().begin().await?;
87        if let Some(scope) = backbone_orm::org_scope::current_org_scope() {
88            backbone_orm::org_scope::bind_org_scope_on(&mut tx, &scope).await?;
89        }
90        Ok(tx)
91    }
92
93    /// Hold `lines` for `ref` at `warehouse` until `expires_at`. Idempotent
94    /// per (ref, item): existing held lines for the ref are replaced. Every
95    /// line must fit the fresh availability triangle minus held
96    /// reservations, or the whole hold refuses (all-or-nothing).
97    pub async fn hold(
98        &self,
99        ref_type: &str,
100        ref_id: Uuid,
101        warehouse_id: Uuid,
102        lines: &[HoldLine],
103        expires_at: DateTime<Utc>,
104    ) -> Result<Vec<HoldAckLine>, ReservationError> {
105        if lines.is_empty() || lines.iter().any(|l| l.qty <= Decimal::ZERO) {
106            return Err(ReservationError::EmptyHold);
107        }
108        // The availability triangle this hold must fit inside: fresh
109        // per call, held subtracted, at the warehouse grain.
110        let availability = self
111            .availability_after_holds(warehouse_id)
112            .await?;
113        let mut tx = self.scoped_tx().await?;
114        // Replace semantics: drop the ref's held lines first (a no-op on
115        // the first hold).
116        Repo::release_for_ref(&mut tx, ref_type, ref_id, Utc::now()).await?;
117        let mut acks = Vec::with_capacity(lines.len());
118        for line in lines {
119            let available = availability
120                .iter()
121                .find(|(item, _)| *item == line.item_id)
122                .map(|(_, qty)| *qty)
123                .unwrap_or(Decimal::ZERO);
124            if line.qty > available {
125                return Err(ReservationError::InsufficientForHold {
126                    item: line.item_id,
127                    asked: line.qty,
128                    available,
129                });
130            }
131            // The insert's own id is generated; read it back for the ack.
132            let id: Uuid = sqlx::query_scalar(
133                r#"INSERT INTO inventory.stock_reservations
134                     (ref_type, ref_id, warehouse_id, item_id, qty, status, expires_at)
135                   VALUES ($1, $2, $3, $4, $5, 'held', $6)
136                   RETURNING id"#,
137            )
138            .bind(ref_type)
139            .bind(ref_id)
140            .bind(warehouse_id)
141            .bind(line.item_id)
142            .bind(line.qty)
143            .bind(expires_at)
144            .fetch_one(&mut *tx)
145            .await?;
146            acks.push(HoldAckLine {
147                item_id: line.item_id,
148                reservation_id: id,
149                remaining_after_hold: available - line.qty,
150            });
151        }
152        tx.commit().await?;
153        Ok(acks)
154    }
155
156    /// Release the ref's held lines (the cart-close hook). Idempotent.
157    pub async fn release(&self, ref_type: &str, ref_id: Uuid) -> Result<usize, ReservationError> {
158        let mut tx = self.scoped_tx().await?;
159        let released = Repo::release_for_ref(&mut tx, ref_type, ref_id, Utc::now()).await?;
160        tx.commit().await?;
161        Ok(released.len())
162    }
163
164    /// Consume the ref's held lines (the order-confirm bridge: the real
165    /// move's reservation takes over). Idempotent.
166    pub async fn consume(&self, ref_type: &str, ref_id: Uuid) -> Result<usize, ReservationError> {
167        let mut tx = self.scoped_tx().await?;
168        let consumed = Repo::consume_for_ref(&mut tx, ref_type, ref_id, Utc::now()).await?;
169        tx.commit().await?;
170        Ok(consumed.len())
171    }
172
173    /// The ref's held lines (inspect).
174    pub async fn held_for_ref(
175        &self,
176        ref_type: &str,
177        ref_id: Uuid,
178    ) -> Result<
179        Vec<crate::infrastructure::persistence::stock_reservation_repository::HeldReservation>,
180        ReservationError,
181    > {
182        let mut conn = self.rpool().acquire().await?;
183        if let Some(scope) = backbone_orm::org_scope::current_org_scope() {
184            // A short-lived read connection with the fence bound, the
185            // module's uniform read posture.
186            let _ = backbone_orm::org_scope::bind_org_scope_on(&mut conn, &scope).await;
187        }
188        Ok(Repo::held_for_ref(&mut conn, ref_type, ref_id).await?)
189    }
190
191    /// The expiry sweep — the abandonment policy. Bounded batches; the
192    /// caller (a host job pass) loops until it answers zero.
193    pub async fn sweep_expired(&self, limit: i64) -> Result<usize, ReservationError> {
194        let mut tx = self.scoped_tx().await?;
195        let expired = Repo::expire_stale(&mut tx, Utc::now(), limit).await?;
196        tx.commit().await?;
197        Ok(expired.len())
198    }
199
200    /// The fresh triangle minus held reservations per item at one warehouse:
201    /// `(quantity - reserved_quantity)` summed over the warehouse's internal
202    /// locations, then the held ledger subtracted at the warehouse grain.
203    /// The availability read's own arithmetic, re-expressed here so the
204    /// hold check and the read can never diverge.
205    async fn availability_after_holds(
206        &self,
207        warehouse_id: Uuid,
208    ) -> Result<Vec<(Uuid, Decimal)>, ReservationError> {
209        let mut conn = self.rpool().acquire().await?;
210        if let Some(scope) = backbone_orm::org_scope::current_org_scope() {
211            let _ = backbone_orm::org_scope::bind_org_scope_on(&mut conn, &scope).await;
212        }
213        let rows: Vec<(Uuid, Decimal)> = sqlx::query_as(
214            r#"WITH triangle AS (
215                   SELECT q.item_id,
216                          COALESCE(SUM(q.quantity - q.reserved_quantity), 0) AS avail
217                     FROM inventory.stock_quants q
218                     JOIN inventory.locations l ON l.id = q.location_id
219                    WHERE l.warehouse_id = $1
220                      AND l.usage::text = 'internal'
221                      AND (q.metadata->>'deleted_at') IS NULL
222                      AND (l.metadata->>'deleted_at') IS NULL
223                    GROUP BY q.item_id
224               ),
225               held AS (
226                   SELECT item_id, COALESCE(SUM(qty), 0) AS held
227                     FROM inventory.stock_reservations
228                    WHERE warehouse_id = $1 AND status = 'held' AND expires_at > now()
229                    GROUP BY item_id
230               )
231               SELECT t.item_id, t.avail - COALESCE(h.held, 0)
232                 FROM triangle t LEFT JOIN held h ON h.item_id = t.item_id"#,
233        )
234        .bind(warehouse_id)
235        .fetch_all(&mut *conn)
236        .await?;
237        Ok(rows)
238    }
239}
240
241impl From<ReservationRepoError> for ReservationError {
242    fn from(e: ReservationRepoError) -> Self {
243        match e {
244            ReservationRepoError::Db(db) => ReservationError::Db(db),
245        }
246    }
247}
backbone_inventory/application/service/inventory_reservation.rs

backbone_inventory/application/service/
inventory_reservation.rs
