backbone-inventory 0.17.2

Inventory: stock ledger of record (SLE) + moving-average valuation; emits COGS/asset AccountingPost (Delivery/Receipt -> GL)
Documentation
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
574
575
576
577
578
579
580
581
582
583
584
585
586
587
588
589
590
591
592
593
594
595
596
597
598
599
600
601
602
603
604
605
606
607
608
609
610
611
612
613
614
615
616
617
618
619
620
621
622
623
624
625
626
627
628
629
630
631
632
633
634
635
636
637
638
639
640
641
//! Procurement service (hand-authored, user-owned): `stock.route` / `stock.rule` /
//! `stock.warehouse.orderpoint` — the pull/push procurement configuration and the reorder
//! rung the daily scheduler drives (spec: docs/odoo/inventory/stock/stock-business-logic.md
//! §6 routes/rules, §7 orderpoints + scheduler, §11 R6/R11, §12 T11).
//!
//! What lives here and what deliberately does not:
//!
//! - **Rule selection** (`search_rule`) — highest-sequence active pull-capable rule for a
//!   demand location (§6). Feeds every move this service mints.
//! - **`run_pull` / `run_push`** (Odoo `_run_pull` / `_run_push`) — mint DRAFT moves only.
//!   Confirm/assign/done are the stock-move engine's verbs; the minted row enters the same
//!   lifecycle every other move does. `run_push` is the entry the move engine calls on
//!   `_action_confirm` to propagate a chain step (§1).
//! - **Orderpoint computes** (T11) — on hand / forecast / to-order / lead days, written back
//!   as stored computes on the orderpoint row. `qty_to_order = max(forecast, item_max_qty) -
//!   forecast`, floored at zero; a positive `qty_to_order_manual` overrides the computed figure
//!   for the order the scheduler mints.
//! - **Rule destination (R13-rule)** — a rule must not route into a `view` location; the
//!   service pre-checks it so a caller gets the typed error instead of a constraint 500.
//! - **Orderpoint uniqueness (R6)** — the SERVICE half over the unique index
//!   `(item_id, location_id)`; the index is the DB backstop.
//!
//! This file holds no SQL — every statement lives in
//! `infrastructure/persistence/procurement_repository.rs` (the module's 4-layer rule).
//!
//! Tenancy (ADR-0029): the module is tenant-agnostic. Own transactions re-bind the caller's
//! ambient org scope before writing (the composed decorator owns isolation); the module itself
//! carries no tenant value.

use async_trait::async_trait;
use rust_decimal::Decimal;
use sqlx::PgPool;
use std::sync::Arc;
use uuid::Uuid;

use crate::domain::entity::MoveState;
use crate::infrastructure::persistence::procurement_repository::{
    ForecastComponents, NewMoveRow, OrderpointComputes, OrderpointRow, ProcurementRepository,
    RuleRow,
};

use super::inventory_events::{InventoryEvent, InventoryEventSink, OrderpointTriggered};
use super::inventory_write_service::legacy_company_echo;

// --- errors -------------------------------------------------------------------

#[derive(Debug)]
pub enum ProcurementError {
    /// R13-rule: a rule must not push into a `view` location.
    RuleDestIsView { location_id: Uuid },
    /// R6: an orderpoint already exists for this (item, location).
    OrderpointExists { item_id: Uuid, location_id: Uuid },
    /// No active pull rule matches the demand (route selection found nothing).
    NoRuleForDemand { item_id: Uuid, location_id: Uuid },
    /// The referenced row does not exist.
    NotFound(Uuid),
    Db(sqlx::Error),
}

impl ProcurementError {
    pub fn code(&self) -> String {
        // Codes are the stable error_codes declared in schema/hooks/stock.hook.yaml.
        match self {
            ProcurementError::RuleDestIsView { .. } => "rule_dest_is_view".into(),
            ProcurementError::OrderpointExists { .. } => "orderpoint_exists".into(),
            ProcurementError::NoRuleForDemand { .. } => "no_rule_for_demand".into(),
            ProcurementError::NotFound(_) => "not_found".into(),
            ProcurementError::Db(_) => "internal_error".into(),
        }
    }
    pub fn http_status(&self) -> u16 {
        match self {
            ProcurementError::NotFound(_) => 404,
            ProcurementError::Db(_) => 500,
            _ => 422,
        }
    }
}

