stateset-db 1.22.0

//! PostgreSQL inventory repository implementation

use super::map_db_error;
use chrono::{DateTime, Utc};
use rust_decimal::Decimal;
use sqlx::FromRow;
use sqlx::postgres::PgPool;
use stateset_core::{
    AdjustInventory, BatchResult, CommerceError, CreateInventoryItem, InventoryBalance,
    InventoryFilter, InventoryItem, InventoryRepository, InventoryReservation,
    InventoryTransaction, LocationStock, ReservationStatus, ReserveInventory, Result, StockLevel,
    TransactionType, validate_batch_size, validate_quantity,
};
use uuid::Uuid;

/// PostgreSQL implementation of `InventoryRepository`
#[derive(Debug, Clone)]
pub struct PgInventoryRepository {
    pool: PgPool,
}

#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum ReservationConfirmOutcome {
    Confirmed,
    Expired,
}

#[derive(FromRow)]
struct InventoryItemRow {
    id: i64,
    sku: String,
    name: String,
    description: Option<String>,
    unit_of_measure: String,
    is_active: bool,
    created_at: DateTime<Utc>,
    updated_at: DateTime<Utc>,
}

#[derive(FromRow)]
struct InventoryBalanceRow {
    id: i64,
    item_id: i64,
    location_id: i32,
    quantity_on_hand: Decimal,
    quantity_allocated: Decimal,
    quantity_available: Decimal,
    reorder_point: Option<Decimal>,
    safety_stock: Option<Decimal>,
    version: i32,
    last_counted_at: Option<DateTime<Utc>>,
    updated_at: DateTime<Utc>,
}

#[derive(FromRow)]
struct ReservationRow {
    id: Uuid,
    item_id: i64,
    location_id: i32,
    quantity: Decimal,
    status: String,
    reference_type: String,
    reference_id: String,
    expires_at: Option<DateTime<Utc>>,
    created_at: DateTime<Utc>,
}

#[derive(FromRow)]
struct TransactionRow {
    id: i64,
    item_id: i64,
    location_id: i32,
    transaction_type: String,
    quantity: Decimal,
    reference_type: Option<String>,
    reference_id: Option<String>,
    reason: Option<String>,
    created_by: Option<String>,
    created_at: DateTime<Utc>,
}

impl PgInventoryRepository {
    pub const fn new(pool: PgPool) -> Self {
        Self { pool }
    }

    fn row_to_item(row: InventoryItemRow) -> InventoryItem {
        InventoryItem {
            id: row.id,
            sku: row.sku,
            name: row.name,
            description: row.description,
            unit_of_measure: row.unit_of_measure,
            is_active: row.is_active,
            created_at: row.created_at,
            updated_at: row.updated_at,
        }
    }

    const fn row_to_balance(row: InventoryBalanceRow) -> InventoryBalance {
        InventoryBalance {
            id: row.id,
            item_id: row.item_id,
            location_id: row.location_id,
            quantity_on_hand: row.quantity_on_hand,
            quantity_allocated: row.quantity_allocated,
            quantity_available: row.quantity_available,
            reorder_point: row.reorder_point,
            safety_stock: row.safety_stock,
            version: row.version,
            last_counted_at: row.last_counted_at,
            updated_at: row.updated_at,
        }
    }

    fn row_to_reservation(row: ReservationRow) -> Result<InventoryReservation> {
        let status: ReservationStatus = row.status.parse().map_err(|e| {
            CommerceError::DatabaseError(format!(
                "Invalid inventory_reservation.status '{}': {}",
                row.status, e
            ))
        })?;

        Ok(InventoryReservation {
            id: row.id,
            item_id: row.item_id,
            location_id: row.location_id,
            quantity: row.quantity,
            status,
            reference_type: row.reference_type,
            reference_id: row.reference_id,
            expires_at: row.expires_at,
            created_at: row.created_at,
        })
    }

    async fn expire_reservation_in_tx(
        tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
        reservation_id: Uuid,
        item_id: i64,
        location_id: i32,
        quantity: Decimal,
        now: DateTime<Utc>,
    ) -> Result<()> {
        let balance: (i32,) = sqlx::query_as(
            "SELECT version FROM inventory_balances WHERE item_id = $1 AND location_id = $2",
        )
        .bind(item_id)
        .bind(location_id)
        .fetch_one(tx.as_mut())
        .await
        .map_err(map_db_error)?;

        let current_version = balance.0;

        let result = sqlx::query(
            r#"
            UPDATE inventory_balances
            SET quantity_allocated = quantity_allocated - $1,
                quantity_available = quantity_on_hand - quantity_allocated + $1,
                version = version + 1,
                updated_at = $2
            WHERE item_id = $3 AND location_id = $4 AND version = $5
            "#,
        )
        .bind(quantity)
        .bind(now)
        .bind(item_id)
        .bind(location_id)
        .bind(current_version)
        .execute(tx.as_mut())
        .await
        .map_err(map_db_error)?;

        if result.rows_affected() == 0 {
            return Err(CommerceError::VersionConflict {
                entity: "inventory_balance".to_string(),
                id: format!("{}:{}", item_id, location_id),
                expected_version: current_version,
            });
        }

        sqlx::query("UPDATE inventory_reservations SET status = 'expired' WHERE id = $1")
            .bind(reservation_id)
            .execute(tx.as_mut())
            .await
            .map_err(map_db_error)?;

        Ok(())
    }

    async fn expire_reservations_for_item_in_tx(
        tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
        item_id: i64,
        location_id: i32,
        now: DateTime<Utc>,
    ) -> Result<()> {
        let reservations: Vec<(Uuid, Decimal)> = sqlx::query_as(
            "SELECT id, quantity FROM inventory_reservations
             WHERE item_id = $1 AND location_id = $2
               AND status IN ('pending', 'confirmed', 'allocated')
               AND expires_at IS NOT NULL AND expires_at < $3",
        )
        .bind(item_id)
        .bind(location_id)
        .bind(now)
        .fetch_all(tx.as_mut())
        .await
        .map_err(map_db_error)?;

        for (reservation_id, quantity) in reservations {
            Self::expire_reservation_in_tx(tx, reservation_id, item_id, location_id, quantity, now)
                .await?;
        }

        Ok(())
    }

