use super::map_db_error;
use chrono::{DateTime, Utc};
use rust_decimal::Decimal;
use sqlx::postgres::PgPool;
use sqlx::{FromRow, QueryBuilder};
use stateset_core::{
BatchResult, CommerceError, CreateProduct, CreateProductVariant, Product, ProductFilter,
ProductId, ProductRepository, ProductStatus, ProductType, ProductVariant, Result,
UpdateProduct, validate_batch_size,
};
use uuid::Uuid;
#[derive(Debug, Clone)]
pub struct PgProductRepository {
pool: PgPool,
}
#[derive(FromRow)]
struct ProductRow {
id: Uuid,
name: String,
slug: String,
description: String,
status: String,
product_type: String,
attributes: serde_json::Value,
seo: Option<serde_json::Value>,
created_at: DateTime<Utc>,
updated_at: DateTime<Utc>,
version: i32,
}
#[derive(FromRow)]
struct VariantRow {
id: Uuid,
product_id: Uuid,
sku: String,
name: String,
price: Decimal,
compare_at_price: Option<Decimal>,
cost: Option<Decimal>,
barcode: Option<String>,
weight: Option<Decimal>,
weight_unit: Option<String>,
options: serde_json::Value,
is_default: bool,
is_active: bool,
created_at: DateTime<Utc>,
updated_at: DateTime<Utc>,
}
impl PgProductRepository {
pub const fn new(pool: PgPool) -> Self {
Self { pool }
}
fn row_to_product(row: ProductRow) -> Result<Product> {
let ProductRow {
id,
name,
slug,
description,
status,
product_type,
attributes,
seo,
created_at,
updated_at,
version: _,
} = row;
let status: ProductStatus = status.parse().map_err(|e| {
CommerceError::DatabaseError(format!("Invalid product.status '{}': {}", status, e))
})?;
let product_type: ProductType = product_type.parse().map_err(|e| {
CommerceError::DatabaseError(format!(
"Invalid product.product_type '{}': {}",
product_type, e
))
})?;
let attributes = serde_json::from_value(attributes).map_err(|e| {
CommerceError::DatabaseError(format!("Invalid JSON for product.attributes: {}", e))
})?;
let seo = seo.map(serde_json::from_value).transpose().map_err(|e| {
CommerceError::DatabaseError(format!("Invalid JSON for product.seo: {}", e))
})?;
Ok(Product {
id: ProductId::from(id),
name,
slug,
description,
status,
product_type,
attributes,
seo,
created_at,
updated_at,
})
}
fn row_to_variant(row: VariantRow) -> Result<ProductVariant> {
let VariantRow {
id,
product_id,
sku,
name,
price,
compare_at_price,
cost,
barcode,
weight,
weight_unit,
options,
is_default,
is_active,
created_at,
updated_at,
} = row;
let options = serde_json::from_value(options).map_err(|e| {
CommerceError::DatabaseError(format!("Invalid JSON for product_variant.options: {}", e))
})?;
Ok(ProductVariant {
id,
product_id: ProductId::from(product_id),
sku,
name,
price,
compare_at_price,
cost,
barcode,
weight,
weight_unit,
options,
is_default,
is_active,
created_at,
updated_at,
})
}
pub async fn create_async(&self, input: CreateProduct) -> Result<Product> {
let id = ProductId::new();
let now = Utc::now();
let slug = input.slug.clone().unwrap_or_else(|| Product::generate_slug(&input.name));
let description = input.description.clone().unwrap_or_default();
let product_type = input.product_type.unwrap_or_default();
let attributes = input.attributes.clone().unwrap_or_default();
let attributes_json = serde_json::to_value(&attributes).unwrap_or_default();
let seo_json = input.seo.as_ref().map(|s| serde_json::to_value(s).unwrap_or_default());
sqlx::query(
r#"
INSERT INTO products (id, name, slug, description, status, product_type, attributes, seo, created_at, updated_at)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)
"#,
)
.bind(id.into_uuid())
.bind(&input.name)
.bind(&slug)
.bind(&description)
.bind(ProductStatus::Draft.to_string())
.bind(product_type.to_string())
.bind(&attributes_json)
.bind(&seo_json)
.bind(now)
.bind(now)
.execute(&self.pool)
.await
.map_err(map_db_error)?;
if let Some(variant_inputs) = &input.variants {
for (i, vi) in variant_inputs.iter().enumerate() {
self.add_variant_async(id.into_uuid(), vi.clone(), i == 0).await?;
}
}
Ok(Product {
id,
name: input.name,
slug,
description,
status: ProductStatus::Draft,
product_type,
attributes,
seo: input.seo,
created_at: now,
updated_at: now,
})
}
pub async fn get_async(&self, id: ProductId) -> Result<Option<Product>> {
let row = sqlx::query_as::<_, ProductRow>("SELECT * FROM products WHERE id = $1")
.bind(id.into_uuid())
.fetch_optional(&self.pool)
.await
.map_err(map_db_error)?;
row.map(Self::row_to_product).transpose()
}
pub async fn get_by_slug_async(&self, slug: &str) -> Result<Option<Product>> {
let row = sqlx::query_as::<_, ProductRow>("SELECT * FROM products WHERE slug = $1")
.bind(slug)
.fetch_optional(&self.pool)
.await
.map_err(map_db_error)?;
row.map(Self::row_to_product).transpose()
}
pub async fn update_async(&self, id: ProductId, input: UpdateProduct) -> Result<Product> {
let now = Utc::now();
let existing_row = sqlx::query_as::<_, ProductRow>("SELECT * FROM products WHERE id = $1")
.bind(id.into_uuid())
.fetch_optional(&self.pool)
.await
.map_err(map_db_error)?
.ok_or(CommerceError::ProductNotFound(id.into_uuid()))?;
let current_version = existing_row.version;
let existing = Self::row_to_product(existing_row)?;
let new_name = input.name.unwrap_or(existing.name);
let new_slug = input.slug.unwrap_or(existing.slug);
let new_description = input.description.unwrap_or(existing.description);
let new_status = input.status.unwrap_or(existing.status);
let new_attributes = input.attributes.unwrap_or(existing.attributes);
let new_seo = input.seo.or(existing.seo);
let attributes_json = serde_json::to_value(&new_attributes).unwrap_or_default();
let seo_json = new_seo.as_ref().map(|s| serde_json::to_value(s).unwrap_or_default());
let result = sqlx::query(
r#"
UPDATE products
SET name = $1, slug = $2, description = $3, status = $4,
attributes = $5, seo = $6, updated_at = $7, version = version + 1
WHERE id = $8 AND version = $9
"#,
)
.bind(&new_name)
.bind(&new_slug)
.bind(&new_description)
.bind(new_status.to_string())
.bind(&attributes_json)
.bind(&seo_json)
.bind(now)
.bind(id.into_uuid())
.bind(current_version)
.execute(&self.pool)
.await
.map_err(map_db_error)?;
if result.rows_affected() == 0 {
return Err(CommerceError::VersionConflict {
entity: "product".to_string(),
id: id.to_string(),
expected_version: current_version,
});
}
self.get_async(id).await?.ok_or(CommerceError::ProductNotFound(id.into_uuid()))
}
pub async fn list_async(&self, filter: ProductFilter) -> Result<Vec<Product>> {
let ProductFilter {
status,
product_type,
search,
category,
min_price,
max_price,
in_stock,
limit,
offset,
after_cursor: _,
} = filter;
let mut builder = QueryBuilder::new("SELECT * FROM products WHERE 1=1");
if let Some(status) = status {
builder.push(" AND status = ").push_bind(status.to_string());
} else {
builder.push(" AND status != 'archived'");
}
if let Some(product_type) = product_type {
builder.push(" AND product_type = ").push_bind(product_type.to_string());
}
if let Some(search) = search {
let pattern = format!("%{}%", search);
builder
.push(" AND (name ILIKE ")
.push_bind(pattern.clone())
.push(" OR description ILIKE ")
.push_bind(pattern)
.push(')');
}
if let Some(category) = category {
builder.push(
" AND EXISTS (SELECT 1 FROM jsonb_array_elements(attributes) attr \
WHERE (attr->>'name' = 'category' OR attr->>'group' = 'category') \
AND attr->>'value' = ",
);
builder.push_bind(category).push(')');
}
if min_price.is_some() || max_price.is_some() {
builder.push(
" AND EXISTS (SELECT 1 FROM product_variants pv \
WHERE pv.product_id = products.id AND pv.is_active = true",
);
if let Some(min_price) = min_price {
builder.push(" AND pv.price >= ").push_bind(min_price);
}
if let Some(max_price) = max_price {
builder.push(" AND pv.price <= ").push_bind(max_price);
}
builder.push(')');
}
if let Some(in_stock) = in_stock {
if in_stock {
builder.push(
" AND EXISTS (SELECT 1 FROM product_variants pv_stock \
JOIN inventory_items ii ON ii.sku = pv_stock.sku \
JOIN inventory_balances ib ON ib.item_id = ii.id \
WHERE pv_stock.product_id = products.id \
AND pv_stock.is_active = true \
AND ib.quantity_available > 0)",
);
} else {
builder.push(
" AND NOT EXISTS (SELECT 1 FROM product_variants pv_stock \
JOIN inventory_items ii ON ii.sku = pv_stock.sku \
JOIN inventory_balances ib ON ib.item_id = ii.id \
WHERE pv_stock.product_id = products.id \
AND pv_stock.is_active = true \
AND ib.quantity_available > 0)",
);
}
}
builder.push(" ORDER BY created_at DESC");
builder.push(" LIMIT ").push_bind(super::effective_limit(limit));
if let Some(offset) = offset {
builder.push(" OFFSET ").push_bind(offset as i64);
}
let rows = builder
.build_query_as::<ProductRow>()
.fetch_all(&self.pool)
.await
.map_err(map_db_error)?;
let mut products = Vec::with_capacity(rows.len());
for row in rows {
products.push(Self::row_to_product(row)?);
}
Ok(products)
}
pub async fn delete_async(&self, id: ProductId) -> Result<()> {
sqlx::query("UPDATE products SET status = 'archived', updated_at = $1 WHERE id = $2")
.bind(Utc::now())
.bind(id.into_uuid())
.execute(&self.pool)
.await
.map_err(map_db_error)?;
Ok(())
}
async fn add_variant_async(
&self,
product_id: Uuid,
input: CreateProductVariant,
is_default: bool,
) -> Result<ProductVariant> {
let id = Uuid::new_v4();
let now = Utc::now();
let options = input.options.clone().unwrap_or_default();
let options_json = serde_json::to_value(&options).unwrap_or_default();
sqlx::query(
r#"
INSERT INTO product_variants (id, product_id, sku, name, price, compare_at_price, cost,
barcode, weight, weight_unit, options, is_default, is_active,
created_at, updated_at)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15)
"#,
)
.bind(id)
.bind(product_id)
.bind(&input.sku)
.bind(input.name.as_deref().unwrap_or(&input.sku))
.bind(input.price)
.bind(input.compare_at_price)
.bind(input.cost)
.bind(&input.barcode)
.bind(input.weight)
.bind(&input.weight_unit)
.bind(&options_json)
.bind(input.is_default.unwrap_or(is_default))
.bind(true)
.bind(now)
.bind(now)
.execute(&self.pool)
.await
.map_err(map_db_error)?;
Ok(ProductVariant {
id,
product_id: ProductId::from(product_id),
sku: input.sku,
name: input.name.unwrap_or_default(),
price: input.price,
compare_at_price: input.compare_at_price,
cost: input.cost,
barcode: input.barcode,
weight: input.weight,
weight_unit: input.weight_unit,
options,
is_default: input.is_default.unwrap_or(is_default),
is_active: true,
created_at: now,
updated_at: now,
})
}
pub async fn add_variant_public_async(
&self,
product_id: ProductId,
input: CreateProductVariant,
) -> Result<ProductVariant> {
self.add_variant_async(product_id.into_uuid(), input, false).await
}
pub async fn get_variant_async(&self, id: Uuid) -> Result<Option<ProductVariant>> {
let row = sqlx::query_as::<_, VariantRow>("SELECT * FROM product_variants WHERE id = $1")
.bind(id)
.fetch_optional(&self.pool)
.await
.map_err(map_db_error)?;
row.map(Self::row_to_variant).transpose()
}
pub async fn get_variant_by_sku_async(&self, sku: &str) -> Result<Option<ProductVariant>> {
let row = sqlx::query_as::<_, VariantRow>("SELECT * FROM product_variants WHERE sku = $1")
.bind(sku)
.fetch_optional(&self.pool)
.await
.map_err(map_db_error)?;
row.map(Self::row_to_variant).transpose()
}
pub async fn update_variant_async(
&self,
id: Uuid,
input: CreateProductVariant,
) -> Result<ProductVariant> {
let now = Utc::now();
let current_version: i32 =
sqlx::query_scalar("SELECT version FROM product_variants WHERE id = $1")
.bind(id)
.fetch_one(&self.pool)
.await
.map_err(|e| match e {
sqlx::Error::RowNotFound => CommerceError::ProductVariantNotFound(id),
e => map_db_error(e),
})?;
let options_json =
serde_json::to_value(input.options.clone().unwrap_or_default()).unwrap_or_default();
let result = sqlx::query(
r#"
UPDATE product_variants
SET sku = $1, name = $2, price = $3, compare_at_price = $4, cost = $5,
barcode = $6, weight = $7, weight_unit = $8, options = $9, updated_at = $10, version = version + 1
WHERE id = $11 AND version = $12
"#,
)
.bind(&input.sku)
.bind(input.name.as_deref().unwrap_or(&input.sku))
.bind(input.price)
.bind(input.compare_at_price)
.bind(input.cost)
.bind(&input.barcode)
.bind(input.weight)
.bind(&input.weight_unit)
.bind(&options_json)
.bind(now)
.bind(id)
.bind(current_version)
.execute(&self.pool)
.await
.map_err(map_db_error)?;
if result.rows_affected() == 0 {
return Err(CommerceError::VersionConflict {
entity: "product_variant".to_string(),
id: id.to_string(),
expected_version: current_version,
});
}
self.get_variant_async(id).await?.ok_or(CommerceError::ProductVariantNotFound(id))
}
pub async fn delete_variant_async(&self, id: Uuid) -> Result<()> {
sqlx::query("DELETE FROM product_variants WHERE id = $1")
.bind(id)
.execute(&self.pool)
.await
.map_err(map_db_error)?;
Ok(())
}
pub async fn get_variants_async(&self, product_id: ProductId) -> Result<Vec<ProductVariant>> {
let rows =
sqlx::query_as::<_, VariantRow>("SELECT * FROM product_variants WHERE product_id = $1")
.bind(product_id.into_uuid())
.fetch_all(&self.pool)
.await
.map_err(map_db_error)?;
let mut variants = Vec::with_capacity(rows.len());
for row in rows {
variants.push(Self::row_to_variant(row)?);
}
Ok(variants)
}
pub async fn count_async(&self, filter: ProductFilter) -> Result<u64> {
let ProductFilter {
status,
product_type,
search,
category,
min_price,
max_price,
in_stock,
limit: _,
offset: _,
after_cursor: _,
} = filter;
let mut builder = QueryBuilder::new("SELECT COUNT(*) FROM products WHERE 1=1");
if let Some(status) = status {
builder.push(" AND status = ").push_bind(status.to_string());
} else {
builder.push(" AND status != 'archived'");
}
if let Some(product_type) = product_type {
builder.push(" AND product_type = ").push_bind(product_type.to_string());
}
if let Some(search) = search {
let pattern = format!("%{}%", search);
builder
.push(" AND (name ILIKE ")
.push_bind(pattern.clone())
.push(" OR description ILIKE ")
.push_bind(pattern)
.push(')');
}
if let Some(category) = category {
builder.push(
" AND EXISTS (SELECT 1 FROM jsonb_array_elements(attributes) attr \
WHERE (attr->>'name' = 'category' OR attr->>'group' = 'category') \
AND attr->>'value' = ",
);
builder.push_bind(category).push(')');
}
if min_price.is_some() || max_price.is_some() {
builder.push(
" AND EXISTS (SELECT 1 FROM product_variants pv \
WHERE pv.product_id = products.id AND pv.is_active = true",
);
if let Some(min_price) = min_price {
builder.push(" AND pv.price >= ").push_bind(min_price);
}
if let Some(max_price) = max_price {
builder.push(" AND pv.price <= ").push_bind(max_price);
}
builder.push(')');
}
if let Some(in_stock) = in_stock {
if in_stock {
builder.push(
" AND EXISTS (SELECT 1 FROM product_variants pv_stock \
JOIN inventory_items ii ON ii.sku = pv_stock.sku \
JOIN inventory_balances ib ON ib.item_id = ii.id \
WHERE pv_stock.product_id = products.id \
AND pv_stock.is_active = true \
AND ib.quantity_available > 0)",
);
} else {
builder.push(
" AND NOT EXISTS (SELECT 1 FROM product_variants pv_stock \
JOIN inventory_items ii ON ii.sku = pv_stock.sku \
JOIN inventory_balances ib ON ib.item_id = ii.id \
WHERE pv_stock.product_id = products.id \
AND pv_stock.is_active = true \
AND ib.quantity_available > 0)",
);
}
}
let count: (i64,) =
builder.build_query_as().fetch_one(&self.pool).await.map_err(map_db_error)?;
Ok(count.0 as u64)
}
pub async fn create_batch_async(
&self,
inputs: Vec<CreateProduct>,
) -> Result<BatchResult<Product>> {
validate_batch_size(&inputs)?;
let mut result = BatchResult::with_capacity(inputs.len());
for (index, input) in inputs.into_iter().enumerate() {
match self.create_async(input).await {
Ok(product) => result.record_success(product),
Err(e) => result.record_failure(index, None, &e),
}
}
Ok(result)
}
pub async fn create_batch_atomic_async(
&self,
inputs: Vec<CreateProduct>,
) -> Result<Vec<Product>> {
validate_batch_size(&inputs)?;
let mut tx = self.pool.begin().await.map_err(map_db_error)?;
let mut products = Vec::with_capacity(inputs.len());
for input in inputs {
let id = ProductId::new();
let now = Utc::now();
let slug = input.slug.clone().unwrap_or_else(|| Product::generate_slug(&input.name));
let description = input.description.clone().unwrap_or_default();
let product_type = input.product_type.unwrap_or_default();
let attributes = input.attributes.clone().unwrap_or_default();
let attributes_json = serde_json::to_value(&attributes).unwrap_or_default();
let seo_json = input.seo.as_ref().map(|s| serde_json::to_value(s).unwrap_or_default());
sqlx::query(
r#"
INSERT INTO products (id, name, slug, description, status, product_type, attributes, seo, created_at, updated_at)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10)
"#,
)
.bind(id.into_uuid())
.bind(&input.name)
.bind(&slug)
.bind(&description)
.bind(ProductStatus::Draft.to_string())
.bind(product_type.to_string())
.bind(&attributes_json)
.bind(&seo_json)
.bind(now)
.bind(now)
.execute(tx.as_mut())
.await
.map_err(map_db_error)?;
if let Some(variant_inputs) = &input.variants {
for (i, vi) in variant_inputs.iter().enumerate() {
let variant_id = Uuid::new_v4();
let options = vi.options.clone().unwrap_or_default();
let options_json = serde_json::to_value(&options).unwrap_or_default();
sqlx::query(
r#"
INSERT INTO product_variants (id, product_id, sku, name, price, compare_at_price, cost,
barcode, weight, weight_unit, options, is_default, is_active,
created_at, updated_at)
VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9, $10, $11, $12, $13, $14, $15)
"#,
)
.bind(variant_id)
.bind(id.into_uuid())
.bind(&vi.sku)
.bind(vi.name.as_deref().unwrap_or(&vi.sku))
.bind(vi.price)
.bind(vi.compare_at_price)
.bind(vi.cost)
.bind(&vi.barcode)
.bind(vi.weight)
.bind(&vi.weight_unit)
.bind(&options_json)
.bind(vi.is_default.unwrap_or(i == 0))
.bind(true)
.bind(now)
.bind(now)
.execute(tx.as_mut())
.await
.map_err(map_db_error)?;
}
}
products.push(Product {
id,
name: input.name,
slug,
description,
status: ProductStatus::Draft,
product_type,
attributes,
seo: input.seo,
created_at: now,
updated_at: now,
});
}
tx.commit().await.map_err(map_db_error)?;
Ok(products)
}
pub async fn update_batch_async(
&self,
updates: Vec<(ProductId, UpdateProduct)>,
) -> Result<BatchResult<Product>> {
validate_batch_size(&updates)?;
let mut result = BatchResult::with_capacity(updates.len());
for (index, (id, input)) in updates.into_iter().enumerate() {
match self.update_async(id, input).await {
Ok(product) => result.record_success(product),
Err(e) => result.record_failure(index, Some(id.to_string()), &e),
}
}
Ok(result)
}
pub async fn update_batch_atomic_async(
&self,
updates: Vec<(ProductId, UpdateProduct)>,
) -> Result<Vec<Product>> {
validate_batch_size(&updates)?;
let mut tx = self.pool.begin().await.map_err(map_db_error)?;
let mut products = Vec::with_capacity(updates.len());
for (id, input) in updates {
let now = Utc::now();
let existing_row =
sqlx::query_as::<_, ProductRow>("SELECT * FROM products WHERE id = $1 FOR UPDATE")
.bind(id.into_uuid())
.fetch_optional(tx.as_mut())
.await
.map_err(map_db_error)?
.ok_or(CommerceError::ProductNotFound(id.into_uuid()))?;
let current_version = existing_row.version;
let existing = Self::row_to_product(existing_row)?;
let new_name = input.name.unwrap_or(existing.name);
let new_slug = input.slug.unwrap_or(existing.slug);
let new_description = input.description.unwrap_or(existing.description);
let new_status = input.status.unwrap_or(existing.status);
let new_attributes = input.attributes.unwrap_or(existing.attributes);
let new_seo = input.seo.or(existing.seo);
let attributes_json = serde_json::to_value(&new_attributes).unwrap_or_default();
let seo_json = new_seo.as_ref().map(|s| serde_json::to_value(s).unwrap_or_default());
let result = sqlx::query(
r#"
UPDATE products
SET name = $1, slug = $2, description = $3, status = $4,
attributes = $5, seo = $6, updated_at = $7, version = version + 1
WHERE id = $8 AND version = $9
"#,
)
.bind(&new_name)
.bind(&new_slug)
.bind(&new_description)
.bind(new_status.to_string())
.bind(&attributes_json)
.bind(&seo_json)
.bind(now)
.bind(id.into_uuid())
.bind(current_version)
.execute(tx.as_mut())
.await
.map_err(map_db_error)?;
if result.rows_affected() == 0 {
return Err(CommerceError::VersionConflict {
entity: "product".to_string(),
id: id.to_string(),
expected_version: current_version,
});
}
products.push(Product {
id,
name: new_name,
slug: new_slug,
description: new_description,
status: new_status,
product_type: existing.product_type,
attributes: new_attributes,
seo: new_seo,
created_at: existing.created_at,
updated_at: now,
});
}
tx.commit().await.map_err(map_db_error)?;
Ok(products)
}
pub async fn delete_batch_async(&self, ids: Vec<ProductId>) -> Result<BatchResult<ProductId>> {
validate_batch_size(&ids)?;
let mut result = BatchResult::with_capacity(ids.len());
for (index, id) in ids.into_iter().enumerate() {
match self.delete_async(id).await {
Ok(()) => result.record_success(id),
Err(e) => result.record_failure(index, Some(id.to_string()), &e),
}
}
Ok(result)
}
pub async fn delete_batch_atomic_async(&self, ids: Vec<ProductId>) -> Result<()> {
validate_batch_size(&ids)?;
if ids.is_empty() {
return Ok(());
}
let mut tx = self.pool.begin().await.map_err(map_db_error)?;
let now = Utc::now();
let uuid_ids: Vec<Uuid> = ids.iter().map(|id| id.into_uuid()).collect();
sqlx::query("UPDATE products SET status = 'archived', updated_at = $1 WHERE id = ANY($2)")
.bind(now)
.bind(&uuid_ids)
.execute(tx.as_mut())
.await
.map_err(map_db_error)?;
tx.commit().await.map_err(map_db_error)?;
Ok(())
}
pub async fn get_batch_async(&self, ids: Vec<ProductId>) -> Result<Vec<Product>> {
validate_batch_size(&ids)?;
if ids.is_empty() {
return Ok(Vec::new());
}
let uuid_ids: Vec<Uuid> = ids.iter().map(|id| id.into_uuid()).collect();
let rows = sqlx::query_as::<_, ProductRow>("SELECT * FROM products WHERE id = ANY($1)")
.bind(&uuid_ids)
.fetch_all(&self.pool)
.await
.map_err(map_db_error)?;
let mut products = Vec::with_capacity(rows.len());
for row in rows {
products.push(Self::row_to_product(row)?);
}
Ok(products)
}
}
impl ProductRepository for PgProductRepository {
fn create(&self, input: CreateProduct) -> Result<Product> {
super::block_on(self.create_async(input))
}
fn get(&self, id: ProductId) -> Result<Option<Product>> {
super::block_on(self.get_async(id))
}
fn get_by_slug(&self, slug: &str) -> Result<Option<Product>> {
super::block_on(self.get_by_slug_async(slug))
}
fn update(&self, id: ProductId, input: UpdateProduct) -> Result<Product> {
super::block_on(self.update_async(id, input))
}
fn list(&self, filter: ProductFilter) -> Result<Vec<Product>> {
super::block_on(self.list_async(filter))
}
fn delete(&self, id: ProductId) -> Result<()> {
super::block_on(self.delete_async(id))
}
fn add_variant(
&self,
product_id: ProductId,
variant: CreateProductVariant,
) -> Result<ProductVariant> {
super::block_on(self.add_variant_public_async(product_id, variant))
}
fn get_variant(&self, id: Uuid) -> Result<Option<ProductVariant>> {
super::block_on(self.get_variant_async(id))
}
fn get_variant_by_sku(&self, sku: &str) -> Result<Option<ProductVariant>> {
super::block_on(self.get_variant_by_sku_async(sku))
}
fn update_variant(&self, id: Uuid, variant: CreateProductVariant) -> Result<ProductVariant> {
super::block_on(self.update_variant_async(id, variant))
}
fn delete_variant(&self, id: Uuid) -> Result<()> {
super::block_on(self.delete_variant_async(id))
}
fn get_variants(&self, product_id: ProductId) -> Result<Vec<ProductVariant>> {
super::block_on(self.get_variants_async(product_id))
}
fn count(&self, filter: ProductFilter) -> Result<u64> {
super::block_on(self.count_async(filter))
}
fn create_batch(&self, inputs: Vec<CreateProduct>) -> Result<BatchResult<Product>> {
super::block_on(self.create_batch_async(inputs))
}
fn create_batch_atomic(&self, inputs: Vec<CreateProduct>) -> Result<Vec<Product>> {
super::block_on(self.create_batch_atomic_async(inputs))
}
fn update_batch(
&self,
updates: Vec<(ProductId, UpdateProduct)>,
) -> Result<BatchResult<Product>> {
super::block_on(self.update_batch_async(updates))
}
fn update_batch_atomic(
&self,
updates: Vec<(ProductId, UpdateProduct)>,
) -> Result<Vec<Product>> {
super::block_on(self.update_batch_atomic_async(updates))
}
fn delete_batch(&self, ids: Vec<ProductId>) -> Result<BatchResult<ProductId>> {
super::block_on(self.delete_batch_async(ids))
}
fn delete_batch_atomic(&self, ids: Vec<ProductId>) -> Result<()> {
super::block_on(self.delete_batch_atomic_async(ids))
}
fn get_batch(&self, ids: Vec<ProductId>) -> Result<Vec<Product>> {
super::block_on(self.get_batch_async(ids))
}
}