impl std::fmt::Display for ProcurementError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        match self {
            ProcurementError::RuleDestIsView { .. } => write!(f, "rule_dest_is_view: a rule destination cannot be a view location (R13)"),
            ProcurementError::OrderpointExists { item_id, location_id } => write!(f, "orderpoint_exists: a reordering rule already covers item {item_id} at location {location_id} (R6)"),
            ProcurementError::NoRuleForDemand { item_id, location_id } => write!(f, "no_rule_for_demand: no active pull rule covers item {item_id} demanded at {location_id}"),
            ProcurementError::NotFound(id) => write!(f, "not_found: {id}"),
            ProcurementError::Db(e) => write!(f, "internal_error: {e}"),
        }
    }
}
impl std::error::Error for ProcurementError {}
impl From<sqlx::Error> for ProcurementError {
    fn from(e: sqlx::Error) -> Self {
        ProcurementError::Db(e)
    }
}

/// A move-lifecycle verb failure crossing the [`MovePipeline`] boundary. Deliberately a flat
/// `{code, message}` — the implementing engine's own error taxonomy stays on its side of the
/// port; the scheduler only logs and continues with the next claimed move.
#[derive(Debug, Clone)]
pub struct MovePipelineError {
    pub code: String,
    pub message: String,
}

impl std::fmt::Display for MovePipelineError {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "{}: {}", self.code, self.message)
    }
}
impl std::error::Error for MovePipelineError {}

/// The stock-move engine's lifecycle verbs, seen from the scheduler (the port the engine
/// implements — the same fail-closed shape as the module's other deferred ports: there is no
/// silent no-op implementation, a job cannot run without a real engine behind it).
///
/// `confirm` is `_action_confirm` (draft → waiting/confirmed, which itself fires `run_push`
/// for transparent push rules); `assign` is `_action_assign` (confirmed/partially_available →
/// partially_available/assigned — the reservation apex write). Both receive the caller's
/// connection: the scheduler drives them inside its per-batch transaction, so a batch's
/// transitions commit or roll back together.
#[async_trait]
pub trait MovePipeline: Send + Sync {
    async fn confirm(
        &self,
        conn: &mut sqlx::PgConnection,
        move_id: Uuid,
    ) -> Result<MoveState, MovePipelineError>;

    async fn assign(
        &self,
        conn: &mut sqlx::PgConnection,
        move_id: Uuid,
    ) -> Result<MoveState, MovePipelineError>;
}

// --- inputs -------------------------------------------------------------------

#[derive(Debug, Clone)]
pub struct NewRoute {
    pub name: String,
    pub active: bool,
    pub sequence: i32,
}

#[derive(Debug, Clone)]
pub struct NewRouteRule {
    pub name: String,
    pub sequence: i32,
    pub action: String,
    pub auto: String,
    pub procure_method: String,
    pub delay: i32,
    pub location_src_id: Option<Uuid>,
    pub location_dest_id: Uuid,
    pub picking_type_id: Uuid,
    pub route_id: Uuid,
    pub warehouse_id: Option<Uuid>,
    pub propagate_cancel: bool,
}

#[derive(Debug, Clone)]
pub struct NewOrderpoint {
    pub name: String,
    pub trigger: String,
    pub item_id: Uuid,
    pub location_id: Uuid,
    pub warehouse_id: Uuid,
    pub item_min_qty: Decimal,
    pub item_max_qty: Decimal,
    pub route_id: Option<Uuid>,
}

/// A demand that needs supply pulled to a location — the entry a confirm engine (sale-stock)
/// or a buying bridge calls; implemented here, consumed by later passes.
#[derive(Debug, Clone)]
pub struct ProcurementRequest {
    pub item_id: Uuid,
    pub quantity: Decimal,
    /// Where the stock is needed (the rule whose destination covers this location supplies it).
    pub location_id: Uuid,
    pub warehouse_id: Option<Uuid>,
    /// Explicit route candidates; None = every active route.
    pub route_ids: Option<Vec<Uuid>>,
    pub origin: String,
    /// The orderpoint this order answers (present when the scheduler minted it).
    pub orderpoint_id: Option<Uuid>,
}

/// What one reorder rung produced — the durable record behind the `OrderpointTriggered` event.
#[derive(Debug, Clone)]
pub struct ReorderOutcome {
    pub orderpoint_id: Uuid,
    pub item_id: Uuid,
    pub qty_to_order: Decimal,
    pub forecast_qty: Decimal,
    /// The minted replenishment move (draft until the pipeline confirms it).
    pub move_id: Uuid,
}