    pub(crate) async fn reserve_in_tx(
        &self,
        tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
        input: &ReserveInventory,
    ) -> Result<InventoryReservation> {
        validate_quantity(input.quantity)?;

        let now = Utc::now();
        let location_id = input.location_id.unwrap_or(1);

        let item: (i64,) = sqlx::query_as("SELECT id FROM inventory_items WHERE sku = $1")
            .bind(&input.sku)
            .fetch_optional(tx.as_mut())
            .await
            .map_err(map_db_error)?
            .ok_or_else(|| CommerceError::InventoryItemNotFound(input.sku.clone()))?;

        let item_id = item.0;

        Self::expire_reservations_for_item_in_tx(tx, item_id, location_id, now).await?;

        // Lock the balance row FOR UPDATE so concurrent reservers serialize on it.
        // Without the lock, two transactions could both read sufficient availability
        // and both succeed, overselling stock (TOCTOU). This mirrors the atomic
        // re-check the SQLite backend performs in its UPDATE WHERE clause.
        let balance: (Decimal, i32) = sqlx::query_as(
            "SELECT quantity_available, version FROM inventory_balances WHERE item_id = $1 AND location_id = $2 FOR UPDATE",
        )
        .bind(item_id)
        .bind(location_id)
        .fetch_one(tx.as_mut())
        .await
        .map_err(map_db_error)?;

        let (available, current_version) = balance;

        if available < input.quantity {
            return Err(CommerceError::InsufficientStock {
                sku: input.sku.clone(),
                requested: input.quantity.to_string(),
                available: available.to_string(),
            });
        }

        let id = Uuid::new_v4();
        let expires_at = input.expires_in_seconds.map(|s| now + chrono::Duration::seconds(s));

        sqlx::query(
            r#"
            INSERT INTO inventory_reservations (id, item_id, location_id, quantity, status,
                                                reference_type, reference_id, expires_at, created_at)
            VALUES ($1, $2, $3, $4, 'pending', $5, $6, $7, $8)
            "#,
        )
        .bind(id)
        .bind(item_id)
        .bind(location_id)
        .bind(input.quantity)
        .bind(&input.reference_type)
        .bind(&input.reference_id)
        .bind(expires_at)
        .bind(now)
        .execute(tx.as_mut())
        .await
        .map_err(map_db_error)?;

        // Guard the UPDATE with both the optimistic version check and an atomic
        // stock guard (`quantity_available >= $qty`). Defence-in-depth alongside the
        // FOR UPDATE lock above: a 0-row result under the matching version can only
        // mean the stock guard tripped, which we surface as an oversell-safe
        // InsufficientStock error rather than silently overselling.
        let result = sqlx::query(
            r#"
            UPDATE inventory_balances
            SET quantity_allocated = quantity_allocated + $1,
                quantity_available = quantity_on_hand - quantity_allocated - $1,
                version = version + 1,
                updated_at = $2
            WHERE item_id = $3 AND location_id = $4 AND version = $5
              AND quantity_available >= $1
            "#,
        )
        .bind(input.quantity)
        .bind(now)
        .bind(item_id)
        .bind(location_id)
        .bind(current_version)
        .execute(tx.as_mut())
        .await
        .map_err(map_db_error)?;

        if result.rows_affected() == 0 {
            // The row is locked FOR UPDATE and the version matched our read, so a
            // 0-row update indicates the stock guard failed (insufficient stock),
            // not a lost optimistic-lock race.
            return Err(CommerceError::InsufficientStock {
                sku: input.sku.clone(),
                requested: input.quantity.to_string(),
                available: available.to_string(),
            });
        }

        Ok(InventoryReservation {
            id,
            item_id,
            location_id,
            quantity: input.quantity,
            status: ReservationStatus::Pending,
            reference_type: input.reference_type.clone(),
            reference_id: input.reference_id.clone(),
            expires_at,
            created_at: now,
        })
    }

    pub(crate) async fn list_reservation_ids_by_reference_in_tx(
        &self,
        tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
        reference_type: &str,
        reference_id: &str,
    ) -> Result<Vec<Uuid>> {
        let rows: Vec<(Uuid,)> = sqlx::query_as(
            "SELECT id FROM inventory_reservations WHERE reference_type = $1 AND reference_id = $2 ORDER BY created_at",
        )
        .bind(reference_type)
        .bind(reference_id)
        .fetch_all(tx.as_mut())
        .await
        .map_err(map_db_error)?;

        Ok(rows.into_iter().map(|(id,)| id).collect())
    }

    pub(crate) async fn release_reservation_in_tx(
        &self,
        tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
        reservation_id: Uuid,
    ) -> Result<()> {
        let now = Utc::now();

        let res = sqlx::query_as::<_, ReservationRow>(
            "SELECT * FROM inventory_reservations WHERE id = $1",
        )
        .bind(reservation_id)
        .fetch_optional(tx.as_mut())
        .await
        .map_err(map_db_error)?
        .ok_or(CommerceError::ReservationNotFound(reservation_id))?;

        let status: ReservationStatus = res.status.parse().map_err(|e| {
            CommerceError::DatabaseError(format!(
                "Invalid inventory_reservation.status '{}': {}",
                res.status, e
            ))
        })?;

        if status == ReservationStatus::Released
            || status == ReservationStatus::Cancelled
            || status == ReservationStatus::Expired
        {
            return Ok(());
        }

        if let Some(expires_at) = res.expires_at {
            if expires_at < now {
                Self::expire_reservation_in_tx(
                    tx,
                    reservation_id,
                    res.item_id,
                    res.location_id,
                    res.quantity,
                    now,
                )
                .await?;
                return Ok(());
            }
        }

        let balance: (i32,) = sqlx::query_as(
            "SELECT version FROM inventory_balances WHERE item_id = $1 AND location_id = $2",
        )
        .bind(res.item_id)
        .bind(res.location_id)
        .fetch_one(tx.as_mut())
        .await
        .map_err(map_db_error)?;

        let current_version = balance.0;

        let result = sqlx::query(
            r#"
            UPDATE inventory_balances
            SET quantity_allocated = quantity_allocated - $1,
                quantity_available = quantity_on_hand - quantity_allocated + $1,
                version = version + 1,
                updated_at = $2
            WHERE item_id = $3 AND location_id = $4 AND version = $5
            "#,
        )
        .bind(res.quantity)
        .bind(now)
        .bind(res.item_id)
        .bind(res.location_id)
        .bind(current_version)
        .execute(tx.as_mut())
        .await
        .map_err(map_db_error)?;

        if result.rows_affected() == 0 {
            return Err(CommerceError::VersionConflict {
                entity: "inventory_balance".to_string(),
                id: format!("{}:{}", res.item_id, res.location_id),
                expected_version: current_version,
            });
        }

        sqlx::query("UPDATE inventory_reservations SET status = 'released' WHERE id = $1")
            .bind(reservation_id)
            .execute(tx.as_mut())
            .await
            .map_err(map_db_error)?;

        Ok(())
    }

