backbone-inventory 0.17.2

//! Repository for StockMove 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 stock-move SQL: the guarded 7-state lifecycle writes
//! (draft/waiting/confirmed/partially_available/assigned/done/cancel — spec stock-business-logic
//! §1), chain linkage over `move_orig_ids`/`move_dest_ids`, and the T1 picking-state projection
//! recompute (the transfer's state is a STORED COMPUTE over its moves — never a machine of its
//! own; this repo re-derives it from the move rows after every state change).
//!
//! 4-layer rule: services orchestrate, repositories hold SQL. Write methods take the CALLER'S
//! connection (the movement's transaction); the caller has already relayed the ambient org
//! scope onto it. State values bind as `$n::move_state` (unqualified — the enum type is created
//! unqualified in migrations) and decode through `::text`.

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

use backbone_orm::org_scope;

use crate::domain::entity::StockMove;

use super::picking_batch_projection_repository::PickingBatchProjectionRepository;

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

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

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

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

/// The exact row a draft move insert writes. `state` starts `draft`; the pipeline
/// (`_action_confirm` / `_action_assign` / `_action_done` / `_action_cancel`) is the only writer
/// of any other state.
pub struct NewMoveRow<'a> {
    pub id: Uuid,
    pub name: &'a str,
    pub item_id: Uuid,
    pub demand_qty: Decimal,
    pub price_unit: Decimal,
    /// "make_to_stock" | "make_to_order" | "mts_else_mto".
    pub procure_method: &'a str,
    pub picking_id: Option<Uuid>,
    pub origin: Option<&'a str>,
    pub location_id: Uuid,
    pub location_dest_id: Uuid,
    pub partner_id: Option<Uuid>,
    pub warehouse_id: Option<Uuid>,
    pub orderpoint_id: Option<Uuid>,
    pub move_orig_ids: Vec<Uuid>,
    pub move_dest_ids: Vec<Uuid>,
    pub is_inventory: bool,
    pub scrapped: bool,
    /// Explicit total value the move carries — a document reversal valued at its original
    /// amount. `None` on ordinary moves (the valuation core derives the carry).
    pub forced_value: Option<Decimal>,
}

/// A move row as the engine reads it. `state` / `procure_method` decode as text (the enum lives in
/// the DB; string comparison against the snake_case vocabulary is the module's proven pattern).
#[derive(Clone)]
pub struct MoveRow {
    pub id: Uuid,
    pub name: String,
    pub state: String,
    /// GL settlement state of the move's OWN AccountingPost leg, decoded as text
    /// ("not_applicable" | "pending" | "posted" | "failed"). Distinct from the lifecycle
    /// `state`: a `done` move can carry a `failed` GL leg (the physical movement stands,
    /// the GL leg is missing and re-drivable).
    pub posting_state: String,
    pub date: chrono::DateTime<chrono::Utc>,
    pub item_id: Uuid,
    pub demand_qty: Decimal,
    pub quantity: Decimal,
    pub price_unit: Decimal,
    pub procure_method: String,
    pub picking_id: Option<Uuid>,
    pub origin: Option<String>,
    /// The procurement rule that minted this move (the `_run_pull` / `_run_push` provenance —
    /// moves minted through a rule carry it, door-minted and hand-minted moves do not).
    pub rule_id: Option<Uuid>,
    pub location_id: Uuid,
    pub location_dest_id: Uuid,
    pub partner_id: Option<Uuid>,
    pub warehouse_id: Option<Uuid>,
    pub orderpoint_id: Option<Uuid>,
    pub move_orig_ids: Vec<Uuid>,
    pub move_dest_ids: Vec<Uuid>,
    pub is_inventory: bool,
    pub scrapped: bool,
    pub propagate_cancel: bool,
    /// Explicit total value the move carries (a document reversal valued at its original
    /// amount); `None` on ordinary moves.
    pub forced_value: Option<Decimal>,
}

