package repository import ( "database/sql" "fmt" "github.com/shopspring/decimal" "github.com/wmms/wmms/internal/model" "github.com/wmms/wmms/pkg/database" ) // 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)` // 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)` const upsertStockWarningRuleSQL = `INSERT INTO stock_warning_rules (organization_id, warehouse_id, material_id, minimum_available, enabled) SELECT o.id, w.id, ?, ?, ? FROM organizations o JOIN warehouses w ON w.organization_id = o.id AND w.code = 'DEFAULT' WHERE o.code = 'DEFAULT' 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 过滤,是为了和前端下拉直接选出的大类选项对上。 func (r *StockRepo) GetAllStock(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 1=1` + stockedMaterialFilterSQL var countArgs []interface{} 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 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 1=1` + stockedMaterialFilterSQL var args []interface{} 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:物料申请和 // 出库表单是按物料逐个查询的,无论有没有入过库都必须给出答案。 func (r *StockRepo) GetByMaterialID(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 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.id = ? GROUP BY m.id, m.material_code, m.name, m.spec, m.unit, pc.code, c.code` err := database.DB.QueryRow(query, 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(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 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 1=1` + stockedMaterialFilterSQL + ` GROUP BY m.id ` + stockWarningHavingSQL + ` ) warning_materials` var total int64 err := database.DB.QueryRow(countQuery).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 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 1=1` + 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, 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 } func (r *StockRepo) CreateLog(log *model.StockLog) error { query := `INSERT INTO stock_logs (material_id, type, quantity, related_order_type, related_order_id) VALUES (?, ?, ?, ?, ?)` result, err := database.DB.Exec(query, log.MaterialID, log.Type, log.Quantity, log.RelatedOrderType, log.RelatedOrderID) if err != nil { return err } id, err := result.LastInsertId() if err != nil { return err } log.ID = id return nil } func (r *StockRepo) GetLogsByMaterialID(materialID int64, limit, offset int) ([]model.StockLog, int64, error) { countQuery := `SELECT COUNT(*) FROM stock_logs WHERE material_id = ?` var total int64 err := database.DB.QueryRow(countQuery, 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 material_id = ? ORDER BY id LIMIT ? OFFSET ?` rows, err := database.DB.Query(query, 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(limit, offset int) ([]model.StockLog, int64, error) { countQuery := `SELECT COUNT(*) FROM stock_logs` var total int64 err := database.DB.QueryRow(countQuery).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) as running_balance FROM stock_logs ) as t ORDER BY id DESC LIMIT ? OFFSET ?` rows, err := database.DB.Query(query, 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(limit, offset int) ([]model.StockLog, int64, error) { var total int64 if err := database.DB.QueryRow(`SELECT COUNT(*) FROM stock_logs`).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 ORDER BY id DESC LIMIT ? OFFSET ?`, 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 material_id = ?`, mid).Scan(&delta) var totalOnHand decimal.Decimal if err := database.DB.QueryRow(`SELECT COALESCE(SUM(on_hand), 0) FROM inventory_balances WHERE material_id = ?`, 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 across inventory dimensions. func (r *StockRepo) GetStock(materialID int64) (decimal.Decimal, error) { var stock decimal.Decimal err := database.DB.QueryRow(`SELECT COALESCE(SUM(on_hand), 0) FROM inventory_balances WHERE material_id = ?`, materialID).Scan(&stock) return stock, err } // GetStockForUpdateTx locks the legacy default location balance used by current inbound/outbound flows. func (r *StockRepo) GetStockForUpdateTx(tx *sql.Tx, materialID int64) (decimal.Decimal, error) { query := `SELECT ib.on_hand FROM inventory_balances ib JOIN organizations o ON o.id = ib.organization_id AND o.code = 'DEFAULT' JOIN warehouses w ON w.id = ib.warehouse_id AND w.code = 'DEFAULT' JOIN warehouse_locations l ON l.id = ib.location_id AND l.code = 'MAIN' WHERE ib.material_id = ? FOR UPDATE` var stock decimal.Decimal err := tx.QueryRow(query, materialID).Scan(&stock) if err == sql.ErrNoRows { return decimal.Zero, nil } return stock, err } // UpdateStockTx writes the legacy inbound/outbound change to the independent default inventory balance. func (r *StockRepo) UpdateStockTx(tx *sql.Tx, materialID int64, delta decimal.Decimal) error { query := `INSERT INTO inventory_balances (organization_id, warehouse_id, location_id, material_id, on_hand) SELECT o.id, w.id, l.id, ?, ? FROM organizations o JOIN warehouses w ON w.organization_id = o.id AND w.code = 'DEFAULT' JOIN warehouse_locations l ON l.warehouse_id = w.id AND l.code = 'MAIN' WHERE o.code = 'DEFAULT' ON DUPLICATE KEY UPDATE on_hand = on_hand + VALUES(on_hand), version = version + 1` result, err := tx.Exec(query, materialID, delta) if err != nil { return err } affected, err := result.RowsAffected() if err != nil { return err } if affected == 0 { return fmt.Errorf("default organization, warehouse, or location is missing") } return nil } const defaultWarehouseExistsSQL = `SELECT COUNT(*) FROM organizations o JOIN warehouses w ON w.organization_id = o.id AND w.code = 'DEFAULT' WHERE o.code = 'DEFAULT'` // 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, materialID int64, minimumAvailable decimal.Decimal, enabled bool) error { result, err := q.Exec(upsertStockWarningRuleSQL, materialID, minimumAvailable, enabled) if err != nil { return err } affected, err := result.RowsAffected() if err != nil { return err } if affected == 0 { // MySQL 在两种情况下都返回 0 影响行:语句什么都没插入(找不到默认组织/仓库), // 以及传入的值和已存的行完全一致。所以要先区分清楚再决定是否报错。 var exists int if err := q.QueryRow(defaultWarehouseExistsSQL).Scan(&exists); err != nil { return err } if exists == 0 { return fmt.Errorf("default organization or warehouse is missing") } } return nil } // UpdateWarningStock 把默认仓库的预警阈值写入库存域。 func (r *StockRepo) UpdateWarningStock(materialID int64, minimumAvailable decimal.Decimal, enabled bool) error { return upsertStockWarningRule(database.DB, materialID, minimumAvailable, enabled) } // UpdateWarningStockTx 在入库事务内写入预警阈值,保证物料的第一个阈值和它的第一行库存 // 同时落库。 func (r *StockRepo) UpdateWarningStockTx(tx *sql.Tx, materialID int64, minimumAvailable decimal.Decimal, enabled bool) error { return upsertStockWarningRule(tx, materialID, minimumAvailable, enabled) } func (r *StockRepo) DeleteLogs(orderType string, orderID int64) error { query := `DELETE FROM stock_logs WHERE related_order_type = ? AND related_order_id = ?` _, err := database.DB.Exec(query, orderType, orderID) return err } // CreateLogTx creates a stock log within a transaction func (r *StockRepo) CreateLogTx(tx *sql.Tx, log *model.StockLog) error { query := `INSERT INTO stock_logs (material_id, type, quantity, related_order_type, related_order_id) VALUES (?, ?, ?, ?, ?)` result, err := tx.Exec(query, log.MaterialID, log.Type, log.Quantity, log.RelatedOrderType, log.RelatedOrderID) if err != nil { return err } id, err := result.LastInsertId() if err != nil { return err } log.ID = id return nil } // DeleteLogsTx deletes stock logs within a transaction func (r *StockRepo) DeleteLogsTx(tx *sql.Tx, orderType string, orderID int64) error { query := `DELETE FROM stock_logs WHERE related_order_type = ? AND related_order_id = ?` _, err := tx.Exec(query, orderType, orderID) return err } func (r *StockRepo) GetInventoryBalanceDetails(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.material_id = ? ORDER BY ib.id` rows, err := database.DB.Query(query, 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() }