backbone_inventory/infrastructure/persistence/

stock_ledger_entry_repository.rs

//! Repository for StockLedgerEntry 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 append-only Stock Ledger SQL
//! (4-layer rule: services orchestrate, repositories hold the SQL).
//!
//! Thin newtype over `backbone_orm::GenericCrudRepository<StockLedgerEntry, backbone_orm::SoftDelete>`.
//! All standard CRUD methods are available via `Deref`.

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

use crate::domain::entity::StockLedgerEntry;

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

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

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

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

/// The exact row an immutable Stock Ledger Entry insert writes.
///
/// Mirrors the raw column shape rather than the `StockLedgerEntry` entity: `voucher_type` is
/// supplied as a free string and cast at the DB (`$12::voucher_type`) so a bad value fails as a DB
/// error, and `status` is a literal 'active' (the SLE is append-only — a correction is a new
/// entry, never an edit).
pub struct NewSleRow<'a> {
    pub item_id: Uuid,
    pub warehouse_id: Uuid,
    pub posting_date: chrono::NaiveDate,
    /// Signed: positive on an inflow, negative on an outflow.
    pub actual_qty: Decimal,
    pub qty_after_txn: Decimal,
    pub incoming_rate: Decimal,
    pub valuation_rate: Decimal,
    pub stock_value: Decimal,
    /// Signed value delta this entry contributes — what the GL post mirrors.
    pub stock_value_difference: Decimal,
    /// "purchase_receipt" | "delivery_note" | "stock_entry" | "stock_reconciliation".
    pub voucher_type: &'a str,
    pub voucher_id: Uuid,
    pub voucher_no: &'a str,
    pub sle_no: i32,
}

/// Hand-written StockLedgerEntry SQL. Lives here (not in the write service) per the module's 4-layer
/// rule: services orchestrate and own the unit of work, repositories hold the SQL.
impl StockLedgerEntryRepository {
    /// Append one immutable SLE.
    ///
    /// Takes the CALLER'S connection: the SLE and the Bin balance it describes must commit as one
    /// unit, on the transaction holding the bin's `FOR UPDATE`. The caller has already relayed the
    /// ambient org scope onto it — don't re-bind here.
    pub async fn insert_sle(
        &self,
        conn: &mut sqlx::PgConnection,
        e: &NewSleRow<'_>,
    ) -> Result<(), sqlx::Error> {
        sqlx::query(
            r#"INSERT INTO inventory.stock_ledger_entries
                (id, item_id, warehouse_id, posting_date, actual_qty, qty_after_txn,
                 incoming_rate, valuation_rate, stock_value, stock_value_difference, voucher_type,
                 voucher_id, voucher_no, sle_no, status)
               VALUES ($1,$2,$3,$4,$5,$6,$7,$8,$9,$10,$11::voucher_type,$12,$13,$14,'active')"#,
        )
        .bind(Uuid::new_v4()).bind(e.item_id).bind(e.warehouse_id).bind(e.posting_date)
        .bind(e.actual_qty).bind(e.qty_after_txn).bind(e.incoming_rate).bind(e.valuation_rate)
        .bind(e.stock_value).bind(e.stock_value_difference).bind(e.voucher_type).bind(e.voucher_id)
        .bind(e.voucher_no).bind(e.sle_no)
        .execute(conn)
        .await?;
        Ok(())
    }

    /// The highest `sle_no` a voucher has already used — so a cancellation can continue the sequence
    /// with its compensating entries instead of colliding on the
    /// `(voucher_type, voucher_id, sle_no)` unique index. Takes the caller's connection (rides the
    /// movement's transaction). 0 when the voucher has no entries yet.
    pub async fn fetch_max_sle_no(
        &self,
        conn: &mut sqlx::PgConnection,
        voucher_type: &str,
        voucher_id: Uuid,
    ) -> Result<i32, sqlx::Error> {
        let max: Option<i32> = sqlx::query_scalar(
            r#"SELECT MAX(sle_no) FROM inventory.stock_ledger_entries
               WHERE voucher_type=$1::voucher_type AND voucher_id=$2 AND (metadata->>'deleted_at') IS NULL"#,
        )
        .bind(voucher_type).bind(voucher_id)
        .fetch_one(conn)
        .await?;
        Ok(max.unwrap_or(0))
    }