/// Hand-written stock-move SQL — the lifecycle writes. Guarded: state transitions carry the
/// expected source state in their WHERE clause, so a stale read cannot silently overwrite a move
/// another caller already advanced (the optimistic guard behind the declared state machine in
/// schema/hooks/stock.hook.yaml; ADR-0015 service pre-check + the WHERE clause as the backstop).
impl StockMoveRepository {
    /// Insert a draft move.
    pub async fn insert_move(
        &self,
        conn: &mut sqlx::PgConnection,
        m: &NewMoveRow<'_>,
    ) -> Result<(), sqlx::Error> {
        sqlx::query(
            r#"INSERT INTO inventory.stock_moves
                 (id, name, state, item_id, demand_qty, price_unit, procure_method, picking_id,
                  origin, location_id, location_dest_id, partner_id, warehouse_id,
                  orderpoint_id, move_orig_ids, move_dest_ids, is_inventory, scrapped, forced_value)
               VALUES ($1,$2,'draft'::move_state,$3,$4,$5,$6::procure_method,$7,$8,$9,$10,$11,
                       $12,$13,$14,$15,$16,$17,$18)"#,
        )
        .bind(m.id).bind(m.name).bind(m.item_id).bind(m.demand_qty).bind(m.price_unit)
        .bind(m.procure_method).bind(m.picking_id).bind(m.origin)
        .bind(m.location_id).bind(m.location_dest_id).bind(m.partner_id)
        .bind(m.warehouse_id).bind(m.orderpoint_id)
        .bind(&m.move_orig_ids).bind(&m.move_dest_ids)
        .bind(m.is_inventory).bind(m.scrapped).bind(m.forced_value)
        .execute(conn)
        .await?;
        Ok(())
    }

    /// Fetch one move. Returns `None` when absent or soft-deleted.
    pub async fn fetch_move(
        &self,
        conn: &mut sqlx::PgConnection,
        move_id: Uuid,
    ) -> Result<Option<MoveRow>, sqlx::Error> {
        let row = sqlx::query(
            r#"SELECT id, name, state::text AS state, posting_state::text AS posting_state, date, item_id, demand_qty, quantity,
                      price_unit, procure_method::text AS procure_method, picking_id, origin, rule_id,
                      location_id, location_dest_id, partner_id, warehouse_id,
                      orderpoint_id, move_orig_ids, move_dest_ids, is_inventory, scrapped,
                      propagate_cancel, forced_value
               FROM inventory.stock_moves
               WHERE id=$1 AND (metadata->>'deleted_at') IS NULL"#,
        )
        .bind(move_id)
        .fetch_optional(&mut *conn)
        .await?;
        Ok(row.map(move_row_of))
    }

    /// Fetch several moves by id (chain propagation). Order follows the input ids.
    pub async fn fetch_moves(
        &self,
        conn: &mut sqlx::PgConnection,
        move_ids: &[Uuid],
    ) -> Result<Vec<MoveRow>, sqlx::Error> {
        let rows = sqlx::query(
            r#"SELECT id, name, state::text AS state, posting_state::text AS posting_state, date, item_id, demand_qty, quantity,
                      price_unit, procure_method::text AS procure_method, picking_id, origin, rule_id,
                      location_id, location_dest_id, partner_id, warehouse_id,
                      orderpoint_id, move_orig_ids, move_dest_ids, is_inventory, scrapped,
                      propagate_cancel, forced_value
               FROM inventory.stock_moves
               WHERE id = ANY($1) AND (metadata->>'deleted_at') IS NULL"#,
        )
        .bind(move_ids)
        .fetch_all(&mut *conn)
        .await?;
        Ok(rows.into_iter().map(move_row_of).collect())
    }

    /// Fetch the moves minted for one voucher (`origin` — voucher doors stamp the
    /// voucher number there so a crashed submit can find and resume its line moves).
    pub async fn fetch_moves_by_origin(
        &self,
        conn: &mut sqlx::PgConnection,
        origin: &str,
    ) -> Result<Vec<MoveRow>, sqlx::Error> {
        let rows = sqlx::query(
            r#"SELECT id, name, state::text AS state, posting_state::text AS posting_state, date, item_id, demand_qty, quantity,
                      price_unit, procure_method::text AS procure_method, picking_id, origin, rule_id,
                      location_id, location_dest_id, partner_id, warehouse_id,
                      orderpoint_id, move_orig_ids, move_dest_ids, is_inventory, scrapped,
                      propagate_cancel, forced_value
               FROM inventory.stock_moves
               WHERE origin=$1 AND (metadata->>'deleted_at') IS NULL
               ORDER BY name"#,
        )
        .bind(origin)
        .fetch_all(&mut *conn)
        .await?;
        Ok(rows.into_iter().map(move_row_of).collect())
    }