    pub(crate) async fn confirm_reservation_in_tx_with_now(
        &self,
        tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
        reservation_id: Uuid,
        now: DateTime<Utc>,
    ) -> Result<ReservationConfirmOutcome> {
        let res = sqlx::query_as::<_, ReservationRow>(
            "SELECT * FROM inventory_reservations WHERE id = $1",
        )
        .bind(reservation_id)
        .fetch_optional(tx.as_mut())
        .await
        .map_err(map_db_error)?
        .ok_or(CommerceError::ReservationNotFound(reservation_id))?;

        let status: ReservationStatus = res.status.parse().map_err(|e| {
            CommerceError::DatabaseError(format!(
                "Invalid inventory_reservation.status '{}': {}",
                res.status, e
            ))
        })?;

        if status == ReservationStatus::Released || status == ReservationStatus::Cancelled {
            return Ok(ReservationConfirmOutcome::Confirmed);
        }
        if status == ReservationStatus::Expired {
            return Ok(ReservationConfirmOutcome::Expired);
        }

        if let Some(expires_at) = res.expires_at {
            if expires_at < now {
                Self::expire_reservation_in_tx(
                    tx,
                    reservation_id,
                    res.item_id,
                    res.location_id,
                    res.quantity,
                    now,
                )
                .await?;
                return Ok(ReservationConfirmOutcome::Expired);
            }
        }

        sqlx::query("UPDATE inventory_reservations SET status = 'confirmed' WHERE id = $1")
            .bind(reservation_id)
            .execute(tx.as_mut())
            .await
            .map_err(map_db_error)?;

        Ok(ReservationConfirmOutcome::Confirmed)
    }

    pub(crate) async fn expire_reservation_if_needed_in_tx(
        &self,
        tx: &mut sqlx::Transaction<'_, sqlx::Postgres>,
        reservation_id: Uuid,
        now: DateTime<Utc>,
    ) -> Result<bool> {
        let res = sqlx::query_as::<_, ReservationRow>(
            "SELECT * FROM inventory_reservations WHERE id = $1",
        )
        .bind(reservation_id)
        .fetch_optional(tx.as_mut())
        .await
        .map_err(map_db_error)?
        .ok_or(CommerceError::ReservationNotFound(reservation_id))?;

        let status: ReservationStatus = res.status.parse().map_err(|e| {
            CommerceError::DatabaseError(format!(
                "Invalid inventory_reservation.status '{}': {}",
                res.status, e
            ))
        })?;

        if status == ReservationStatus::Released || status == ReservationStatus::Cancelled {
            return Ok(false);
        }

        if status == ReservationStatus::Expired {
            return Ok(true);
        }

        if let Some(expires_at) = res.expires_at {
            if expires_at < now {
                Self::expire_reservation_in_tx(
                    tx,
                    reservation_id,
                    res.item_id,
                    res.location_id,
                    res.quantity,
                    now,
                )
                .await?;
                return Ok(true);
            }
        }

        Ok(false)
    }

    fn row_to_transaction(row: TransactionRow) -> Result<InventoryTransaction> {
        let transaction_type: TransactionType = row.transaction_type.parse().map_err(|e| {
            CommerceError::DatabaseError(format!(
                "Invalid inventory_transaction.transaction_type '{}': {}",
                row.transaction_type, e
            ))
        })?;

        Ok(InventoryTransaction {
            id: row.id,
            item_id: row.item_id,
            location_id: row.location_id,
            transaction_type,
            quantity: row.quantity,
            reference_type: row.reference_type,
            reference_id: row.reference_id,
            reason: row.reason,
            created_by: row.created_by,
            created_at: row.created_at,
        })
    }

    /// Create an inventory item (async)
    pub async fn create_item_async(&self, input: CreateInventoryItem) -> Result<InventoryItem> {
        let now = Utc::now();

        let mut tx = self.pool.begin().await.map_err(map_db_error)?;

        let row: (i64,) = sqlx::query_as(
            r#"
            INSERT INTO inventory_items (sku, name, description, unit_of_measure, is_active, created_at, updated_at)
            VALUES ($1, $2, $3, $4, $5, $6, $7)
            RETURNING id
            "#,
        )
        .bind(&input.sku)
        .bind(&input.name)
        .bind(&input.description)
        .bind(input.unit_of_measure.as_deref().unwrap_or("EA"))
        .bind(true)
        .bind(now)
        .bind(now)
        .fetch_one(tx.as_mut())
        .await
        .map_err(map_db_error)?;

        let id = row.0;
        let location_id = input.location_id.unwrap_or(1);

        // Create initial balance
        let initial_qty = input.initial_quantity.unwrap_or(Decimal::ZERO);
        sqlx::query(
            r#"
            INSERT INTO inventory_balances (item_id, location_id, quantity_on_hand, quantity_allocated,
                                            quantity_available, reorder_point, safety_stock, version, updated_at)
            VALUES ($1, $2, $3, 0, $4, $5, $6, 1, $7)
            "#,
        )
        .bind(id)
        .bind(location_id)
        .bind(initial_qty)
        .bind(initial_qty)
        .bind(input.reorder_point)
        .bind(input.safety_stock)
        .bind(now)
        .execute(tx.as_mut())
        .await
        .map_err(map_db_error)?;

        tx.commit().await.map_err(map_db_error)?;

        Ok(InventoryItem {
            id,
            sku: input.sku,
            name: input.name,
            description: input.description,
            unit_of_measure: input.unit_of_measure.unwrap_or_else(|| "EA".to_string()),
            is_active: true,
            created_at: now,
            updated_at: now,
        })
    }

    /// Get inventory item by ID (async)
    pub async fn get_item_async(&self, id: i64) -> Result<Option<InventoryItem>> {
        let row =
            sqlx::query_as::<_, InventoryItemRow>("SELECT * FROM inventory_items WHERE id = $1")
                .bind(id)
                .fetch_optional(&self.pool)
                .await
                .map_err(map_db_error)?;

        Ok(row.map(Self::row_to_item))
    }

    /// Get inventory item by SKU (async)
    pub async fn get_item_by_sku_async(&self, sku: &str) -> Result<Option<InventoryItem>> {
        let row =
            sqlx::query_as::<_, InventoryItemRow>("SELECT * FROM inventory_items WHERE sku = $1")
                .bind(sku)
                .fetch_optional(&self.pool)
                .await
                .map_err(map_db_error)?;

        Ok(row.map(Self::row_to_item))
    }

