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;
#[derive(Debug)]
pub enum ProcurementError {
RuleDestIsView { location_id: Uuid },
OrderpointExists { item_id: Uuid, location_id: Uuid },
NoRuleForDemand { item_id: Uuid, location_id: Uuid },
NotFound(Uuid),
Db(sqlx::Error),
}
impl ProcurementError {
pub fn code(&self) -> String {
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)
}
}
#[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 {}
#[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>;
}
#[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>,
}
#[derive(Debug, Clone)]
pub struct ProcurementRequest {
pub item_id: Uuid,
pub quantity: Decimal,
pub location_id: Uuid,
pub warehouse_id: Option<Uuid>,
pub route_ids: Option<Vec<Uuid>>,
pub origin: String,
pub orderpoint_id: Option<Uuid>,
}
#[derive(Debug, Clone)]
pub struct ReorderOutcome {
pub orderpoint_id: Uuid,
pub item_id: Uuid,
pub qty_to_order: Decimal,
pub forecast_qty: Decimal,
pub move_id: Uuid,
}
#[derive(Debug, Clone, Copy, Default)]
pub struct AssignSweepReport {
pub claimed: usize,
pub assigned: usize,
pub partially_available: usize,
pub failed: usize,
}
#[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 }
}
fn rpool(&self) -> PgPool {
crate::request_pool::current().unwrap_or_else(|| self.db_pool.clone())
}
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)
}
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)
}
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(())
}
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?)
}
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?)
}
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)
}
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
}
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)
}
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),
})
}
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)
}
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,
}))
}
pub fn emit_orderpoint_triggered(&self, o: &ReorderOutcome) {
self.sink.publish(InventoryEvent::OrderpointTriggered(OrderpointTriggered {
orderpoint_id: o.orderpoint_id,
company_id: legacy_company_echo(),
item_id: o.item_id,
qty_to_order: o.qty_to_order,
forecast_qty: o.forecast_qty,
}));
}
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
}
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()
}
}
#[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,
}
}
}