    /// Guarded state transition: `UPDATE ... WHERE id=$1 AND state=$expected`. Returns Ok(false)
    /// when zero rows matched — the move is not in the expected state (the caller maps that to a
    /// domain error). This is the declared state machine's write valve: states only move through
    /// here with their declared source.
    pub async fn transition_state(
        &self,
        conn: &mut sqlx::PgConnection,
        move_id: Uuid,
        expected_from: &str,
        to: &str,
    ) -> Result<bool, sqlx::Error> {
        let res = sqlx::query(
            r#"UPDATE inventory.stock_moves
               SET state = $3::move_state
               WHERE id=$1 AND state = $2::move_state AND (metadata->>'deleted_at') IS NULL"#,
        )
        .bind(move_id).bind(expected_from).bind(to)
        .execute(&mut *conn)
        .await?;
        Ok(res.rows_affected() == 1)
    }

    /// The done write: `state='done'`, the done quantity, and the effective-date stamp (the
    /// `date` column is the schema's effective-date surface — set to the processing instant on
    /// done; there is no separate `date_done` column on moves). Guarded on the expected source
    /// state like [`Self::transition_state`].
    pub async fn mark_done(
        &self,
        conn: &mut sqlx::PgConnection,
        move_id: Uuid,
        expected_from: &str,
        done_qty: Decimal,
        when: chrono::DateTime<chrono::Utc>,
    ) -> Result<bool, sqlx::Error> {
        let res = sqlx::query(
            r#"UPDATE inventory.stock_moves
               SET state='done'::move_state, quantity=$3, date=$4
               WHERE id=$1 AND state = $2::move_state AND (metadata->>'deleted_at') IS NULL"#,
        )
        .bind(move_id).bind(expected_from).bind(done_qty).bind(when)
        .execute(&mut *conn)
        .await?;
        Ok(res.rows_affected() == 1)
    }

    /// Attach a move to its grouping transfer (the picking-assignment write: the move's
    /// `picking_id` is set, and the projection re-derives on the caller's next reproject —
    /// this write only moves the pointer). Runs inside the caller's transaction; the ambient
    /// org scope is already relayed onto it by the service that resolved the group.
    pub async fn set_picking(
        &self,
        conn: &mut sqlx::PgConnection,
        move_id: Uuid,
        picking_id: Uuid,
    ) -> Result<(), sqlx::Error> {
        sqlx::query(
            r#"UPDATE inventory.stock_moves
               SET picking_id = $2
               WHERE id = $1 AND (metadata->>'deleted_at') IS NULL"#,
        )
        .bind(move_id)
        .bind(picking_id)
        .execute(&mut *conn)
        .await?;
        Ok(())
    }

    /// Link a backorder child to its parent: the parent's `move_dest_ids` gains the child, the
    /// child's `move_orig_ids` gains the parent (the chain vocabulary `move_orig_ids`/
    /// `move_dest_ids` — spec §1/§4; done-qty propagation and `waiting` release walk these).
    pub async fn link_chain(
        &self,
        conn: &mut sqlx::PgConnection,
        parent_id: Uuid,
        child_id: Uuid,
    ) -> Result<(), sqlx::Error> {
        sqlx::query(
            r#"UPDATE inventory.stock_moves
               SET move_dest_ids = array_append(move_dest_ids, $2)
                   WHERE id=$1 AND NOT ($2 = ANY(move_dest_ids))"#,
        )
        .bind(parent_id).bind(child_id)
        .execute(&mut *conn)
        .await?;
        sqlx::query(
            r#"UPDATE inventory.stock_moves
               SET move_orig_ids = array_append(move_orig_ids, $2)
                   WHERE id=$1 AND NOT ($2 = ANY(move_orig_ids))"#,
        )
        .bind(child_id).bind(parent_id)
        .execute(&mut *conn)
        .await?;
        Ok(())
    }
}

/// The GL-leg settlement writes (the move's OWN AccountingPost outcome — the same posture the
/// voucher doors keep on their headers). The `pending` marker is written INSIDE the done
/// transaction, atomically with the physical movement; the `posted`/`failed` reconciles run on
/// the pool AFTER the movement commits (the GL leg is eventually consistent — a crash between
/// commit and the reconcile leaves `pending`, which the repost verb re-drives; accounting
/// dedupes on `(source_type, source_id, posting_type)` so a re-drive can never double post).
impl StockMoveRepository {
    /// Arm the GL leg: `posting_state = pending`, inside the movement's own transaction. Called
    /// by the done path only when it actually built an envelope for this move — moves that post
    /// no GL of their own never leave `not_applicable`.
    pub async fn set_posting_pending(
        &self,
        conn: &mut PgConnection,
        move_id: Uuid,
    ) -> Result<(), sqlx::Error> {
        sqlx::query(
            r#"UPDATE inventory.stock_moves
               SET posting_state='pending'::gl_posting_state
               WHERE id=$1"#,
        )
        .bind(move_id)
        .execute(&mut *conn)
        .await?;
        Ok(())
    }

