backbone_inventory/infrastructure/persistence/

quant_repository.rs

//! Repository for Quant entities
//!
//! Originally generated by metaphor-schema; now **user-owned** — this exact path is declared under
//! `user_owned` in `metaphor.codegen.yaml`, so the generator skips it wholesale. The custom methods
//! below hold the hand-written quant SQL: the reservation triangle's apex (spec stock-business-logic
//! §3/§5.1). `stock_quants.reserved_quantity` is the AUTHORITATIVE reservation counter — these
//! methods are its single writer family; the move-line quantity is a mirror and the move state an
//! aggregate, both driven by the engine, never by a second counter here.
//!
//! 4-layer rule: services orchestrate, repositories hold SQL. Every write method takes the CALLER'S
//! connection (the movement's transaction), and the caller has already relayed the ambient org
//! scope onto it.

use rust_decimal::Decimal;
use sqlx::{PgPool, Row};
use uuid::Uuid;

use crate::domain::entity::Quant;

/// Table name for Quant entities
pub const TABLE_NAME: &str = "inventory.stock_quants";

/// Repository for Quant entities.
///
/// All standard CRUD, soft-delete, pagination, and bulk methods are
/// provided automatically via `Deref` to `backbone_orm::GenericCrudRepository`.
pub struct QuantRepository(
    backbone_orm::GenericCrudRepository<Quant, backbone_orm::SoftDelete>,
);

impl std::ops::Deref for QuantRepository {
    type Target = backbone_orm::GenericCrudRepository<Quant, backbone_orm::SoftDelete>;
    fn deref(&self) -> &Self::Target { &self.0 }
}

impl QuantRepository {
    /// Create a new repository instance.
    pub fn new(pool: PgPool) -> Self {
        Self(backbone_orm::GenericCrudRepository::new(pool, TABLE_NAME))
    }
}

/// A quant row as the reservation engine sees it.
pub struct QuantRow {
    pub id: Uuid,
    pub item_id: Uuid,
    pub location_id: Uuid,
    pub lot_id: Option<Uuid>,
    pub package_id: Option<Uuid>,
    pub owner_id: Option<Uuid>,
    pub quantity: Decimal,
    pub reserved_quantity: Decimal,
    pub in_date: Option<chrono::DateTime<chrono::Utc>>,
}

impl QuantRow {
    /// `available = quantity - reserved` — always a READ off the authoritative pair, never a
    /// third writer (spec T2 / stock.hook.yaml rule T2-avail).
    pub fn available(&self) -> Decimal {
        self.quantity - self.reserved_quantity
    }
}

/// The dimension tuple that identifies exactly ONE quant row: item x location x lot x package x
/// owner (spec §3). NULL dimensions are matched with `IS NOT DISTINCT FROM` — the table's unique
/// index treats NULLs as distinct, so NULL-dimension dedup is the ENGINE's responsibility
/// (stock.hook.yaml R12), carried by the advisory lock + insert-guard below.
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub struct QuantDims {
    pub item_id: Uuid,
    pub location_id: Uuid,
    pub lot_id: Option<Uuid>,
    pub package_id: Option<Uuid>,
    pub owner_id: Option<Uuid>,
}

impl QuantDims {
    /// A stable text key for the transaction-scoped advisory lock that serializes creation of the
    /// quant row for this dimension tuple (two competing first-writers would otherwise both pass
    /// the NOT EXISTS guard — NULLs are distinct in the partial unique index).
    fn lock_key(&self) -> String {
        let opt = |v: Option<Uuid>| v.map(|u| u.to_string()).unwrap_or_else(|| "NULL".into());
        format!(
            "quant:{}:{}:{}:{}:{}",
            self.item_id, self.location_id, opt(self.lot_id), opt(self.package_id), opt(self.owner_id)
        )
    }
}

/// Hand-written quant SQL — the reservation apex. Lives here (not in the move engine) per the
/// module's 4-layer rule: services orchestrate, repositories hold SQL.
impl QuantRepository {
    /// Lock the quant row for a dimension tuple `FOR UPDATE`, creating a zeroed row if absent —
    /// the reservation-engine analogue of `BinRepository::lock_or_init`.
    ///
    /// **This row lock is what serializes competing reservations**: two `_action_assign` calls
    /// racing for the same stock queue on this row; the second reads the first's committed
    /// reservation once the lock is granted (READ COMMITTED re-evaluates the row under lock), so
    /// `reserved_quantity` can never exceed what the on-hand actually allowed (R22).
    ///
    /// Creation is serialized by a transaction-scoped advisory lock on the dimension key because
    /// the partial unique index `(item, location, lot, package, owner) WHERE deleted_at IS NULL`
    /// does NOT dedup NULL dimensions — without the advisory lock two first-writers could both
    /// insert. The lock auto-releases at commit/rollback of the caller's transaction.
    pub async fn lock_or_init(
        &self,
        conn: &mut sqlx::PgConnection,
        dims: QuantDims,
    ) -> Result<QuantRow, sqlx::Error> {
        Self::lock_or_init_on(conn, dims).await
    }