    /// Get stock level for a SKU (async)
    pub async fn get_stock_async(&self, sku: &str) -> Result<Option<StockLevel>> {
        let item_row =
            sqlx::query_as::<_, InventoryItemRow>("SELECT * FROM inventory_items WHERE sku = $1")
                .bind(sku)
                .fetch_optional(&self.pool)
                .await
                .map_err(map_db_error)?;

        let item = match item_row {
            Some(r) => r,
            None => return Ok(None),
        };

        let balance_rows = sqlx::query_as::<_, InventoryBalanceRow>(
            "SELECT * FROM inventory_balances WHERE item_id = $1",
        )
        .bind(item.id)
        .fetch_all(&self.pool)
        .await
        .map_err(map_db_error)?;

        let mut total_on_hand = Decimal::ZERO;
        let mut total_allocated = Decimal::ZERO;
        let mut total_available = Decimal::ZERO;
        let mut locations = Vec::new();

        for b in balance_rows {
            total_on_hand += b.quantity_on_hand;
            total_allocated += b.quantity_allocated;
            total_available += b.quantity_available;
            locations.push(LocationStock {
                location_id: b.location_id,
                location_name: None,
                on_hand: b.quantity_on_hand,
                allocated: b.quantity_allocated,
                available: b.quantity_available,
            });
        }

        Ok(Some(StockLevel {
            sku: item.sku,
            name: item.name,
            total_on_hand,
            total_allocated,
            total_available,
            locations,
        }))
    }

    /// Get balance at specific location (async)
    pub async fn get_balance_async(
        &self,
        item_id: i64,
        location_id: i32,
    ) -> Result<Option<InventoryBalance>> {
        let row = sqlx::query_as::<_, InventoryBalanceRow>(
            "SELECT * FROM inventory_balances WHERE item_id = $1 AND location_id = $2",
        )
        .bind(item_id)
        .bind(location_id)
        .fetch_optional(&self.pool)
        .await
        .map_err(map_db_error)?;

        Ok(row.map(Self::row_to_balance))
    }

    /// Adjust inventory (async)
    pub async fn adjust_async(&self, input: AdjustInventory) -> Result<InventoryTransaction> {
        let now = Utc::now();
        let location_id = input.location_id.unwrap_or(1);

        // Get item ID
        let item: (i64,) = sqlx::query_as("SELECT id FROM inventory_items WHERE sku = $1")
            .bind(&input.sku)
            .fetch_one(&self.pool)
            .await
            .map_err(|_| CommerceError::InventoryItemNotFound(input.sku.clone()))?;

        let item_id = item.0;

        // Get current quantities/version for optimistic locking
        let balance: (Decimal, Decimal, i32) = sqlx::query_as(
            "SELECT quantity_on_hand, quantity_allocated, version
             FROM inventory_balances WHERE item_id = $1 AND location_id = $2",
        )
        .bind(item_id)
        .bind(location_id)
        .fetch_one(&self.pool)
        .await
        .map_err(map_db_error)?;

        let (quantity_on_hand, quantity_allocated, current_version) = balance;
        let new_on_hand = quantity_on_hand + input.quantity;
        let new_available = new_on_hand - quantity_allocated;

        if new_on_hand < Decimal::ZERO {
            return Err(CommerceError::InsufficientStock {
                sku: input.sku.clone(),
                requested: input.quantity.abs().to_string(),
                available: quantity_on_hand.to_string(),
            });
        }
        if new_available < Decimal::ZERO {
            let available = quantity_on_hand - quantity_allocated;
            return Err(CommerceError::InsufficientStock {
                sku: input.sku.clone(),
                requested: input.quantity.abs().to_string(),
                available: available.to_string(),
            });
        }

        // Update balance with optimistic locking
        let result = sqlx::query(
            r#"
            UPDATE inventory_balances
            SET quantity_on_hand = quantity_on_hand + $1,
                quantity_available = quantity_on_hand + $1 - quantity_allocated,
                version = version + 1,
                updated_at = $2
            WHERE item_id = $3 AND location_id = $4 AND version = $5
            "#,
        )
        .bind(input.quantity)
        .bind(now)
        .bind(item_id)
        .bind(location_id)
        .bind(current_version)
        .execute(&self.pool)
        .await
        .map_err(map_db_error)?;

        if result.rows_affected() == 0 {
            return Err(CommerceError::VersionConflict {
                entity: "inventory_balance".to_string(),
                id: format!("{}:{}", item_id, location_id),
                expected_version: current_version,
            });
        }

        // Record transaction
        let tx_type = "adjustment";

        let tx_row: (i64,) = sqlx::query_as(
            r#"
            INSERT INTO inventory_transactions (item_id, location_id, transaction_type, quantity,
                                                reference_type, reference_id, reason, created_at)
            VALUES ($1, $2, $3, $4, $5, $6, $7, $8)
            RETURNING id
            "#,
        )
        .bind(item_id)
        .bind(location_id)
        .bind(tx_type)
        .bind(input.quantity)
        .bind(&input.reference_type)
        .bind(&input.reference_id)
        .bind(&input.reason)
        .bind(now)
        .fetch_one(&self.pool)
        .await
        .map_err(map_db_error)?;

        Ok(InventoryTransaction {
            id: tx_row.0,
            item_id,
            location_id,
            transaction_type: TransactionType::Adjustment,
            quantity: input.quantity,
            reference_type: input.reference_type,
            reference_id: input.reference_id,
            reason: Some(input.reason),
            created_by: None,
            created_at: now,
        })
    }

