backbone_inventory/infrastructure/persistence/
stock_adjustment_repository.rs

1//! Quant-driven adjustment SQL (hand-authored, user-owned).
2//!
3//! The physical half of the ONE adjustment door (spec stock-business-logic.md §5.2):
4//! there is no `stock.inventory` model — a counted quantity STAGES on the quant row
5//! (`inventory_quantity`), the diff is a stored compute (`inventory_diff_quantity`,
6//! T4), and applying flips the quant to the count while the service mints the
7//! `is_inventory` move through the move mint in [`super::stock_picking_repository`].
8//!
9//! Guards carried here: R24 (no staging/applying while the quant holds
10//! reservations — checked by the service from the locked row this file returns),
11//! R13 (only internal locations hold countable stock).
12//!
13//! Per the module's 4-layer rule the statements live here and take the caller's
14//! connection, so the stage/apply writes commit as one unit with the move mint,
15//! the SLE append, and the Bin reblance.
16
17use rust_decimal::Decimal;
18use sqlx::{PgConnection, Row};
19use uuid::Uuid;
20
21/// A locked quant + its location facts (the guard inputs the service checks).
22pub struct QuantCountRow {
23    pub id: Uuid,
24    pub item_id: Uuid,
25    pub location_id: Uuid,
26    pub quantity: Decimal,
27    pub reserved_quantity: Decimal,
28    pub inventory_quantity: Option<Decimal>,
29    pub inventory_diff_quantity: Option<Decimal>,
30    pub inventory_quantity_set: bool,
31    pub inventory_date: Option<chrono::NaiveDate>,
32    /// Location usage as text (`internal` / `view` / ... — R13 check).
33    pub location_usage: String,
34    /// The location's warehouse — the grain the valuation engine (Bin/SLE) keys on.
35    pub location_warehouse_id: Option<Uuid>,
36}
37
38/// How a count addresses a quant: by id, or by the (item, location) pair with all
39/// optional dimensions NULL — the bin-compatible quant grain.
40#[derive(Debug, Clone, Copy)]
41pub enum QuantSelector {
42    ById(Uuid),
43    ItemLocation { item_id: Uuid, location_id: Uuid },
44}
45
46impl QuantSelector {
47    /// (where-fragment, binds) for the selector. Returns the SQL predicate and the
48    /// one-or-two bound values.
49    fn where_clause(&self) -> (&'static str, Vec<Uuid>) {
50        match self {
51            QuantSelector::ById(id) => ("q.id = $1", vec![*id]),
52            QuantSelector::ItemLocation { item_id, location_id } => (
53                "q.item_id = $1 AND q.location_id = $2 AND q.lot_id IS NULL AND q.package_id IS NULL AND q.owner_id IS NULL",
54                vec![*item_id, *location_id],
55            ),
56        }
57    }
58}
59
60/// The quant staging/apply SQL. Stateless; every method takes the caller's connection.
61#[derive(Default)]
62pub struct StockAdjustmentRepository;
63
64impl StockAdjustmentRepository {
65    pub fn new() -> Self {
66        Self
67    }
68
69    /// Lock the addressed quant `FOR UPDATE` (the reservation-apex row — every
70    /// count stage/apply serializes here). `None` when no quant matches; the
71    /// service then either errors (apply) or initializes one (stage).
72    pub async fn lock_quant(
73        &self,
74        conn: &mut PgConnection,
75        selector: QuantSelector,
76    ) -> Result<Option<QuantCountRow>, sqlx::Error> {
77        let (clause, binds) = selector.where_clause();
78        let sql = format!(
79            r#"SELECT q.id, q.item_id, q.location_id, q.quantity, q.reserved_quantity,
80                      q.inventory_quantity, q.inventory_diff_quantity, q.inventory_quantity_set,
81                      q.inventory_date,
82                      l.usage::text AS location_usage,
83                      l.warehouse_id AS location_warehouse_id
84               FROM inventory.stock_quants q
85               JOIN inventory.locations l ON l.id = q.location_id
86               WHERE {clause} AND (q.metadata->>'deleted_at') IS NULL
87               FOR UPDATE OF q"#
88        );
89        let mut q = sqlx::query(&sql);
90        for b in binds {
91            q = q.bind(b);
92        }
93        let row = q.fetch_optional(conn).await?;
94        Ok(row.map(|r| QuantCountRow {
95            id: r.get("id"),
96            item_id: r.get("item_id"),
97            location_id: r.get("location_id"),
98            quantity: r.get("quantity"),
99            reserved_quantity: r.get("reserved_quantity"),
100            inventory_quantity: r.get("inventory_quantity"),
101            inventory_diff_quantity: r.get("inventory_diff_quantity"),
102            inventory_quantity_set: r.get("inventory_quantity_set"),
103            inventory_date: r.get("inventory_date"),
104            location_usage: r.get("location_usage"),
105            location_warehouse_id: r.get("location_warehouse_id"),
106        }))
107    }
108
109    /// Initialize the zeroed quant for (item, location) — the count surface exists
110    /// even before any stock does. Returns its id; the caller has already verified
111    /// the location is internal (R13).
112    pub async fn init_quant(
113        &self,
114        conn: &mut PgConnection,
115        item_id: Uuid,
116        location_id: Uuid,
117    ) -> Result<Uuid, sqlx::Error> {
118        let id = Uuid::new_v4();
119        sqlx::query(
120            r#"INSERT INTO inventory.stock_quants
121                 (id, item_id, location_id, quantity, reserved_quantity, available_quantity,
122                  inventory_quantity_set, sn_duplicated)
123               VALUES ($1, $2, $3, 0, 0, 0, FALSE, FALSE)"#,
124        )
125        .bind(id)
126        .bind(item_id)
127        .bind(location_id)
128        .execute(conn)
129        .await?;
130        Ok(id)
131    }
132
133    /// Stage a counted quantity on a quant (T4): record the count and its stored
134    /// compute (`inventory_diff_quantity = counted - quantity`), and raise the
135    /// `inventory_quantity_set` gate that Apply demands. The pending-count partial
136    /// index (`WHERE inventory_quantity_set = true`) is the worklist this feeds.
137    pub async fn stage_count(
138        &self,
139        conn: &mut PgConnection,
140        quant_id: Uuid,
141        counted: Decimal,
142        diff: Decimal,
143    ) -> Result<(), sqlx::Error> {
144        sqlx::query(
145            r#"UPDATE inventory.stock_quants SET
146                 inventory_quantity = $2,
147                 inventory_diff_quantity = $3,
148                 inventory_quantity_set = TRUE,
149                 inventory_date = CURRENT_DATE
150               WHERE id = $1"#,
151        )
152        .bind(quant_id)
153        .bind(counted)
154        .bind(diff)
155        .execute(conn)
156        .await?;
157        Ok(())
158    }
159
160    /// Consume the staged count: clear the count fields and drop the
161    /// `inventory_quantity_set` gate. This deliberately does NOT write `quantity` — the
162    /// `is_inventory` move the apply minted through the engine pipeline is the one writer
163    /// that reconciled the quant to the count (its two-step sync maintains `quantity`,
164    /// `reserved_quantity`, and the T2 stored compute together). Consuming the staging is
165    /// what makes re-application a no-op: a second apply finds the gate down and mints
166    /// nothing.
167    pub async fn consume_staged_count(
168        &self,
169        conn: &mut PgConnection,
170        quant_id: Uuid,
171    ) -> Result<(), sqlx::Error> {
172        sqlx::query(
173            r#"UPDATE inventory.stock_quants SET
174                 inventory_quantity = NULL,
175                 inventory_diff_quantity = NULL,
176                 inventory_quantity_set = FALSE,
177                 inventory_date = NULL
178               WHERE id = $1"#,
179        )
180        .bind(quant_id)
181        .execute(conn)
182        .await?;
183        Ok(())
184    }
185
186    /// All quants at a location with a staged-but-unapplied count (the pending-count
187    /// worklist the partial index `WHERE inventory_quantity_set = true` serves).
188    pub async fn staged_counts_at_location(
189        &self,
190        conn: &mut PgConnection,
191        location_id: Uuid,
192    ) -> Result<Vec<StagedCountRow>, sqlx::Error> {
193        let rows = sqlx::query(
194            r#"SELECT id, item_id, quantity, inventory_quantity, inventory_diff_quantity
195               FROM inventory.stock_quants
196               WHERE location_id = $1 AND inventory_quantity_set
197                 AND (metadata->>'deleted_at') IS NULL
198               ORDER BY id"#,
199        )
200        .bind(location_id)
201        .fetch_all(conn)
202        .await?;
203        Ok(rows
204            .iter()
205            .map(|r| StagedCountRow {
206                quant_id: r.get("id"),
207                item_id: r.get("item_id"),
208                on_hand_qty: r.get("quantity"),
209                counted_qty: r.get("inventory_quantity"),
210                diff_qty: r.get("inventory_diff_quantity"),
211            })
212            .collect())
213    }
214}
215
216/// One row of the pending-count worklist (the staged counts at a location).
217pub struct StagedCountRow {
218    pub quant_id: Uuid,
219    pub item_id: Uuid,
220    pub on_hand_qty: Decimal,
221    pub counted_qty: Option<Decimal>,
222    pub diff_qty: Option<Decimal>,
223}
backbone_inventory/infrastructure/persistence/stock_adjustment_repository.rs

backbone_inventory/infrastructure/persistence/
stock_adjustment_repository.rs