    /// [`Self::lock_or_init`] for a caller that holds no repository: every path that may create
    /// a quant goes through this one, so they all take the same creation lock.
    pub async fn lock_or_init_on(
        conn: &mut sqlx::PgConnection,
        dims: QuantDims,
    ) -> Result<QuantRow, sqlx::Error> {
        sqlx::query("SELECT pg_advisory_xact_lock(hashtextextended($1, 0))")
            .bind(dims.lock_key())
            .execute(&mut *conn)
            .await?;
        if let Some(row) = Self::fetch_for_update(conn, dims).await? {
            return Ok(row);
        }
        sqlx::query(
            r#"INSERT INTO inventory.stock_quants
                 (id, item_id, location_id, lot_id, package_id, owner_id, quantity,
                  reserved_quantity, available_quantity)
               SELECT $1,$2,$3,$4,$5,$6,0,0,0
               WHERE NOT EXISTS (
                 SELECT 1 FROM inventory.stock_quants q
                 WHERE q.item_id=$2 AND q.location_id=$3
                   AND q.lot_id IS NOT DISTINCT FROM $4
                   AND q.package_id IS NOT DISTINCT FROM $5
                   AND q.owner_id IS NOT DISTINCT FROM $6
                   AND (q.metadata->>'deleted_at') IS NULL
               )"#,
        )
        .bind(Uuid::new_v4()).bind(dims.item_id).bind(dims.location_id)
        .bind(dims.lot_id).bind(dims.package_id).bind(dims.owner_id)
        .execute(&mut *conn)
        .await?;
        // Either this call won the insert race, or the advisory lock means it never races: the
        // re-select under FOR UPDATE returns exactly one row either way.
        Self::fetch_for_update(conn, dims)
            .await?
            .ok_or_else(|| sqlx::Error::RowNotFound)
    }

    /// The `FOR UPDATE` re-select used by [`Self::lock_or_init`]. NULL-safe on every dimension.
    async fn fetch_for_update(
        conn: &mut sqlx::PgConnection,
        dims: QuantDims,
    ) -> Result<Option<QuantRow>, sqlx::Error> {
        let row = sqlx::query(
            r#"SELECT id, item_id, location_id, lot_id, package_id, owner_id, quantity,
                      reserved_quantity, in_date
               FROM inventory.stock_quants
               WHERE item_id=$1 AND location_id=$2
                 AND lot_id IS NOT DISTINCT FROM $3
                 AND package_id IS NOT DISTINCT FROM $4
                 AND owner_id IS NOT DISTINCT FROM $5
                 AND (metadata->>'deleted_at') IS NULL
               FOR UPDATE"#,
        )
        .bind(dims.item_id).bind(dims.location_id)
        .bind(dims.lot_id).bind(dims.package_id).bind(dims.owner_id)
        .fetch_optional(&mut *conn)
        .await?;
        Ok(row.map(|r| quant_row_of(r)))
    }

    /// The RESERVED step of `_synchronize_quant` (spec §3): move `delta` onto
    /// `reserved_quantity` — the authoritative counter. `available_quantity` (the stored T2
    /// compute) is rewritten in the SAME statement as the derived shadow `quantity - reserved`,
    /// so it can never drift. The caller holds the row's `FOR UPDATE` from [`Self::lock_or_init`].
    ///
    /// `delta > 0` reserves; `delta < 0` releases. The table CHECK (`reserved_quantity >= 0`)
    /// backstops release-over-reserve (a raw-SQL writer cannot underflow it — ADR-0015).
    pub async fn adjust_reserved(
        &self,
        conn: &mut sqlx::PgConnection,
        quant_id: Uuid,
        delta: Decimal,
    ) -> Result<QuantRow, sqlx::Error> {
        let row = sqlx::query(
            r#"UPDATE inventory.stock_quants
               SET reserved_quantity = reserved_quantity + $2,
                   available_quantity = quantity - (reserved_quantity + $2)
               WHERE id = $1 AND (metadata->>'deleted_at') IS NULL
               RETURNING id, item_id, location_id, lot_id, package_id, owner_id, quantity,
                         reserved_quantity, in_date"#,
        )
        .bind(quant_id).bind(delta)
        .fetch_one(&mut *conn)
        .await?;
        Ok(quant_row_of(row))
    }