/// Counters the scheduler reports per run.
#[derive(Debug, Clone, Copy, Default)]
pub struct AssignSweepReport {
    pub claimed: usize,
    pub assigned: usize,
    pub partially_available: usize,
    pub failed: usize,
}

// --- service ------------------------------------------------------------------

#[derive(Clone)]
pub struct ProcurementService {
    db_pool: PgPool,
    sink: Arc<dyn InventoryEventSink>,
}

impl ProcurementService {
    pub fn new(db_pool: PgPool) -> Self {
        Self::with_sink(db_pool, Arc::new(super::inventory_events::LoggingSink))
    }

    pub fn with_sink(db_pool: PgPool, sink: Arc<dyn InventoryEventSink>) -> Self {
        Self { db_pool, sink }
    }

    /// The database this call runs on: the composer's request pool when one
    /// is bound (a tenant mount, or a relay consumer wrapped by the host),
    /// else the composed pool (ADR-0029 pool law).
    fn rpool(&self) -> PgPool {
        crate::request_pool::current().unwrap_or_else(|| self.db_pool.clone())
    }

    // -- routes ---------------------------------------------------------------

    /// Create a route (an ordered rule collection).
    ///
    /// The write rides its own transaction re-bound to the caller's ambient org scope
    /// (ADR-0029) — the scope is task-local and a fresh pool transaction carries none of it;
    /// the composed decorator's fill stamps the acting unit on the INSERT. Undecorated (module
    /// tests, jobs) the transaction stays plain.
    pub async fn create_route(&self, r: NewRoute) -> Result<Uuid, ProcurementError> {
        let id = Uuid::new_v4();
        let mut tx = self.rpool().begin().await?;
        super::inventory_write_service::relay_ambient_scope(&mut tx).await?;
        ProcurementRepository::insert_route(&mut *tx, id, &r.name, r.active, r.sequence).await?;
        tx.commit().await?;
        Ok(id)
    }

    // -- rules ----------------------------------------------------------------

    /// Create a procurement rule after the R13-rule destination check (never route INTO a view
    /// location). The DB trigger is the backstop for raw writers; this check exists so a
    /// service caller gets the typed error instead of a constraint 500.
    ///
    /// The pre-check read and the insert ride one transaction re-bound to the caller's ambient
    /// org scope (ADR-0029); undecorated (module tests, jobs) the transaction stays plain.
    pub async fn create_rule(&self, r: NewRouteRule) -> Result<Uuid, ProcurementError> {
        let mut tx = self.rpool().begin().await?;
        super::inventory_write_service::relay_ambient_scope(&mut tx).await?;
        self.check_rule_destination(&mut tx, r.location_dest_id).await?;

        let id = Uuid::new_v4();
        ProcurementRepository::insert_rule(
            &mut *tx,
            id,
            &r.name,
            r.sequence,
            &r.action,
            &r.auto,
            &r.procure_method,
            r.delay,
            r.location_src_id,
            r.location_dest_id,
            r.picking_type_id,
            r.route_id,
            r.warehouse_id,
            r.propagate_cancel,
        )
        .await?;
        tx.commit().await?;
        Ok(id)
    }

    /// The R13-rule service pre-check: never route INTO a view location. Runs on the caller's
    /// connection so the read shares the caller's transaction.
    async fn check_rule_destination(
        &self,
        conn: &mut sqlx::PgConnection,
        location_dest_id: Uuid,
    ) -> Result<(), ProcurementError> {
        if let Some(usage) = ProcurementRepository::location_usage(&mut *conn, location_dest_id).await? {
            if usage == "view" {
                return Err(ProcurementError::RuleDestIsView { location_id: location_dest_id });
            }
        }
        Ok(())
    }

    // -- rule selection + move minting (§6) ------------------------------------

    /// `_search_rule`: the highest-sequence active pull-capable rule covering the demand
    /// location. Public — the confirm engine and later bridges select rules through the same
    /// ordering the scheduler does.
    pub async fn search_rule(
        &self,
        demand_location_id: Uuid,
        route_ids: Option<Vec<Uuid>>,
    ) -> Result<Option<RuleRow>, ProcurementError> {
        Ok(ProcurementRepository::search_rule(&self.rpool(), demand_location_id, route_ids).await?)
    }