    /// The total value one move carried off its source bin: the NEGATED sum of the move's
    /// signed SLE value deltas (a move's rows ride `voucher_type='stock_entry'` with
    /// `voucher_no` = the move name; an outflow's delta is negative, so negating yields the
    /// positive value the move took out — the COGS of a delivery-shaped move). A voucher door
    /// resuming a crashed submit uses this to recover an already-landed line's value without
    /// re-reading the bins. 0 when the move has no rows.
    pub async fn sum_move_value(
        &self,
        pool: &PgPool,
        move_name: &str,
    ) -> Result<Decimal, sqlx::Error> {
        let row = backbone_orm::org_scope::fetch_one_row_scoped(
            pool,
            sqlx::query(
                r#"SELECT COALESCE(-SUM(stock_value_difference), 0) AS carried
                   FROM inventory.stock_ledger_entries
                   WHERE voucher_no=$1 AND voucher_type='stock_entry'
                     AND (metadata->>'deleted_at') IS NULL"#,
            )
            .bind(move_name),
        )
        .await?;
        Ok(row.get("carried"))
    }

    /// The value legs ONE move minted, split by direction: `(out_value, in_value)` — the
    /// negated sum of the move's negative `stock_value_difference` rows (what the OUT leg
    /// carried off the source bin) and the sum of its positive rows (what the IN leg blended
    /// into the destination). Keyed on `voucher_id` = the move id (the exact grain the mint
    /// wrote; move names are descriptive, ids are unique). A GL repost uses this to rebuild the
    /// move's envelope from the committed ledger without re-reading the bins. `(0, 0)` when the
    /// move minted no rows.
    pub async fn move_leg_values(
        &self,
        pool: &PgPool,
        move_id: Uuid,
    ) -> Result<(Decimal, Decimal), sqlx::Error> {
        let row = backbone_orm::org_scope::fetch_one_row_scoped(
            pool,
            sqlx::query(
                r#"SELECT
                       COALESCE(-SUM(stock_value_difference) FILTER (WHERE stock_value_difference < 0), 0) AS out_value,
                       COALESCE(SUM(stock_value_difference) FILTER (WHERE stock_value_difference > 0), 0) AS in_value
                   FROM inventory.stock_ledger_entries
                   WHERE voucher_type='stock_entry' AND voucher_id=$1
                     AND (metadata->>'deleted_at') IS NULL"#,
            )
            .bind(move_id),
        )
        .await?;
        Ok((row.get("out_value"), row.get("in_value")))
    }

    /// The value ONE move blended into its destination bin: the sum of the move's positive
    /// signed SLE value deltas, on the caller's connection (a landed-cost validation reads
    /// every target move's carried value inside its read phase). This is the `value` basis of
    /// a landed-cost split. 0 when the move minted no rows.
    pub async fn sum_move_in_value(
        &self,
        conn: &mut sqlx::PgConnection,
        move_id: Uuid,
    ) -> Result<Decimal, sqlx::Error> {
        let row: Option<Decimal> = sqlx::query_scalar(
            r#"SELECT COALESCE(SUM(stock_value_difference), 0)
               FROM inventory.stock_ledger_entries
               WHERE voucher_type='stock_entry' AND voucher_id=$1
                 AND stock_value_difference > 0 AND (metadata->>'deleted_at') IS NULL"#,
        )
        .bind(move_id)
        .fetch_one(&mut *conn)
        .await?;
        Ok(row.unwrap_or(Decimal::ZERO))
    }

    /// The landed-cost revaluation rows one landed cost minted (`voucher_type='landed_cost'`,
    /// one value-only row per target move line): the voucher name and the signed value delta
    /// each row carries. A landed-cost GL repost rebuilds its debit legs from exactly these
    /// rows — the minted ledger, never a re-computation. Takes the caller's connection.
    pub async fn fetch_lc_revaluations(
        &self,
        conn: &mut sqlx::PgConnection,
        lc_id: Uuid,
    ) -> Result<Vec<(String, Decimal)>, sqlx::Error> {
        let rows = sqlx::query(
            r#"SELECT voucher_no, stock_value_difference FROM inventory.stock_ledger_entries
               WHERE voucher_type='landed_cost' AND voucher_id=$1
                 AND (metadata->>'deleted_at') IS NULL
               ORDER BY sle_no"#,
        )
        .bind(lc_id)
        .fetch_all(&mut *conn)
        .await?;
        Ok(rows
            .into_iter()
            .map(|r| (r.get::<String, _>("voucher_no"), r.get::<Decimal, _>("stock_value_difference")))
            .collect())
    }