    /// The AVAILABLE step of `_synchronize_quant` (spec §3): physically move stock —
    /// `qty_delta` onto `quantity` (negative = stock leaves this quant), plus a simultaneous
    /// `reserved_delta` (normally the release of the reservation this move consumed). The stored
    /// T2 compute is rewritten in lockstep. Stamp `in_date` on first arrival (FIFO ordering
    /// within the lot) when the quant transitions to a positive on-hand from non-positive.
    pub async fn apply_qty(
        &self,
        conn: &mut sqlx::PgConnection,
        quant_id: Uuid,
        qty_delta: Decimal,
        reserved_delta: Decimal,
    ) -> Result<QuantRow, sqlx::Error> {
        let row = sqlx::query(
            r#"UPDATE inventory.stock_quants
               SET quantity = quantity + $2,
                   reserved_quantity = reserved_quantity + $3,
                   available_quantity = (quantity + $2) - (reserved_quantity + $3),
                   in_date = CASE
                       WHEN in_date IS NULL AND (quantity + $2) > 0 THEN NOW()
                       ELSE in_date END
               WHERE id = $1 AND (metadata->>'deleted_at') IS NULL
               RETURNING id, item_id, location_id, lot_id, package_id, owner_id, quantity,
                         reserved_quantity, in_date"#,
        )
        .bind(quant_id).bind(qty_delta).bind(reserved_delta)
        .fetch_one(&mut *conn)
        .await?;
        Ok(quant_row_of(row))
    }

    /// Reservation candidates at one (item, location): the quants with free availability,
    /// oldest arrival first (`in_date` FIFO ordering — the default removal strategy; explicit
    /// fifo/lifo/fefo strategies layer on top when the tracking surface lands). Takes the
    /// caller's transaction: `_action_assign` locks each candidate as it reserves it.
    pub async fn fetch_reservation_candidates(
        &self,
        conn: &mut sqlx::PgConnection,
        item_id: Uuid,
        location_id: Uuid,
    ) -> Result<Vec<QuantRow>, sqlx::Error> {
        let rows = sqlx::query(
            r#"SELECT id, item_id, location_id, lot_id, package_id, owner_id, quantity,
                      reserved_quantity, in_date
               FROM inventory.stock_quants
               WHERE item_id=$1 AND location_id=$2
                 AND quantity - reserved_quantity > 0
                 AND (metadata->>'deleted_at') IS NULL
               ORDER BY in_date NULLS LAST, id"#,
        )
        .bind(item_id).bind(location_id)
        .fetch_all(&mut *conn)
        .await?;
        Ok(rows.into_iter().map(quant_row_of).collect())
    }

    /// Re-derive one quant's authoritative `reserved_quantity` from the MIRROR arm: the sum of
    /// the live move-line quantities matching this quant's dimension tuple (state not
    /// done/cancel). This is the triangle's self-heal — the reserved step of
    /// `_synchronize_quant` after a move's lines flip out of the live set (done or cancel),
    /// including the partial-validate case where an operator shrinks a line below what it had
    /// reserved (the residual would otherwise stay held by no move). The mirror can only ever
    /// reflect reservations the engine minted, so the SUM is the truth by construction.
    ///
    /// Takes the caller's transaction; the caller holds the quant's `FOR UPDATE` from
    /// [`Self::lock_or_init`]. Returns the recomputed row.
    pub async fn recompute_reserved_from_mirror(
        &self,
        conn: &mut sqlx::PgConnection,
        quant_id: Uuid,
        dims: QuantDims,
    ) -> Result<QuantRow, sqlx::Error> {
        let row = sqlx::query(
            r#"UPDATE inventory.stock_quants q
               SET reserved_quantity = m.mirror_qty,
                   available_quantity = q.quantity - m.mirror_qty
               FROM (
                   SELECT COALESCE(SUM(l.quantity), 0) AS mirror_qty
                   FROM inventory.stock_move_lines l
                   WHERE l.item_id = $2
                     AND l.location_id = $3
                     AND l.lot_id IS NOT DISTINCT FROM $4
                     AND l.package_id IS NOT DISTINCT FROM $5
                     AND l.owner_id IS NOT DISTINCT FROM $6
                     AND l.state::text NOT IN ('done', 'cancel')
                     AND (l.metadata->>'deleted_at') IS NULL
               ) m
               WHERE q.id = $1 AND (q.metadata->>'deleted_at') IS NULL
               RETURNING q.id, q.item_id, q.location_id, q.lot_id, q.package_id, q.owner_id,
                         q.quantity, q.reserved_quantity, q.in_date"#,
        )
        .bind(quant_id)
        .bind(dims.item_id)
        .bind(dims.location_id)
        .bind(dims.lot_id)
        .bind(dims.package_id)
        .bind(dims.owner_id)
        .fetch_one(&mut *conn)
        .await?;
        Ok(quant_row_of(row))
    }
}

