backbone_inventory/infrastructure/persistence/
stock_move_repository.rs

1//! Repository for StockMove entities
2//!
3//! Originally generated by metaphor-schema; now **user-owned** — this exact path is declared under
4//! `user_owned` in `metaphor.codegen.yaml`, so the generator skips it wholesale. The custom methods
5//! below hold the hand-written stock-move SQL: the guarded 7-state lifecycle writes
6//! (draft/waiting/confirmed/partially_available/assigned/done/cancel — spec stock-business-logic
7//! §1), chain linkage over `move_orig_ids`/`move_dest_ids`, and the T1 picking-state projection
8//! recompute (the transfer's state is a STORED COMPUTE over its moves — never a machine of its
9//! own; this repo re-derives it from the move rows after every state change).
10//!
11//! 4-layer rule: services orchestrate, repositories hold SQL. Write methods take the CALLER'S
12//! connection (the movement's transaction); the caller has already relayed the ambient org
13//! scope onto it. State values bind as `$n::move_state` (unqualified — the enum type is created
14//! unqualified in migrations) and decode through `::text`.
15
16use rust_decimal::Decimal;
17use sqlx::{PgConnection, PgPool, Row};
18use uuid::Uuid;
19
20use backbone_orm::org_scope;
21
22use crate::domain::entity::StockMove;
23
24use super::picking_batch_projection_repository::PickingBatchProjectionRepository;
25
26/// Table name for StockMove entities
27pub const TABLE_NAME: &str = "inventory.stock_moves";
28
29/// Repository for StockMove entities.
30///
31/// All standard CRUD, soft-delete, pagination, and bulk methods are
32/// provided automatically via `Deref` to `backbone_orm::GenericCrudRepository`.
33pub struct StockMoveRepository(
34    backbone_orm::GenericCrudRepository<StockMove, backbone_orm::SoftDelete>,
35);
36
37impl std::ops::Deref for StockMoveRepository {
38    type Target = backbone_orm::GenericCrudRepository<StockMove, backbone_orm::SoftDelete>;
39    fn deref(&self) -> &Self::Target { &self.0 }
40}
41
42impl StockMoveRepository {
43    /// Create a new repository instance.
44    pub fn new(pool: PgPool) -> Self {
45        Self(backbone_orm::GenericCrudRepository::new(pool, TABLE_NAME))
46    }
47}
48
49/// The exact row a draft move insert writes. `state` starts `draft`; the pipeline
50/// (`_action_confirm` / `_action_assign` / `_action_done` / `_action_cancel`) is the only writer
51/// of any other state.
52pub struct NewMoveRow<'a> {
53    pub id: Uuid,
54    pub name: &'a str,
55    pub item_id: Uuid,
56    pub demand_qty: Decimal,
57    pub price_unit: Decimal,
58    /// "make_to_stock" | "make_to_order" | "mts_else_mto".
59    pub procure_method: &'a str,
60    pub picking_id: Option<Uuid>,
61    pub origin: Option<&'a str>,
62    pub location_id: Uuid,
63    pub location_dest_id: Uuid,
64    pub partner_id: Option<Uuid>,
65    pub warehouse_id: Option<Uuid>,
66    pub orderpoint_id: Option<Uuid>,
67    pub move_orig_ids: Vec<Uuid>,
68    pub move_dest_ids: Vec<Uuid>,
69    pub is_inventory: bool,
70    pub scrapped: bool,
71    /// Explicit total value the move carries — a document reversal valued at its original
72    /// amount. `None` on ordinary moves (the valuation core derives the carry).
73    pub forced_value: Option<Decimal>,
74}
75
76/// A move row as the engine reads it. `state` / `procure_method` decode as text (the enum lives in
77/// the DB; string comparison against the snake_case vocabulary is the module's proven pattern).
78#[derive(Clone)]
79pub struct MoveRow {
80    pub id: Uuid,
81    pub name: String,
82    pub state: String,
83    /// GL settlement state of the move's OWN AccountingPost leg, decoded as text
84    /// ("not_applicable" | "pending" | "posted" | "failed"). Distinct from the lifecycle
85    /// `state`: a `done` move can carry a `failed` GL leg (the physical movement stands,
86    /// the GL leg is missing and re-drivable).
87    pub posting_state: String,
88    pub date: chrono::DateTime<chrono::Utc>,
89    pub item_id: Uuid,
90    pub demand_qty: Decimal,
91    pub quantity: Decimal,
92    pub price_unit: Decimal,
93    pub procure_method: String,
94    pub picking_id: Option<Uuid>,
95    pub origin: Option<String>,
96    /// The procurement rule that minted this move (the `_run_pull` / `_run_push` provenance —
97    /// moves minted through a rule carry it, door-minted and hand-minted moves do not).
98    pub rule_id: Option<Uuid>,
99    pub location_id: Uuid,
100    pub location_dest_id: Uuid,
101    pub partner_id: Option<Uuid>,
102    pub warehouse_id: Option<Uuid>,
103    pub orderpoint_id: Option<Uuid>,
104    pub move_orig_ids: Vec<Uuid>,
105    pub move_dest_ids: Vec<Uuid>,
106    pub is_inventory: bool,
107    pub scrapped: bool,
108    pub propagate_cancel: bool,
109    /// Explicit total value the move carries (a document reversal valued at its original
110    /// amount); `None` on ordinary moves.
111    pub forced_value: Option<Decimal>,
112}
113
114/// Hand-written stock-move SQL — the lifecycle writes. Guarded: state transitions carry the
115/// expected source state in their WHERE clause, so a stale read cannot silently overwrite a move
116/// another caller already advanced (the optimistic guard behind the declared state machine in
117/// schema/hooks/stock.hook.yaml; ADR-0015 service pre-check + the WHERE clause as the backstop).
118impl StockMoveRepository {
119    /// Insert a draft move.
120    pub async fn insert_move(
121        &self,
122        conn: &mut sqlx::PgConnection,
123        m: &NewMoveRow<'_>,
124    ) -> Result<(), sqlx::Error> {
125        sqlx::query(
126            r#"INSERT INTO inventory.stock_moves
127                 (id, name, state, item_id, demand_qty, price_unit, procure_method, picking_id,
128                  origin, location_id, location_dest_id, partner_id, warehouse_id,
129                  orderpoint_id, move_orig_ids, move_dest_ids, is_inventory, scrapped, forced_value)
130               VALUES ($1,$2,'draft'::move_state,$3,$4,$5,$6::procure_method,$7,$8,$9,$10,$11,
131                       $12,$13,$14,$15,$16,$17,$18)"#,
132        )
133        .bind(m.id).bind(m.name).bind(m.item_id).bind(m.demand_qty).bind(m.price_unit)
134        .bind(m.procure_method).bind(m.picking_id).bind(m.origin)
135        .bind(m.location_id).bind(m.location_dest_id).bind(m.partner_id)
136        .bind(m.warehouse_id).bind(m.orderpoint_id)
137        .bind(&m.move_orig_ids).bind(&m.move_dest_ids)
138        .bind(m.is_inventory).bind(m.scrapped).bind(m.forced_value)
139        .execute(conn)
140        .await?;
141        Ok(())
142    }
143
144    /// Fetch one move. Returns `None` when absent or soft-deleted.
145    pub async fn fetch_move(
146        &self,
147        conn: &mut sqlx::PgConnection,
148        move_id: Uuid,
149    ) -> Result<Option<MoveRow>, sqlx::Error> {
150        let row = sqlx::query(
151            r#"SELECT id, name, state::text AS state, posting_state::text AS posting_state, date, item_id, demand_qty, quantity,
152                      price_unit, procure_method::text AS procure_method, picking_id, origin, rule_id,
153                      location_id, location_dest_id, partner_id, warehouse_id,
154                      orderpoint_id, move_orig_ids, move_dest_ids, is_inventory, scrapped,
155                      propagate_cancel, forced_value
156               FROM inventory.stock_moves
157               WHERE id=$1 AND (metadata->>'deleted_at') IS NULL"#,
158        )
159        .bind(move_id)
160        .fetch_optional(&mut *conn)
161        .await?;
162        Ok(row.map(move_row_of))
163    }
164
165    /// Fetch several moves by id (chain propagation). Order follows the input ids.
166    pub async fn fetch_moves(
167        &self,
168        conn: &mut sqlx::PgConnection,
169        move_ids: &[Uuid],
170    ) -> Result<Vec<MoveRow>, sqlx::Error> {
171        let rows = sqlx::query(
172            r#"SELECT id, name, state::text AS state, posting_state::text AS posting_state, date, item_id, demand_qty, quantity,
173                      price_unit, procure_method::text AS procure_method, picking_id, origin, rule_id,
174                      location_id, location_dest_id, partner_id, warehouse_id,
175                      orderpoint_id, move_orig_ids, move_dest_ids, is_inventory, scrapped,
176                      propagate_cancel, forced_value
177               FROM inventory.stock_moves
178               WHERE id = ANY($1) AND (metadata->>'deleted_at') IS NULL"#,
179        )
180        .bind(move_ids)
181        .fetch_all(&mut *conn)
182        .await?;
183        Ok(rows.into_iter().map(move_row_of).collect())
184    }
185
186    /// Fetch the moves minted for one voucher (`origin` — voucher doors stamp the
187    /// voucher number there so a crashed submit can find and resume its line moves).
188    pub async fn fetch_moves_by_origin(
189        &self,
190        conn: &mut sqlx::PgConnection,
191        origin: &str,
192    ) -> Result<Vec<MoveRow>, sqlx::Error> {
193        let rows = sqlx::query(
194            r#"SELECT id, name, state::text AS state, posting_state::text AS posting_state, date, item_id, demand_qty, quantity,
195                      price_unit, procure_method::text AS procure_method, picking_id, origin, rule_id,
196                      location_id, location_dest_id, partner_id, warehouse_id,
197                      orderpoint_id, move_orig_ids, move_dest_ids, is_inventory, scrapped,
198                      propagate_cancel, forced_value
199               FROM inventory.stock_moves
200               WHERE origin=$1 AND (metadata->>'deleted_at') IS NULL
201               ORDER BY name"#,
202        )
203        .bind(origin)
204        .fetch_all(&mut *conn)
205        .await?;
206        Ok(rows.into_iter().map(move_row_of).collect())
207    }
208
209    /// Guarded state transition: `UPDATE ... WHERE id=$1 AND state=$expected`. Returns Ok(false)
210    /// when zero rows matched — the move is not in the expected state (the caller maps that to a
211    /// domain error). This is the declared state machine's write valve: states only move through
212    /// here with their declared source.
213    pub async fn transition_state(
214        &self,
215        conn: &mut sqlx::PgConnection,
216        move_id: Uuid,
217        expected_from: &str,
218        to: &str,
219    ) -> Result<bool, sqlx::Error> {
220        let res = sqlx::query(
221            r#"UPDATE inventory.stock_moves
222               SET state = $3::move_state
223               WHERE id=$1 AND state = $2::move_state AND (metadata->>'deleted_at') IS NULL"#,
224        )
225        .bind(move_id).bind(expected_from).bind(to)
226        .execute(&mut *conn)
227        .await?;
228        Ok(res.rows_affected() == 1)
229    }
230
231    /// The done write: `state='done'`, the done quantity, and the effective-date stamp (the
232    /// `date` column is the schema's effective-date surface — set to the processing instant on
233    /// done; there is no separate `date_done` column on moves). Guarded on the expected source
234    /// state like [`Self::transition_state`].
235    pub async fn mark_done(
236        &self,
237        conn: &mut sqlx::PgConnection,
238        move_id: Uuid,
239        expected_from: &str,
240        done_qty: Decimal,
241        when: chrono::DateTime<chrono::Utc>,
242    ) -> Result<bool, sqlx::Error> {
243        let res = sqlx::query(
244            r#"UPDATE inventory.stock_moves
245               SET state='done'::move_state, quantity=$3, date=$4
246               WHERE id=$1 AND state = $2::move_state AND (metadata->>'deleted_at') IS NULL"#,
247        )
248        .bind(move_id).bind(expected_from).bind(done_qty).bind(when)
249        .execute(&mut *conn)
250        .await?;
251        Ok(res.rows_affected() == 1)
252    }
253
254    /// Attach a move to its grouping transfer (the picking-assignment write: the move's
255    /// `picking_id` is set, and the projection re-derives on the caller's next reproject —
256    /// this write only moves the pointer). Runs inside the caller's transaction; the ambient
257    /// org scope is already relayed onto it by the service that resolved the group.
258    pub async fn set_picking(
259        &self,
260        conn: &mut sqlx::PgConnection,
261        move_id: Uuid,
262        picking_id: Uuid,
263    ) -> Result<(), sqlx::Error> {
264        sqlx::query(
265            r#"UPDATE inventory.stock_moves
266               SET picking_id = $2
267               WHERE id = $1 AND (metadata->>'deleted_at') IS NULL"#,
268        )
269        .bind(move_id)
270        .bind(picking_id)
271        .execute(&mut *conn)
272        .await?;
273        Ok(())
274    }
275
276    /// Link a backorder child to its parent: the parent's `move_dest_ids` gains the child, the
277    /// child's `move_orig_ids` gains the parent (the chain vocabulary `move_orig_ids`/
278    /// `move_dest_ids` — spec §1/§4; done-qty propagation and `waiting` release walk these).
279    pub async fn link_chain(
280        &self,
281        conn: &mut sqlx::PgConnection,
282        parent_id: Uuid,
283        child_id: Uuid,
284    ) -> Result<(), sqlx::Error> {
285        sqlx::query(
286            r#"UPDATE inventory.stock_moves
287               SET move_dest_ids = array_append(move_dest_ids, $2)
288                   WHERE id=$1 AND NOT ($2 = ANY(move_dest_ids))"#,
289        )
290        .bind(parent_id).bind(child_id)
291        .execute(&mut *conn)
292        .await?;
293        sqlx::query(
294            r#"UPDATE inventory.stock_moves
295               SET move_orig_ids = array_append(move_orig_ids, $2)
296                   WHERE id=$1 AND NOT ($2 = ANY(move_orig_ids))"#,
297        )
298        .bind(child_id).bind(parent_id)
299        .execute(&mut *conn)
300        .await?;
301        Ok(())
302    }
303}
304
305/// The GL-leg settlement writes (the move's OWN AccountingPost outcome — the same posture the
306/// voucher doors keep on their headers). The `pending` marker is written INSIDE the done
307/// transaction, atomically with the physical movement; the `posted`/`failed` reconciles run on
308/// the pool AFTER the movement commits (the GL leg is eventually consistent — a crash between
309/// commit and the reconcile leaves `pending`, which the repost verb re-drives; accounting
310/// dedupes on `(source_type, source_id, posting_type)` so a re-drive can never double post).
311impl StockMoveRepository {
312    /// Arm the GL leg: `posting_state = pending`, inside the movement's own transaction. Called
313    /// by the done path only when it actually built an envelope for this move — moves that post
314    /// no GL of their own never leave `not_applicable`.
315    pub async fn set_posting_pending(
316        &self,
317        conn: &mut PgConnection,
318        move_id: Uuid,
319    ) -> Result<(), sqlx::Error> {
320        sqlx::query(
321            r#"UPDATE inventory.stock_moves
322               SET posting_state='pending'::gl_posting_state
323               WHERE id=$1"#,
324        )
325        .bind(move_id)
326        .execute(&mut *conn)
327        .await?;
328        Ok(())
329    }
330
331    /// The pending→posted reconcile from the accounting ack. Guarded on
332    /// `posting_state <> 'posted'` so a re-drive that accounting answered from its dedupe
333    /// (returning the ORIGINAL journal) can never flip a settled state.
334    ///
335    /// Runs `org_scope::execute_scoped` on the pool: the write rides the ambient org scope's
336    /// request-dedicated connection (the composing service sets it per request) so it passes
337    /// the decorator's fence (ADR-0029).
338    pub async fn mark_posting_posted(
339        &self,
340        pool: &PgPool,
341        move_id: Uuid,
342    ) -> Result<(), sqlx::Error> {
343        org_scope::execute_scoped(
344            pool,
345            sqlx::query(
346                r#"UPDATE inventory.stock_moves
347                   SET posting_state='posted'::gl_posting_state
348                   WHERE id=$1 AND posting_state <> 'posted'::gl_posting_state"#,
349            )
350            .bind(move_id),
351        )
352        .await?;
353        Ok(())
354    }
355
356    /// Record a GL rejection. The physical movement is NOT rolled back — it really happened;
357    /// the move parks in `failed` and is re-drivable via `repost_move_gl`.
358    ///
359    /// Same org-scoped pool write as [`Self::mark_posting_posted`].
360    pub async fn mark_posting_failed(
361        &self,
362        pool: &PgPool,
363        move_id: Uuid,
364    ) -> Result<(), sqlx::Error> {
365        org_scope::execute_scoped(
366            pool,
367            sqlx::query(
368                r#"UPDATE inventory.stock_moves
369                   SET posting_state='failed'::gl_posting_state
370                   WHERE id=$1"#,
371            )
372            .bind(move_id),
373        )
374        .await?;
375        Ok(())
376    }
377
378    /// Retire an unsettled GL leg back to `not_applicable` — used by the repost verb when the
379    /// envelope can no longer be built (the directive no longer supplies the accounts the leg
380    /// shape needs), so a stuck `pending`/`failed` move whose GL is genuinely not postable under
381    /// the current configuration settles instead of living on the sweep worklist forever. Never
382    /// touches a `posted` move.
383    pub async fn mark_posting_not_applicable(
384        &self,
385        pool: &PgPool,
386        move_id: Uuid,
387    ) -> Result<(), sqlx::Error> {
388        org_scope::execute_scoped(
389            pool,
390            sqlx::query(
391                r#"UPDATE inventory.stock_moves
392                   SET posting_state='not_applicable'::gl_posting_state
393                   WHERE id=$1 AND posting_state IN ('pending'::gl_posting_state, 'failed'::gl_posting_state)"#,
394            )
395            .bind(move_id),
396        )
397        .await?;
398        Ok(())
399    }
400}
401
402/// The location facts a move's guards and valuation core need: usage (supplier/view/internal/
403/// customer/inventory/production/transit — the routing + reservation rules key off it), and the
404/// warehouse a valuation bin resolves to. `usage` decodes as text like every other enum here.
405pub struct LocationFacts {
406    pub id: Uuid,
407    pub usage: String,
408    pub warehouse_id: Option<Uuid>,
409    /// Per-location valuation-account override: when set, an inventory GL leg that touches this
410    /// location resolves to it ahead of the caller-supplied (door-header / directive) account —
411    /// the smallest-first override of the account-resolution chain. `None` = no override.
412    pub valuation_account_id: Option<Uuid>,
413}
414
415/// Location reads the move pipeline owns (guards R9/R13 + the bin-warehouse resolution in
416/// `_action_done`'s valuation core). One round trip for both endpoints.
417impl StockMoveRepository {
418    /// Fetch a move's two endpoint locations. Returns `(src, dest)` as read; `None` for an absent
419    /// or soft-deleted endpoint (the caller maps that to `LocationNotFound`).
420    pub async fn fetch_move_locations(
421        &self,
422        conn: &mut sqlx::PgConnection,
423        location_id: Uuid,
424        location_dest_id: Uuid,
425    ) -> Result<(Option<LocationFacts>, Option<LocationFacts>), sqlx::Error> {
426        let rows = sqlx::query(
427            r#"SELECT id, usage::text AS usage, warehouse_id, valuation_account_id
428               FROM inventory.locations
429               WHERE id IN ($1, $2) AND (metadata->>'deleted_at') IS NULL"#,
430        )
431        .bind(location_id)
432        .bind(location_dest_id)
433        .fetch_all(&mut *conn)
434        .await?;
435        let find = |id: Uuid| {
436            rows.iter().find(|r| r.get::<Uuid, _>("id") == id).map(|r| LocationFacts {
437                id: r.get("id"),
438                usage: r.get("usage"),
439                warehouse_id: r.get("warehouse_id"),
440                valuation_account_id: r.get("valuation_account_id"),
441            })
442        };
443        Ok((find(location_id), find(location_dest_id)))
444    }
445}
446
447/// The T1 picking-state projection (spec §2, ADR-0016 `projection` shape).
448///
449/// `transfers.state` is a STORED COMPUTE over its moves — `_get_relevant_state_among_moves`: the
450/// LEAST-advanced non-cancelled move state (cancelled moves drop out of the aggregation entirely;
451/// all-cancelled shows `cancel`). The picking has no state machine of its own anywhere; every
452/// move-state change re-runs this recompute. The SQL lives on the MOVE repository because the
453/// projection derives entirely from move rows — the transfer merely receives the result.
454impl StockMoveRepository {
455    /// Recompute + write one transfer's projected state from its moves. Returns the new projected
456    /// state (text), or `None` when the transfer has no live moves. Also stamps the transfer's
457    /// `date_done` when the projection first reaches `done`.
458    pub async fn reproject_picking(
459        &self,
460        conn: &mut sqlx::PgConnection,
461        picking_id: Uuid,
462    ) -> Result<Option<String>, sqlx::Error> {
463        let row = sqlx::query(
464            r#"WITH agg AS (
465                   SELECT
466                       COUNT(*) FILTER (WHERE state <> 'cancel'::move_state) AS live_count,
467                       COUNT(*) AS total_count,
468                       MIN(CASE state WHEN 'draft'::move_state THEN 0
469                                      WHEN 'waiting'::move_state THEN 1
470                                      WHEN 'confirmed'::move_state THEN 2
471                                      WHEN 'partially_available'::move_state THEN 2
472                                      WHEN 'assigned'::move_state THEN 3
473                                      WHEN 'done'::move_state THEN 4
474                                      ELSE 5 END) FILTER (WHERE state <> 'cancel'::move_state) AS rank
475                   FROM inventory.stock_moves
476                   WHERE picking_id = $1 AND (metadata->>'deleted_at') IS NULL
477               )
478               UPDATE inventory.transfers t
479               SET state = CASE agg.rank
480                       WHEN 0 THEN 'draft'::transfer_state
481                       WHEN 1 THEN 'waiting'::transfer_state
482                       WHEN 2 THEN 'confirmed'::transfer_state
483                       WHEN 3 THEN 'assigned'::transfer_state
484                       WHEN 4 THEN 'done'::transfer_state
485                       ELSE 'cancel'::transfer_state END,
486                   date_done = CASE WHEN agg.rank = 4 THEN COALESCE(t.date_done, NOW())
487                                    ELSE t.date_done END
488               FROM agg
489               WHERE t.id = $1
490               RETURNING t.state::text"#,
491        )
492        .bind(picking_id)
493        .fetch_optional(&mut *conn)
494        .await?;
495        // SB-1 cascade: the picking's state just changed, so its batch's stored compute
496        // must re-derive in the SAME transaction — the batch never lags its members. The
497        // projection SQL lives on the batch repository (its single owner); the batch
498        // repository is stateless, so a throwaway instance borrows it here. No-ops when
499        // the picking belongs to no batch.
500        if row.is_some() {
501            PickingBatchProjectionRepository::default()
502                .reproject_batch_of_picking(conn, picking_id)
503                .await?;
504        }
505        Ok(row.map(|r| r.get::<String, _>("state")))
506    }
507}
508
509fn move_row_of(r: sqlx::postgres::PgRow) -> MoveRow {
510    MoveRow {
511        id: r.get("id"),
512        name: r.get("name"),
513        state: r.get("state"),
514        posting_state: r.get("posting_state"),
515        date: r.get("date"),
516        item_id: r.get("item_id"),
517        demand_qty: r.get("demand_qty"),
518        quantity: r.get("quantity"),
519        price_unit: r.get("price_unit"),
520        procure_method: r.get("procure_method"),
521        picking_id: r.get("picking_id"),
522        origin: r.get("origin"),
523        rule_id: r.get("rule_id"),
524        location_id: r.get("location_id"),
525        location_dest_id: r.get("location_dest_id"),
526        partner_id: r.get("partner_id"),
527        warehouse_id: r.get("warehouse_id"),
528        orderpoint_id: r.get("orderpoint_id"),
529        move_orig_ids: r.get("move_orig_ids"),
530        move_dest_ids: r.get("move_dest_ids"),
531        is_inventory: r.get("is_inventory"),
532        scrapped: r.get("scrapped"),
533        propagate_cancel: r.get("propagate_cancel"),
534        forced_value: r.get("forced_value"),
535    }
536}
537
538backbone_core::impl_crud_repository!(StockMoveRepository, StockMove, soft_delete);
backbone_inventory/infrastructure/persistence/stock_move_repository.rs

backbone_inventory/infrastructure/persistence/
stock_move_repository.rs