    /// Reserve inventory (async)
    pub async fn reserve_async(&self, input: ReserveInventory) -> Result<InventoryReservation> {
        // Validate quantity is positive to prevent negative reservations
        // that could inflate available stock
        validate_quantity(input.quantity)?;

        let now = Utc::now();
        let location_id = input.location_id.unwrap_or(1);

        let mut tx = self.pool.begin().await.map_err(map_db_error)?;

        // Get item ID
        let item: (i64,) = sqlx::query_as("SELECT id FROM inventory_items WHERE sku = $1")
            .bind(&input.sku)
            .fetch_one(tx.as_mut())
            .await
            .map_err(|_| CommerceError::InventoryItemNotFound(input.sku.clone()))?;

        let item_id = item.0;

        Self::expire_reservations_for_item_in_tx(&mut tx, item_id, location_id, now).await?;

        // Check availability and get current version for optimistic locking.
        // Lock the balance row FOR UPDATE so concurrent reservers serialize on it:
        // without the lock two transactions could both read sufficient availability
        // and both succeed, overselling stock (TOCTOU). This mirrors the atomic
        // re-check the SQLite backend performs in its UPDATE WHERE clause.
        let balance: (Decimal, i32) = sqlx::query_as(
            "SELECT quantity_available, version FROM inventory_balances WHERE item_id = $1 AND location_id = $2 FOR UPDATE",
        )
        .bind(item_id)
        .bind(location_id)
        .fetch_one(tx.as_mut())
        .await
        .map_err(map_db_error)?;

        let (available, current_version) = balance;

        if available < input.quantity {
            return Err(CommerceError::InsufficientStock {
                sku: input.sku,
                requested: input.quantity.to_string(),
                available: available.to_string(),
            });
        }

        // Create reservation
        let id = Uuid::new_v4();
        let expires_at = input.expires_in_seconds.map(|s| now + chrono::Duration::seconds(s));

        sqlx::query(
            r#"
            INSERT INTO inventory_reservations (id, item_id, location_id, quantity, status,
                                                reference_type, reference_id, expires_at, created_at)
            VALUES ($1, $2, $3, $4, 'pending', $5, $6, $7, $8)
            "#,
        )
        .bind(id)
        .bind(item_id)
        .bind(location_id)
        .bind(input.quantity)
        .bind(&input.reference_type)
        .bind(&input.reference_id)
        .bind(expires_at)
        .bind(now)
        .execute(tx.as_mut())
        .await
        .map_err(map_db_error)?;

        // Update allocation with optimistic locking. Guard the UPDATE with both the
        // optimistic version check and an atomic stock guard (`quantity_available >= $qty`)
        // as defence-in-depth alongside the FOR UPDATE lock above: a 0-row result under
        // the matching version can only mean the stock guard tripped, which we surface as
        // an oversell-safe InsufficientStock error rather than silently overselling.
        let result = sqlx::query(
            r#"
            UPDATE inventory_balances
            SET quantity_allocated = quantity_allocated + $1,
                quantity_available = quantity_on_hand - quantity_allocated - $1,
                version = version + 1,
                updated_at = $2
            WHERE item_id = $3 AND location_id = $4 AND version = $5
              AND quantity_available >= $1
            "#,
        )
        .bind(input.quantity)
        .bind(now)
        .bind(item_id)
        .bind(location_id)
        .bind(current_version)
        .execute(tx.as_mut())
        .await
        .map_err(map_db_error)?;

        if result.rows_affected() == 0 {
            // The row is locked FOR UPDATE and the version matched our read, so a
            // 0-row update indicates the stock guard failed (insufficient stock),
            // not a lost optimistic-lock race.
            return Err(CommerceError::InsufficientStock {
                sku: input.sku.clone(),
                requested: input.quantity.to_string(),
                available: available.to_string(),
            });
        }

        tx.commit().await.map_err(map_db_error)?;

        Ok(InventoryReservation {
            id,
            item_id,
            location_id,
            quantity: input.quantity,
            status: ReservationStatus::Pending,
            reference_type: input.reference_type,
            reference_id: input.reference_id,
            expires_at,
            created_at: now,
        })
    }

    /// Get a reservation by ID (async)
    pub async fn get_reservation_async(
        &self,
        reservation_id: Uuid,
    ) -> Result<Option<InventoryReservation>> {
        let row = sqlx::query_as::<_, ReservationRow>(
            "SELECT * FROM inventory_reservations WHERE id = $1",
        )
        .bind(reservation_id)
        .fetch_optional(&self.pool)
        .await
        .map_err(map_db_error)?;

        match row {
            Some(row) => Ok(Some(Self::row_to_reservation(row)?)),
            None => Ok(None),
        }
    }

    /// List reservations by reference (async)
    pub async fn list_reservations_by_reference_async(
        &self,
        reference_type: &str,
        reference_id: &str,
    ) -> Result<Vec<InventoryReservation>> {
        let rows = sqlx::query_as::<_, ReservationRow>(
            "SELECT * FROM inventory_reservations WHERE reference_type = $1 AND reference_id = $2 ORDER BY created_at",
        )
        .bind(reference_type)
        .bind(reference_id)
        .fetch_all(&self.pool)
        .await
        .map_err(map_db_error)?;

        let mut reservations = Vec::with_capacity(rows.len());
        for row in rows {
            reservations.push(Self::row_to_reservation(row)?);
        }
        Ok(reservations)
    }

    /// Release a reservation (async)
    pub async fn release_reservation_async(&self, reservation_id: Uuid) -> Result<()> {
        let now = Utc::now();

        // Get reservation
        let mut tx = self.pool.begin().await.map_err(map_db_error)?;
        let res = sqlx::query_as::<_, ReservationRow>(
            "SELECT * FROM inventory_reservations WHERE id = $1",
        )
        .bind(reservation_id)
        .fetch_optional(tx.as_mut())
        .await
        .map_err(map_db_error)?
        .ok_or(CommerceError::ReservationNotFound(reservation_id))?;

        let status: ReservationStatus = res.status.parse().map_err(|e| {
            CommerceError::DatabaseError(format!(
                "Invalid inventory_reservation.status '{}': {}",
                res.status, e
            ))
        })?;
        if status == ReservationStatus::Released
            || status == ReservationStatus::Cancelled
            || status == ReservationStatus::Expired
        {
            tx.commit().await.map_err(map_db_error)?;
            return Ok(());
        }

        if let Some(expires_at) = res.expires_at {
            if expires_at < now {
                Self::expire_reservation_in_tx(
                    &mut tx,
                    reservation_id,
                    res.item_id,
                    res.location_id,
                    res.quantity,
                    now,
                )
                .await?;
                tx.commit().await.map_err(map_db_error)?;
                return Ok(());
            }
        }

        // Get current version for optimistic locking
        let balance: (i32,) = sqlx::query_as(
            "SELECT version FROM inventory_balances WHERE item_id = $1 AND location_id = $2",
        )
        .bind(res.item_id)
        .bind(res.location_id)
        .fetch_one(tx.as_mut())
        .await
        .map_err(map_db_error)?;

        let current_version = balance.0;

        // Release allocation with optimistic locking
        let result = sqlx::query(
            r#"
            UPDATE inventory_balances
            SET quantity_allocated = quantity_allocated - $1,
                quantity_available = quantity_on_hand - quantity_allocated + $1,
                version = version + 1,
                updated_at = $2
            WHERE item_id = $3 AND location_id = $4 AND version = $5
            "#,
        )
        .bind(res.quantity)
        .bind(now)
        .bind(res.item_id)
        .bind(res.location_id)
        .bind(current_version)
        .execute(tx.as_mut())
        .await
        .map_err(map_db_error)?;

        if result.rows_affected() == 0 {
            return Err(CommerceError::VersionConflict {
                entity: "inventory_balance".to_string(),
                id: format!("{}:{}", res.item_id, res.location_id),
                expected_version: current_version,
            });
        }

        // Update reservation status
        sqlx::query("UPDATE inventory_reservations SET status = 'released' WHERE id = $1")
            .bind(reservation_id)
            .execute(tx.as_mut())
            .await
            .map_err(map_db_error)?;

        tx.commit().await.map_err(map_db_error)?;

        Ok(())
    }