    /// `_run_pull`: mint the inbound DRAFT move a selected pull rule prescribes — FROM the
    /// rule's source (or the operation type's default source when the rule names none) TO the
    /// rule's destination, at the requested quantity. The move enters the lifecycle as a draft;
    /// the caller (scheduler or confirm engine) drives it through the pipeline.
    pub async fn run_pull(
        &self,
        conn: &mut sqlx::PgConnection,
        rule: &RuleRow,
        req: &ProcurementRequest,
    ) -> Result<Uuid, ProcurementError> {
        let src = match rule.location_src_id {
            Some(s) => s,
            None => {
                sqlx::query_scalar::<_, Uuid>(
                    r#"SELECT default_location_src_id FROM inventory.operation_types
                       WHERE id = $1 AND (metadata->>'deleted_at') IS NULL"#,
                )
                .bind(rule.picking_type_id)
                .fetch_one(&mut *conn)
                .await?
            }
        };
        let row = NewMoveRow {
            name: format!("PULL/{}/{}", rule.name, &Uuid::new_v4().simple().to_string()[..8]),
            item_id: req.item_id,
            demand_qty: req.quantity,
            procure_method: rule.procure_method.clone(),
            origin: req.origin.clone(),
            location_id: src,
            location_dest_id: rule.location_dest_id,
            rule_id: Some(rule.id),
            warehouse_id: rule.warehouse_id.or(req.warehouse_id),
            orderpoint_id: req.orderpoint_id,
            move_orig_ids: Vec::new(),
            move_dest_ids: Vec::new(),
            propagate_cancel: rule.propagate_cancel,
        };
        Ok(ProcurementRepository::insert_move(&mut *conn, &row).await?)
    }

    /// `_run_push`: propagate a confirmed move FROM the rule's source — mint the chained
    /// downstream move (rule source → the upstream's destination) and link it onto the
    /// upstream's `move_dest_ids`. Called by the move engine on `_action_confirm` for
    /// `transparent` push rules; `manual` rules wait for a human.
    pub async fn run_push(
        &self,
        conn: &mut sqlx::PgConnection,
        rule: &RuleRow,
        upstream_move: &crate::domain::entity::StockMove,
    ) -> Result<Uuid, ProcurementError> {
        let src = rule.location_src_id.unwrap_or(upstream_move.location_id);
        let row = NewMoveRow {
            name: format!("PUSH/{}/{}", rule.name, &Uuid::new_v4().simple().to_string()[..8]),
            item_id: upstream_move.item_id,
            demand_qty: upstream_move.demand_qty - upstream_move.quantity,
            procure_method: rule.procure_method.clone(),
            origin: upstream_move.name.clone(),
            location_id: src,
            location_dest_id: upstream_move.location_dest_id,
            rule_id: Some(rule.id),
            warehouse_id: rule.warehouse_id.or(upstream_move.warehouse_id),
            orderpoint_id: upstream_move.orderpoint_id,
            move_orig_ids: vec![upstream_move.id],
            move_dest_ids: Vec::new(),
            propagate_cancel: rule.propagate_cancel,
        };
        let downstream = ProcurementRepository::insert_move(&mut *conn, &row).await?;
        ProcurementRepository::link_move_chain(&mut *conn, upstream_move.id, downstream).await?;
        Ok(downstream)
    }

    /// The procurement-group entry point: resolve the rule for a demand and mint the pull
    /// move. This is the surface a sale-stock confirm engine calls per storable line and a
    /// purchase bridge (`_run_buy`) attaches to — implemented here, consumed by later passes.
    pub async fn run_procurement(
        &self,
        conn: &mut sqlx::PgConnection,
        req: &ProcurementRequest,
    ) -> Result<Uuid, ProcurementError> {
        let rule = self
            .search_rule(req.location_id, req.route_ids.clone())
            .await?
            .ok_or(ProcurementError::NoRuleForDemand {
                item_id: req.item_id,
                location_id: req.location_id,
            })?;
        self.run_pull(conn, &rule, req).await
    }

    // -- orderpoints (§7, T11) --------------------------------------------------