    /// Whether a voucher already minted an SLE row under an exact `voucher_no` — the
    /// crash-resume probe for deterministic-name inserts (a landed-cost revaluation row
    /// `{lc_number}/{move_name}/{idx}` that already landed in a prior crashed attempt must be
    /// skipped, never re-minted). Takes the caller's connection (rides the caller's unit of
    /// work).
    pub async fn has_sle_named(
        &self,
        conn: &mut sqlx::PgConnection,
        voucher_type: &str,
        voucher_id: Uuid,
        voucher_no: &str,
    ) -> Result<bool, sqlx::Error> {
        let n: i64 = sqlx::query_scalar(
            r#"SELECT COUNT(*) FROM inventory.stock_ledger_entries
               WHERE voucher_type=$1::voucher_type AND voucher_id=$2 AND voucher_no=$3
                 AND (metadata->>'deleted_at') IS NULL"#,
        )
        .bind(voucher_type)
        .bind(voucher_id)
        .bind(voucher_no)
        .fetch_one(&mut *conn)
        .await?;
        Ok(n > 0)
    }

    /// Read-only FIFO ATTRIBUTION of a bin's current on-hand onto its inbound moves: for each
    /// move, how many of the units it blended in are still on hand. This answers "how much of
    /// this receipt is left to revalue" for landed costs.
    ///
    /// The arithmetic (per (item, warehouse)): order the IN rows FIFO by
    /// `(posting_date, created_at, id)`; the units still on hand are the NEWEST ones, so a
    /// move's remaining share is `min(qty_m, max(0, on_hand − qty of IN rows strictly before
    /// it))`. OUT rows never enter the formula — their effect is already carried by the bin's
    /// current `actual_qty`. Value-only rows (qty 0, e.g. landed-cost adjustments) never move
    /// the attribution.
    ///
    /// **This is attribution, NOT FIFO costing** — the moving-average SLE engine stays the
    /// single valuation writer; nothing here feeds a rate or a value.
    pub async fn remaining_qty_for_moves(
        &self,
        pool: &PgPool,
        move_ids: &[Uuid],
    ) -> Result<std::collections::HashMap<Uuid, Decimal>, sqlx::Error> {
        if move_ids.is_empty() { return Ok(std::collections::HashMap::new()); }
        let rows = backbone_orm::org_scope::fetch_all_rows_scoped(
            pool,
            sqlx::query(
                r#"WITH ins AS (
                       SELECT s.voucher_id AS move_id, s.item_id, s.warehouse_id,
                              s.actual_qty AS qty,
                              COALESCE(SUM(s.actual_qty) OVER (
                                  PARTITION BY s.item_id, s.warehouse_id
                                  ORDER BY s.posting_date, (s.metadata->>'created_at'), s.id
                                  ROWS BETWEEN UNBOUNDED PRECEDING AND 1 PRECEDING
                              ), 0) AS prior_in
                       FROM inventory.stock_ledger_entries s
                       WHERE s.voucher_type='stock_entry'
                         AND s.actual_qty > 0 AND (s.metadata->>'deleted_at') IS NULL
                   )
                   SELECT ins.move_id,
                          LEAST(ins.qty, GREATEST(COALESCE(b.actual_qty, 0) - ins.prior_in, 0)) AS remaining
                   FROM ins
                   LEFT JOIN inventory.bins b
                          ON b.item_id=ins.item_id
                         AND b.warehouse_id=ins.warehouse_id
                         AND (b.metadata->>'deleted_at') IS NULL
                   WHERE ins.move_id = ANY($1)"#,
            )
            .bind(move_ids),
        )
        .await?;
        Ok(rows
            .into_iter()
            .map(|r| (r.get::<Uuid, _>("move_id"), r.get::<Decimal, _>("remaining")))
            .collect())
    }
}

backbone_core::impl_crud_repository!(StockLedgerEntryRepository, StockLedgerEntry, soft_delete);