Skip to main content

backbone_inventory/infrastructure/persistence/
picking_batch_projection_repository.rs

1//! Picking-batch projection SQL (hand-authored, user-owned) — the stock.picking.batch
2//! satellite (SB-1).
3//!
4//! Distinct from the generated `picking_batch_repository.rs` (the GenericCrud newtype the
5//! read routes use): this file owns the BATCH SURFACE'S OWN SQL that is not generic CRUD —
6//!   - the batch header mint (`state` starts at `draft` and is from then on ONLY ever
7//!     written by [`Self::reproject_batch`] — the SB-1 stored compute, spec
8//!     misc-features.md SB / ADR-0016 `projection`),
9//!   - the membership writes (`transfers.batch_id` set/clear + the `had_members`
10//!     discriminator stamp — the only two columns the membership path owns),
11//!   - the SB-1 projection recompute: the batch's state derives from its member
12//!     pickings' states exactly the way the transfer's derives from its moves
13//!     (LEAST-advanced live member; cancelled members drop out; no live member and
14//!     nothing left to pick → `cancel`). The picking-projection cascade in
15//!     `StockMoveRepository::reproject_picking` calls back into
16//!     [`Self::reproject_batch_of_picking`] after every member-state change, so the
17//!     recompute has exactly ONE implementation and ONE writer family — never a second
18//!     writer beside the move engine.
19//!   - the projection PROBE reads (header + member picking states): the projected
20//!     `picking_batches.state` is READ here, never asserted or hand-set.
21//!
22//! State values are snake_case; enum parameters are bound as text with explicit casts and
23//! read back with `::text` (the module-wide enum lesson). Statements live here and take
24//! the caller's connection; services orchestrate.
25
26use sqlx::{PgConnection, Row};
27use uuid::Uuid;
28
29/// Header fields of one picking batch (the projection probe surface).
30#[derive(Debug, Clone)]
31pub struct BatchHeaderRow {
32    pub id: Uuid,
33    pub name: String,
34    pub state: String,
35    pub is_wave: bool,
36    pub user_id: Option<Uuid>,
37    pub scheduled_date: chrono::DateTime<chrono::Utc>,
38    /// True once the batch has ever grouped a picking — the SB-1 discriminator.
39    pub had_members: bool,
40}
41
42/// State of one member picking, as the projection probe reads it.
43#[derive(Debug, Clone)]
44pub struct BatchMemberRow {
45    pub id: Uuid,
46    pub name: String,
47    /// draft / waiting / confirmed / assigned / done / cancel (as text).
48    pub state: String,
49}
50
51/// A batch header insert. `state` starts at `draft`; membership writes and move
52/// transitions reproject — the stored value is always the projection of the member
53/// picking states.
54pub struct NewBatchRow<'a> {
55    pub id: Uuid,
56    pub name: &'a str,
57    pub is_wave: bool,
58    pub user_id: Option<Uuid>,
59}
60
61/// The batch mint/membership/probe SQL. Stateless — every method takes the caller's
62/// connection so the header insert, the membership write, and the projection recompute
63/// commit as one unit.
64#[derive(Default)]
65pub struct PickingBatchProjectionRepository;
66
67impl PickingBatchProjectionRepository {
68    pub fn new() -> Self {
69        Self
70    }
71
72    /// Insert the batch header (a draft grouping point). Leaks the raw `sqlx::Error`
73    /// deliberately so the service can turn a unique violation on the name — the composing
74    /// decorator's org-scoped arbiter — into the typed duplicate-name error.
75    pub async fn insert_batch(
76        &self,
77        conn: &mut PgConnection,
78        b: &NewBatchRow<'_>,
79    ) -> Result<(), sqlx::Error> {
80        sqlx::query(
81            r#"INSERT INTO inventory.picking_batches
82                 (id, name, is_wave, user_id, scheduled_date, state, had_members)
83               VALUES ($1, $2, $3, $4, NOW(), 'draft'::picking_batch_state, FALSE)"#,
84        )
85        .bind(b.id)
86        .bind(b.name)
87        .bind(b.is_wave)
88        .bind(b.user_id)
89        .execute(conn)
90        .await?;
91        Ok(())
92    }
93
94    /// Read one batch header (probe surface — the projected state is READ here, never
95    /// asserted by callers).
96    pub async fn fetch_batch(
97        &self,
98        conn: &mut PgConnection,
99        batch_id: Uuid,
100    ) -> Result<Option<BatchHeaderRow>, sqlx::Error> {
101        let row = sqlx::query(
102            r#"SELECT id, name, state::text AS state, is_wave, user_id,
103                      scheduled_date, had_members
104               FROM inventory.picking_batches
105               WHERE id = $1 AND (metadata->>'deleted_at') IS NULL"#,
106        )
107        .bind(batch_id)
108        .fetch_optional(conn)
109        .await?;
110        Ok(row.map(|r| BatchHeaderRow {
111            id: r.get("id"),
112            name: r.get("name"),
113            state: r.get("state"),
114            is_wave: r.get("is_wave"),
115            user_id: r.get("user_id"),
116            scheduled_date: r.get("scheduled_date"),
117            had_members: r.get("had_members"),
118        }))
119    }
120
121    /// Read a batch's member pickings (the projection inputs, for the probe surface).
122    pub async fn fetch_members(
123        &self,
124        conn: &mut PgConnection,
125        batch_id: Uuid,
126    ) -> Result<Vec<BatchMemberRow>, sqlx::Error> {
127        let rows = sqlx::query(
128            r#"SELECT id, name, state::text AS state
129               FROM inventory.transfers
130               WHERE batch_id = $1 AND (metadata->>'deleted_at') IS NULL
131               ORDER BY scheduled_date, id"#,
132        )
133        .bind(batch_id)
134        .fetch_all(conn)
135        .await?;
136        Ok(rows
137            .iter()
138            .map(|r| BatchMemberRow {
139                id: r.get("id"),
140                name: r.get("name"),
141                state: r.get("state"),
142            })
143            .collect())
144    }
145
146    /// Attach a picking to its batch: set the membership pointer and stamp the SB-1
147    /// discriminator (`had_members` — once true, an empty batch auto-cancels instead of
148    /// falling back to `draft`). The caller reprojects right after; this write only
149    /// moves the pointer.
150    pub async fn add_member(
151        &self,
152        conn: &mut PgConnection,
153        batch_id: Uuid,
154        picking_id: Uuid,
155    ) -> Result<(), sqlx::Error> {
156        sqlx::query(
157            r#"UPDATE inventory.transfers
158               SET batch_id = $2
159               WHERE id = $1 AND (metadata->>'deleted_at') IS NULL"#,
160        )
161        .bind(picking_id)
162        .bind(batch_id)
163        .execute(&mut *conn)
164        .await?;
165        sqlx::query(
166            r#"UPDATE inventory.picking_batches
167               SET had_members = TRUE
168               WHERE id = $1"#,
169        )
170        .bind(batch_id)
171        .execute(conn)
172        .await?;
173        Ok(())
174    }
175
176    /// Detach a picking from its batch (removal is always allowed — a done or cancelled
177    /// picking still leaves its batch). Returns false when the picking was not a member.
178    pub async fn remove_member(
179        &self,
180        conn: &mut PgConnection,
181        batch_id: Uuid,
182        picking_id: Uuid,
183    ) -> Result<bool, sqlx::Error> {
184        let res = sqlx::query(
185            r#"UPDATE inventory.transfers
186               SET batch_id = NULL
187               WHERE id = $1 AND batch_id = $2 AND (metadata->>'deleted_at') IS NULL"#,
188        )
189        .bind(picking_id)
190        .bind(batch_id)
191        .execute(conn)
192        .await?;
193        Ok(res.rows_affected() == 1)
194    }
195
196    /// The SB-1 projection recompute (spec misc-features.md SB, ADR-0016 `projection`
197    /// shape). `picking_batches.state` is a STORED COMPUTE over its member pickings:
198    /// the LEAST-advanced live member band, with cancelled members dropping out of the
199    /// aggregation entirely —
200    ///   live member draft/confirmed → batch `draft` (nothing ready to wave-pick yet)
201    ///   every live member ≥ waiting → `waiting`
202    ///   every live member ≥ assigned → `ready`
203    ///   every live member done     → `done`
204    ///   no live member, members exist (all cancelled) → `cancel`
205    ///   no member at all, the batch HAS grouped one before → `cancel` (auto-cancel on
206    ///       empty — SB-1: an emptied batch must not linger as a live work item)
207    ///   no member at all, never grouped one → `draft` (a freshly minted batch the
208    ///       operator is about to fill)
209    /// A member's `confirmed` counts at the draft band (Odoo's `_compute_state` has no
210    /// `confirmed` in any prefix: a confirmed-but-unreserved picking is not wave-ready).
211    /// Returns the new projected state (text), or `None` when the batch row is gone.
212    pub async fn reproject_batch(
213        &self,
214        conn: &mut PgConnection,
215        batch_id: Uuid,
216    ) -> Result<Option<String>, sqlx::Error> {
217        let row = sqlx::query(
218            r#"WITH agg AS (
219                   SELECT
220                       COUNT(*) FILTER (WHERE state <> 'cancel'::transfer_state) AS live_count,
221                       COUNT(*) AS total_count,
222                       MIN(CASE state WHEN 'draft'::transfer_state THEN 0
223                                      WHEN 'confirmed'::transfer_state THEN 0
224                                      WHEN 'waiting'::transfer_state THEN 1
225                                      WHEN 'assigned'::transfer_state THEN 2
226                                      WHEN 'done'::transfer_state THEN 3
227                                      ELSE 4 END)
228                           FILTER (WHERE state <> 'cancel'::transfer_state) AS rank
229                   FROM inventory.transfers
230                   WHERE batch_id = $1 AND (metadata->>'deleted_at') IS NULL
231               )
232               UPDATE inventory.picking_batches b
233               SET state = CASE
234                       WHEN agg.live_count > 0 THEN CASE agg.rank
235                           WHEN 0 THEN 'draft'::picking_batch_state
236                           WHEN 1 THEN 'waiting'::picking_batch_state
237                           WHEN 2 THEN 'ready'::picking_batch_state
238                           WHEN 3 THEN 'done'::picking_batch_state
239                           ELSE 'draft'::picking_batch_state END
240                       WHEN agg.total_count > 0 THEN 'cancel'::picking_batch_state
241                       WHEN b.had_members THEN 'cancel'::picking_batch_state
242                       ELSE 'draft'::picking_batch_state END
243               FROM agg
244               WHERE b.id = $1
245               RETURNING b.state::text"#,
246        )
247        .bind(batch_id)
248        .fetch_optional(&mut *conn)
249        .await?;
250        Ok(row.map(|r| r.get::<String, _>("state")))
251    }
252
253    /// Read one picking's batch membership. Outer `None` = the transfer row is absent
254    /// (or fenced out — the ambient org scope's fence hides it); inner `None` = the
255    /// picking belongs to no batch. The membership-guard input for the typed
256    /// already-batched refusal.
257    pub async fn picking_membership(
258        &self,
259        conn: &mut PgConnection,
260        picking_id: Uuid,
261    ) -> Result<Option<Option<Uuid>>, sqlx::Error> {
262        let row = sqlx::query_scalar::<_, Option<Uuid>>(
263            r#"SELECT batch_id FROM inventory.transfers
264               WHERE id = $1 AND (metadata->>'deleted_at') IS NULL"#,
265        )
266        .bind(picking_id)
267        .fetch_optional(&mut *conn)
268        .await?;
269        Ok(row)
270    }
271
272    /// The picking-projection cascade entry: after a picking's own state was re-derived,
273    /// re-derive its batch's (if the picking belongs to one). Called by
274    /// `StockMoveRepository::reproject_picking` on every member-state change so the
275    /// batch never lags its members — the picking and its batch roll-up commit in the
276    /// same transaction.
277    pub async fn reproject_batch_of_picking(
278        &self,
279        conn: &mut PgConnection,
280        picking_id: Uuid,
281    ) -> Result<Option<String>, sqlx::Error> {
282        let batch_id: Option<Uuid> = sqlx::query_scalar(
283            r#"SELECT batch_id FROM inventory.transfers
284               WHERE id = $1 AND (metadata->>'deleted_at') IS NULL"#,
285        )
286        .bind(picking_id)
287        .fetch_optional(&mut *conn)
288        .await?
289        .flatten();
290        match batch_id {
291            Some(b) => self.reproject_batch(conn, b).await,
292            None => Ok(None),
293        }
294    }
295}