    /// Create an orderpoint (min/max replenishment rule) — the service half of R6: a duplicate
    /// (item, location) coverage is the typed `orderpoint_exists` error; the partial unique
    /// index is the DB backstop a raw writer hits instead.
    ///
    /// The duplicate-check read and the insert ride one transaction re-bound to the caller's
    /// ambient org scope (ADR-0029); undecorated (module tests, jobs) it stays plain — the
    /// composed decorator owns isolation.
    pub async fn create_orderpoint(&self, o: NewOrderpoint) -> Result<Uuid, ProcurementError> {
        let mut tx = self.rpool().begin().await?;
        super::inventory_write_service::relay_ambient_scope(&mut tx).await?;
        let existing = ProcurementRepository::orderpoint_exists(&mut *tx, o.item_id, o.location_id).await?;
        if existing > 0 {
            return Err(ProcurementError::OrderpointExists { item_id: o.item_id, location_id: o.location_id });
        }

        let id = Uuid::new_v4();
        let inserted = ProcurementRepository::insert_orderpoint(
            &mut *tx,
            id,
            &o.name,
            &o.trigger,
            o.item_id,
            o.location_id,
            o.warehouse_id,
            o.item_min_qty,
            o.item_max_qty,
            o.route_id,
        )
        .await;
        if let Err(e) = inserted {
            if e.as_database_error().map(|d| d.is_unique_violation()).unwrap_or(false) {
                return Err(ProcurementError::OrderpointExists {
                    item_id: o.item_id,
                    location_id: o.location_id,
                });
            }
            return Err(e.into());
        }
        tx.commit().await?;
        Ok(id)
    }

    /// The T11 computes for one orderpoint, from the forecast aggregation:
    /// `qty_forecast = on_hand + incoming − outgoing` and
    /// `qty_to_order = max(qty_forecast, item_max_qty) − qty_forecast` (floored at zero).
    /// Lead days are the matched rule's delay; the deadline is `today + lead_days`.
    pub async fn compute_orderpoint(
        &self,
        conn: &mut sqlx::PgConnection,
        op: &OrderpointRow,
    ) -> Result<OrderpointComputes, ProcurementError> {
        let f: ForecastComponents =
            ProcurementRepository::forecast_components(&mut *conn, op.item_id, op.location_id)
                .await?;
        let forecast = f.on_hand + f.incoming - f.outgoing;
        let to_order = (op.item_max_qty - forecast).max(Decimal::ZERO);

        let rule = ProcurementRepository::search_rule(
            &mut *conn,
            op.location_id,
            op.route_id.map(|r| vec![r]),
        )
        .await?;
        let delay = match &rule {
            Some(r) => r.delay,
            None => 0,
        };
        let lead_days = Decimal::from(delay);
        let deadline = chrono::Utc::now().date_naive()
            + chrono::Duration::days(i64::from(delay));
        Ok(OrderpointComputes {
            qty_on_hand: f.on_hand,
            qty_forecast: forecast,
            qty_to_order: to_order,
            lead_days,
            deadline_date: Some(deadline),
        })
    }

    /// Recompute and persist one orderpoint's stored computes (the read-model refresh a
    /// replenishment view drives for `manual` trigger rules — humans order, the computes
    /// recommend).
    ///
    /// The orderpoint read rides the caller's ambient org scope (ADR-0029): under the composed
    /// shape the decorator's fence bounds it and the forecast/stamp writes ride the same
    /// bound transaction; undecorated (module tests, jobs) it is plain.
    pub async fn recompute_orderpoint(
        &self,
        orderpoint_id: Uuid,
    ) -> Result<OrderpointComputes, ProcurementError> {
        let mut tx = self.rpool().begin().await?;
        super::inventory_write_service::relay_ambient_scope(&mut tx).await?;
        let op = sqlx::query_as::<_, OrderpointRowDb>(
            r#"SELECT id, name, trigger::text AS trigger, item_id, location_id, warehouse_id,
                      item_min_qty, item_max_qty, route_id, qty_to_order_manual
               FROM inventory.reordering_rules
               WHERE id = $1 AND (metadata->>'deleted_at') IS NULL"#,
        )
        .bind(orderpoint_id)
        .fetch_optional(&mut *tx)
        .await?
        .ok_or(ProcurementError::NotFound(orderpoint_id))?;
        let op = op.into_row();
        let computes = self.compute_orderpoint(&mut tx, &op).await?;
        ProcurementRepository::stamp_orderpoint(&mut *tx, orderpoint_id, &computes).await?;
        tx.commit().await?;
        Ok(computes)
    }

