1use async_trait::async_trait;
31use rust_decimal::Decimal;
32use sqlx::PgPool;
33use std::sync::Arc;
34use uuid::Uuid;
35
36use crate::domain::entity::MoveState;
37use crate::infrastructure::persistence::procurement_repository::{
38 ForecastComponents, NewMoveRow, OrderpointComputes, OrderpointRow, ProcurementRepository,
39 RuleRow,
40};
41
42use super::inventory_events::{InventoryEvent, InventoryEventSink, OrderpointTriggered};
43use super::inventory_write_service::legacy_company_echo;
44
45#[derive(Debug)]
48pub enum ProcurementError {
49 RuleDestIsView { location_id: Uuid },
51 OrderpointExists { item_id: Uuid, location_id: Uuid },
53 NoRuleForDemand { item_id: Uuid, location_id: Uuid },
55 NotFound(Uuid),
57 Db(sqlx::Error),
58}
59
60impl ProcurementError {
61 pub fn code(&self) -> String {
62 match self {
64 ProcurementError::RuleDestIsView { .. } => "rule_dest_is_view".into(),
65 ProcurementError::OrderpointExists { .. } => "orderpoint_exists".into(),
66 ProcurementError::NoRuleForDemand { .. } => "no_rule_for_demand".into(),
67 ProcurementError::NotFound(_) => "not_found".into(),
68 ProcurementError::Db(_) => "internal_error".into(),
69 }
70 }
71 pub fn http_status(&self) -> u16 {
72 match self {
73 ProcurementError::NotFound(_) => 404,
74 ProcurementError::Db(_) => 500,
75 _ => 422,
76 }
77 }
78}
79
80impl std::fmt::Display for ProcurementError {
81 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
82 match self {
83 ProcurementError::RuleDestIsView { .. } => write!(f, "rule_dest_is_view: a rule destination cannot be a view location (R13)"),
84 ProcurementError::OrderpointExists { item_id, location_id } => write!(f, "orderpoint_exists: a reordering rule already covers item {item_id} at location {location_id} (R6)"),
85 ProcurementError::NoRuleForDemand { item_id, location_id } => write!(f, "no_rule_for_demand: no active pull rule covers item {item_id} demanded at {location_id}"),
86 ProcurementError::NotFound(id) => write!(f, "not_found: {id}"),
87 ProcurementError::Db(e) => write!(f, "internal_error: {e}"),
88 }
89 }
90}
91impl std::error::Error for ProcurementError {}
92impl From<sqlx::Error> for ProcurementError {
93 fn from(e: sqlx::Error) -> Self {
94 ProcurementError::Db(e)
95 }
96}
97
98#[derive(Debug, Clone)]
102pub struct MovePipelineError {
103 pub code: String,
104 pub message: String,
105}
106
107impl std::fmt::Display for MovePipelineError {
108 fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
109 write!(f, "{}: {}", self.code, self.message)
110 }
111}
112impl std::error::Error for MovePipelineError {}
113
114#[async_trait]
124pub trait MovePipeline: Send + Sync {
125 async fn confirm(
126 &self,
127 conn: &mut sqlx::PgConnection,
128 move_id: Uuid,
129 ) -> Result<MoveState, MovePipelineError>;
130
131 async fn assign(
132 &self,
133 conn: &mut sqlx::PgConnection,
134 move_id: Uuid,
135 ) -> Result<MoveState, MovePipelineError>;
136}
137
138#[derive(Debug, Clone)]
141pub struct NewRoute {
142 pub name: String,
143 pub active: bool,
144 pub sequence: i32,
145}
146
147#[derive(Debug, Clone)]
148pub struct NewRouteRule {
149 pub name: String,
150 pub sequence: i32,
151 pub action: String,
152 pub auto: String,
153 pub procure_method: String,
154 pub delay: i32,
155 pub location_src_id: Option<Uuid>,
156 pub location_dest_id: Uuid,
157 pub picking_type_id: Uuid,
158 pub route_id: Uuid,
159 pub warehouse_id: Option<Uuid>,
160 pub propagate_cancel: bool,
161}
162
163#[derive(Debug, Clone)]
164pub struct NewOrderpoint {
165 pub name: String,
166 pub trigger: String,
167 pub item_id: Uuid,
168 pub location_id: Uuid,
169 pub warehouse_id: Uuid,
170 pub item_min_qty: Decimal,
171 pub item_max_qty: Decimal,
172 pub route_id: Option<Uuid>,
173}
174
175#[derive(Debug, Clone)]
178pub struct ProcurementRequest {
179 pub item_id: Uuid,
180 pub quantity: Decimal,
181 pub location_id: Uuid,
183 pub warehouse_id: Option<Uuid>,
184 pub route_ids: Option<Vec<Uuid>>,
186 pub origin: String,
187 pub orderpoint_id: Option<Uuid>,
189}
190
191#[derive(Debug, Clone)]
193pub struct ReorderOutcome {
194 pub orderpoint_id: Uuid,
195 pub item_id: Uuid,
196 pub qty_to_order: Decimal,
197 pub forecast_qty: Decimal,
198 pub move_id: Uuid,
200}
201
202#[derive(Debug, Clone, Copy, Default)]
204pub struct AssignSweepReport {
205 pub claimed: usize,
206 pub assigned: usize,
207 pub partially_available: usize,
208 pub failed: usize,
209}
210
211#[derive(Clone)]
214pub struct ProcurementService {
215 db_pool: PgPool,
216 sink: Arc<dyn InventoryEventSink>,
217}
218
219impl ProcurementService {
220 pub fn new(db_pool: PgPool) -> Self {
221 Self::with_sink(db_pool, Arc::new(super::inventory_events::LoggingSink))
222 }
223
224 pub fn with_sink(db_pool: PgPool, sink: Arc<dyn InventoryEventSink>) -> Self {
225 Self { db_pool, sink }
226 }
227
228 fn rpool(&self) -> PgPool {
232 crate::request_pool::current().unwrap_or_else(|| self.db_pool.clone())
233 }
234
235 pub async fn create_route(&self, r: NewRoute) -> Result<Uuid, ProcurementError> {
244 let id = Uuid::new_v4();
245 let mut tx = self.rpool().begin().await?;
246 super::inventory_write_service::relay_ambient_scope(&mut tx).await?;
247 ProcurementRepository::insert_route(&mut *tx, id, &r.name, r.active, r.sequence).await?;
248 tx.commit().await?;
249 Ok(id)
250 }
251
252 pub async fn create_rule(&self, r: NewRouteRule) -> Result<Uuid, ProcurementError> {
261 let mut tx = self.rpool().begin().await?;
262 super::inventory_write_service::relay_ambient_scope(&mut tx).await?;
263 self.check_rule_destination(&mut tx, r.location_dest_id).await?;
264
265 let id = Uuid::new_v4();
266 ProcurementRepository::insert_rule(
267 &mut *tx,
268 id,
269 &r.name,
270 r.sequence,
271 &r.action,
272 &r.auto,
273 &r.procure_method,
274 r.delay,
275 r.location_src_id,
276 r.location_dest_id,
277 r.picking_type_id,
278 r.route_id,
279 r.warehouse_id,
280 r.propagate_cancel,
281 )
282 .await?;
283 tx.commit().await?;
284 Ok(id)
285 }
286
287 async fn check_rule_destination(
290 &self,
291 conn: &mut sqlx::PgConnection,
292 location_dest_id: Uuid,
293 ) -> Result<(), ProcurementError> {
294 if let Some(usage) = ProcurementRepository::location_usage(&mut *conn, location_dest_id).await? {
295 if usage == "view" {
296 return Err(ProcurementError::RuleDestIsView { location_id: location_dest_id });
297 }
298 }
299 Ok(())
300 }
301
302 pub async fn search_rule(
308 &self,
309 demand_location_id: Uuid,
310 route_ids: Option<Vec<Uuid>>,
311 ) -> Result<Option<RuleRow>, ProcurementError> {
312 Ok(ProcurementRepository::search_rule(&self.rpool(), demand_location_id, route_ids).await?)
313 }
314
315 pub async fn run_pull(
320 &self,
321 conn: &mut sqlx::PgConnection,
322 rule: &RuleRow,
323 req: &ProcurementRequest,
324 ) -> Result<Uuid, ProcurementError> {
325 let src = match rule.location_src_id {
326 Some(s) => s,
327 None => {
328 sqlx::query_scalar::<_, Uuid>(
329 r#"SELECT default_location_src_id FROM inventory.operation_types
330 WHERE id = $1 AND (metadata->>'deleted_at') IS NULL"#,
331 )
332 .bind(rule.picking_type_id)
333 .fetch_one(&mut *conn)
334 .await?
335 }
336 };
337 let row = NewMoveRow {
338 name: format!("PULL/{}/{}", rule.name, &Uuid::new_v4().simple().to_string()[..8]),
339 item_id: req.item_id,
340 demand_qty: req.quantity,
341 procure_method: rule.procure_method.clone(),
342 origin: req.origin.clone(),
343 location_id: src,
344 location_dest_id: rule.location_dest_id,
345 rule_id: Some(rule.id),
346 warehouse_id: rule.warehouse_id.or(req.warehouse_id),
347 orderpoint_id: req.orderpoint_id,
348 move_orig_ids: Vec::new(),
349 move_dest_ids: Vec::new(),
350 propagate_cancel: rule.propagate_cancel,
351 };
352 Ok(ProcurementRepository::insert_move(&mut *conn, &row).await?)
353 }
354
355 pub async fn run_push(
360 &self,
361 conn: &mut sqlx::PgConnection,
362 rule: &RuleRow,
363 upstream_move: &crate::domain::entity::StockMove,
364 ) -> Result<Uuid, ProcurementError> {
365 let src = rule.location_src_id.unwrap_or(upstream_move.location_id);
366 let row = NewMoveRow {
367 name: format!("PUSH/{}/{}", rule.name, &Uuid::new_v4().simple().to_string()[..8]),
368 item_id: upstream_move.item_id,
369 demand_qty: upstream_move.demand_qty - upstream_move.quantity,
370 procure_method: rule.procure_method.clone(),
371 origin: upstream_move.name.clone(),
372 location_id: src,
373 location_dest_id: upstream_move.location_dest_id,
374 rule_id: Some(rule.id),
375 warehouse_id: rule.warehouse_id.or(upstream_move.warehouse_id),
376 orderpoint_id: upstream_move.orderpoint_id,
377 move_orig_ids: vec![upstream_move.id],
378 move_dest_ids: Vec::new(),
379 propagate_cancel: rule.propagate_cancel,
380 };
381 let downstream = ProcurementRepository::insert_move(&mut *conn, &row).await?;
382 ProcurementRepository::link_move_chain(&mut *conn, upstream_move.id, downstream).await?;
383 Ok(downstream)
384 }
385
386 pub async fn run_procurement(
390 &self,
391 conn: &mut sqlx::PgConnection,
392 req: &ProcurementRequest,
393 ) -> Result<Uuid, ProcurementError> {
394 let rule = self
395 .search_rule(req.location_id, req.route_ids.clone())
396 .await?
397 .ok_or(ProcurementError::NoRuleForDemand {
398 item_id: req.item_id,
399 location_id: req.location_id,
400 })?;
401 self.run_pull(conn, &rule, req).await
402 }
403
404 pub async fn create_orderpoint(&self, o: NewOrderpoint) -> Result<Uuid, ProcurementError> {
414 let mut tx = self.rpool().begin().await?;
415 super::inventory_write_service::relay_ambient_scope(&mut tx).await?;
416 let existing = ProcurementRepository::orderpoint_exists(&mut *tx, o.item_id, o.location_id).await?;
417 if existing > 0 {
418 return Err(ProcurementError::OrderpointExists { item_id: o.item_id, location_id: o.location_id });
419 }
420
421 let id = Uuid::new_v4();
422 let inserted = ProcurementRepository::insert_orderpoint(
423 &mut *tx,
424 id,
425 &o.name,
426 &o.trigger,
427 o.item_id,
428 o.location_id,
429 o.warehouse_id,
430 o.item_min_qty,
431 o.item_max_qty,
432 o.route_id,
433 )
434 .await;
435 if let Err(e) = inserted {
436 if e.as_database_error().map(|d| d.is_unique_violation()).unwrap_or(false) {
437 return Err(ProcurementError::OrderpointExists {
438 item_id: o.item_id,
439 location_id: o.location_id,
440 });
441 }
442 return Err(e.into());
443 }
444 tx.commit().await?;
445 Ok(id)
446 }
447
448 pub async fn compute_orderpoint(
453 &self,
454 conn: &mut sqlx::PgConnection,
455 op: &OrderpointRow,
456 ) -> Result<OrderpointComputes, ProcurementError> {
457 let f: ForecastComponents =
458 ProcurementRepository::forecast_components(&mut *conn, op.item_id, op.location_id)
459 .await?;
460 let forecast = f.on_hand + f.incoming - f.outgoing;
461 let to_order = (op.item_max_qty - forecast).max(Decimal::ZERO);
462
463 let rule = ProcurementRepository::search_rule(
464 &mut *conn,
465 op.location_id,
466 op.route_id.map(|r| vec![r]),
467 )
468 .await?;
469 let delay = match &rule {
470 Some(r) => r.delay,
471 None => 0,
472 };
473 let lead_days = Decimal::from(delay);
474 let deadline = chrono::Utc::now().date_naive()
475 + chrono::Duration::days(i64::from(delay));
476 Ok(OrderpointComputes {
477 qty_on_hand: f.on_hand,
478 qty_forecast: forecast,
479 qty_to_order: to_order,
480 lead_days,
481 deadline_date: Some(deadline),
482 })
483 }
484
485 pub async fn recompute_orderpoint(
493 &self,
494 orderpoint_id: Uuid,
495 ) -> Result<OrderpointComputes, ProcurementError> {
496 let mut tx = self.rpool().begin().await?;
497 super::inventory_write_service::relay_ambient_scope(&mut tx).await?;
498 let op = sqlx::query_as::<_, OrderpointRowDb>(
499 r#"SELECT id, name, trigger::text AS trigger, item_id, location_id, warehouse_id,
500 item_min_qty, item_max_qty, route_id, qty_to_order_manual
501 FROM inventory.reordering_rules
502 WHERE id = $1 AND (metadata->>'deleted_at') IS NULL"#,
503 )
504 .bind(orderpoint_id)
505 .fetch_optional(&mut *tx)
506 .await?
507 .ok_or(ProcurementError::NotFound(orderpoint_id))?;
508 let op = op.into_row();
509 let computes = self.compute_orderpoint(&mut tx, &op).await?;
510 ProcurementRepository::stamp_orderpoint(&mut *tx, orderpoint_id, &computes).await?;
511 tx.commit().await?;
512 Ok(computes)
513 }
514
515 pub async fn reorder_one(
523 &self,
524 conn: &mut sqlx::PgConnection,
525 op: &OrderpointRow,
526 ) -> Result<Option<ReorderOutcome>, ProcurementError> {
527 let computes = self.compute_orderpoint(&mut *conn, op).await?;
528 ProcurementRepository::stamp_orderpoint(&mut *conn, op.id, &computes).await?;
529 if computes.qty_forecast >= op.item_min_qty {
530 return Ok(None);
531 }
532 let rule = ProcurementRepository::search_rule(
533 &mut *conn,
534 op.location_id,
535 op.route_id.map(|r| vec![r]),
536 )
537 .await?
538 .ok_or(ProcurementError::NoRuleForDemand {
539 item_id: op.item_id,
540 location_id: op.location_id,
541 })?;
542 let qty = if op.qty_to_order_manual > Decimal::ZERO {
543 op.qty_to_order_manual
544 } else {
545 computes.qty_to_order
546 };
547 let req = ProcurementRequest {
548 item_id: op.item_id,
549 quantity: qty,
550 location_id: op.location_id,
551 warehouse_id: Some(op.warehouse_id),
552 route_ids: op.route_id.map(|r| vec![r]),
553 origin: format!("orderpoint:{}", op.name),
554 orderpoint_id: Some(op.id),
555 };
556 let move_id = self.run_pull(&mut *conn, &rule, &req).await?;
557 Ok(Some(ReorderOutcome {
558 orderpoint_id: op.id,
559 item_id: op.item_id,
560 qty_to_order: qty,
561 forecast_qty: computes.qty_forecast,
562 move_id,
563 }))
564 }
565
566 pub fn emit_orderpoint_triggered(&self, o: &ReorderOutcome) {
569 self.sink.publish(InventoryEvent::OrderpointTriggered(OrderpointTriggered {
570 orderpoint_id: o.orderpoint_id,
571 company_id: legacy_company_echo(),
574 item_id: o.item_id,
575 qty_to_order: o.qty_to_order,
576 forecast_qty: o.forecast_qty,
577 }));
578 }
579
580 pub async fn assign_one(
587 &self,
588 conn: &mut sqlx::PgConnection,
589 pipeline: &dyn MovePipeline,
590 move_id: Uuid,
591 ) -> Result<MoveState, MovePipelineError> {
592 pipeline.assign(conn, move_id).await
593 }
594
595 pub async fn housekeep_quants(
599 &self,
600 conn: &mut sqlx::PgConnection,
601 limit: i64,
602 ) -> Result<u64, sqlx::Error> {
603 ProcurementRepository::housekeep_quants(conn, limit).await
604 }
605
606 pub fn sink(&self) -> Arc<dyn InventoryEventSink> {
607 self.sink.clone()
608 }
609}
610
611#[derive(sqlx::FromRow)]
613struct OrderpointRowDb {
614 id: Uuid,
615 name: String,
616 r#trigger: String,
617 item_id: Uuid,
618 location_id: Uuid,
619 warehouse_id: Uuid,
620 item_min_qty: Decimal,
621 item_max_qty: Decimal,
622 route_id: Option<Uuid>,
623 qty_to_order_manual: Decimal,
624}
625
626impl OrderpointRowDb {
627 fn into_row(self) -> OrderpointRow {
628 OrderpointRow {
629 id: self.id,
630 name: self.name,
631 trigger: self.r#trigger,
632 item_id: self.item_id,
633 location_id: self.location_id,
634 warehouse_id: self.warehouse_id,
635 item_min_qty: self.item_min_qty,
636 item_max_qty: self.item_max_qty,
637 route_id: self.route_id,
638 qty_to_order_manual: self.qty_to_order_manual,
639 }
640 }
641}