    /// The pending→posted reconcile from the accounting ack. Guarded on
    /// `posting_state <> 'posted'` so a re-drive that accounting answered from its dedupe
    /// (returning the ORIGINAL journal) can never flip a settled state.
    ///
    /// Runs `org_scope::execute_scoped` on the pool: the write rides the ambient org scope's
    /// request-dedicated connection (the composing service sets it per request) so it passes
    /// the decorator's fence (ADR-0029).
    pub async fn mark_posting_posted(
        &self,
        pool: &PgPool,
        move_id: Uuid,
    ) -> Result<(), sqlx::Error> {
        org_scope::execute_scoped(
            pool,
            sqlx::query(
                r#"UPDATE inventory.stock_moves
                   SET posting_state='posted'::gl_posting_state
                   WHERE id=$1 AND posting_state <> 'posted'::gl_posting_state"#,
            )
            .bind(move_id),
        )
        .await?;
        Ok(())
    }

    /// Record a GL rejection. The physical movement is NOT rolled back — it really happened;
    /// the move parks in `failed` and is re-drivable via `repost_move_gl`.
    ///
    /// Same org-scoped pool write as [`Self::mark_posting_posted`].
    pub async fn mark_posting_failed(
        &self,
        pool: &PgPool,
        move_id: Uuid,
    ) -> Result<(), sqlx::Error> {
        org_scope::execute_scoped(
            pool,
            sqlx::query(
                r#"UPDATE inventory.stock_moves
                   SET posting_state='failed'::gl_posting_state
                   WHERE id=$1"#,
            )
            .bind(move_id),
        )
        .await?;
        Ok(())
    }

    /// Retire an unsettled GL leg back to `not_applicable` — used by the repost verb when the
    /// envelope can no longer be built (the directive no longer supplies the accounts the leg
    /// shape needs), so a stuck `pending`/`failed` move whose GL is genuinely not postable under
    /// the current configuration settles instead of living on the sweep worklist forever. Never
    /// touches a `posted` move.
    pub async fn mark_posting_not_applicable(
        &self,
        pool: &PgPool,
        move_id: Uuid,
    ) -> Result<(), sqlx::Error> {
        org_scope::execute_scoped(
            pool,
            sqlx::query(
                r#"UPDATE inventory.stock_moves
                   SET posting_state='not_applicable'::gl_posting_state
                   WHERE id=$1 AND posting_state IN ('pending'::gl_posting_state, 'failed'::gl_posting_state)"#,
            )
            .bind(move_id),
        )
        .await?;
        Ok(())
    }
}

/// The location facts a move's guards and valuation core need: usage (supplier/view/internal/
/// customer/inventory/production/transit — the routing + reservation rules key off it), and the
/// warehouse a valuation bin resolves to. `usage` decodes as text like every other enum here.
pub struct LocationFacts {
    pub id: Uuid,
    pub usage: String,
    pub warehouse_id: Option<Uuid>,
    /// Per-location valuation-account override: when set, an inventory GL leg that touches this
    /// location resolves to it ahead of the caller-supplied (door-header / directive) account —
    /// the smallest-first override of the account-resolution chain. `None` = no override.
    pub valuation_account_id: Option<Uuid>,
}

/// Location reads the move pipeline owns (guards R9/R13 + the bin-warehouse resolution in
/// `_action_done`'s valuation core). One round trip for both endpoints.
impl StockMoveRepository {
    /// Fetch a move's two endpoint locations. Returns `(src, dest)` as read; `None` for an absent
    /// or soft-deleted endpoint (the caller maps that to `LocationNotFound`).
    pub async fn fetch_move_locations(
        &self,
        conn: &mut sqlx::PgConnection,
        location_id: Uuid,
        location_dest_id: Uuid,
    ) -> Result<(Option<LocationFacts>, Option<LocationFacts>), sqlx::Error> {
        let rows = sqlx::query(
            r#"SELECT id, usage::text AS usage, warehouse_id, valuation_account_id
               FROM inventory.locations
               WHERE id IN ($1, $2) AND (metadata->>'deleted_at') IS NULL"#,
        )
        .bind(location_id)
        .bind(location_dest_id)
        .fetch_all(&mut *conn)
        .await?;
        let find = |id: Uuid| {
            rows.iter().find(|r| r.get::<Uuid, _>("id") == id).map(|r| LocationFacts {
                id: r.get("id"),
                usage: r.get("usage"),
                warehouse_id: r.get("warehouse_id"),
                valuation_account_id: r.get("valuation_account_id"),
            })
        };
        Ok((find(location_id), find(location_dest_id)))
    }
}