    /// The reorder rung (`_run_orderpoints`, §7 task 1) for ONE claimed orderpoint: refresh
    /// the T11 computes, and when `qty_forecast < item_min_qty`, mint the replenishment move of
    /// `qty_to_order` (a positive `qty_to_order_manual` overrides the computed figure) and
    /// publish `OrderpointTriggered`. Returns the outcome when the rule fired.
    ///
    /// Runs inside the caller's transaction — the scheduler commits per batch and emits the
    /// event after that commit (an event never precedes its durable record).
    pub async fn reorder_one(
        &self,
        conn: &mut sqlx::PgConnection,
        op: &OrderpointRow,
    ) -> Result<Option<ReorderOutcome>, ProcurementError> {
        let computes = self.compute_orderpoint(&mut *conn, op).await?;
        ProcurementRepository::stamp_orderpoint(&mut *conn, op.id, &computes).await?;
        if computes.qty_forecast >= op.item_min_qty {
            return Ok(None);
        }
        let rule = ProcurementRepository::search_rule(
            &mut *conn,
            op.location_id,
            op.route_id.map(|r| vec![r]),
        )
        .await?
        .ok_or(ProcurementError::NoRuleForDemand {
            item_id: op.item_id,
            location_id: op.location_id,
        })?;
        let qty = if op.qty_to_order_manual > Decimal::ZERO {
            op.qty_to_order_manual
        } else {
            computes.qty_to_order
        };
        let req = ProcurementRequest {
            item_id: op.item_id,
            quantity: qty,
            location_id: op.location_id,
            warehouse_id: Some(op.warehouse_id),
            route_ids: op.route_id.map(|r| vec![r]),
            origin: format!("orderpoint:{}", op.name),
            orderpoint_id: Some(op.id),
        };
        let move_id = self.run_pull(&mut *conn, &rule, &req).await?;
        Ok(Some(ReorderOutcome {
            orderpoint_id: op.id,
            item_id: op.item_id,
            qty_to_order: qty,
            forecast_qty: computes.qty_forecast,
            move_id,
        }))
    }

    /// Publish the reorder event (after the batch commit). Kept separate from
    /// [`Self::reorder_one`] so the transaction boundary stays explicit.
    pub fn emit_orderpoint_triggered(&self, o: &ReorderOutcome) {
        self.sink.publish(InventoryEvent::OrderpointTriggered(OrderpointTriggered {
            orderpoint_id: o.orderpoint_id,
            // Legacy twin (ADR-0029): filled from the ambient org scope's company echo for
            // consumers that still read a tenant off the wire. No module statement keys on it.
            company_id: legacy_company_echo(),
            item_id: o.item_id,
            qty_to_order: o.qty_to_order,
            forecast_qty: o.forecast_qty,
        }));
    }

    // -- assign sweep (§7 task 2) ----------------------------------------------

    /// `_run_scheduler_assign` for ONE claimed move: hand it to the move engine's assign verb
    /// on the caller's connection. The sweep (claiming, ordering, accounting) lives in the
    /// scheduler job; this is the per-move step so the engine contract is exercisable in
    /// isolation.
    pub async fn assign_one(
        &self,
        conn: &mut sqlx::PgConnection,
        pipeline: &dyn MovePipeline,
        move_id: Uuid,
    ) -> Result<MoveState, MovePipelineError> {
        pipeline.assign(conn, move_id).await
    }

    /// The quant vacuum tail (§7 task 3) — see the repository for the R21-shaped predicate.
    /// Returns the raw driver error: this is a pure delegation with no domain decision, and
    /// the scheduler (its only caller) reports infrastructure failures as-is.
    pub async fn housekeep_quants(
        &self,
        conn: &mut sqlx::PgConnection,
        limit: i64,
    ) -> Result<u64, sqlx::Error> {
        ProcurementRepository::housekeep_quants(conn, limit).await
    }

    pub fn sink(&self) -> Arc<dyn InventoryEventSink> {
        self.sink.clone()
    }
}

/// Query-shaped orderpoint row (private) — maps straight onto [`OrderpointRow`].
#[derive(sqlx::FromRow)]
struct OrderpointRowDb {
    id: Uuid,
    name: String,
    r#trigger: String,
    item_id: Uuid,
    location_id: Uuid,
    warehouse_id: Uuid,
    item_min_qty: Decimal,
    item_max_qty: Decimal,
    route_id: Option<Uuid>,
    qty_to_order_manual: Decimal,
}

impl OrderpointRowDb {
    fn into_row(self) -> OrderpointRow {
        OrderpointRow {
            id: self.id,
            name: self.name,
            trigger: self.r#trigger,
            item_id: self.item_id,
            location_id: self.location_id,
            warehouse_id: self.warehouse_id,
            item_min_qty: self.item_min_qty,
            item_max_qty: self.item_max_qty,
            route_id: self.route_id,
            qty_to_order_manual: self.qty_to_order_manual,
        }
    }
}