1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
//! Repository for Quant entities
//!
//! Originally generated by metaphor-schema; now **user-owned** — this exact path is declared under
//! `user_owned` in `metaphor.codegen.yaml`, so the generator skips it wholesale. The custom methods
//! below hold the hand-written quant SQL: the reservation triangle's apex (spec stock-business-logic
//! §3/§5.1). `stock_quants.reserved_quantity` is the AUTHORITATIVE reservation counter — these
//! methods are its single writer family; the move-line quantity is a mirror and the move state an
//! aggregate, both driven by the engine, never by a second counter here.
//!
//! 4-layer rule: services orchestrate, repositories hold SQL. Every write method takes the CALLER'S
//! connection (the movement's transaction), and the caller has already relayed the ambient org
//! scope onto it.
use rust_decimal::Decimal;
use sqlx::{PgPool, Row};
use uuid::Uuid;
use crate::domain::entity::Quant;
/// Table name for Quant entities
pub const TABLE_NAME: &str = "inventory.stock_quants";
/// Repository for Quant entities.
///
/// All standard CRUD, soft-delete, pagination, and bulk methods are
/// provided automatically via `Deref` to `backbone_orm::GenericCrudRepository`.
pub struct QuantRepository(
backbone_orm::GenericCrudRepository<Quant, backbone_orm::SoftDelete>,
);
impl std::ops::Deref for QuantRepository {
type Target = backbone_orm::GenericCrudRepository<Quant, backbone_orm::SoftDelete>;
fn deref(&self) -> &Self::Target { &self.0 }
}
impl QuantRepository {
/// Create a new repository instance.
pub fn new(pool: PgPool) -> Self {
Self(backbone_orm::GenericCrudRepository::new(pool, TABLE_NAME))
}
}
/// A quant row as the reservation engine sees it.
pub struct QuantRow {
pub id: Uuid,
pub item_id: Uuid,
pub location_id: Uuid,
pub lot_id: Option<Uuid>,
pub package_id: Option<Uuid>,
pub owner_id: Option<Uuid>,
pub quantity: Decimal,
pub reserved_quantity: Decimal,
pub in_date: Option<chrono::DateTime<chrono::Utc>>,
}
impl QuantRow {
/// `available = quantity - reserved` — always a READ off the authoritative pair, never a
/// third writer (spec T2 / stock.hook.yaml rule T2-avail).
pub fn available(&self) -> Decimal {
self.quantity - self.reserved_quantity
}
}
/// The dimension tuple that identifies exactly ONE quant row: item x location x lot x package x
/// owner (spec §3). NULL dimensions are matched with `IS NOT DISTINCT FROM` — the table's unique
/// index treats NULLs as distinct, so NULL-dimension dedup is the ENGINE's responsibility
/// (stock.hook.yaml R12), carried by the advisory lock + insert-guard below.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct QuantDims {
pub item_id: Uuid,
pub location_id: Uuid,
pub lot_id: Option<Uuid>,
pub package_id: Option<Uuid>,
pub owner_id: Option<Uuid>,
}
impl QuantDims {
/// A stable text key for the transaction-scoped advisory lock that serializes creation of the
/// quant row for this dimension tuple (two competing first-writers would otherwise both pass
/// the NOT EXISTS guard — NULLs are distinct in the partial unique index).
fn lock_key(&self) -> String {
let opt = |v: Option<Uuid>| v.map(|u| u.to_string()).unwrap_or_else(|| "NULL".into());
format!(
"quant:{}:{}:{}:{}:{}",
self.item_id, self.location_id, opt(self.lot_id), opt(self.package_id), opt(self.owner_id)
)
}
}
/// Hand-written quant SQL — the reservation apex. Lives here (not in the move engine) per the
/// module's 4-layer rule: services orchestrate, repositories hold SQL.
impl QuantRepository {
/// Lock the quant row for a dimension tuple `FOR UPDATE`, creating a zeroed row if absent —
/// the reservation-engine analogue of `BinRepository::lock_or_init`.
///
/// **This row lock is what serializes competing reservations**: two `_action_assign` calls
/// racing for the same stock queue on this row; the second reads the first's committed
/// reservation once the lock is granted (READ COMMITTED re-evaluates the row under lock), so
/// `reserved_quantity` can never exceed what the on-hand actually allowed (R22).
///
/// Creation is serialized by a transaction-scoped advisory lock on the dimension key because
/// the partial unique index `(item, location, lot, package, owner) WHERE deleted_at IS NULL`
/// does NOT dedup NULL dimensions — without the advisory lock two first-writers could both
/// insert. The lock auto-releases at commit/rollback of the caller's transaction.
pub async fn lock_or_init(
&self,
conn: &mut sqlx::PgConnection,
dims: QuantDims,
) -> Result<QuantRow, sqlx::Error> {
sqlx::query("SELECT pg_advisory_xact_lock(hashtextextended($1, 0))")
.bind(dims.lock_key())
.execute(&mut *conn)
.await?;
if let Some(row) = Self::fetch_for_update(conn, dims).await? {
return Ok(row);
}
sqlx::query(
r#"INSERT INTO inventory.stock_quants
(id, item_id, location_id, lot_id, package_id, owner_id, quantity,
reserved_quantity, available_quantity)
SELECT $1,$2,$3,$4,$5,$6,0,0,0
WHERE NOT EXISTS (
SELECT 1 FROM inventory.stock_quants q
WHERE q.item_id=$2 AND q.location_id=$3
AND q.lot_id IS NOT DISTINCT FROM $4
AND q.package_id IS NOT DISTINCT FROM $5
AND q.owner_id IS NOT DISTINCT FROM $6
AND (q.metadata->>'deleted_at') IS NULL
)"#,
)
.bind(Uuid::new_v4()).bind(dims.item_id).bind(dims.location_id)
.bind(dims.lot_id).bind(dims.package_id).bind(dims.owner_id)
.execute(&mut *conn)
.await?;
// Either this call won the insert race, or the advisory lock means it never races: the
// re-select under FOR UPDATE returns exactly one row either way.
Self::fetch_for_update(conn, dims)
.await?
.ok_or_else(|| sqlx::Error::RowNotFound)
}
/// The `FOR UPDATE` re-select used by [`Self::lock_or_init`]. NULL-safe on every dimension.
async fn fetch_for_update(
conn: &mut sqlx::PgConnection,
dims: QuantDims,
) -> Result<Option<QuantRow>, sqlx::Error> {
let row = sqlx::query(
r#"SELECT id, item_id, location_id, lot_id, package_id, owner_id, quantity,
reserved_quantity, in_date
FROM inventory.stock_quants
WHERE item_id=$1 AND location_id=$2
AND lot_id IS NOT DISTINCT FROM $3
AND package_id IS NOT DISTINCT FROM $4
AND owner_id IS NOT DISTINCT FROM $5
AND (metadata->>'deleted_at') IS NULL
FOR UPDATE"#,
)
.bind(dims.item_id).bind(dims.location_id)
.bind(dims.lot_id).bind(dims.package_id).bind(dims.owner_id)
.fetch_optional(&mut *conn)
.await?;
Ok(row.map(|r| quant_row_of(r)))
}
/// The RESERVED step of `_synchronize_quant` (spec §3): move `delta` onto
/// `reserved_quantity` — the authoritative counter. `available_quantity` (the stored T2
/// compute) is rewritten in the SAME statement as the derived shadow `quantity - reserved`,
/// so it can never drift. The caller holds the row's `FOR UPDATE` from [`Self::lock_or_init`].
///
/// `delta > 0` reserves; `delta < 0` releases. The table CHECK (`reserved_quantity >= 0`)
/// backstops release-over-reserve (a raw-SQL writer cannot underflow it — ADR-0015).
pub async fn adjust_reserved(
&self,
conn: &mut sqlx::PgConnection,
quant_id: Uuid,
delta: Decimal,
) -> Result<QuantRow, sqlx::Error> {
let row = sqlx::query(
r#"UPDATE inventory.stock_quants
SET reserved_quantity = reserved_quantity + $2,
available_quantity = quantity - (reserved_quantity + $2)
WHERE id = $1 AND (metadata->>'deleted_at') IS NULL
RETURNING id, item_id, location_id, lot_id, package_id, owner_id, quantity,
reserved_quantity, in_date"#,
)
.bind(quant_id).bind(delta)
.fetch_one(&mut *conn)
.await?;
Ok(quant_row_of(row))
}
/// The AVAILABLE step of `_synchronize_quant` (spec §3): physically move stock —
/// `qty_delta` onto `quantity` (negative = stock leaves this quant), plus a simultaneous
/// `reserved_delta` (normally the release of the reservation this move consumed). The stored
/// T2 compute is rewritten in lockstep. Stamp `in_date` on first arrival (FIFO ordering
/// within the lot) when the quant transitions to a positive on-hand from non-positive.
pub async fn apply_qty(
&self,
conn: &mut sqlx::PgConnection,
quant_id: Uuid,
qty_delta: Decimal,
reserved_delta: Decimal,
) -> Result<QuantRow, sqlx::Error> {
let row = sqlx::query(
r#"UPDATE inventory.stock_quants
SET quantity = quantity + $2,
reserved_quantity = reserved_quantity + $3,
available_quantity = (quantity + $2) - (reserved_quantity + $3),
in_date = CASE
WHEN in_date IS NULL AND (quantity + $2) > 0 THEN NOW()
ELSE in_date END
WHERE id = $1 AND (metadata->>'deleted_at') IS NULL
RETURNING id, item_id, location_id, lot_id, package_id, owner_id, quantity,
reserved_quantity, in_date"#,
)
.bind(quant_id).bind(qty_delta).bind(reserved_delta)
.fetch_one(&mut *conn)
.await?;
Ok(quant_row_of(row))
}
/// Reservation candidates at one (item, location): the quants with free availability,
/// oldest arrival first (`in_date` FIFO ordering — the default removal strategy; explicit
/// fifo/lifo/fefo strategies layer on top when the tracking surface lands). Takes the
/// caller's transaction: `_action_assign` locks each candidate as it reserves it.
pub async fn fetch_reservation_candidates(
&self,
conn: &mut sqlx::PgConnection,
item_id: Uuid,
location_id: Uuid,
) -> Result<Vec<QuantRow>, sqlx::Error> {
let rows = sqlx::query(
r#"SELECT id, item_id, location_id, lot_id, package_id, owner_id, quantity,
reserved_quantity, in_date
FROM inventory.stock_quants
WHERE item_id=$1 AND location_id=$2
AND quantity - reserved_quantity > 0
AND (metadata->>'deleted_at') IS NULL
ORDER BY in_date NULLS LAST, id"#,
)
.bind(item_id).bind(location_id)
.fetch_all(&mut *conn)
.await?;
Ok(rows.into_iter().map(quant_row_of).collect())
}
/// Re-derive one quant's authoritative `reserved_quantity` from the MIRROR arm: the sum of
/// the live move-line quantities matching this quant's dimension tuple (state not
/// done/cancel). This is the triangle's self-heal — the reserved step of
/// `_synchronize_quant` after a move's lines flip out of the live set (done or cancel),
/// including the partial-validate case where an operator shrinks a line below what it had
/// reserved (the residual would otherwise stay held by no move). The mirror can only ever
/// reflect reservations the engine minted, so the SUM is the truth by construction.
///
/// Takes the caller's transaction; the caller holds the quant's `FOR UPDATE` from
/// [`Self::lock_or_init`]. Returns the recomputed row.
pub async fn recompute_reserved_from_mirror(
&self,
conn: &mut sqlx::PgConnection,
quant_id: Uuid,
dims: QuantDims,
) -> Result<QuantRow, sqlx::Error> {
let row = sqlx::query(
r#"UPDATE inventory.stock_quants q
SET reserved_quantity = m.mirror_qty,
available_quantity = q.quantity - m.mirror_qty
FROM (
SELECT COALESCE(SUM(l.quantity), 0) AS mirror_qty
FROM inventory.stock_move_lines l
WHERE l.item_id = $2
AND l.location_id = $3
AND l.lot_id IS NOT DISTINCT FROM $4
AND l.package_id IS NOT DISTINCT FROM $5
AND l.owner_id IS NOT DISTINCT FROM $6
AND l.state::text NOT IN ('done', 'cancel')
AND (l.metadata->>'deleted_at') IS NULL
) m
WHERE q.id = $1 AND (q.metadata->>'deleted_at') IS NULL
RETURNING q.id, q.item_id, q.location_id, q.lot_id, q.package_id, q.owner_id,
q.quantity, q.reserved_quantity, q.in_date"#,
)
.bind(quant_id)
.bind(dims.item_id)
.bind(dims.location_id)
.bind(dims.lot_id)
.bind(dims.package_id)
.bind(dims.owner_id)
.fetch_one(&mut *conn)
.await?;
Ok(quant_row_of(row))
}
}
/// On-hand aggregation at one location — the `available = quantity - reserved` READ (T2), summed
/// over the location's quant grain. Pool-based (each read its own unit of work), riding the
/// ambient org scope like every read in the suite.
pub struct QuantOnHandRow {
pub on_hand_qty: Decimal,
pub reserved_qty: Decimal,
pub quant_count: i64,
}
impl QuantRepository {
/// SUM(quantity) / SUM(reserved_quantity) over one (item, location). Zeroed row when
/// no quant exists (an unreceived item is unavailable, not an error).
pub async fn fetch_on_hand(
&self,
pool: &PgPool,
item_id: Uuid,
location_id: Uuid,
) -> Result<QuantOnHandRow, sqlx::Error> {
let row = backbone_orm::company_scope::fetch_one_row_scoped(
pool,
sqlx::query(
r#"SELECT
COALESCE(SUM(quantity), 0) AS on_hand_qty,
COALESCE(SUM(reserved_quantity), 0) AS reserved_qty,
COUNT(*) AS quant_count
FROM inventory.stock_quants
WHERE item_id=$1 AND location_id=$2
AND (metadata->>'deleted_at') IS NULL"#,
)
.bind(item_id).bind(location_id),
)
.await?;
Ok(QuantOnHandRow {
on_hand_qty: row.get("on_hand_qty"),
reserved_qty: row.get("reserved_qty"),
quant_count: row.get("quant_count"),
})
}
/// One warehouse-scope availability row (the shop-pivot read): SUM(quantity) /
/// SUM(reserved_quantity) over every quant whose location belongs to the pivot warehouse and
/// holds stock (usage `internal`). `available` is NOT read off the stored
/// `available_quantity` column — the caller computes `on_hand − reserved` per call, so the
/// number is fresh off the reservation triangle regardless of any stored-compute staleness.
///
/// Batch over `item_ids = ANY($1)`; items with no quant rows in the warehouse are simply
/// absent from the result (the caller projects them as zeroed, unavailable rows — an
/// unreceived item is unavailable, not an error). Pool-based and org-scoped: the read rides
/// the ambient org scope the composing service set per request (ADR-0029).
pub async fn fetch_warehouse_on_hand(
&self,
pool: &PgPool,
item_ids: &[Uuid],
warehouse_id: Uuid,
) -> Result<Vec<QuantWarehouseRow>, sqlx::Error> {
let rows = backbone_orm::company_scope::fetch_all_rows_scoped(
pool,
sqlx::query(
r#"SELECT
q.item_id AS item_id,
COALESCE(SUM(q.quantity), 0) AS on_hand_qty,
COALESCE(SUM(q.reserved_quantity), 0) AS reserved_qty
FROM inventory.stock_quants q
JOIN inventory.locations l ON l.id = q.location_id
WHERE q.item_id = ANY($1)
AND l.warehouse_id = $2
AND l.usage = 'internal'::location_usage
AND (q.metadata->>'deleted_at') IS NULL
AND (l.metadata->>'deleted_at') IS NULL
GROUP BY q.item_id"#,
)
.bind(item_ids)
.bind(warehouse_id),
)
.await?;
Ok(rows
.into_iter()
.map(|r| QuantWarehouseRow {
item_id: r.get("item_id"),
on_hand_qty: r.get("on_hand_qty"),
reserved_qty: r.get("reserved_qty"),
})
.collect())
}
}
/// One item's warehouse-scope on-hand pair (the shop-pivot availability read).
pub struct QuantWarehouseRow {
pub item_id: Uuid,
pub on_hand_qty: Decimal,
pub reserved_qty: Decimal,
}
fn quant_row_of(r: sqlx::postgres::PgRow) -> QuantRow {
QuantRow {
id: r.get("id"),
item_id: r.get("item_id"),
location_id: r.get("location_id"),
lot_id: r.get("lot_id"),
package_id: r.get("package_id"),
owner_id: r.get("owner_id"),
quantity: r.get("quantity"),
reserved_quantity: r.get("reserved_quantity"),
in_date: r.get("in_date"),
}
}
backbone_core::impl_crud_repository!(QuantRepository, Quant, soft_delete);