1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
//! 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);