/// The T1 picking-state projection (spec §2, ADR-0016 `projection` shape).
///
/// `transfers.state` is a STORED COMPUTE over its moves — `_get_relevant_state_among_moves`: the
/// LEAST-advanced non-cancelled move state (cancelled moves drop out of the aggregation entirely;
/// all-cancelled shows `cancel`). The picking has no state machine of its own anywhere; every
/// move-state change re-runs this recompute. The SQL lives on the MOVE repository because the
/// projection derives entirely from move rows — the transfer merely receives the result.
impl StockMoveRepository {
    /// Recompute + write one transfer's projected state from its moves. Returns the new projected
    /// state (text), or `None` when the transfer has no live moves. Also stamps the transfer's
    /// `date_done` when the projection first reaches `done`.
    pub async fn reproject_picking(
        &self,
        conn: &mut sqlx::PgConnection,
        picking_id: Uuid,
    ) -> Result<Option<String>, sqlx::Error> {
        let row = sqlx::query(
            r#"WITH agg AS (
                   SELECT
                       COUNT(*) FILTER (WHERE state <> 'cancel'::move_state) AS live_count,
                       COUNT(*) AS total_count,
                       MIN(CASE state WHEN 'draft'::move_state THEN 0
                                      WHEN 'waiting'::move_state THEN 1
                                      WHEN 'confirmed'::move_state THEN 2
                                      WHEN 'partially_available'::move_state THEN 2
                                      WHEN 'assigned'::move_state THEN 3
                                      WHEN 'done'::move_state THEN 4
                                      ELSE 5 END) FILTER (WHERE state <> 'cancel'::move_state) AS rank
                   FROM inventory.stock_moves
                   WHERE picking_id = $1 AND (metadata->>'deleted_at') IS NULL
               )
               UPDATE inventory.transfers t
               SET state = CASE agg.rank
                       WHEN 0 THEN 'draft'::transfer_state
                       WHEN 1 THEN 'waiting'::transfer_state
                       WHEN 2 THEN 'confirmed'::transfer_state
                       WHEN 3 THEN 'assigned'::transfer_state
                       WHEN 4 THEN 'done'::transfer_state
                       ELSE 'cancel'::transfer_state END,
                   date_done = CASE WHEN agg.rank = 4 THEN COALESCE(t.date_done, NOW())
                                    ELSE t.date_done END
               FROM agg
               WHERE t.id = $1
               RETURNING t.state::text"#,
        )
        .bind(picking_id)
        .fetch_optional(&mut *conn)
        .await?;
        // SB-1 cascade: the picking's state just changed, so its batch's stored compute
        // must re-derive in the SAME transaction — the batch never lags its members. The
        // projection SQL lives on the batch repository (its single owner); the batch
        // repository is stateless, so a throwaway instance borrows it here. No-ops when
        // the picking belongs to no batch.
        if row.is_some() {
            PickingBatchProjectionRepository::default()
                .reproject_batch_of_picking(conn, picking_id)
                .await?;
        }
        Ok(row.map(|r| r.get::<String, _>("state")))
    }
}

fn move_row_of(r: sqlx::postgres::PgRow) -> MoveRow {
    MoveRow {
        id: r.get("id"),
        name: r.get("name"),
        state: r.get("state"),
        posting_state: r.get("posting_state"),
        date: r.get("date"),
        item_id: r.get("item_id"),
        demand_qty: r.get("demand_qty"),
        quantity: r.get("quantity"),
        price_unit: r.get("price_unit"),
        procure_method: r.get("procure_method"),
        picking_id: r.get("picking_id"),
        origin: r.get("origin"),
        rule_id: r.get("rule_id"),
        location_id: r.get("location_id"),
        location_dest_id: r.get("location_dest_id"),
        partner_id: r.get("partner_id"),
        warehouse_id: r.get("warehouse_id"),
        orderpoint_id: r.get("orderpoint_id"),
        move_orig_ids: r.get("move_orig_ids"),
        move_dest_ids: r.get("move_dest_ids"),
        is_inventory: r.get("is_inventory"),
        scrapped: r.get("scrapped"),
        propagate_cancel: r.get("propagate_cancel"),
        forced_value: r.get("forced_value"),
    }
}

backbone_core::impl_crud_repository!(StockMoveRepository, StockMove, soft_delete);