/// On-hand aggregation at one location — the `available = quantity - reserved` READ (T2), summed
/// over the location's quant grain. Pool-based (each read its own unit of work), riding the
/// ambient org scope like every read in the suite.
pub struct QuantOnHandRow {
    pub on_hand_qty: Decimal,
    pub reserved_qty: Decimal,
    pub quant_count: i64,
}

impl QuantRepository {
    /// SUM(quantity) / SUM(reserved_quantity) over one (item, location). Zeroed row when
    /// no quant exists (an unreceived item is unavailable, not an error).
    pub async fn fetch_on_hand(
        &self,
        pool: &PgPool,
        item_id: Uuid,
        location_id: Uuid,
    ) -> Result<QuantOnHandRow, sqlx::Error> {
        let row = backbone_orm::org_scope::fetch_one_row_scoped(
            pool,
            sqlx::query(
                r#"SELECT
                       COALESCE(SUM(quantity), 0) AS on_hand_qty,
                       COALESCE(SUM(reserved_quantity), 0) AS reserved_qty,
                       COUNT(*) AS quant_count
                   FROM inventory.stock_quants
                   WHERE item_id=$1 AND location_id=$2
                     AND (metadata->>'deleted_at') IS NULL"#,
            )
            .bind(item_id).bind(location_id),
        )
        .await?;
        Ok(QuantOnHandRow {
            on_hand_qty: row.get("on_hand_qty"),
            reserved_qty: row.get("reserved_qty"),
            quant_count: row.get("quant_count"),
        })
    }

    /// One warehouse-scope availability row (the shop-pivot read): SUM(quantity) /
    /// SUM(reserved_quantity) over every quant whose location belongs to the pivot warehouse and
    /// holds stock (usage `internal`). `available` is NOT read off the stored
    /// `available_quantity` column — the caller computes `on_hand − reserved` per call, so the
    /// number is fresh off the reservation triangle regardless of any stored-compute staleness.
    ///
    /// Batch over `item_ids = ANY($1)`; items with no quant rows in the warehouse are simply
    /// absent from the result (the caller projects them as zeroed, unavailable rows — an
    /// unreceived item is unavailable, not an error). Pool-based and org-scoped: the read rides
    /// the ambient org scope the composing service set per request (ADR-0029).
    pub async fn fetch_warehouse_on_hand(
        &self,
        pool: &PgPool,
        item_ids: &[Uuid],
        warehouse_id: Uuid,
    ) -> Result<Vec<QuantWarehouseRow>, sqlx::Error> {
        let rows = backbone_orm::org_scope::fetch_all_rows_scoped(
            pool,
            sqlx::query(
                r#"SELECT
                       q.item_id AS item_id,
                       COALESCE(SUM(q.quantity), 0) AS on_hand_qty,
                       COALESCE(SUM(q.reserved_quantity), 0) AS reserved_qty
                   FROM inventory.stock_quants q
                   JOIN inventory.locations l ON l.id = q.location_id
                   WHERE q.item_id = ANY($1)
                     AND l.warehouse_id = $2
                     AND l.usage = 'internal'::location_usage
                     AND (q.metadata->>'deleted_at') IS NULL
                     AND (l.metadata->>'deleted_at') IS NULL
                   GROUP BY q.item_id"#,
            )
            .bind(item_ids)
            .bind(warehouse_id),
        )
        .await?;
        Ok(rows
            .into_iter()
            .map(|r| QuantWarehouseRow {
                item_id: r.get("item_id"),
                on_hand_qty: r.get("on_hand_qty"),
                reserved_qty: r.get("reserved_qty"),
            })
            .collect())
    }
}

/// One item's warehouse-scope on-hand pair (the shop-pivot availability read).
pub struct QuantWarehouseRow {
    pub item_id: Uuid,
    pub on_hand_qty: Decimal,
    pub reserved_qty: Decimal,
}

fn quant_row_of(r: sqlx::postgres::PgRow) -> QuantRow {
    QuantRow {
        id: r.get("id"),
        item_id: r.get("item_id"),
        location_id: r.get("location_id"),
        lot_id: r.get("lot_id"),
        package_id: r.get("package_id"),
        owner_id: r.get("owner_id"),
        quantity: r.get("quantity"),
        reserved_quantity: r.get("reserved_quantity"),
        in_date: r.get("in_date"),
    }
}

backbone_core::impl_crud_repository!(QuantRepository, Quant, soft_delete);