package repository import ( "database/sql" "errors" "github.com/shopspring/decimal" "github.com/wmms/wmms/internal/model" "github.com/wmms/wmms/pkg/database" ) // ErrInsufficientStock 表示在指定库存维度上条件扣减失败:余额行不存在(按 0 处理) // 或余额不足以覆盖本次扣减。它由 DeductStockTx 返回,Service 再翻译成对外的 // ErrInsufficientStock。 var ErrInsufficientStock = errors.New("库存不足") // obsoleteMaterialFilterSQL 把已作废的物料挡在库存视图之外。作废要求现存量已经为 0, // 所以"不再补货"的物料是走作废退役,而不是永远挂在预警列表里。 const obsoleteMaterialFilterSQL = ` AND m.status <> 'obsolete'` // stockedMaterialFilterSQL 在此基础上再挡掉从未入库过的物料。"是否存在 // inventory_balances 行"正是"进过仓库"和"只存在于主数据"的分界线;被领用到 0 的行 // 依然算数,所以已耗尽的物料仍然可见。余额行与物料必须同属一家公司,避免把别家公司 // 的余额当成"本公司进过库"。 const stockedMaterialFilterSQL = obsoleteMaterialFilterSQL + ` AND EXISTS (SELECT 1 FROM inventory_balances ib2 WHERE ib2.material_id = m.id AND ib2.organization_id = m.organization_id)` // stockWarningHavingSQL 必须与 service.ClassifyStock 保持一致:只要预警处于启用状态, // 现存量触达阈值就预警。阈值 0 是合法配置,含义是"库存降到 0 时预警",不再被排除在外。 // // 没有配过规则的物料靠 COALESCE(..., FALSE) 挡掉:无 swr 行时 MAX(swr.enabled) 返回 NULL, // 兜底成 FALSE 即"不参与预警"。这样"从没配过"和"被手动关闭"在判定上统一为不预警, // 前端则靠 warning_configured 把两者区分开来做预填。 // // enabled 的判断仍放在 HAVING 而不是 join 条件里:被禁用的规则必须参与分组,否则它会 // 塌缩成"没配过规则",warning_configured 就报错了。 const stockWarningHavingSQL = `HAVING COALESCE(MAX(swr.enabled), FALSE) = TRUE AND COALESCE(SUM(ib.on_hand), 0) <= COALESCE(MAX(swr.minimum_available), 0)` // upsertStockWarningRuleSQL 把预警阈值写到显式的公司、仓库维度上。维度由 InventoryScope // 提供,不再从 DEFAULT/MAIN 兜底;缺少这些维度会由外键约束挡下而不是静默写入默认公司。 const upsertStockWarningRuleSQL = `INSERT INTO stock_warning_rules (organization_id, warehouse_id, material_id, minimum_available, enabled) VALUES (?, ?, ?, ?, ?) ON DUPLICATE KEY UPDATE minimum_available = VALUES(minimum_available), enabled = VALUES(enabled)` type StockRepo struct{} func NewStockRepo() *StockRepo { return &StockRepo{} } // GetAllStock 的 category 收的是大类编码(material_categories 里 parent_id 为空的那层)。 // 按大类 code 而不是 category_id 过滤,是为了和前端下拉直接选出的大类选项对上。 // 列表、分页总数都按 organizationID 过滤,余额与预警规则也限定在同一公司。 func (r *StockRepo) GetAllStock(organizationID int64, search, category string, limit, offset int) ([]model.Stock, int64, error) { // COUNT 和主查询是两条独立 SQL,筛选条件必须成对出现,否则分页总数会和实际行数 // 对不上、末尾多出空白页。主查询本来就 JOIN 了分类表,这里为按大类过滤补上同样的 JOIN; // 两张表都按主键关联,不会放大行数,COUNT 的语义不变。 countQuery := `SELECT COUNT(*) FROM materials m LEFT JOIN material_categories c ON c.id = m.category_id LEFT JOIN material_categories pc ON pc.id = c.parent_id WHERE m.organization_id = ?` + stockedMaterialFilterSQL countArgs := []interface{}{organizationID} if search != "" { countQuery += ` AND (m.material_code LIKE ? OR m.name LIKE ?)` searchPattern := "%" + search + "%" countArgs = append(countArgs, searchPattern, searchPattern) } if category != "" { countQuery += ` AND pc.code = ?` countArgs = append(countArgs, category) } var total int64 err := database.DB.QueryRow(countQuery, countArgs...).Scan(&total) if err != nil { return nil, 0, err } query := `SELECT m.id, m.material_code, m.name, m.spec, m.unit, COALESCE(SUM(ib.on_hand), 0), COALESCE(MAX(swr.minimum_available), 0), COALESCE(MAX(swr.enabled), FALSE), MAX(swr.material_id) IS NOT NULL, IFNULL(pc.code, ''), IFNULL(c.code, '') FROM materials m LEFT JOIN inventory_balances ib ON ib.material_id = m.id AND ib.organization_id = m.organization_id LEFT JOIN stock_warning_rules swr ON swr.organization_id = ib.organization_id AND swr.warehouse_id = ib.warehouse_id AND swr.material_id = m.id LEFT JOIN material_categories c ON c.id = m.category_id LEFT JOIN material_categories pc ON pc.id = c.parent_id WHERE m.organization_id = ?` + stockedMaterialFilterSQL args := []interface{}{organizationID} if search != "" { query += ` AND (m.material_code LIKE ? OR m.name LIKE ?)` searchPattern := "%" + search + "%" args = append(args, searchPattern, searchPattern) } if category != "" { query += ` AND pc.code = ?` args = append(args, category) } query += ` GROUP BY m.id, m.material_code, m.name, m.spec, m.unit, pc.code, c.code ORDER BY m.id LIMIT ? OFFSET ?` args = append(args, limit, offset) rows, err := database.DB.Query(query, args...) if err != nil { return nil, 0, err } defer rows.Close() var stocks []model.Stock for rows.Next() { var s model.Stock err := rows.Scan(&s.MaterialID, &s.MaterialCode, &s.Name, &s.Spec, &s.Unit, &s.TotalOnHand, &s.WarningStock, &s.WarningEnabled, &s.WarningConfigured, &s.MajorCategory, &s.MinorCategory) if err != nil { return nil, 0, err } stocks = append(stocks, s) } return stocks, total, nil } // GetByMaterialID 返回单个物料的库存。这里刻意不套 stockedMaterialFilterSQL:物料申请和 // 出库表单是按物料逐个查询的,无论有没有入过库都必须给出答案。 // // 组织条件是硬约束:物料不属于当前公司时命中 sql.ErrNoRows,接口回 404,公司 A 不能 // 凭物料 ID 读到公司 B 的库存。余额也只汇总本公司维度。 func (r *StockRepo) GetByMaterialID(organizationID, materialID int64) (*model.Stock, error) { s := &model.Stock{} query := `SELECT m.id, m.material_code, m.name, m.spec, m.unit, COALESCE(SUM(ib.on_hand), 0), COALESCE(MAX(swr.minimum_available), 0), COALESCE(MAX(swr.enabled), FALSE), MAX(swr.material_id) IS NOT NULL, IFNULL(pc.code, ''), IFNULL(c.code, '') FROM materials m LEFT JOIN inventory_balances ib ON ib.material_id = m.id AND ib.organization_id = m.organization_id LEFT JOIN stock_warning_rules swr ON swr.organization_id = ib.organization_id AND swr.warehouse_id = ib.warehouse_id AND swr.material_id = m.id LEFT JOIN material_categories c ON c.id = m.category_id LEFT JOIN material_categories pc ON pc.id = c.parent_id WHERE m.organization_id = ? AND m.id = ? GROUP BY m.id, m.material_code, m.name, m.spec, m.unit, pc.code, c.code` err := database.DB.QueryRow(query, organizationID, materialID).Scan(&s.MaterialID, &s.MaterialCode, &s.Name, &s.Spec, &s.Unit, &s.TotalOnHand, &s.WarningStock, &s.WarningEnabled, &s.WarningConfigured, &s.MajorCategory, &s.MinorCategory) if err != nil { return nil, err } return s, nil } func (r *StockRepo) GetWarningStock(organizationID int64, limit, offset int) ([]model.Stock, int64, error) { countQuery := `SELECT COUNT(*) FROM ( SELECT m.id FROM materials m LEFT JOIN inventory_balances ib ON ib.material_id = m.id AND ib.organization_id = m.organization_id LEFT JOIN stock_warning_rules swr ON swr.organization_id = ib.organization_id AND swr.warehouse_id = ib.warehouse_id AND swr.material_id = m.id WHERE m.organization_id = ?` + stockedMaterialFilterSQL + ` GROUP BY m.id ` + stockWarningHavingSQL + ` ) warning_materials` var total int64 err := database.DB.QueryRow(countQuery, organizationID).Scan(&total) if err != nil { return nil, 0, err } query := `SELECT m.id, m.material_code, m.name, m.spec, m.unit, COALESCE(SUM(ib.on_hand), 0), COALESCE(MAX(swr.minimum_available), 0), COALESCE(MAX(swr.enabled), FALSE), MAX(swr.material_id) IS NOT NULL, IFNULL(pc.code, ''), IFNULL(c.code, '') FROM materials m LEFT JOIN inventory_balances ib ON ib.material_id = m.id AND ib.organization_id = m.organization_id LEFT JOIN stock_warning_rules swr ON swr.organization_id = ib.organization_id AND swr.warehouse_id = ib.warehouse_id AND swr.material_id = m.id LEFT JOIN material_categories c ON c.id = m.category_id LEFT JOIN material_categories pc ON pc.id = c.parent_id WHERE m.organization_id = ?` + stockedMaterialFilterSQL + ` GROUP BY m.id, m.material_code, m.name, m.spec, m.unit, pc.code, c.code ` + stockWarningHavingSQL + ` ORDER BY m.id LIMIT ? OFFSET ?` rows, err := database.DB.Query(query, organizationID, limit, offset) if err != nil { return nil, 0, err } defer rows.Close() var stocks []model.Stock for rows.Next() { var s model.Stock err := rows.Scan(&s.MaterialID, &s.MaterialCode, &s.Name, &s.Spec, &s.Unit, &s.TotalOnHand, &s.WarningStock, &s.WarningEnabled, &s.WarningConfigured, &s.MajorCategory, &s.MinorCategory) if err != nil { return nil, 0, err } stocks = append(stocks, s) } return stocks, total, nil } // stockLogDimensionArgs 把流水所属的公司、仓库、库位维度按 INSERT 顺序展开, // 三个维度都是显式落库,不由数据库默认值或历史编码兜底。 func stockLogDimensionArgs(log *model.StockLog) []interface{} { return []interface{}{log.OrganizationID, log.WarehouseID, log.LocationID, log.MaterialID, log.Type, log.Quantity, log.RelatedOrderType, log.RelatedOrderID} } const stockLogInsertSQL = `INSERT INTO stock_logs (organization_id, warehouse_id, location_id, material_id, type, quantity, related_order_type, related_order_id) VALUES (?, ?, ?, ?, ?, ?, ?, ?)` func (r *StockRepo) CreateLog(log *model.StockLog) error { result, err := database.DB.Exec(stockLogInsertSQL, stockLogDimensionArgs(log)...) if err != nil { return err } id, err := result.LastInsertId() if err != nil { return err } log.ID = id return nil } func (r *StockRepo) GetLogsByMaterialID(organizationID, materialID int64, limit, offset int) ([]model.StockLog, int64, error) { countQuery := `SELECT COUNT(*) FROM stock_logs WHERE organization_id = ? AND material_id = ?` var total int64 err := database.DB.QueryRow(countQuery, organizationID, materialID).Scan(&total) if err != nil { return nil, 0, err } query := `SELECT id, material_id, type, quantity, related_order_type, related_order_id, created_at FROM stock_logs WHERE organization_id = ? AND material_id = ? ORDER BY id LIMIT ? OFFSET ?` rows, err := database.DB.Query(query, organizationID, materialID, limit, offset) if err != nil { return nil, 0, err } defer rows.Close() var logs []model.StockLog for rows.Next() { var log model.StockLog err := rows.Scan(&log.ID, &log.MaterialID, &log.Type, &log.Quantity, &log.RelatedOrderType, &log.RelatedOrderID, &log.CreatedAt) if err != nil { return nil, 0, err } logs = append(logs, log) } return logs, total, nil } func (r *StockRepo) GetAllLogs(organizationID int64, limit, offset int) ([]model.StockLog, int64, error) { countQuery := `SELECT COUNT(*) FROM stock_logs WHERE organization_id = ?` var total int64 err := database.DB.QueryRow(countQuery, organizationID).Scan(&total) if err != nil { return nil, 0, err } // Compute running_balance as cumulative sum of (in=+quantity, out=-quantity) ordered by id ASC, // then reverse to show newest first while preserving correct cumulative balance for each row. // 结存只在同一公司内累加,不能先全局算完再过滤公司。 query := ` SELECT id, material_id, type, quantity, related_order_type, related_order_id, created_at FROM ( SELECT *, (SELECT COALESCE(SUM(CASE WHEN type='in' THEN quantity ELSE -quantity END), 0) FROM stock_logs s2 WHERE s2.id <= stock_logs.id AND s2.organization_id = stock_logs.organization_id) as running_balance FROM stock_logs WHERE organization_id = ? ) as t ORDER BY id DESC LIMIT ? OFFSET ?` rows, err := database.DB.Query(query, organizationID, limit, offset) if err != nil { return nil, 0, err } defer rows.Close() var logs []model.StockLog for rows.Next() { var log model.StockLog err := rows.Scan(&log.ID, &log.MaterialID, &log.Type, &log.Quantity, &log.RelatedOrderType, &log.RelatedOrderID, &log.CreatedAt) if err != nil { return nil, 0, err } logs = append(logs, log) } return logs, total, nil } // GetAllLogsWithBalance returns logs ordered newest-first with correct running balance. // Balance = stock level BEFORE this log's change was applied. // Strategy: Walk rows from oldest to newest within the page, accumulating balance. // The starting balance for the oldest row is: total on-hand - sum(all deltas for this material newer than the page). // 全部计算都在当前公司的流水与余额范围内完成,不跨公司汇总。 func (r *StockRepo) GetAllLogsWithBalance(organizationID int64, limit, offset int) ([]model.StockLog, int64, error) { var total int64 if err := database.DB.QueryRow(`SELECT COUNT(*) FROM stock_logs WHERE organization_id = ?`, organizationID).Scan(&total); err != nil { return nil, 0, err } // Fetch page rows ordered newest-first (id DESC). rows, err := database.DB.Query(` SELECT id, material_id, type, quantity, related_order_type, related_order_id, created_at FROM stock_logs WHERE organization_id = ? ORDER BY id DESC LIMIT ? OFFSET ?`, organizationID, limit, offset) if err != nil { return nil, 0, err } defer rows.Close() type rowData struct { log model.StockLog delta int } var pageRows []rowData materialIDs := make(map[int64]bool) for rows.Next() { var r rowData if err := rows.Scan(&r.log.ID, &r.log.MaterialID, &r.log.Type, &r.log.Quantity, &r.log.RelatedOrderType, &r.log.RelatedOrderID, &r.log.CreatedAt); err != nil { return nil, 0, err } if r.log.Type == "in" { r.delta = r.log.Quantity } else { r.delta = -r.log.Quantity } pageRows = append(pageRows, r) materialIDs[r.log.MaterialID] = true } if len(pageRows) == 0 { return []model.StockLog{}, total, nil } // Get initialBalance: total on-hand of this material minus all deltas for this material. // This gives the balance BEFORE the oldest log in this page was applied. initialBalances := make(map[int64]decimal.Decimal) for mid := range materialIDs { var delta int database.DB.QueryRow(` SELECT COALESCE(SUM(CASE WHEN type='in' THEN quantity ELSE -quantity END), 0) FROM stock_logs WHERE organization_id = ? AND material_id = ?`, organizationID, mid).Scan(&delta) var totalOnHand decimal.Decimal if err := database.DB.QueryRow(`SELECT COALESCE(SUM(on_hand), 0) FROM inventory_balances WHERE organization_id = ? AND material_id = ?`, organizationID, mid).Scan(&totalOnHand); err != nil { return nil, 0, err } initialBalances[mid] = totalOnHand.Sub(decimal.NewFromInt(int64(delta))) } // Process from OLDEST to NEWEST within the page (reverse order of our DESC result). // balance_for_oldest = total_on_hand - newer_than_page_delta // balance_after_row = balance_before_row + delta for i := len(pageRows) - 1; i >= 0; i-- { r := &pageRows[i] bal := initialBalances[r.log.MaterialID] if r.log.Type == "in" { bal = bal.Add(decimal.NewFromInt(int64(r.log.Quantity))) } else { bal = bal.Sub(decimal.NewFromInt(int64(r.log.Quantity))) } r.log.RunningBalance = bal initialBalances[r.log.MaterialID] = bal } var logs []model.StockLog for _, r := range pageRows { logs = append(logs, r.log) } return logs, total, nil } // GetStock returns the current on-hand balance for a material within one inventory scope. // 余额行不存在时按 0 处理(SUM 无行返回 0)。 func (r *StockRepo) GetStock(scope model.InventoryScope, materialID int64) (decimal.Decimal, error) { var stock decimal.Decimal err := database.DB.QueryRow(`SELECT COALESCE(SUM(on_hand), 0) FROM inventory_balances WHERE organization_id = ? AND warehouse_id = ? AND location_id = ? AND material_id = ?`, scope.OrganizationID, scope.WarehouseID, scope.LocationID, materialID).Scan(&stock) return stock, err } // GetStockForUpdateTx locks the balance row of one material within the given scope. // 余额行不存在时按 0 处理——没有行意味着可用量为 0,而不是"从别的公司借"。 func (r *StockRepo) GetStockForUpdateTx(tx *sql.Tx, scope model.InventoryScope, materialID int64) (decimal.Decimal, error) { query := `SELECT on_hand FROM inventory_balances WHERE organization_id = ? AND warehouse_id = ? AND location_id = ? AND material_id = ? FOR UPDATE` var stock decimal.Decimal err := tx.QueryRow(query, scope.OrganizationID, scope.WarehouseID, scope.LocationID, materialID).Scan(&stock) if err == sql.ErrNoRows { return decimal.Zero, nil } return stock, err } // UpdateStockTx 按显式库存维度增量写入余额(delta 可为负)。余额行不存在时用现有 // 维度唯一键创建,不再从 DEFAULT/MAIN 兜底;维度对应的仓库、库位或物料不存在时由 // 外键约束挡下,而不是静默写进默认公司。 func (r *StockRepo) UpdateStockTx(tx *sql.Tx, scope model.InventoryScope, materialID int64, delta decimal.Decimal) error { query := `INSERT INTO inventory_balances (organization_id, warehouse_id, location_id, material_id, on_hand) VALUES (?, ?, ?, ?, ?) ON DUPLICATE KEY UPDATE on_hand = on_hand + VALUES(on_hand), version = version + 1` _, err := tx.Exec(query, scope.OrganizationID, scope.WarehouseID, scope.LocationID, materialID, delta) return err } // DeductStockTx 在显式库存维度上条件扣减:只有当余额行存在且 on_hand >= quantity 时 // 才更新。影响 0 行表示余额行不存在(按 0 处理)或余额不足,一律返回 ErrInsufficientStock, // 因此不会透支,也不会锁到别家公司的余额。UPDATE ... WHERE 同时承担行锁,保证并发扣减 // 不会读到过期余额。 func (r *StockRepo) DeductStockTx(tx *sql.Tx, scope model.InventoryScope, materialID int64, quantity decimal.Decimal) error { query := `UPDATE inventory_balances SET on_hand = on_hand - ?, version = version + 1 WHERE organization_id = ? AND warehouse_id = ? AND location_id = ? AND material_id = ? AND on_hand >= ?` result, err := tx.Exec(query, quantity, scope.OrganizationID, scope.WarehouseID, scope.LocationID, materialID, quantity) if err != nil { return err } affected, err := result.RowsAffected() if err != nil { return err } if affected == 0 { return ErrInsufficientStock } return nil } // execQuerier 同时被 *sql.DB 和 *sql.Tx 满足,让预警阈值的写入逻辑可以在入库事务里复用。 type execQuerier interface { Exec(query string, args ...interface{}) (sql.Result, error) QueryRow(query string, args ...interface{}) *sql.Row } func upsertStockWarningRule(q execQuerier, scope model.InventoryScope, materialID int64, minimumAvailable decimal.Decimal, enabled bool) error { // 维度显式来自 scope:组织的仓库、库位缺失时由外键约束报错,不再回落到 DEFAULT/MAIN。 // 传入的值与已存行完全一致时 MySQL 返回 0 影响行,这是"没变"而不是失败,因此不再 // 借此判定配置缺失——那种情况由 ResolveInventoryScope 在调用前拦下。 _, err := q.Exec(upsertStockWarningRuleSQL, scope.OrganizationID, scope.WarehouseID, materialID, minimumAvailable, enabled) return err } // UpdateWarningStock 把指定库存维度的预警阈值写入库存域。 func (r *StockRepo) UpdateWarningStock(scope model.InventoryScope, materialID int64, minimumAvailable decimal.Decimal, enabled bool) error { return upsertStockWarningRule(database.DB, scope, materialID, minimumAvailable, enabled) } // UpdateWarningStockTx 在入库事务内写入预警阈值,保证物料的第一个阈值和它的第一行库存 // 同时落库。 func (r *StockRepo) UpdateWarningStockTx(tx *sql.Tx, scope model.InventoryScope, materialID int64, minimumAvailable decimal.Decimal, enabled bool) error { return upsertStockWarningRule(tx, scope, materialID, minimumAvailable, enabled) } // DeleteLogs 删除指定公司的库存流水:组织、关联类型、单据 ID 三个条件同时约束, // 否则凭别家公司的单据 ID 就能删掉它的流水。 func (r *StockRepo) DeleteLogs(organizationID int64, orderType string, orderID int64) error { query := `DELETE FROM stock_logs WHERE organization_id = ? AND related_order_type = ? AND related_order_id = ?` _, err := database.DB.Exec(query, organizationID, orderType, orderID) return err } // CreateLogTx creates a stock log within a transaction func (r *StockRepo) CreateLogTx(tx *sql.Tx, log *model.StockLog) error { result, err := tx.Exec(stockLogInsertSQL, stockLogDimensionArgs(log)...) if err != nil { return err } id, err := result.LastInsertId() if err != nil { return err } log.ID = id return nil } // DeleteLogsTx 在事务内删除指定公司的库存流水,同样同时约束组织、关联类型和单据 ID。 func (r *StockRepo) DeleteLogsTx(tx *sql.Tx, organizationID int64, orderType string, orderID int64) error { query := `DELETE FROM stock_logs WHERE organization_id = ? AND related_order_type = ? AND related_order_id = ?` _, err := tx.Exec(query, organizationID, orderType, orderID) return err } // GetInventoryBalanceDetails 只返回当前公司的库存维度明细。调用方已经先确认物料属于 // 当前公司(否则回 404),这里再用组织条件收口,避免凭物料 ID 看到别家公司的仓库、库位。 func (r *StockRepo) GetInventoryBalanceDetails(organizationID, materialID int64) ([]model.InventoryBalanceDetail, error) { query := ` SELECT ib.id, ib.organization_id, o.name, ib.warehouse_id, w.name, ib.location_id, l.name, ib.material_id, ib.on_hand, ib.reserved, ib.locked, ib.in_transit, ib.updated_at FROM inventory_balances ib JOIN organizations o ON o.id = ib.organization_id JOIN warehouses w ON w.id = ib.warehouse_id JOIN warehouse_locations l ON l.id = ib.location_id WHERE ib.organization_id = ? AND ib.material_id = ? ORDER BY ib.id` rows, err := database.DB.Query(query, organizationID, materialID) if err != nil { return nil, err } defer rows.Close() var details []model.InventoryBalanceDetail for rows.Next() { var d model.InventoryBalanceDetail if err := rows.Scan(&d.ID, &d.OrganizationID, &d.Organization, &d.WarehouseID, &d.Warehouse, &d.LocationID, &d.Location, &d.MaterialID, &d.OnHand, &d.Reserved, &d.Locked, &d.InTransit, &d.UpdatedAt); err != nil { return nil, err } details = append(details, d) } return details, rows.Err() }