    /// Confirm a reservation (async)
    pub async fn confirm_reservation_async(&self, reservation_id: Uuid) -> Result<()> {
        let now = Utc::now();
        let mut tx = self.pool.begin().await.map_err(map_db_error)?;

        let res = sqlx::query_as::<_, ReservationRow>(
            "SELECT * FROM inventory_reservations WHERE id = $1",
        )
        .bind(reservation_id)
        .fetch_optional(tx.as_mut())
        .await
        .map_err(map_db_error)?
        .ok_or(CommerceError::ReservationNotFound(reservation_id))?;

        let status: ReservationStatus = res.status.parse().map_err(|e| {
            CommerceError::DatabaseError(format!(
                "Invalid inventory_reservation.status '{}': {}",
                res.status, e
            ))
        })?;

        if status == ReservationStatus::Released || status == ReservationStatus::Cancelled {
            tx.commit().await.map_err(map_db_error)?;
            return Ok(());
        }
        if status == ReservationStatus::Expired {
            tx.commit().await.map_err(map_db_error)?;
            return Err(CommerceError::ReservationExpired(reservation_id));
        }

        if let Some(expires_at) = res.expires_at {
            if expires_at < now {
                Self::expire_reservation_in_tx(
                    &mut tx,
                    reservation_id,
                    res.item_id,
                    res.location_id,
                    res.quantity,
                    now,
                )
                .await?;
                tx.commit().await.map_err(map_db_error)?;
                return Err(CommerceError::ReservationExpired(reservation_id));
            }
        }

        sqlx::query("UPDATE inventory_reservations SET status = 'confirmed' WHERE id = $1")
            .bind(reservation_id)
            .execute(tx.as_mut())
            .await
            .map_err(map_db_error)?;

        tx.commit().await.map_err(map_db_error)?;

        Ok(())
    }

    /// List inventory items (async)
    pub async fn list_async(&self, filter: InventoryFilter) -> Result<Vec<InventoryItem>> {
        let limit = super::effective_limit(filter.limit);
        let offset = filter.offset.unwrap_or(0) as i64;

        let rows = sqlx::query_as::<_, InventoryItemRow>(
            "SELECT * FROM inventory_items WHERE is_active = TRUE ORDER BY created_at DESC LIMIT $1 OFFSET $2",
        )
        .bind(limit)
        .bind(offset)
        .fetch_all(&self.pool)
        .await
        .map_err(map_db_error)?;

        Ok(rows.into_iter().map(Self::row_to_item).collect())
    }

    /// Get items below reorder point (async)
    pub async fn get_reorder_needed_async(&self) -> Result<Vec<StockLevel>> {
        let rows = sqlx::query_as::<_, InventoryItemRow>(
            r#"
            SELECT i.* FROM inventory_items i
            JOIN inventory_balances b ON i.id = b.item_id
            WHERE b.quantity_available < COALESCE(b.reorder_point, 0)
            AND i.is_active = TRUE
            "#,
        )
        .fetch_all(&self.pool)
        .await
        .map_err(map_db_error)?;

        let mut result = Vec::new();
        for row in rows {
            if let Some(stock) = self.get_stock_async(&row.sku).await? {
                result.push(stock);
            }
        }

        Ok(result)
    }

    /// Get transaction history (async)
    pub async fn get_transactions_async(
        &self,
        item_id: i64,
        limit: u32,
    ) -> Result<Vec<InventoryTransaction>> {
        let rows = sqlx::query_as::<_, TransactionRow>(
            "SELECT * FROM inventory_transactions WHERE item_id = $1 ORDER BY created_at DESC LIMIT $2",
        )
        .bind(item_id)
        .bind(limit as i64)
        .fetch_all(&self.pool)
        .await
        .map_err(map_db_error)?;

        let mut transactions = Vec::with_capacity(rows.len());
        for row in rows {
            transactions.push(Self::row_to_transaction(row)?);
        }
        Ok(transactions)
    }

    /// Record an inventory transaction (async)
    pub async fn record_transaction_async(
        &self,
        transaction: InventoryTransaction,
    ) -> Result<InventoryTransaction> {
        let row: (i64,) = sqlx::query_as(
            r#"
            INSERT INTO inventory_transactions (
                item_id,
                location_id,
                transaction_type,
                quantity,
                reference_type,
                reference_id,
                reason,
                created_by,
                created_at
            ) VALUES ($1, $2, $3, $4, $5, $6, $7, $8, $9)
            RETURNING id
            "#,
        )
        .bind(transaction.item_id)
        .bind(transaction.location_id)
        .bind(transaction.transaction_type.to_string())
        .bind(transaction.quantity)
        .bind(transaction.reference_type.clone())
        .bind(&transaction.reference_id)
        .bind(transaction.reason.clone())
        .bind(transaction.created_by.clone())
        .bind(transaction.created_at)
        .fetch_one(&self.pool)
        .await
        .map_err(map_db_error)?;

        Ok(InventoryTransaction { id: row.0, ..transaction })
    }

    // ========================================================================
    // Batch Operations (async)
    // ========================================================================

    /// Create multiple inventory items - partial success allowed (async)
    pub async fn create_item_batch_async(
        &self,
        inputs: Vec<CreateInventoryItem>,
    ) -> Result<BatchResult<InventoryItem>> {
        validate_batch_size(&inputs)?;

        let mut result = BatchResult::with_capacity(inputs.len());

        for (index, input) in inputs.into_iter().enumerate() {
            match self.create_item_async(input).await {
                Ok(item) => result.record_success(item),
                Err(e) => result.record_failure(index, None, &e),
            }
        }

        Ok(result)
    }

