backbone_inventory/application/service/inventory_move_engine.rs
1//! The converged stock-move engine (hand-authored, user-owned): `_action_confirm` /
2//! `_action_assign` / `_action_done` / `_action_cancel` over quant-grain reservation.
3//!
4//! Spec: `docs/odoo/inventory/stock/stock-business-logic.md` §1 (7-state lifecycle), §2 (T1
5//! picking projection), §3 (reservation triangle), §4 (`_action_done` pipeline); the declared
6//! contract is `schema/hooks/stock.hook.yaml`. The MOVE owns the lifecycle
7//! (draft/waiting/confirmed/partially_available/assigned/done/cancel); the transfer/picking state
8//! is a PROJECTION re-derived on every move change (never its own state machine, ADR-0016).
9//!
10//! **The reservation triangle** (§3): `stock_quants.reserved_quantity` is AUTHORITATIVE — written
11//! only under the quant row's `FOR UPDATE` (`QuantRepository`); each move LINE mirrors the grain it
12//! reserved (item x src x dest x lot x package x owner x qty); the move STATE aggregates the mirror
13//! (`assigned` when the mirror meets demand, `partially_available` when positive but short).
14//! `available = quantity - reserved` is a READ everywhere (T2), never a second writer.
15//!
16//! **`_action_done`** (§4): guards R22 (physical stock suffices for every draw) + R23 (positive
17//! qty) + R24-at-move-grain (done needs lines) + R9 (src != dest); then per line the TWO-STEP
18//! `_synchronize_quant` — the reserved step releases the line's reservation, the available step
19//! moves the physical quantity (src quant decremented, dest quant incremented); done-qty propagates
20//! along `move_orig_ids`/`move_dest_ids` (a `waiting` child releases to `confirmed` once ALL its
21//! parents are done); a partial done (qty < demand) mints the BACKORDER move per policy; `date` is
22//! stamped to the processing instant and `state='done'`; the transfer reprojects (T1).
23//!
24//! **Valuation (money path — unchanged contract).** The pipeline mints Stock Ledger Entries through
25//! the EXISTING `StockLedgerEntryRepository` + the Bin moving-average reblende, and posts the GL
26//! legs through the existing `AccountingPost` seam (`inventory_gl.rs`). There is NO
27//! stock.valuation.layer and no second ledger: receipt blends the rate (`value += qty·price; rate =
28//! value/qty`), delivery consumes the current average (`cogs = qty·rate; rate unchanged`),
29//! transfer is value-neutral at the carried rate — the W1-proven legs. The V7 ordering invariant
30//! (stock-account §2.1: OUT valued BEFORE, IN valued AFTER) is structural here: the OUT leg's value
31//! is computed from the src Bin snapshot taken under the canonical FOR UPDATE locks BEFORE the dest
32//! reblend runs; the IN leg then reblends the destination with the carried value.
33//!
34//! **Voucher identity note.** Move-minted SLE rows ride `voucher_type='stock_entry'` (the generic
35//! stock-operation door) with `voucher_id = move_id` — the `voucher_type` enum has no
36//! `stock_move` variant, and none is needed: consumers discriminate by the id columns, not the
37//! type string. GL legs are emitted with `idempotency_key = move_id` under
38//! `source_type='inventory'`, `source_id = move_id`, `posting_type='original'` — the same
39//! producer name the voucher doors use, disambiguated by the source id (each post's source is
40//! exactly one document: a voucher OR a move, never both). The move's GL settlement is recorded
41//! on the move itself: `_action_done` arms `posting_state='pending'` in the movement's own
42//! transaction when it built an envelope, and reconciles to `posted`/`failed` once the sink
43//! answers. A rejected post does NOT roll the physical movement back — the move parks in
44//! `failed` (a durable, sweepable record of the missing GL leg) and `repost_move_gl` re-drives
45//! it; accounting's dedupe on `(company, source_type, source_id, posting_type)` makes the
46//! re-drive idempotent. Moves that post no GL of their own (voucher-door legs whose GL is owned
47//! by the voucher, value-neutral warehouse-to-warehouse shapes, directives without the needed
48//! accounts) stay `not_applicable`.
49//!
50//! Per the module's 4-layer rule this file holds no SQL — the statements live on
51//! `QuantRepository` / `StockMoveRepository` / `StockMoveLineRepository` / `BinRepository` /
52//! `StockLedgerEntryRepository`, whose write methods take THIS service's transaction so the quant
53//! flips, the Bin reblende, the SLE rows and the move state commit as one unit.
54
55use async_trait::async_trait;
56use chrono::Utc;
57use rust_decimal::Decimal;
58use uuid::Uuid;
59
60use crate::domain::entity::MoveState;
61use crate::infrastructure::persistence::{MoveRow, NewMoveLineRow, NewMoveRow, QuantDims};
62
63use super::inventory_events::{
64 BackorderCreated, InventoryEvent, MoveAssigned, MoveCancelled, MoveConfirmed, MoveDone,
65 TransferProjected,
66};
67use super::inventory_gl::{AccountingPostEnvelope, GlPostLine, GlPostSink};
68use super::inventory_posture::Posture;
69use super::inventory_write_service::{
70 legacy_company_echo, money, rate6, relay_ambient_scope, InventoryError, InventoryWriteService,
71 SubmitOutcome,
72};
73use super::procurement_service::{MovePipeline, MovePipelineError};
74
75// --- input vocabulary ---------------------------------------------------------
76
77/// The move-creation input (draft insert; the pipeline advances state).
78#[derive(Debug, Clone)]
79pub struct NewStockMove {
80 pub name: String,
81 pub item_id: Uuid,
82 pub demand_qty: Decimal,
83 /// Unit valuation price the SLE mint uses on the IN leg (external receipts). 0 = carry the
84 /// current average (value-neutral receive).
85 pub price_unit: Decimal,
86 /// "make_to_stock" | "make_to_order" | "mts_else_mto".
87 pub procure_method: String,
88 pub picking_id: Option<Uuid>,
89 pub origin: Option<String>,
90 pub location_id: Uuid,
91 pub location_dest_id: Uuid,
92 pub partner_id: Option<Uuid>,
93 pub warehouse_id: Option<Uuid>,
94 pub orderpoint_id: Option<Uuid>,
95 pub move_orig_ids: Vec<Uuid>,
96 pub move_dest_ids: Vec<Uuid>,
97 pub is_inventory: bool,
98 pub scrapped: bool,
99 /// Explicit total value the move carries — a document reversal valued at its original
100 /// amount. When set, the valuation core uses it on the value-bearing leg(s) instead of the
101 /// average/price-derived carry, so reversing a voucher restores the estate to exactly its
102 /// pre-movement state. `None` on ordinary moves.
103 pub forced_value: Option<Decimal>,
104}
105
106/// Backorder policy on partial validate (spec §4 step 4: `create_backorder ∈ {always, never,
107/// delayed}` — the operation type's vocabulary). `Always` mints the backorder CONFIRMED and
108/// reserves it right away (reserve on mint); `Delayed` mints it CONFIRMED but defers the
109/// reservation to the scheduler's assign sweep; `Never` leaves the remainder unbackordered.
110#[derive(Debug, Clone, Copy, PartialEq, Eq)]
111pub enum BackorderPolicy {
112 Always,
113 Never,
114 Delayed,
115}
116
117/// GL accounts for the legs `_action_done` may post. Accounts are inventory/composition config —
118/// the same posture as the intake contract (the caller supplies them; they are not move columns).
119/// A leg whose accounts are absent is simply not posted (`gl_posted=false`), the physical movement
120/// is unaffected.
121#[derive(Debug, Clone, Default)]
122pub struct MoveGlDirective {
123 /// COGS debit account for OUT legs (internal -> customer).
124 pub cogs_account_id: Option<Uuid>,
125 /// Inventory valuation account (credit side of OUT, debit side of IN).
126 pub inventory_account_id: Option<Uuid>,
127 /// GR/IR clearing account (credit side of external IN legs).
128 pub grir_account_id: Option<Uuid>,
129 /// Adjustment account for `is_inventory` moves (the value-diff counterleg).
130 pub adjustment_account_id: Option<Uuid>,
131 /// Ledger currency of the post.
132 pub currency: String,
133}
134
135/// Outcome of `_action_done`.
136#[derive(Debug, Clone)]
137pub struct MoveDoneOutcome {
138 pub move_id: Uuid,
139 pub done_qty: Decimal,
140 /// The minted backorder move (remaining demand), if any.
141 pub backorder_move_id: Option<Uuid>,
142 /// SLE rows minted by the valuation core (0 for value-neutral same-warehouse moves).
143 pub sle_count: i32,
144 /// The value the OUT leg carried off the source bin (COGS for a delivery shape).
145 pub out_value: Decimal,
146 /// The value the IN leg blended into the destination bin (the line amount for a receipt
147 /// shape).
148 pub in_value: Decimal,
149 pub gl_posted: bool,
150 pub gl_amount: Decimal,
151}
152
153/// Outcome of `_action_assign`.
154#[derive(Debug, Clone)]
155pub struct MoveAssignOutcome {
156 pub move_id: Uuid,
157 /// The move's post-assign state: `assigned` (fully reserved), `partially_available` (some
158 /// reserved), `confirmed` (nothing reservable at the source location).
159 pub state: String,
160 pub reserved_qty: Decimal,
161}
162
163/// The move lifecycle engine. An `impl InventoryWriteService` chunk over the vocabulary in
164/// [`super::inventory_write_service`] — the converged stock pipeline the routes/scheduler surfaces
165/// drive.
166impl InventoryWriteService {
167 // ---- create -------------------------------------------------------------
168
169 /// Create a DRAFT move (spec §1: `draft → waiting/confirmed` happens on confirm, never at
170 /// insert). Guards: R9 (src != dest), non-negative demand (R23 at the door), R13 (a quant
171 /// surface can never be a view location — checked for both endpoints that hold stock).
172 pub async fn create_move(&self, m: NewStockMove) -> Result<Uuid, InventoryError> {
173 if m.demand_qty < Decimal::ZERO || m.price_unit < Decimal::ZERO {
174 return Err(InventoryError::NegativeQuantity);
175 }
176 if m.location_id == m.location_dest_id {
177 return Err(InventoryError::SameLocation { move_id: Uuid::new_v4(), location_id: m.location_id });
178 }
179 let id = Uuid::new_v4();
180 let mut tx = self.rpool().begin().await?;
181 // Re-bind the caller's ambient org scope onto this transaction before the location
182 // reads (ADR-0029) — the scope is task-local and a fresh pool transaction carries none
183 // of it. Under the composed shape the decorator's fence bounds every read here;
184 // undecorated (module tests, jobs) the transaction stays plain.
185 relay_ambient_scope(&mut tx).await?;
186 let locs = self.moves.fetch_move_locations(&mut tx, m.location_id, m.location_dest_id).await?;
187 let (src, dst) = match locs {
188 (Some(s), Some(d)) => (s, d),
189 (None, _) => return Err(InventoryError::LocationNotFound(m.location_id)),
190 (_, None) => return Err(InventoryError::LocationNotFound(m.location_dest_id)),
191 };
192 for loc in [&src, &dst] {
193 if loc.usage == "view" {
194 return Err(InventoryError::ViewLocationHoldsNoStock { location_id: loc.id });
195 }
196 }
197 self.moves.insert_move(&mut tx, &NewMoveRow {
198 id,
199 name: &m.name,
200 item_id: m.item_id,
201 demand_qty: m.demand_qty,
202 price_unit: m.price_unit,
203 procure_method: &m.procure_method,
204 picking_id: m.picking_id,
205 origin: m.origin.as_deref(),
206 location_id: m.location_id,
207 location_dest_id: m.location_dest_id,
208 partner_id: m.partner_id,
209 warehouse_id: m.warehouse_id,
210 orderpoint_id: m.orderpoint_id,
211 move_orig_ids: m.move_orig_ids.clone(),
212 move_dest_ids: m.move_dest_ids.clone(),
213 is_inventory: m.is_inventory,
214 scrapped: m.scrapped,
215 forced_value: m.forced_value,
216 }).await?;
217 // Chain bookkeeping: a move naming its `move_orig_ids` parents gets the REVERSE link
218 // written too (each parent's move_dest_ids gains this move) — the done-qty propagation
219 // and waiting-release walk the parent's `move_dest_ids`, so both directions must exist
220 // (`link_chain` is idempotent per pair).
221 for parent in &m.move_orig_ids {
222 self.moves.link_chain(&mut tx, *parent, id).await?;
223 }
224 if let Some(picking) = m.picking_id {
225 self.moves.reproject_picking(&mut tx, picking).await?;
226 }
227 tx.commit().await?;
228 Ok(id)
229 }
230
231 // ---- _action_confirm ------------------------------------------------------
232
233 /// `draft → confirmed`, or `draft → waiting` while any `move_orig_ids` parent is not done
234 /// (spec §1: `waiting` is the state a chained move sits in until its parents land).
235 /// `procure_method` gates supply: `make_to_order` / `mts_else_mto` moves are minted by the
236 /// procurement (rule) engine, not here — this method only owns the state gate.
237 ///
238 /// The move read rides the caller's ambient org scope (ADR-0029): under the composed shape
239 /// the decorator's fence bounds it; undecorated (module tests, jobs) it is plain.
240 pub async fn action_confirm(&self, move_id: Uuid) -> Result<String, InventoryError> {
241 let mut tx = self.rpool().begin().await?;
242 relay_ambient_scope(&mut tx).await?;
243 let mv = self.moves.fetch_move(&mut tx, move_id).await?
244 .ok_or(InventoryError::NotFound(move_id))?;
245 if mv.state != "draft" {
246 return Err(InventoryError::WrongMoveState { move_id, action: "confirm", current: mv.state });
247 }
248 let to = self.confirm_core(&mut tx, &mv).await?;
249 tx.commit().await?;
250 if to == "confirmed" {
251 self.sink.publish(InventoryEvent::MoveConfirmed(MoveConfirmed {
252 move_id, company_id: legacy_company_echo(), item_id: mv.item_id,
253 demand_qty: mv.demand_qty, picking_id: mv.picking_id,
254 }));
255 }
256 Ok(to)
257 }
258
259 /// The confirm verb's core on the caller's connection (the ambient org scope already bound
260 /// — ADR-0029): the parents-done check, the guarded transition, and the picking reproject.
261 /// Shared by the public verb (own transaction) and the [`MovePipeline`] port the scheduler
262 /// drives on its per-batch connection.
263 async fn confirm_core(
264 &self,
265 conn: &mut sqlx::PgConnection,
266 mv: &MoveRow,
267 ) -> Result<String, InventoryError> {
268 let parents_done = if mv.move_orig_ids.is_empty() {
269 true
270 } else {
271 let parents = self.moves.fetch_moves(conn, &mv.move_orig_ids).await?;
272 parents.iter().all(|p| p.state == "done")
273 };
274 let to = if parents_done { "confirmed" } else { "waiting" };
275 let ok = self.moves.transition_state(conn, mv.id, "draft", to).await?;
276 if !ok {
277 return Err(InventoryError::WrongMoveState { move_id: mv.id, action: "confirm", current: "raced".into() });
278 }
279 if let Some(picking) = mv.picking_id {
280 self.moves.reproject_picking(conn, picking).await?;
281 }
282 Ok(to.into())
283 }
284
285 // ---- _action_assign -------------------------------------------------------
286
287 /// Reserve against quants at the source location (spec §1 `_action_assign`): the authoritative
288 /// `reserved_quantity` is written under each quant's `FOR UPDATE` (competing reservations
289 /// serialize there — one winner per unit of stock, R22/R25), a mirror move-line is minted per
290 /// reserved quant grain, and the move state aggregates the mirror.
291 ///
292 /// The move read rides the caller's ambient org scope (ADR-0029): under the composed shape
293 /// the decorator's fence bounds it; undecorated (module tests, jobs) it is plain.
294 pub async fn action_assign(&self, move_id: Uuid) -> Result<MoveAssignOutcome, InventoryError> {
295 let mut tx = self.rpool().begin().await?;
296 relay_ambient_scope(&mut tx).await?;
297 let mv = self.moves.fetch_move(&mut tx, move_id).await?
298 .ok_or(InventoryError::NotFound(move_id))?;
299 if mv.state != "confirmed" && mv.state != "partially_available" {
300 return Err(InventoryError::WrongMoveState { move_id, action: "assign", current: mv.state });
301 }
302 let (to, total) = self.assign_core(&mut tx, &mv).await?;
303 let picking = mv.picking_id;
304 tx.commit().await?;
305 if to == "assigned" {
306 self.sink.publish(InventoryEvent::MoveAssigned(MoveAssigned {
307 move_id, company_id: legacy_company_echo(), picking_id: picking,
308 }));
309 }
310 Ok(MoveAssignOutcome { move_id, state: to, reserved_qty: total })
311 }
312
313 /// The assign verb's core on the caller's connection (the ambient org scope already bound —
314 /// ADR-0029): the reservation loop, the mirror mint, the aggregate state transition, the
315 /// line re-mirror and the picking reproject. Returns `(new_state, reserved_qty)`. Shared by
316 /// the public verb and the [`MovePipeline`] port.
317 async fn assign_core(
318 &self,
319 conn: &mut sqlx::PgConnection,
320 mv: &MoveRow,
321 ) -> Result<(String, Decimal), InventoryError> {
322 let locs = self.moves.fetch_move_locations(conn, mv.location_id, mv.location_dest_id).await?;
323 let src = locs.0.ok_or(InventoryError::LocationNotFound(mv.location_id))?;
324 // Reservation only draws from INTERNAL stock; an inbound move (supplier source) has nothing
325 // to reserve and stays `confirmed`.
326 let mut reserved = Decimal::ZERO;
327 if src.usage == "internal" {
328 let already = self.move_lines.sum_mirror_qty(conn, mv.id).await?;
329 let mut remaining = mv.demand_qty - already;
330 let candidates = self.quants.fetch_reservation_candidates(conn, mv.item_id, mv.location_id).await?;
331 for cand in candidates {
332 if remaining <= Decimal::ZERO { break; }
333 let dims = QuantDims {
334 item_id: cand.item_id, location_id: cand.location_id,
335 lot_id: cand.lot_id, package_id: cand.package_id, owner_id: cand.owner_id,
336 };
337 // Lock the quant row, re-read the free availability under the lock (READ COMMITTED
338 // sees the competitor's committed reservation), then reserve the min(free, need).
339 let locked = self.quants.lock_or_init(conn, dims).await?;
340 let free = locked.available();
341 if free <= Decimal::ZERO { continue; }
342 let take = if free < remaining { free } else { remaining };
343 self.quants.adjust_reserved(conn, locked.id, take).await?;
344 self.move_lines.insert_line(conn, &NewMoveLineRow {
345 id: Uuid::new_v4(),
346 quantity: take,
347 lot_id: cand.lot_id,
348 package_id: cand.package_id,
349 result_package_id: None,
350 owner_id: cand.owner_id,
351 move_id: mv.id,
352 picking_id: mv.picking_id,
353 location_id: mv.location_id,
354 location_dest_id: mv.location_dest_id,
355 item_id: mv.item_id,
356 state: mv.state.as_str(), // transient: re-mirrored to the post-assign state below
357 }).await?;
358 reserved += take;
359 remaining -= take;
360 }
361 reserved += already;
362 } else {
363 // An INBOUND move (non-internal source — supplier/production) has no stock to
364 // reserve: its supply is unconditionally available, so `_action_assign` mints the
365 // execution line for the full remaining demand (Odoo's incoming-move readiness) —
366 // the line records what will land, with no quant touched at the source.
367 let already = self.move_lines.sum_mirror_qty(conn, mv.id).await?;
368 if already < mv.demand_qty {
369 self.move_lines.insert_line(conn, &NewMoveLineRow {
370 id: Uuid::new_v4(),
371 quantity: mv.demand_qty - already,
372 lot_id: None,
373 package_id: None,
374 result_package_id: None,
375 owner_id: None,
376 move_id: mv.id,
377 picking_id: mv.picking_id,
378 location_id: mv.location_id,
379 location_dest_id: mv.location_dest_id,
380 item_id: mv.item_id,
381 state: mv.state.as_str(), // transient: re-mirrored to the post-assign state below
382 }).await?;
383 }
384 reserved = mv.demand_qty;
385 }
386 let total = if mv.demand_qty < reserved { mv.demand_qty } else { reserved };
387 let to = if total >= mv.demand_qty && mv.demand_qty > Decimal::ZERO {
388 "assigned"
389 } else if total > Decimal::ZERO {
390 "partially_available"
391 } else {
392 "confirmed"
393 };
394 let ok = self.moves.transition_state(conn, mv.id, mv.state.as_str(), to).await?;
395 if !ok {
396 return Err(InventoryError::WrongMoveState { move_id: mv.id, action: "assign", current: "raced".into() });
397 }
398 self.move_lines.mirror_state(conn, mv.id, to).await?;
399 if let Some(picking) = mv.picking_id {
400 self.moves.reproject_picking(conn, picking).await?;
401 }
402 Ok((to.into(), total))
403 }
404
405 // ---- unreserve (the triangle's release arm) --------------------------------
406
407 /// Release every reservation the move holds (spec §1: cancel frees reservations first; assign
408 /// retries also release-then-re.reserve). The mirror lines zero out; the authoritative
409 /// `reserved_quantity` drops by exactly what the lines held.
410 ///
411 /// The move read rides the caller's ambient org scope (ADR-0029): under the composed shape
412 /// the decorator's fence bounds it; undecorated (module tests, jobs) it is plain.
413 pub async fn unreserve_move(&self, move_id: Uuid) -> Result<Decimal, InventoryError> {
414 let mut tx = self.rpool().begin().await?;
415 relay_ambient_scope(&mut tx).await?;
416 let mv = self.moves.fetch_move(&mut tx, move_id).await?
417 .ok_or(InventoryError::NotFound(move_id))?;
418 let released = self.unreserve_lines(&mut tx, &mv).await?;
419 tx.commit().await?;
420 Ok(released)
421 }
422
423 /// Internal: release all live reservations of a move inside the caller's transaction, by
424 /// zeroing the mirror lines and re-deriving each touched quant's authoritative counter from
425 /// the live mirror (the triangle's self-heal — a quant can never hold a reservation no live
426 /// line claims). Returns the quantity the lines held. The zeroed lines stay as the record of
427 /// what the move had reserved.
428 async fn unreserve_lines(
429 &self,
430 tx: &mut sqlx::PgConnection,
431 mv: &MoveRow,
432 ) -> Result<Decimal, InventoryError> {
433 let lines = self.move_lines.fetch_lines_for_move(tx, mv.id).await?;
434 let mut released = Decimal::ZERO;
435 let mut seen_dims: Vec<QuantDims> = Vec::new();
436 for line in lines {
437 if line.quantity > Decimal::ZERO {
438 released += line.quantity;
439 }
440 self.move_lines.set_quantity(tx, line.id, Decimal::ZERO).await?;
441 let dims = QuantDims {
442 item_id: line.item_id, location_id: line.location_id,
443 lot_id: line.lot_id, package_id: line.package_id, owner_id: line.owner_id,
444 };
445 if !seen_dims.contains(&dims) { seen_dims.push(dims); }
446 }
447 for dims in seen_dims {
448 let quant = self.quants.lock_or_init(tx, dims).await?;
449 self.quants.recompute_reserved_from_mirror(tx, quant.id, dims).await?;
450 }
451 Ok(released)
452 }
453
454 // ---- _action_done ----------------------------------------------------------
455
456 /// The mutation core (spec §4). See the module docs above for the full pipeline; the guards,
457 /// the two-step quant sync, the V7-ordered valuation, the chain propagation, the backorder
458 /// split and the projection all live here.
459 ///
460 /// The move read rides the caller's ambient org scope (ADR-0029): under the composed shape
461 /// the decorator's fence bounds it; undecorated (module tests, jobs) it is plain.
462 pub async fn action_done(
463 &self,
464 move_id: Uuid,
465 backorder: BackorderPolicy,
466 gl: &MoveGlDirective,
467 sink: &dyn GlPostSink,
468 ) -> Result<MoveDoneOutcome, InventoryError> {
469 let mut tx = self.rpool().begin().await?;
470 relay_ambient_scope(&mut tx).await?;
471 let mv = self.moves.fetch_move(&mut tx, move_id).await?
472 .ok_or(InventoryError::NotFound(move_id))?;
473 match mv.state.as_str() {
474 "assigned" | "partially_available" | "confirmed" => {}
475 other => return Err(InventoryError::WrongMoveState { move_id, action: "done", current: other.into() }),
476 }
477
478 let lines = self.move_lines.fetch_lines_for_move(&mut tx, move_id).await?;
479 if lines.is_empty() {
480 return Err(InventoryError::MoveLinesRequired { move_id }); // R24 at move grain
481 }
482 let mut done_qty = Decimal::ZERO;
483 for l in &lines {
484 if l.quantity < Decimal::ZERO {
485 return Err(InventoryError::NegativeQuantity); // R23
486 }
487 done_qty += l.quantity;
488 }
489 if done_qty <= Decimal::ZERO && mv.demand_qty > Decimal::ZERO {
490 return Err(InventoryError::MoveLinesRequired { move_id }); // nothing to validate
491 }
492
493 let locs = self.moves.fetch_move_locations(&mut tx, mv.location_id, mv.location_dest_id).await?;
494 let src = locs.0.ok_or(InventoryError::LocationNotFound(mv.location_id))?;
495 let dst = locs.1.ok_or(InventoryError::LocationNotFound(mv.location_dest_id))?;
496
497 // -- the done write FIRST, so the lines drop out of the live-reservation mirror before
498 // the reserved step re-derives the quant's counter (the two steps below are ordered
499 // by exactly this dependency) -------------------------------------------------------
500 let now = Utc::now();
501 let ok = self.moves.mark_done(&mut tx, move_id, mv.state.as_str(), done_qty, now).await?;
502 if !ok {
503 return Err(InventoryError::WrongMoveState { move_id, action: "done", current: "raced".into() });
504 }
505 self.move_lines.mirror_state(&mut tx, move_id, "done").await?;
506
507 // -- Step 1 of _synchronize_quant (the RESERVED step): re-derive each touched quant's
508 // authoritative reserved_quantity from the live mirror. Flipping this move's lines to
509 // `done` dropped them out of the SUM, so the reservation releases here — including any
510 // residual a shrunk line would otherwise strand. One recompute per DISTINCT source
511 // dimension tuple; quants that flip to zero on-hand keep their zeroed counter.
512 let mut seen_dims: Vec<QuantDims> = Vec::new();
513 for line in &lines {
514 if line.quantity <= Decimal::ZERO { continue; }
515 let src_dims = QuantDims {
516 item_id: line.item_id, location_id: line.location_id,
517 lot_id: line.lot_id, package_id: line.package_id, owner_id: line.owner_id,
518 };
519 if seen_dims.contains(&src_dims) { continue; }
520 seen_dims.push(src_dims);
521 let quant = self.quants.lock_or_init(&mut tx, src_dims).await?;
522 self.quants.recompute_reserved_from_mirror(&mut tx, quant.id, src_dims).await?;
523 }
524
525 // -- Step 2 of _synchronize_quant (the AVAILABLE step): the physical flip, guarded by R22
526 // at quant grain — the PHYSICAL on-hand must cover the draw (the reserved <= available
527 // invariant itself is held by the row lock + the CHECK backstop). Only an INTERNAL
528 // source is drawn from: the quant estate tracks our own stock, so an inbound move
529 // (supplier/production source) has no source quant to decrement — its supply materializes
530 // at the destination quant below.
531 let draw_from_src = src.usage == "internal";
532 for line in &lines {
533 if line.quantity <= Decimal::ZERO { continue; }
534 if draw_from_src {
535 let src_dims = QuantDims {
536 item_id: line.item_id, location_id: line.location_id,
537 lot_id: line.lot_id, package_id: line.package_id, owner_id: line.owner_id,
538 };
539 let src_quant = self.quants.lock_or_init(&mut tx, src_dims).await?;
540 if src_quant.quantity < line.quantity {
541 return Err(InventoryError::InsufficientStock {
542 item_id: mv.item_id,
543 warehouse_id: mv.warehouse_id.unwrap_or(mv.location_id),
544 available: src_quant.quantity,
545 requested: line.quantity,
546 });
547 }
548 self.quants.apply_qty(&mut tx, src_quant.id, -line.quantity, Decimal::ZERO).await?;
549 }
550 // Only an INTERNAL destination materializes a quant, the exact twin
551 // of the draw gate above: the quant estate tracks our own stock,
552 // so virtual counterparts (vendors, customers, production, transit,
553 // loss) keep NO balance — a move through them records its lines and
554 // its valuation, never a standing quant. The owner ruling of 2026-09-29;
555 // the previous shape (IN recorded, OUT dropped) left the counterparts
556 // as running tallies nothing could read.
557 if dst.usage == "internal" {
558 let dst_dims = QuantDims {
559 item_id: line.item_id, location_id: line.location_dest_id,
560 lot_id: line.lot_id, package_id: line.result_package_id.or(line.package_id), owner_id: line.owner_id,
561 };
562 let dst_quant = self.quants.lock_or_init(&mut tx, dst_dims).await?;
563 self.quants.apply_qty(&mut tx, dst_quant.id, line.quantity, Decimal::ZERO).await?;
564 }
565 }
566
567 // -- valuation core (V7: OUT valued before, IN after) --------------------------------
568 let (sle_count, out_value, in_value) = self
569 .mint_move_valuation(&mut tx, &mv, &src, &dst, done_qty)
570 .await?;
571
572 // -- GL arming (inside the movement's transaction) -----------------------------------
573 // Build the envelope BEFORE commit and arm the GL leg in the SAME transaction as the
574 // physical movement: a move either lands done with its GL leg `pending`, or not at all.
575 // A move that builds no envelope (voucher-door legs whose GL is owned by the voucher,
576 // value-neutral shapes, a directive without the accounts this leg needs, a `periodic`
577 // posture whose real-time posts are suppressed) stays `not_applicable` — the engine
578 // posts nothing on its behalf.
579 let posture = self.posting_posture_on(&mut tx).await?;
580 let envelope = self.move_gl_envelope(&mv, &src, &dst, gl, out_value, in_value, &posture);
581 if envelope.is_some() {
582 self.moves.set_posting_pending(&mut tx, move_id).await?;
583 }
584
585 // -- backorder split (partial validate) ----------------------------------------------
586 // Policy vocabulary (the operation type's `create_backorder`): the remainder becomes its
587 // own move on the SAME picking (the transfer stays open below `done` while it lives).
588 // Both minting policies CONFIRM the backorder — a confirmed move is what the
589 // scheduler's assign sweep and a re-validate can drive; a draft one is invisible to
590 // both. `Always` then reserves right away (reserve on mint); `Delayed` leaves the
591 // reservation to the scheduler's assign sweep (deferred reservation).
592 let mut backorder_move_id = None;
593 if done_qty < mv.demand_qty && backorder != BackorderPolicy::Never {
594 let remaining = mv.demand_qty - done_qty;
595 let child = Uuid::new_v4();
596 let backorder_name = format!("{}/BO", mv.name);
597 self.moves.insert_move(&mut tx, &NewMoveRow {
598 id: child,
599 name: &backorder_name,
600 item_id: mv.item_id,
601 demand_qty: remaining,
602 price_unit: mv.price_unit,
603 procure_method: &mv.procure_method,
604 picking_id: mv.picking_id,
605 origin: mv.origin.as_deref(),
606 location_id: mv.location_id,
607 location_dest_id: mv.location_dest_id,
608 partner_id: mv.partner_id,
609 warehouse_id: mv.warehouse_id,
610 orderpoint_id: mv.orderpoint_id,
611 move_orig_ids: vec![move_id],
612 move_dest_ids: mv.move_dest_ids.clone(),
613 is_inventory: mv.is_inventory,
614 scrapped: mv.scrapped,
615 forced_value: None, // fresh demand: ordinary valuation, never the parent's reversal carry
616 }).await?;
617 self.moves.link_chain(&mut tx, move_id, child).await?;
618 let child_row = self.moves.fetch_move(&mut tx, child).await?
619 .ok_or(InventoryError::NotFound(child))?;
620 self.confirm_core(&mut tx, &child_row).await?;
621 if backorder == BackorderPolicy::Always {
622 let fresh = self.moves.fetch_move(&mut tx, child).await?
623 .ok_or(InventoryError::NotFound(child))?;
624 if matches!(fresh.state.as_str(), "confirmed" | "partially_available") {
625 self.assign_core(&mut tx, &fresh).await?;
626 }
627 }
628 backorder_move_id = Some(child);
629 }
630
631 // -- done-qty propagation: release waiting children whose parents are all done --------
632 let mut released_children: Vec<crate::infrastructure::persistence::MoveRow> = Vec::new();
633 if !mv.move_dest_ids.is_empty() {
634 let children = self.moves.fetch_moves(&mut tx, &mv.move_dest_ids).await?;
635 for ch in &children {
636 if ch.state != "waiting" { continue; }
637 let parents = self.moves.fetch_moves(&mut tx, &ch.move_orig_ids).await?;
638 if parents.iter().all(|p| p.state == "done") {
639 let ok = self.moves.transition_state(&mut tx, ch.id, "waiting", "confirmed").await?;
640 if ok {
641 released_children.push(ch.clone());
642 }
643 }
644 }
645 }
646
647 let mut projected_state: Option<String> = None;
648 if let Some(picking) = mv.picking_id {
649 projected_state = self.moves.reproject_picking(&mut tx, picking).await?;
650 }
651 let picking = mv.picking_id;
652 tx.commit().await?;
653
654 // -- GL post (eventually consistent — the physical movement already committed) ---------
655 // Failure posture: a rejection does NOT roll the physical movement back and does NOT
656 // strand the hole silently — the move is marked `failed` (the durable, sweepable record
657 // of the missing GL leg) while the error still surfaces to the caller, and
658 // `repost_move_gl` re-drives the leg. A crash between this commit and the reconcile
659 // leaves the move `pending`; the same repost heals it. Accounting dedupes on the
660 // envelope's source identity, so a re-drive can never double post.
661 let mut gl_posted = false;
662 let mut gl_amount = Decimal::ZERO;
663 if let Some(env) = envelope {
664 debug_assert!(env.is_balanced());
665 gl_amount = env.lines.iter().map(|l| l.debit).sum();
666 match sink.post(&env).await {
667 Ok(_) => {
668 self.moves.mark_posting_posted(&self.rpool(), move_id).await?;
669 gl_posted = true;
670 }
671 Err(rej) => {
672 let _ = self.moves.mark_posting_failed(&self.rpool(), move_id).await;
673 return Err(InventoryError::GlRejected { code: rej.code, message: rej.message });
674 }
675 }
676 }
677
678 // -- events ----------------------------------------------------------------------------
679 self.sink.publish(InventoryEvent::MoveDone(MoveDone {
680 move_id, company_id: legacy_company_echo(), item_id: mv.item_id,
681 quantity: done_qty, price_unit: mv.price_unit, is_inventory: mv.is_inventory,
682 }));
683 if let Some(child) = backorder_move_id {
684 self.sink.publish(InventoryEvent::BackorderCreated(BackorderCreated {
685 backorder_id: child, company_id: legacy_company_echo(), origin_id: move_id,
686 }));
687 }
688 for ch in released_children {
689 self.sink.publish(InventoryEvent::MoveConfirmed(MoveConfirmed {
690 move_id: ch.id, company_id: legacy_company_echo(), item_id: ch.item_id,
691 demand_qty: ch.demand_qty, picking_id: ch.picking_id,
692 }));
693 }
694 if let Some(p) = picking {
695 if let Some(state) = projected_state {
696 self.sink.publish(InventoryEvent::TransferProjected(TransferProjected {
697 transfer_id: p, company_id: legacy_company_echo(), state, previous_state: mv.state.clone(),
698 }));
699 }
700 }
701 Ok(MoveDoneOutcome { move_id, done_qty, backorder_move_id, sle_count, out_value, in_value, gl_posted, gl_amount })
702 }
703
704 // ---- _action_cancel --------------------------------------------------------
705
706 /// `* → cancel` (spec §1): frees the reservation first, then propagates to the chained
707 /// children (`move_dest_ids`) unless `propagate_cancel=false` — never into a done move.
708 ///
709 /// The move read rides the caller's ambient org scope (ADR-0029): under the composed shape
710 /// the decorator's fence bounds it; undecorated (module tests, jobs) it is plain.
711 pub async fn action_cancel(&self, move_id: Uuid) -> Result<Decimal, InventoryError> {
712 let mut tx = self.rpool().begin().await?;
713 relay_ambient_scope(&mut tx).await?;
714 let mv = self.moves.fetch_move(&mut tx, move_id).await?
715 .ok_or(InventoryError::NotFound(move_id))?;
716 match mv.state.as_str() {
717 "draft" | "waiting" | "confirmed" | "partially_available" | "assigned" => {}
718 other => return Err(InventoryError::WrongMoveState { move_id, action: "cancel", current: other.into() }),
719 }
720 let released = self.unreserve_lines(&mut tx, &mv).await?;
721 let ok = self.moves.transition_state(&mut tx, move_id, mv.state.as_str(), "cancel").await?;
722 if !ok {
723 return Err(InventoryError::WrongMoveState { move_id, action: "cancel", current: "raced".into() });
724 }
725 self.move_lines.mirror_state(&mut tx, move_id, "cancel").await?;
726 if mv.propagate_cancel && !mv.move_dest_ids.is_empty() {
727 let children = self.moves.fetch_moves(&mut tx, &mv.move_dest_ids).await?;
728 for ch in children {
729 if matches!(ch.state.as_str(), "done" | "cancel") { continue; }
730 let _ = self.moves.transition_state(&mut tx, ch.id, ch.state.as_str(), "cancel").await?;
731 self.move_lines.mirror_state(&mut tx, ch.id, "cancel").await?;
732 }
733 }
734 if let Some(picking) = mv.picking_id {
735 self.moves.reproject_picking(&mut tx, picking).await?;
736 }
737 tx.commit().await?;
738 self.sink.publish(InventoryEvent::MoveCancelled(MoveCancelled {
739 move_id, company_id: legacy_company_echo(), released_qty: released,
740 }));
741 Ok(released)
742 }
743
744 // ---- GL repost: re-drive a stuck move GL leg ---------------------------------
745
746 /// Re-drive the GL leg of a done move whose `posting_state` is `pending` or `failed` — a
747 /// rejected or interrupted AccountingPost from `_action_done`. The physical movement
748 /// (quants, Bin, SLE rows) already happened and is never re-touched: the envelope is rebuilt
749 /// from the move's committed ledger legs plus the caller's account directive, re-emitted
750 /// under the move's ORIGINAL source identity (`source_type='inventory'`, `source_id` = move
751 /// id, `posting_type='original'`), and accounting's dedupe on that identity makes the
752 /// re-drive idempotent — a leg that actually landed before the status write was lost returns
753 /// the original journal instead of a second one.
754 ///
755 /// Short-circuits exactly like the voucher doors' repost verbs: `posted` → nothing to drive;
756 /// `not_applicable` → this move owns no GL leg (a voucher door posted its voucher's
757 /// envelope; a value-neutral shape posts nothing), so a sweep can call this blindly. If the
758 /// rebuilt envelope is `None` (the directive no longer supplies the accounts this leg shape
759 /// needs), the unsettled leg retires to `not_applicable` — the GL is genuinely not postable
760 /// under the current configuration and the move leaves the sweep worklist.
761 ///
762 /// On a fresh rejection the move parks in `failed` again and the error surfaces, as in
763 /// `_action_done`.
764 ///
765 /// The move read rides the caller's ambient org scope (ADR-0029): under the composed shape
766 /// the decorator's fence bounds it; undecorated (module tests, jobs) it is plain.
767 pub async fn repost_move_gl(
768 &self,
769 move_id: Uuid,
770 gl: &MoveGlDirective,
771 sink: &dyn GlPostSink,
772 ) -> Result<SubmitOutcome, InventoryError> {
773 let mv = {
774 let mut tx = self.rpool().begin().await?;
775 relay_ambient_scope(&mut tx).await?;
776 let mv = self.moves.fetch_move(&mut tx, move_id).await?
777 .ok_or(InventoryError::NotFound(move_id))?;
778 tx.commit().await?;
779 mv
780 };
781 match mv.posting_state.as_str() {
782 "posted" => return Ok(SubmitOutcome {
783 voucher_id: move_id, posted: true, journal_id: None, post_id: None,
784 gl_amount: Decimal::ZERO,
785 }),
786 "not_applicable" => return Ok(SubmitOutcome {
787 voucher_id: move_id, posted: false, journal_id: None, post_id: None,
788 gl_amount: Decimal::ZERO,
789 }),
790 _ => {} // pending | failed → re-drive below
791 }
792 if mv.state != "done" {
793 // posting_state only leaves not_applicable inside a done transaction, so this is a
794 // defensive guard — but a hand-mangled row should fail loudly, not re-post.
795 return Err(InventoryError::WrongMoveState { move_id, action: "repost_gl", current: mv.state });
796 }
797 let locs = {
798 let mut tx = self.rpool().begin().await?;
799 relay_ambient_scope(&mut tx).await?;
800 let locs = self.moves.fetch_move_locations(&mut tx, mv.location_id, mv.location_dest_id).await?;
801 tx.commit().await?;
802 locs
803 };
804 let src = locs.0.ok_or(InventoryError::LocationNotFound(mv.location_id))?;
805 let dst = locs.1.ok_or(InventoryError::LocationNotFound(mv.location_dest_id))?;
806 // The SAME posture the original done-transaction consulted: a `periodic` posture's
807 // move legs retire to `not_applicable` (the real-time post is genuinely not wanted
808 // under the current configuration), exactly like a directive that no longer supplies
809 // the accounts.
810 let posture = self.posting_posture().await?;
811 if posture.periodic {
812 self.moves.mark_posting_not_applicable(&self.rpool(), move_id).await?;
813 return Ok(SubmitOutcome {
814 voucher_id: move_id, posted: false, journal_id: None, post_id: None,
815 gl_amount: Decimal::ZERO,
816 });
817 }
818 let (out_value, in_value) = self.sles.move_leg_values(&self.rpool(), move_id).await?;
819 let envelope = self.move_gl_envelope(&mv, &src, &dst, gl, out_value, in_value, &posture);
820 let Some(env) = envelope else {
821 self.moves.mark_posting_not_applicable(&self.rpool(), move_id).await?;
822 return Ok(SubmitOutcome {
823 voucher_id: move_id, posted: false, journal_id: None, post_id: None,
824 gl_amount: Decimal::ZERO,
825 });
826 };
827 debug_assert!(env.is_balanced());
828 let gl_amount = env.lines.iter().map(|l| l.debit).sum();
829 match sink.post(&env).await {
830 Ok(ack) => {
831 self.moves.mark_posting_posted(&self.rpool(), move_id).await?;
832 Ok(SubmitOutcome {
833 voucher_id: move_id, posted: true,
834 journal_id: Some(ack.journal_id), post_id: Some(ack.post_id),
835 gl_amount,
836 })
837 }
838 Err(rej) => {
839 let _ = self.moves.mark_posting_failed(&self.rpool(), move_id).await;
840 Err(InventoryError::GlRejected { code: rej.code, message: rej.message })
841 }
842 }
843 }
844
845 // ---- valuation core (private) -----------------------------------------------
846
847 /// The Bin/SLE half of `_action_done` — the moving-average engine over the move's legs, with
848 /// the W1-proven arithmetic (receipt blends, delivery consumes, transfer carries; residual
849 /// flush when a bin drains to zero) and the V7 ordering (OUT valued from the locked src
850 /// snapshot BEFORE the dest reblend). Returns `(sle_count, out_value, in_value)`.
851 ///
852 /// Skips valuation entirely (returns zeros) when neither side resolves to an internal
853 /// warehouse bin, and for same-warehouse moves (the bin grain is (item, warehouse): a move
854 /// inside one warehouse changes no balance — the quant flips above carry the physical truth).
855 async fn mint_move_valuation(
856 &self,
857 tx: &mut sqlx::PgConnection,
858 mv: &crate::infrastructure::persistence::MoveRow,
859 src: &crate::infrastructure::persistence::LocationFacts,
860 dst: &crate::infrastructure::persistence::LocationFacts,
861 qty: Decimal,
862 ) -> Result<(i32, Decimal, Decimal), InventoryError> {
863 let src_wh = if src.usage == "internal" { src.warehouse_id } else { None };
864 let dst_wh = if dst.usage == "internal" { dst.warehouse_id } else { None };
865 if src_wh.is_none() && dst_wh.is_none() { return Ok((0, Decimal::ZERO, Decimal::ZERO)); }
866 if let (Some(a), Some(b)) = (src_wh, dst_wh) {
867 if a == b { return Ok((0, Decimal::ZERO, Decimal::ZERO)); }
868 }
869 let posting_date = Utc::now().date_naive();
870 let mut sle_no = self.sles.fetch_max_sle_no(tx, "stock_entry", mv.id).await?;
871
872 // Lock BOTH bins in canonical warehouse order (the deadlock rule proven by the transfer
873 // path: opposing moves on the same pair serialize instead of deadlocking).
874 let mut legs: Vec<(Uuid, crate::infrastructure::persistence::BinBalanceRow)> = Vec::new();
875 for wh in [src_wh, dst_wh].into_iter().flatten() {
876 let bal = self.bins.lock_or_init(tx, mv.item_id, wh).await?;
877 legs.push((wh, bal));
878 }
879 // `BinBalanceRow` is not Clone; `Decimal` is Copy, so reconstruct on lookup.
880 let balance_of = |wh: Uuid| {
881 legs.iter().find(|(w, _)| *w == wh).map(|(_, b)| crate::infrastructure::persistence::BinBalanceRow {
882 actual_qty: b.actual_qty, valuation_rate: b.valuation_rate, stock_value: b.stock_value,
883 })
884 };
885
886 let mut out_value = Decimal::ZERO;
887 let mut in_value = Decimal::ZERO;
888
889 // OUT leg (V7: valued BEFORE the IN reblend, at the src bin's pre-move average).
890 if let (Some(wh), Some(bal)) = (src_wh, src_wh.and_then(balance_of)) {
891 let out_qty = bal.actual_qty - qty;
892 if out_qty < Decimal::ZERO {
893 return Err(InventoryError::InsufficientStock {
894 item_id: mv.item_id, warehouse_id: wh,
895 available: bal.actual_qty, requested: qty,
896 });
897 }
898 // A forced value (a document reversal at its original amount) overrides the
899 // average/drain-derived carry — and reblends the remaining rate, because the
900 // remaining estate's value changed by an amount the old average does not describe.
901 // Ordinary moves: residual-flush rule (draining the bin to 0 carries its entire
902 // remaining value) or the current 2dp-rounded average.
903 let forced = mv.forced_value.is_some();
904 out_value = match mv.forced_value {
905 Some(fv) => fv,
906 None if out_qty.is_zero() => bal.stock_value,
907 None => money(qty * bal.valuation_rate),
908 };
909 let out_stock_value = bal.stock_value - out_value;
910 let out_rate = if out_qty > Decimal::ZERO {
911 if forced { rate6(out_stock_value / out_qty) } else { bal.valuation_rate }
912 } else { Decimal::ZERO };
913 self.bins.update_balance(tx, mv.item_id, wh, out_qty, out_rate, out_stock_value).await?;
914 sle_no += 1;
915 self.sles.insert_sle(tx, &crate::infrastructure::persistence::NewSleRow {
916 item_id: mv.item_id, warehouse_id: wh, posting_date,
917 actual_qty: -qty, qty_after_txn: out_qty, incoming_rate: Decimal::ZERO,
918 valuation_rate: out_rate, stock_value: out_stock_value,
919 stock_value_difference: -out_value,
920 voucher_type: "stock_entry", voucher_id: mv.id, voucher_no: &mv.name, sle_no,
921 }).await?;
922 }
923
924 // IN leg (V7: valued AFTER — reblends the destination with the carried value).
925 if let (Some(wh), Some(bal)) = (dst_wh, dst_wh.and_then(balance_of)) {
926 // Carried value: a forced value (a document reversal at its original amount)
927 // overrides everything; an external source (receipt) values the inflow at the
928 // move's unit price; an internal transfer carries the OUT value. A zero price on
929 // an external receive carries the destination's current average (value-neutral
930 // receive).
931 let carried = if let Some(fv) = mv.forced_value {
932 fv
933 } else if src.usage != "internal" {
934 if mv.price_unit > Decimal::ZERO { money(qty * mv.price_unit) } else { money(qty * bal.valuation_rate) }
935 } else {
936 out_value
937 };
938 in_value = carried;
939 let in_qty = bal.actual_qty + qty;
940 let in_stock_value = bal.stock_value + carried;
941 let in_rate = if in_qty > Decimal::ZERO { rate6(in_stock_value / in_qty) } else { Decimal::ZERO };
942 self.bins.update_balance(tx, mv.item_id, wh, in_qty, in_rate, in_stock_value).await?;
943 sle_no += 1;
944 self.sles.insert_sle(tx, &crate::infrastructure::persistence::NewSleRow {
945 item_id: mv.item_id, warehouse_id: wh, posting_date,
946 actual_qty: qty, qty_after_txn: in_qty, incoming_rate: mv.price_unit,
947 valuation_rate: in_rate, stock_value: in_stock_value,
948 stock_value_difference: carried,
949 voucher_type: "stock_entry", voucher_id: mv.id, voucher_no: &mv.name, sle_no,
950 }).await?;
951 }
952
953 Ok((sle_no, out_value, in_value))
954 }
955
956 // ---- landed-cost revaluation (the one non-move valuation write) --------------------------
957
958 /// Revalue a DONE receipt move's bin by `delta`: ONE value-only SLE row
959 /// (`voucher_type='landed_cost'`, `voucher_id` = the landed cost, qty 0, value `delta`)
960 /// plus the Bin value reblend (`value += delta; qty unchanged; rate = value/qty`).
961 ///
962 /// **RETROACTIVE-REVALUATION ASYMMETRY — deliberate, preserved.** `delta` is the
963 /// REMAINING-share portion of the move's landed-cost allocation
964 /// (`delta = allocation x remaining_qty / done_qty`, computed by the landed-cost door from a
965 /// read-only FIFO attribution over the SLE history). The already-CONSUMED portion of the
966 /// allocation produces NO correcting entry on this path — no COGS true-up, no journal leg,
967 /// no negative-SLE compensation — even though the consumed units also "cost more" now. This
968 /// one-sided revalue of already-done moves is the reference ERP's behavior and the
969 /// valuation-overlay plan row records it as an explicit keep: the consumed share's cost is
970 /// accepted as recognized-at-the-old-average at consumption time, and only the stock still
971 /// on hand revalues. Both GL sides of the landed-cost post carry only the remaining-portion
972 /// value so the post balances without inventing the missing COGS leg. Reversal is the
973 /// negative-amount landed-cost pattern (swapped legs verbatim), never a cancel.
974 ///
975 /// The one-writer invariant holds: this verb is the ONLY non-engine-internal caller path
976 /// that reaches `insert_sle`/`update_balance`, and it lives HERE, inside the engine, beside
977 /// [`Self::mint_move_valuation`] — the landed-cost door calls the verb, never the writers.
978 /// The engine posts NO GL for this leg (the landed-cost document owns its own envelope).
979 ///
980 /// Crash-resume: the SLE row's name is deterministic
981 /// (`{lc_number}/{move_name}/{move_line_id}` — the move line id is the stable target grain;
982 /// worksheet row ids are transient and must not appear in ledger names), so a re-drive of a
983 /// crashed validation finds the row via [`Self::has_sle_named`] semantics and skips it,
984 /// exactly like a door resuming a half-landed submit. Takes the CALLER'S connection: the
985 /// revaluation SLE, the bin reblend and the landed-cost document's state flip commit as one
986 /// unit. Returns `Ok(())` without minting when `delta` is zero (nothing to revalue) or the
987 /// row already landed.
988 pub(super) async fn adjust_move_value(
989 &self,
990 tx: &mut sqlx::PgConnection,
991 mv: &crate::infrastructure::persistence::MoveRow,
992 move_line_id: Uuid,
993 delta: Decimal,
994 lc_id: Uuid,
995 lc_number: &str,
996 ) -> Result<(), InventoryError> {
997 if delta.is_zero() {
998 return Ok(()); // an all-consumed target line revalues nothing
999 }
1000 // The bin grain is (item, warehouse): resolve the warehouse the move's IN leg blended
1001 // into — the destination location's warehouse, the same resolution mint_move_valuation
1002 // used, so the reblend lands on the bin the original inflow created.
1003 let locs = self.moves.fetch_move_locations(tx, mv.location_id, mv.location_dest_id).await?;
1004 let dst = locs.1.ok_or(InventoryError::LocationNotFound(mv.location_dest_id))?;
1005 if dst.usage != "internal" {
1006 // The target set is the receipt's DONE moves that blended value into a bin; a move
1007 // whose destination is not an internal warehouse minted no valuation to revalue.
1008 return Err(InventoryError::LandedCostNoValuationAccount { move_id: mv.id });
1009 }
1010 let wh = dst.warehouse_id
1011 .ok_or(InventoryError::LandedCostNoValuationAccount { move_id: mv.id })?;
1012 let name = format!("{}/{}/{}", lc_number, mv.name, move_line_id);
1013 if self.sles.has_sle_named(tx, "landed_cost", lc_id, &name).await? {
1014 return Ok(()); // this leg already landed in a prior (crashed) attempt — resume, never double-mint
1015 }
1016 let bal = self.bins.lock_or_init(tx, mv.item_id, wh).await?;
1017 let new_value = bal.stock_value + delta;
1018 let new_qty = bal.actual_qty; // a landed cost moves no quantity, only value
1019 let new_rate = if new_qty > Decimal::ZERO {
1020 rate6(new_value / new_qty)
1021 } else {
1022 Decimal::ZERO
1023 };
1024 self.bins.update_balance(tx, mv.item_id, wh, new_qty, new_rate, new_value).await?;
1025 let sle_no = self.sles.fetch_max_sle_no(tx, "landed_cost", lc_id).await? + 1;
1026 self.sles.insert_sle(tx, &crate::infrastructure::persistence::NewSleRow {
1027 item_id: mv.item_id, warehouse_id: wh,
1028 posting_date: chrono::Utc::now().date_naive(),
1029 actual_qty: Decimal::ZERO, qty_after_txn: new_qty,
1030 incoming_rate: Decimal::ZERO, valuation_rate: new_rate,
1031 stock_value: new_value, stock_value_difference: delta,
1032 voucher_type: "landed_cost", voucher_id: lc_id, voucher_no: &name, sle_no,
1033 }).await?;
1034 Ok(())
1035 }
1036
1037 /// Build the GL envelope for the done move's leg shape, or `None` when the shape posts no GL
1038 /// (cross-warehouse internal transfer: value-neutral), the directive lacks the accounts, the
1039 /// `periodic` posture suppresses real-time stock posts, or the move carries neither value
1040 /// nor quantity (the explicit account-move gate).
1041 fn move_gl_envelope(
1042 &self,
1043 mv: &crate::infrastructure::persistence::MoveRow,
1044 src: &crate::infrastructure::persistence::LocationFacts,
1045 dst: &crate::infrastructure::persistence::LocationFacts,
1046 gl: &MoveGlDirective,
1047 out_value: Decimal,
1048 in_value: Decimal,
1049 posture: &Posture,
1050 ) -> Option<AccountingPostEnvelope> {
1051 // A `periodic` posture posts no real-time stock GL — the closing flow (a later
1052 // increment) owns those legs; the move stays `not_applicable`.
1053 if posture.periodic {
1054 return None;
1055 }
1056 // The explicit account-move gate (`_should_create_account_move` port): no envelope
1057 // when the move carries zero value AND zero qty, or the accounts its leg shape
1058 // requires are unset — both stay `not_applicable`.
1059 let shape_accounts_set = if mv.is_inventory {
1060 gl.inventory_account_id.is_some() && gl.adjustment_account_id.is_some()
1061 } else if src.usage == "internal" && dst.usage != "internal" {
1062 gl.cogs_account_id.is_some() && gl.inventory_account_id.is_some()
1063 } else if src.usage != "internal" && dst.usage == "internal" {
1064 gl.inventory_account_id.is_some() && gl.grir_account_id.is_some()
1065 } else {
1066 false // internal cross-warehouse: value-neutral, never posts
1067 };
1068 if !Self::should_create_account_move(out_value + in_value, mv.quantity, shape_accounts_set) {
1069 return None;
1070 }
1071 // The account-resolution chain for the inventory legs: the location's
1072 // valuation-account override when set, else the directive's account (the
1073 // door-header equivalent). Smallest-first override: a location beats the directive.
1074 let (description, lines) = if mv.is_inventory {
1075 // Adjustment shape (reconciliation vocabulary): the signed value diff between the
1076 // inventory account and the adjustment counterleg. The inventory leg resolves
1077 // the override of whichever side is the internal (counted) location.
1078 let inv = if dst.usage == "internal" { dst.valuation_account_id } else { src.valuation_account_id }
1079 .or(gl.inventory_account_id)?;
1080 let adj = gl.adjustment_account_id?;
1081 if in_value > out_value {
1082 ("Inventory adjustment".to_string(), vec![
1083 GlPostLine::debit(inv, in_value - out_value).with_description("Inventory"),
1084 GlPostLine::credit(adj, in_value - out_value).with_description("Adjustment"),
1085 ])
1086 } else if out_value > in_value {
1087 ("Inventory adjustment".to_string(), vec![
1088 GlPostLine::debit(adj, out_value - in_value).with_description("Adjustment"),
1089 GlPostLine::credit(inv, out_value - in_value).with_description("Inventory"),
1090 ])
1091 } else {
1092 return None;
1093 }
1094 } else if src.usage == "internal" && dst.usage != "internal" {
1095 // OUT (delivery shape): Dr COGS / Cr Inventory — the W1-proven leg. The inventory
1096 // leg prefers the SOURCE location's valuation-account override.
1097 ("Stock issue".to_string(), vec![
1098 GlPostLine::debit(gl.cogs_account_id?, out_value).with_description("COGS"),
1099 GlPostLine::credit(src.valuation_account_id.or(gl.inventory_account_id)?, out_value).with_description("Inventory"),
1100 ])
1101 } else if src.usage != "internal" && dst.usage == "internal" {
1102 // IN (receipt shape): Dr Inventory / Cr GR/IR. The inventory leg prefers the
1103 // DESTINATION location's valuation-account override.
1104 ("Goods receipt".to_string(), vec![
1105 GlPostLine::debit(dst.valuation_account_id.or(gl.inventory_account_id)?, in_value).with_description("Inventory"),
1106 GlPostLine::credit(gl.grir_account_id?, in_value).with_description("GR/IR clearing"),
1107 ])
1108 } else {
1109 // Internal cross-warehouse: value-neutral, no GL (the transfer-path contract).
1110 return None;
1111 };
1112 Some(AccountingPostEnvelope {
1113 idempotency_key: mv.id.to_string(),
1114 // Legacy twin (ADR-0029): filled from the ambient org scope's company echo for
1115 // consumers that still read a tenant off the wire. No module statement keys on it.
1116 company_id: legacy_company_echo(),
1117 branch_id: None,
1118 source_type: "inventory".into(),
1119 source_id: mv.id,
1120 source_reference: Some(mv.name.clone()),
1121 posting_date: Utc::now().date_naive(),
1122 currency: if gl.currency.is_empty() { "IDR".into() } else { gl.currency.clone() },
1123 posting_type: "original".into(),
1124 reverses_post_id: None,
1125 description: Some(description),
1126 lines,
1127 })
1128 }
1129}
1130
1131// --- the scheduler's MovePipeline port ------------------------------------------
1132
1133/// The write service IS the move engine, so it implements the port the daily scheduler drives
1134/// (`procurement_service::MovePipeline`): the per-move confirm/assign verbs on the CALLER'S
1135/// connection — the scheduler runs them inside its per-batch transaction, so a batch's
1136/// transitions commit or roll back together. Failures cross the boundary as a flat
1137/// `{code, message}` (the engine's own error taxonomy stays on this side of the port). Events
1138/// publish immediately through the advisory sink; the batch commit boundary is the scheduler's
1139/// concern.
1140#[async_trait]
1141impl MovePipeline for InventoryWriteService {
1142 async fn confirm(
1143 &self,
1144 conn: &mut sqlx::PgConnection,
1145 move_id: Uuid,
1146 ) -> Result<MoveState, MovePipelineError> {
1147 let pipe = |e: InventoryError| MovePipelineError { code: e.code(), message: e.to_string() };
1148 let mv = self
1149 .moves
1150 .fetch_move(conn, move_id)
1151 .await
1152 .map_err(|e| pipe(InventoryError::Db(e)))?
1153 .ok_or_else(|| pipe(InventoryError::NotFound(move_id)))?;
1154 if mv.state != "draft" {
1155 return Err(pipe(InventoryError::WrongMoveState { move_id, action: "confirm", current: mv.state }));
1156 }
1157 relay_ambient_scope(conn).await.map_err(|e| pipe(InventoryError::Db(e)))?;
1158 let to = self.confirm_core(conn, &mv).await.map_err(pipe)?;
1159 if to == "confirmed" {
1160 self.sink.publish(InventoryEvent::MoveConfirmed(MoveConfirmed {
1161 move_id,
1162 company_id: legacy_company_echo(),
1163 item_id: mv.item_id,
1164 demand_qty: mv.demand_qty,
1165 picking_id: mv.picking_id,
1166 }));
1167 }
1168 to.parse::<MoveState>()
1169 .map_err(|e| MovePipelineError { code: "bad_move_state".into(), message: e })
1170 }
1171
1172 async fn assign(
1173 &self,
1174 conn: &mut sqlx::PgConnection,
1175 move_id: Uuid,
1176 ) -> Result<MoveState, MovePipelineError> {
1177 let pipe = |e: InventoryError| MovePipelineError { code: e.code(), message: e.to_string() };
1178 let mv = self
1179 .moves
1180 .fetch_move(conn, move_id)
1181 .await
1182 .map_err(|e| pipe(InventoryError::Db(e)))?
1183 .ok_or_else(|| pipe(InventoryError::NotFound(move_id)))?;
1184 if mv.state != "confirmed" && mv.state != "partially_available" {
1185 return Err(pipe(InventoryError::WrongMoveState { move_id, action: "assign", current: mv.state }));
1186 }
1187 relay_ambient_scope(conn).await.map_err(|e| pipe(InventoryError::Db(e)))?;
1188 let (to, _reserved) = self.assign_core(conn, &mv).await.map_err(pipe)?;
1189 if to == "assigned" {
1190 self.sink.publish(InventoryEvent::MoveAssigned(MoveAssigned {
1191 move_id, company_id: legacy_company_echo(), picking_id: mv.picking_id,
1192 }));
1193 }
1194 to.parse::<MoveState>()
1195 .map_err(|e| MovePipelineError { code: "bad_move_state".into(), message: e })
1196 }
1197}