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    /// The quant for (item, location) with no lot, package or owner, created zeroed when there is
110    /// none yet — the count surface exists even before any stock does. Returns its id; the caller
111    /// has already verified the location is internal (R13). Created through the quant
112    /// repository's creation lock, so two counts that both found no quant still make one.
113    pub async fn init_quant(
114        &self,
115        conn: &mut PgConnection,
116        item_id: Uuid,
117        location_id: Uuid,
118    ) -> Result<Uuid, sqlx::Error> {
119        let dims = super::quant_repository::QuantDims { item_id, location_id, lot_id: None, package_id: None, owner_id: None };
120        Ok(super::quant_repository::QuantRepository::lock_or_init_on(conn, dims).await?.id)
121    }
122
123    /// Stage a counted quantity on a quant (T4): record the count and its stored
124    /// compute (`inventory_diff_quantity = counted - quantity`), and raise the
125    /// `inventory_quantity_set` gate that Apply demands. The pending-count partial
126    /// index (`WHERE inventory_quantity_set = true`) is the worklist this feeds.
127    pub async fn stage_count(
128        &self,
129        conn: &mut PgConnection,
130        quant_id: Uuid,
131        counted: Decimal,
132        diff: Decimal,
133    ) -> Result<(), sqlx::Error> {
134        sqlx::query(
135            r#"UPDATE inventory.stock_quants SET
136                 inventory_quantity = $2,
137                 inventory_diff_quantity = $3,
138                 inventory_quantity_set = TRUE,
139                 inventory_date = CURRENT_DATE
140               WHERE id = $1"#,
141        )
142        .bind(quant_id)
143        .bind(counted)
144        .bind(diff)
145        .execute(conn)
146        .await?;
147        Ok(())
148    }
149
150    /// Consume the staged count: clear the count fields and drop the
151    /// `inventory_quantity_set` gate. This deliberately does NOT write `quantity` — the
152    /// `is_inventory` move the apply minted through the engine pipeline is the one writer
153    /// that reconciled the quant to the count (its two-step sync maintains `quantity`,
154    /// `reserved_quantity`, and the T2 stored compute together). Consuming the staging is
155    /// what makes re-application a no-op: a second apply finds the gate down and mints
156    /// nothing.
157    pub async fn consume_staged_count(
158        &self,
159        conn: &mut PgConnection,
160        quant_id: Uuid,
161    ) -> Result<(), sqlx::Error> {
162        sqlx::query(
163            r#"UPDATE inventory.stock_quants SET
164                 inventory_quantity = NULL,
165                 inventory_diff_quantity = NULL,
166                 inventory_quantity_set = FALSE,
167                 inventory_date = NULL
168               WHERE id = $1"#,
169        )
170        .bind(quant_id)
171        .execute(conn)
172        .await?;
173        Ok(())
174    }
175
176    /// All quants at a location with a staged-but-unapplied count (the pending-count
177    /// worklist the partial index `WHERE inventory_quantity_set = true` serves).
178    pub async fn staged_counts_at_location(
179        &self,
180        conn: &mut PgConnection,
181        location_id: Uuid,
182    ) -> Result<Vec<StagedCountRow>, sqlx::Error> {
183        let rows = sqlx::query(
184            r#"SELECT id, item_id, quantity, inventory_quantity, inventory_diff_quantity
185               FROM inventory.stock_quants
186               WHERE location_id = $1 AND inventory_quantity_set
187                 AND (metadata->>'deleted_at') IS NULL
188               ORDER BY id"#,
189        )
190        .bind(location_id)
191        .fetch_all(conn)
192        .await?;
193        Ok(rows
194            .iter()
195            .map(|r| StagedCountRow {
196                quant_id: r.get("id"),
197                item_id: r.get("item_id"),
198                on_hand_qty: r.get("quantity"),
199                counted_qty: r.get("inventory_quantity"),
200                diff_qty: r.get("inventory_diff_quantity"),
201            })
202            .collect())
203    }
204}
205
206/// One row of the pending-count worklist (the staged counts at a location).
207pub struct StagedCountRow {
208    pub quant_id: Uuid,
209    pub item_id: Uuid,
210    pub on_hand_qty: Decimal,
211    pub counted_qty: Option<Decimal>,
212    pub diff_qty: Option<Decimal>,
213}
backbone_inventory/infrastructure/persistence/stock_adjustment_repository.rs

backbone_inventory/infrastructure/persistence/
stock_adjustment_repository.rs