    /// Create multiple inventory items - atomic (all-or-nothing) (async)
    pub async fn create_item_batch_atomic_async(
        &self,
        inputs: Vec<CreateInventoryItem>,
    ) -> Result<Vec<InventoryItem>> {
        validate_batch_size(&inputs)?;

        let mut tx = self.pool.begin().await.map_err(map_db_error)?;
        let mut items = Vec::with_capacity(inputs.len());
        let now = Utc::now();

        for input in inputs {
            let row: (i64,) = sqlx::query_as(
                r#"
                INSERT INTO inventory_items (sku, name, description, unit_of_measure, is_active, created_at, updated_at)
                VALUES ($1, $2, $3, $4, $5, $6, $7)
                RETURNING id
                "#,
            )
            .bind(&input.sku)
            .bind(&input.name)
            .bind(&input.description)
            .bind(input.unit_of_measure.as_deref().unwrap_or("EA"))
            .bind(true)
            .bind(now)
            .bind(now)
            .fetch_one(tx.as_mut())
            .await
            .map_err(map_db_error)?;

            let id = row.0;
            let location_id = input.location_id.unwrap_or(1);

            // Create initial balance
            let initial_qty = input.initial_quantity.unwrap_or(Decimal::ZERO);
            sqlx::query(
                r#"
                INSERT INTO inventory_balances (item_id, location_id, quantity_on_hand, quantity_allocated,
                                                quantity_available, reorder_point, safety_stock, version, updated_at)
                VALUES ($1, $2, $3, 0, $4, $5, $6, 1, $7)
                "#,
            )
            .bind(id)
            .bind(location_id)
            .bind(initial_qty)
            .bind(initial_qty)
            .bind(input.reorder_point)
            .bind(input.safety_stock)
            .bind(now)
            .execute(tx.as_mut())
            .await
            .map_err(map_db_error)?;

            items.push(InventoryItem {
                id,
                sku: input.sku,
                name: input.name,
                description: input.description,
                unit_of_measure: input.unit_of_measure.unwrap_or_else(|| "EA".to_string()),
                is_active: true,
                created_at: now,
                updated_at: now,
            });
        }

        tx.commit().await.map_err(map_db_error)?;
        Ok(items)
    }

    /// Adjust multiple inventory quantities - partial success allowed (async)
    pub async fn adjust_batch_async(
        &self,
        adjustments: Vec<AdjustInventory>,
    ) -> Result<BatchResult<InventoryTransaction>> {
        validate_batch_size(&adjustments)?;

        let mut result = BatchResult::with_capacity(adjustments.len());

        for (index, adjustment) in adjustments.into_iter().enumerate() {
            let sku = adjustment.sku.clone();
            match self.adjust_async(adjustment).await {
                Ok(tx) => result.record_success(tx),
                Err(e) => result.record_failure(index, Some(sku), &e),
            }
        }

        Ok(result)
    }

    /// Adjust multiple inventory quantities - atomic (all-or-nothing) (async)
    pub async fn adjust_batch_atomic_async(
        &self,
        adjustments: Vec<AdjustInventory>,
    ) -> Result<Vec<InventoryTransaction>> {
        validate_batch_size(&adjustments)?;

        let mut tx = self.pool.begin().await.map_err(map_db_error)?;
        let mut transactions = Vec::with_capacity(adjustments.len());
        let now = Utc::now();

        for input in adjustments {
            let location_id = input.location_id.unwrap_or(1);

            // Get item ID
            let item: (i64,) = sqlx::query_as("SELECT id FROM inventory_items WHERE sku = $1")
                .bind(&input.sku)
                .fetch_one(tx.as_mut())
                .await
                .map_err(|_| CommerceError::InventoryItemNotFound(input.sku.clone()))?;

            let item_id = item.0;

            // Get current quantities/version for optimistic locking
            let balance: (Decimal, Decimal, i32) = sqlx::query_as(
                "SELECT quantity_on_hand, quantity_allocated, version
                 FROM inventory_balances WHERE item_id = $1 AND location_id = $2",
            )
            .bind(item_id)
            .bind(location_id)
            .fetch_one(tx.as_mut())
            .await
            .map_err(map_db_error)?;

            let (quantity_on_hand, quantity_allocated, current_version) = balance;
            let new_on_hand = quantity_on_hand + input.quantity;
            let new_available = new_on_hand - quantity_allocated;

            if new_on_hand < Decimal::ZERO {
                return Err(CommerceError::InsufficientStock {
                    sku: input.sku.clone(),
                    requested: input.quantity.abs().to_string(),
                    available: quantity_on_hand.to_string(),
                });
            }
            if new_available < Decimal::ZERO {
                let available = quantity_on_hand - quantity_allocated;
                return Err(CommerceError::InsufficientStock {
                    sku: input.sku.clone(),
                    requested: input.quantity.abs().to_string(),
                    available: available.to_string(),
                });
            }

            // Update balance with optimistic locking
            let result = sqlx::query(
                r#"
                UPDATE inventory_balances
                SET quantity_on_hand = quantity_on_hand + $1,
                    quantity_available = quantity_on_hand + $1 - quantity_allocated,
                    version = version + 1,
                    updated_at = $2
                WHERE item_id = $3 AND location_id = $4 AND version = $5
                "#,
            )
            .bind(input.quantity)
            .bind(now)
            .bind(item_id)
            .bind(location_id)
            .bind(current_version)
            .execute(tx.as_mut())
            .await
            .map_err(map_db_error)?;

            if result.rows_affected() == 0 {
                return Err(CommerceError::VersionConflict {
                    entity: "inventory_balance".to_string(),
                    id: format!("{}:{}", item_id, location_id),
                    expected_version: current_version,
                });
            }

            // Record transaction
            let tx_row: (i64,) = sqlx::query_as(
                r#"
                INSERT INTO inventory_transactions (item_id, location_id, transaction_type, quantity,
                                                    reference_type, reference_id, reason, created_at)
                VALUES ($1, $2, 'adjustment', $3, $4, $5, $6, $7)
                RETURNING id
                "#,
            )
            .bind(item_id)
            .bind(location_id)
            .bind(input.quantity)
            .bind(&input.reference_type)
            .bind(&input.reference_id)
            .bind(&input.reason)
            .bind(now)
            .fetch_one(tx.as_mut())
            .await
            .map_err(map_db_error)?;

            transactions.push(InventoryTransaction {
                id: tx_row.0,
                item_id,
                location_id,
                transaction_type: TransactionType::Adjustment,
                quantity: input.quantity,
                reference_type: input.reference_type,
                reference_id: input.reference_id,
                reason: Some(input.reason),
                created_by: None,
                created_at: now,
            });
        }

        tx.commit().await.map_err(map_db_error)?;
        Ok(transactions)
    }

