use backbone_orm::org_scope;
use rust_decimal::Decimal;
use sqlx::PgPool;
use uuid::Uuid;
use crate::domain::entity::CatalogStatus;
use crate::domain::services::uom_tree::{
convert_quantity, ConversionRounding, UomChain, UomConversionError,
};
pub use crate::infrastructure::persistence::ItemHit;
use crate::infrastructure::persistence::{
AttributeRepository, AttributeValueRepository, BrandRepository, ItemGroupRepository,
ItemRepository, ItemVariantRepository, NewAttributeRow, NewAttributeValueRow, NewBrandRow,
NewItemGroupRow, NewItemRow, NewItemVariantRow, NewUomRow, UomRepository,
};
#[derive(Debug)]
pub enum CatalogWriteError {
ItemGroupNotFound(Uuid),
UomNotFound(Uuid),
ParentNotFound(Uuid),
NoUsageFlag,
DuplicateItemCode(String),
DuplicateBarcode(String),
AttributeNotFound(Uuid),
BrandNotFound(Uuid),
ItemNotFound(Uuid),
ItemVariantNotFound(Uuid),
InvalidStatusTransition { from: CatalogStatus, to: CatalogStatus },
DuplicateUomCode(String),
DuplicateBrandCode(String),
DuplicateAttributeCode(String),
DuplicateValueCode(String),
DuplicateSku(String),
NoOptions,
UnknownAttribute(String),
UnknownAttributeValue(String),
RelativeShapeMismatch,
NonPositiveRelativeFactor,
UomCycle { parent: Uuid },
Conversion(UomConversionError),
ProtectedUom { code: String },
UomHasChildren { code: String },
UomInUse { code: String },
Db(sqlx::Error),
}
impl CatalogWriteError {
pub fn code(&self) -> &'static str {
match self {
CatalogWriteError::ItemGroupNotFound(_) => "item_group_not_found",
CatalogWriteError::UomNotFound(_) => "uom_not_found",
CatalogWriteError::ParentNotFound(_) => "parent_not_found",
CatalogWriteError::NoUsageFlag => "no_usage_flag",
CatalogWriteError::DuplicateItemCode(_) => "duplicate_item_code",
CatalogWriteError::DuplicateBarcode(_) => "duplicate_barcode",
CatalogWriteError::AttributeNotFound(_) => "attribute_not_found",
CatalogWriteError::BrandNotFound(_) => "brand_not_found",
CatalogWriteError::ItemNotFound(_) => "item_not_found",
CatalogWriteError::ItemVariantNotFound(_) => "item_variant_not_found",
CatalogWriteError::InvalidStatusTransition { .. } => "invalid_status_transition",
CatalogWriteError::DuplicateUomCode(_) => "duplicate_uom_code",
CatalogWriteError::DuplicateBrandCode(_) => "duplicate_brand_code",
CatalogWriteError::DuplicateAttributeCode(_) => "duplicate_attribute_code",
CatalogWriteError::DuplicateValueCode(_) => "duplicate_value_code",
CatalogWriteError::DuplicateSku(_) => "duplicate_sku",
CatalogWriteError::NoOptions => "no_options",
CatalogWriteError::UnknownAttribute(_) => "unknown_attribute",
CatalogWriteError::UnknownAttributeValue(_) => "unknown_attribute_value",
CatalogWriteError::RelativeShapeMismatch => "relative_shape_mismatch",
CatalogWriteError::NonPositiveRelativeFactor => "non_positive_relative_factor",
CatalogWriteError::UomCycle { .. } => "uom_cycle",
CatalogWriteError::Conversion(e) => e.code(),
CatalogWriteError::ProtectedUom { .. } => "protected_uom",
CatalogWriteError::UomHasChildren { .. } => "uom_has_children",
CatalogWriteError::UomInUse { .. } => "uom_in_use",
CatalogWriteError::Db(_) => "internal_error",
}
}
pub fn http_status(&self) -> u16 {
match self {
CatalogWriteError::Db(_) => 500,
_ => 422,
}
}
}
impl std::fmt::Display for CatalogWriteError {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
write!(f, "{}", self.code())?;
match self {
CatalogWriteError::ItemGroupNotFound(id)
| CatalogWriteError::UomNotFound(id)
| CatalogWriteError::ParentNotFound(id) => write!(f, ": {id}"),
CatalogWriteError::DuplicateItemCode(v)
| CatalogWriteError::DuplicateBarcode(v)
| CatalogWriteError::DuplicateAttributeCode(v)
| CatalogWriteError::DuplicateValueCode(v)
| CatalogWriteError::DuplicateSku(v)
| CatalogWriteError::DuplicateUomCode(v)
| CatalogWriteError::DuplicateBrandCode(v)
| CatalogWriteError::UnknownAttribute(v)
| CatalogWriteError::UnknownAttributeValue(v) => write!(f, ": {v}"),
CatalogWriteError::AttributeNotFound(id)
| CatalogWriteError::BrandNotFound(id)
| CatalogWriteError::ItemNotFound(id)
| CatalogWriteError::ItemVariantNotFound(id) => write!(f, ": {id}"),
CatalogWriteError::InvalidStatusTransition { from, to } => write!(f, ": {from:?} -> {to:?}"),
CatalogWriteError::UomCycle { parent } => write!(f, ": {parent}"),
CatalogWriteError::Conversion(e) => write!(f, ": {e}"),
CatalogWriteError::ProtectedUom { code }
| CatalogWriteError::UomHasChildren { code }
| CatalogWriteError::UomInUse { code } => write!(f, ": {code}"),
_ => Ok(()),
}
}
}
impl std::error::Error for CatalogWriteError {}
impl From<sqlx::Error> for CatalogWriteError {
fn from(e: sqlx::Error) -> Self {
CatalogWriteError::Db(e)
}
}
#[derive(Debug, Clone)]
pub struct NewItemGroup {
pub code: String,
pub name: String,
pub parent_id: Option<Uuid>,
pub is_group: bool,
}
#[derive(Debug, Clone)]
pub struct NewItem {
pub item_code: String,
pub name: String,
pub description: Option<String>,
pub barcode: Option<String>,
pub brand_id: Option<Uuid>,
pub item_group_id: Uuid,
pub default_uom_id: Uuid,
pub item_type: Option<String>,
pub is_sales_item: bool,
pub is_purchase_item: bool,
pub is_stock_item: bool,
pub hsn_code: Option<String>,
pub is_taxable: bool,
pub weight_per_unit: Option<Decimal>,
pub standard_cost: Option<Decimal>,
pub tags: Option<serde_json::Value>,
pub data: Option<serde_json::Value>,
}
pub fn is_physical_item_type(item_type: &str) -> bool {
matches!(item_type, "physical_good" | "bundle" | "rental")
}
#[derive(Debug, Clone)]
pub struct NewAttribute {
pub code: String,
pub name: String,
pub attribute_type: Option<String>,
}
#[derive(Debug, Clone)]
pub struct NewAttributeValue {
pub attribute_id: Uuid,
pub code: String,
pub label: String,
pub label_en: Option<String>,
pub swatch_hex: Option<String>,
pub sort_order: i32,
}
#[derive(Debug, Clone)]
pub struct NewUom {
pub code: String,
pub name: String,
pub uom_type: Option<String>,
pub decimal_places: i32,
pub relative_uom_id: Option<Uuid>,
pub relative_factor: Option<Decimal>,
}
#[derive(Debug, Clone)]
pub struct NewBrand {
pub code: String,
pub name: String,
pub short_description: Option<String>,
pub description: Option<String>,
pub logo_url: Option<String>,
pub sort_order: i32,
}
#[derive(Debug, Clone)]
pub struct NewItemVariant {
pub item_id: Uuid,
pub sku: String,
pub variant_label: Option<String>,
pub options: std::collections::BTreeMap<String, String>,
pub barcode: Option<String>,
pub is_default: bool,
pub weight_per_unit: Option<Decimal>,
}
#[derive(Clone)]
pub struct CatalogWriteService {
db_pool: PgPool,
}
impl CatalogWriteService {
pub fn new(db_pool: PgPool) -> Self {
Self { db_pool }
}
async fn relay_ambient_scope(conn: &mut sqlx::PgConnection) -> Result<(), sqlx::Error> {
if let Some(scope) = org_scope::current_org_scope() {
org_scope::bind_org_scope_on(conn, &scope).await?;
}
Ok(())
}
pub async fn lookup_item(&self, code: &str) -> Result<Option<ItemHit>, CatalogWriteError> {
let items = ItemRepository::new(self.db_pool.clone());
if let Some(hit) = items.find_by_scan_code_scoped(&self.db_pool, code).await? {
return Ok(Some(hit));
}
let variants = ItemVariantRepository::new(self.db_pool.clone());
let hit = variants
.find_variant_by_scan_code_scoped(&self.db_pool, code)
.await?;
Ok(hit)
}
fn is_dup(e: &sqlx::Error, needle: &str) -> bool {
e.as_database_error()
.map(|d| d.is_unique_violation() && d.constraint().unwrap_or("").contains(needle))
.unwrap_or(false)
}
pub async fn create_item_group(&self, g: NewItemGroup) -> Result<Uuid, CatalogWriteError> {
let item_groups = ItemGroupRepository::new(self.db_pool.clone());
let mut tx = self.db_pool.begin().await?;
Self::relay_ambient_scope(&mut tx).await?;
if let Some(pid) = g.parent_id {
if !item_groups.exists_id(&mut *tx, pid).await? {
return Err(CatalogWriteError::ParentNotFound(pid));
}
}
let id = Uuid::new_v4();
let r = item_groups
.insert_item_group(
&mut *tx,
&NewItemGroupRow {
id,
code: &g.code,
name: &g.name,
parent_id: g.parent_id,
is_group: g.is_group,
},
)
.await;
match r {
Ok(_) => {
tx.commit().await?;
Ok(id)
}
Err(e) if Self::is_dup(&e, "code") => Err(CatalogWriteError::DuplicateItemCode(g.code)),
Err(e) => Err(e.into()),
}
}
pub async fn create_item(&self, i: NewItem) -> Result<Uuid, CatalogWriteError> {
let item_type = i.item_type.clone().unwrap_or_else(|| "physical_good".to_string());
let is_stock_item = i.is_stock_item && is_physical_item_type(&item_type);
if !(i.is_sales_item || i.is_purchase_item || is_stock_item) {
return Err(CatalogWriteError::NoUsageFlag);
}
let item_groups = ItemGroupRepository::new(self.db_pool.clone());
let uoms = UomRepository::new(self.db_pool.clone());
let mut tx = self.db_pool.begin().await?;
Self::relay_ambient_scope(&mut tx).await?;
if !item_groups.exists_id(&mut *tx, i.item_group_id).await? {
return Err(CatalogWriteError::ItemGroupNotFound(i.item_group_id));
}
if !uoms.exists_id(&mut *tx, i.default_uom_id).await? {
return Err(CatalogWriteError::UomNotFound(i.default_uom_id));
}
if let Some(bid) = i.brand_id {
let brands = BrandRepository::new(self.db_pool.clone());
if !brands.exists_id(&mut *tx, bid).await? {
return Err(CatalogWriteError::BrandNotFound(bid));
}
}
let id = Uuid::new_v4();
let tags = i.tags.clone().unwrap_or_else(|| serde_json::json!([]));
let data = i.data.clone().unwrap_or_else(|| serde_json::json!({}));
let items = ItemRepository::new(self.db_pool.clone());
let r = items
.insert_item(
&mut *tx,
&NewItemRow {
id,
item_code: &i.item_code,
name: &i.name,
description: i.description.as_deref(),
barcode: i.barcode.as_deref(),
brand_id: i.brand_id,
item_group_id: i.item_group_id,
default_uom_id: i.default_uom_id,
item_type: &item_type,
is_sales_item: i.is_sales_item,
is_purchase_item: i.is_purchase_item,
is_stock_item,
hsn_code: i.hsn_code.as_deref(),
is_taxable: i.is_taxable,
weight_per_unit: i.weight_per_unit,
standard_cost: i.standard_cost,
tags: &tags,
data: &data,
},
)
.await;
match r {
Ok(_) => {
tx.commit().await?;
Ok(id)
}
Err(e) if Self::is_dup(&e, "barcode") => Err(CatalogWriteError::DuplicateBarcode(
i.barcode.unwrap_or_default(),
)),
Err(e) if Self::is_dup(&e, "item_code") || Self::is_dup(&e, "items") => {
Err(CatalogWriteError::DuplicateItemCode(i.item_code))
}
Err(e) => Err(e.into()),
}
}
pub async fn create_uom(&self, u: NewUom) -> Result<Uuid, CatalogWriteError> {
let repo = UomRepository::new(self.db_pool.clone());
let mut tx = self.db_pool.begin().await?;
Self::relay_ambient_scope(&mut tx).await?;
let (relative_uom_id, relative_factor, factor) = match (u.relative_uom_id, u.relative_factor) {
(None, None) => (None, None, Decimal::ONE),
(Some(parent), Some(rf)) => {
if rf <= Decimal::ZERO {
return Err(CatalogWriteError::NonPositiveRelativeFactor);
}
if !repo.exists_id(&mut *tx, parent).await? {
return Err(CatalogWriteError::ParentNotFound(parent));
}
let parent_factor = repo
.find_factor(&mut *tx, parent)
.await?
.ok_or(CatalogWriteError::ParentNotFound(parent))?;
(Some(parent), Some(rf), parent_factor * rf)
}
(Some(_), None) | (None, Some(_)) => {
return Err(CatalogWriteError::RelativeShapeMismatch);
}
};
let id = Uuid::new_v4();
let ut = u.uom_type.clone().unwrap_or_else(|| "count".to_string());
let r = repo
.insert_uom(
&mut *tx,
&NewUomRow {
id,
code: &u.code,
name: &u.name,
uom_type: &ut,
decimal_places: u.decimal_places,
relative_uom_id,
relative_factor,
factor,
},
)
.await;
match r {
Ok(_) => {
repo.recompute_factors(&mut *tx).await?;
tx.commit().await?;
Ok(id)
}
Err(e) if Self::is_dup(&e, "code") => Err(CatalogWriteError::DuplicateUomCode(u.code)),
Err(e) => Err(e.into()),
}
}
pub async fn set_uom_relative(
&self,
uom_id: Uuid,
relative: Option<(Uuid, Decimal)>,
) -> Result<(), CatalogWriteError> {
let (relative_uom_id, relative_factor) = match relative {
None => (None, None),
Some((parent, rf)) => {
if rf <= Decimal::ZERO {
return Err(CatalogWriteError::NonPositiveRelativeFactor);
}
(Some(parent), Some(rf))
}
};
let uoms = UomRepository::new(self.db_pool.clone());
let mut tx = self.db_pool.begin().await?;
Self::relay_ambient_scope(&mut tx).await?;
if !uoms.exists_id(&mut *tx, uom_id).await? {
return Err(CatalogWriteError::UomNotFound(uom_id));
}
if let Some(parent) = relative_uom_id {
if !uoms.exists_id(&mut *tx, parent).await? {
return Err(CatalogWriteError::ParentNotFound(parent));
}
if uoms.is_self_or_descendant(&mut *tx, uom_id, parent).await? {
return Err(CatalogWriteError::UomCycle { parent });
}
}
uoms.set_relative(&mut *tx, uom_id, relative_uom_id, relative_factor)
.await?;
uoms.recompute_factors(&mut *tx).await?;
tx.commit().await?;
Ok(())
}
pub async fn convert_quantity(
&self,
qty: Decimal,
from_uom: Uuid,
to_uom: Uuid,
rounding: ConversionRounding,
) -> Result<Decimal, CatalogWriteError> {
let uoms = UomRepository::new(self.db_pool.clone());
let from_rows = uoms
.load_tree_chain(&self.db_pool, from_uom)
.await?
.ok_or(CatalogWriteError::UomNotFound(from_uom))?;
let to_rows = uoms
.load_tree_chain(&self.db_pool, to_uom)
.await?
.ok_or(CatalogWriteError::UomNotFound(to_uom))?;
let from = UomChain::from_rows(from_uom, from_rows).map_err(CatalogWriteError::Conversion)?;
let to = UomChain::from_rows(to_uom, to_rows).map_err(CatalogWriteError::Conversion)?;
convert_quantity(qty, &from, &to, rounding).map_err(CatalogWriteError::Conversion)
}
pub async fn delete_uom(&self, uom_id: Uuid) -> Result<(), CatalogWriteError> {
let uoms = UomRepository::new(self.db_pool.clone());
let mut tx = self.db_pool.begin().await?;
Self::relay_ambient_scope(&mut tx).await?;
let code: Option<String> = sqlx::query_scalar(
"SELECT code FROM catalog.uoms \
WHERE id = $1 AND (metadata->>'deleted_at') IS NULL",
)
.bind(uom_id)
.fetch_optional(&mut *tx)
.await?;
let code = code.ok_or(CatalogWriteError::UomNotFound(uom_id))?;
if uoms.find_protection(&mut *tx, uom_id).await? != Some(false) {
return Err(CatalogWriteError::ProtectedUom { code });
}
if uoms.count_live_children(&mut *tx, uom_id).await? > 0 {
return Err(CatalogWriteError::UomHasChildren { code });
}
if uoms.exists_live_item_using(&mut *tx, uom_id).await? {
return Err(CatalogWriteError::UomInUse { code });
}
uoms.soft_delete_uom(&mut *tx, uom_id).await?;
tx.commit().await?;
Ok(())
}
pub async fn create_attribute(&self, a: NewAttribute) -> Result<Uuid, CatalogWriteError> {
let id = Uuid::new_v4();
let at = a.attribute_type.clone().unwrap_or_else(|| "other".to_string());
let repo = AttributeRepository::new(self.db_pool.clone());
let mut tx = self.db_pool.begin().await?;
Self::relay_ambient_scope(&mut tx).await?;
let r = repo
.insert_attribute(
&mut *tx,
&NewAttributeRow {
id,
code: &a.code,
name: &a.name,
attribute_type: &at,
},
)
.await;
match r {
Ok(_) => {
tx.commit().await?;
Ok(id)
}
Err(e) if Self::is_dup(&e, "code") => Err(CatalogWriteError::DuplicateAttributeCode(a.code)),
Err(e) => Err(e.into()),
}
}
pub async fn create_attribute_value(&self, v: NewAttributeValue) -> Result<Uuid, CatalogWriteError> {
let attrs = AttributeRepository::new(self.db_pool.clone());
let mut tx = self.db_pool.begin().await?;
Self::relay_ambient_scope(&mut tx).await?;
if !attrs.exists_id(&mut *tx, v.attribute_id).await? {
return Err(CatalogWriteError::AttributeNotFound(v.attribute_id));
}
let id = Uuid::new_v4();
let repo = AttributeValueRepository::new(self.db_pool.clone());
let r = repo
.insert_attribute_value(
&mut *tx,
&NewAttributeValueRow {
id,
attribute_id: v.attribute_id,
code: &v.code,
label: &v.label,
label_en: v.label_en.as_deref(),
swatch_hex: v.swatch_hex.as_deref(),
sort_order: v.sort_order,
},
)
.await;
match r {
Ok(_) => {
tx.commit().await?;
Ok(id)
}
Err(e) if Self::is_dup(&e, "code") => Err(CatalogWriteError::DuplicateValueCode(v.code)),
Err(e) => Err(e.into()),
}
}
pub async fn create_item_variant(&self, v: NewItemVariant) -> Result<Uuid, CatalogWriteError> {
let items = ItemRepository::new(self.db_pool.clone());
if v.options.is_empty() {
return Err(CatalogWriteError::NoOptions);
}
let attr_values = AttributeValueRepository::new(self.db_pool.clone());
let attrs = AttributeRepository::new(self.db_pool.clone());
let mut tx = self.db_pool.begin().await?;
Self::relay_ambient_scope(&mut tx).await?;
if !items.exists_id(&mut *tx, v.item_id).await? {
return Err(CatalogWriteError::ItemNotFound(v.item_id));
}
let mut labels: Vec<String> = Vec::with_capacity(v.options.len());
for (attr_code, val_code) in &v.options {
let row = attr_values
.find_value_with_attribute(&mut *tx, attr_code, val_code)
.await?;
match row {
Some(r) => labels.push(r.label),
None => {
let attr_ok = attrs.find_id_by_code(&mut *tx, attr_code).await?;
return if attr_ok.is_some() {
Err(CatalogWriteError::UnknownAttributeValue(format!("{attr_code}={val_code}")))
} else {
Err(CatalogWriteError::UnknownAttribute(attr_code.clone()))
};
}
}
}
let label = v.variant_label.clone().unwrap_or_else(|| labels.join(" / "));
let options_json = serde_json::to_value(&v.options).unwrap_or(serde_json::json!({}));
let id = Uuid::new_v4();
let variants = ItemVariantRepository::new(self.db_pool.clone());
let r = variants
.insert_variant(
&mut *tx,
&NewItemVariantRow {
id,
item_id: v.item_id,
sku: &v.sku,
variant_label: &label,
options: &options_json,
barcode: v.barcode.as_deref(),
is_default: v.is_default,
weight_per_unit: v.weight_per_unit,
},
)
.await;
if let Err(e) = r {
drop(tx);
return if Self::is_dup(&e, "barcode") {
Err(CatalogWriteError::DuplicateBarcode(v.barcode.unwrap_or_default()))
} else if e.as_database_error().map(|d| d.is_unique_violation()).unwrap_or(false) {
Err(CatalogWriteError::DuplicateSku(v.sku))
} else {
Err(e.into())
};
}
items.set_has_variants_true(&mut *tx, v.item_id).await?;
tx.commit().await?;
Ok(id)
}
pub async fn delete_item_variant(&self, variant_id: Uuid) -> Result<(), CatalogWriteError> {
let variants = ItemVariantRepository::new(self.db_pool.clone());
let mut tx = self.db_pool.begin().await?;
Self::relay_ambient_scope(&mut tx).await?;
let item_id = variants
.find_item_id_for_live(&mut *tx, variant_id)
.await?
.ok_or(CatalogWriteError::ItemVariantNotFound(variant_id))?;
variants.soft_delete_variant(&mut *tx, variant_id).await?;
let remaining = variants.count_live_variants(&mut *tx, item_id).await?;
if remaining == 0 {
let items = ItemRepository::new(self.db_pool.clone());
items.set_has_variants_false(&mut *tx, item_id).await?;
}
tx.commit().await?;
Ok(())
}
pub async fn transition_item_status(
&self,
item_id: Uuid,
target: CatalogStatus,
) -> Result<(), CatalogWriteError> {
let items = ItemRepository::new(self.db_pool.clone());
let mut tx = self.db_pool.begin().await?;
Self::relay_ambient_scope(&mut tx).await?;
let current = items
.find_status(&mut *tx, item_id)
.await?
.ok_or(CatalogWriteError::ItemNotFound(item_id))?;
if !Self::transition_allowed(current, target) {
return Err(CatalogWriteError::InvalidStatusTransition { from: current, to: target });
}
items.set_status(&mut *tx, item_id, target).await?;
tx.commit().await?;
Ok(())
}
fn transition_allowed(from: CatalogStatus, to: CatalogStatus) -> bool {
use CatalogStatus::*;
matches!(
(from, to),
(Active, Inactive) | (Inactive, Active) | (Active, Discontinued) | (Inactive, Discontinued)
)
}
pub async fn create_brand(&self, b: NewBrand) -> Result<Uuid, CatalogWriteError> {
let id = Uuid::new_v4();
let repo = BrandRepository::new(self.db_pool.clone());
let mut tx = self.db_pool.begin().await?;
Self::relay_ambient_scope(&mut tx).await?;
let r = repo
.insert_brand(
&mut *tx,
&NewBrandRow {
id,
code: &b.code,
name: &b.name,
short_description: b.short_description.as_deref(),
description: b.description.as_deref(),
logo_url: b.logo_url.as_deref(),
sort_order: b.sort_order,
},
)
.await;
match r {
Ok(_) => {
tx.commit().await?;
Ok(id)
}
Err(e) if Self::is_dup(&e, "code") => Err(CatalogWriteError::DuplicateBrandCode(b.code)),
Err(e) => Err(e.into()),
}
}
}