    /// Get multiple inventory items by ID (async)
    pub async fn get_item_batch_async(&self, ids: Vec<i64>) -> Result<Vec<InventoryItem>> {
        validate_batch_size(&ids)?;

        if ids.is_empty() {
            return Ok(Vec::new());
        }

        let rows = sqlx::query_as::<_, InventoryItemRow>(
            "SELECT * FROM inventory_items WHERE id = ANY($1)",
        )
        .bind(&ids)
        .fetch_all(&self.pool)
        .await
        .map_err(map_db_error)?;

        Ok(rows.into_iter().map(Self::row_to_item).collect())
    }

    /// Get stock levels for multiple SKUs (async)
    pub async fn get_stock_batch_async(&self, skus: Vec<String>) -> Result<Vec<StockLevel>> {
        validate_batch_size(&skus)?;

        if skus.is_empty() {
            return Ok(Vec::new());
        }

        // Get all items matching the SKUs
        let item_rows = sqlx::query_as::<_, InventoryItemRow>(
            "SELECT * FROM inventory_items WHERE sku = ANY($1)",
        )
        .bind(&skus)
        .fetch_all(&self.pool)
        .await
        .map_err(map_db_error)?;

        if item_rows.is_empty() {
            return Ok(Vec::new());
        }

        // Collect item IDs for batch balance lookup
        let item_ids: Vec<i64> = item_rows.iter().map(|r| r.id).collect();

        // Get all balances for these items
        let balance_rows = sqlx::query_as::<_, InventoryBalanceRow>(
            "SELECT * FROM inventory_balances WHERE item_id = ANY($1)",
        )
        .bind(&item_ids)
        .fetch_all(&self.pool)
        .await
        .map_err(map_db_error)?;

        // Build StockLevel for each item
        let mut results = Vec::with_capacity(item_rows.len());
        for item in item_rows {
            let mut total_on_hand = Decimal::ZERO;
            let mut total_allocated = Decimal::ZERO;
            let mut total_available = Decimal::ZERO;
            let mut locations = Vec::new();

            for b in balance_rows.iter().filter(|b| b.item_id == item.id) {
                total_on_hand += b.quantity_on_hand;
                total_allocated += b.quantity_allocated;
                total_available += b.quantity_available;
                locations.push(LocationStock {
                    location_id: b.location_id,
                    location_name: None,
                    on_hand: b.quantity_on_hand,
                    allocated: b.quantity_allocated,
                    available: b.quantity_available,
                });
            }

            results.push(StockLevel {
                sku: item.sku,
                name: item.name,
                total_on_hand,
                total_allocated,
                total_available,
                locations,
            });
        }

        Ok(results)
    }
}

impl InventoryRepository for PgInventoryRepository {
    fn create_item(&self, input: CreateInventoryItem) -> Result<InventoryItem> {
        super::block_on(self.create_item_async(input))
    }

    fn get_item(&self, id: i64) -> Result<Option<InventoryItem>> {
        super::block_on(self.get_item_async(id))
    }

    fn get_item_by_sku(&self, sku: &str) -> Result<Option<InventoryItem>> {
        super::block_on(self.get_item_by_sku_async(sku))
    }

    fn get_stock(&self, sku: &str) -> Result<Option<StockLevel>> {
        super::block_on(self.get_stock_async(sku))
    }

    fn get_balance(&self, item_id: i64, location_id: i32) -> Result<Option<InventoryBalance>> {
        super::block_on(self.get_balance_async(item_id, location_id))
    }

    fn adjust(&self, input: AdjustInventory) -> Result<InventoryTransaction> {
        super::block_on(self.adjust_async(input))
    }

    fn reserve(&self, input: ReserveInventory) -> Result<InventoryReservation> {
        super::block_on(self.reserve_async(input))
    }

    fn get_reservation(&self, reservation_id: Uuid) -> Result<Option<InventoryReservation>> {
        super::block_on(self.get_reservation_async(reservation_id))
    }

    fn release_reservation(&self, reservation_id: Uuid) -> Result<()> {
        super::block_on(self.release_reservation_async(reservation_id))
    }

    fn confirm_reservation(&self, reservation_id: Uuid) -> Result<()> {
        super::block_on(self.confirm_reservation_async(reservation_id))
    }

    fn list_reservations_by_reference(
        &self,
        reference_type: &str,
        reference_id: &str,
    ) -> Result<Vec<InventoryReservation>> {
        super::block_on(self.list_reservations_by_reference_async(reference_type, reference_id))
    }

    fn list(&self, filter: InventoryFilter) -> Result<Vec<InventoryItem>> {
        super::block_on(self.list_async(filter))
    }

    fn get_reorder_needed(&self) -> Result<Vec<StockLevel>> {
        super::block_on(self.get_reorder_needed_async())
    }

    fn record_transaction(
        &self,
        transaction: InventoryTransaction,
    ) -> Result<InventoryTransaction> {
        super::block_on(self.record_transaction_async(transaction))
    }

    fn get_transactions(&self, item_id: i64, limit: u32) -> Result<Vec<InventoryTransaction>> {
        super::block_on(self.get_transactions_async(item_id, limit))
    }

    // === Batch Operations ===

    fn create_item_batch(
        &self,
        inputs: Vec<CreateInventoryItem>,
    ) -> Result<BatchResult<InventoryItem>> {
        super::block_on(self.create_item_batch_async(inputs))
    }

    fn create_item_batch_atomic(
        &self,
        inputs: Vec<CreateInventoryItem>,
    ) -> Result<Vec<InventoryItem>> {
        super::block_on(self.create_item_batch_atomic_async(inputs))
    }

    fn adjust_batch(
        &self,
        adjustments: Vec<AdjustInventory>,
    ) -> Result<BatchResult<InventoryTransaction>> {
        super::block_on(self.adjust_batch_async(adjustments))
    }

    fn adjust_batch_atomic(
        &self,
        adjustments: Vec<AdjustInventory>,
    ) -> Result<Vec<InventoryTransaction>> {
        super::block_on(self.adjust_batch_atomic_async(adjustments))
    }

    fn get_item_batch(&self, ids: Vec<i64>) -> Result<Vec<InventoryItem>> {
        super::block_on(self.get_item_batch_async(ids))
    }

    fn get_stock_batch(&self, skus: Vec<String>) -> Result<Vec<StockLevel>> {
        super::block_on(self.get_stock_batch_async(skus))
    }
}