material_manage_system/internal/repository/material.go

598 lines
20 KiB
Go
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

package repository
import (
"database/sql"
"errors"
"fmt"
"sort"
"strings"
"github.com/wmms/wmms/internal/model"
"github.com/wmms/wmms/pkg/database"
)
// ErrDuplicateMaterialCode 表示同组织下物料编码已被占用。
var ErrDuplicateMaterialCode = errors.New("物料编码已存在")
// BeginTx starts a new database transaction
func BeginTx() (*sql.Tx, error) {
return database.DB.Begin()
}
// probeLimit 限制自动编码撞号后向后探测的次数:号段被大面积占用时宁可报错,
// 也不要让一次创建长时间空转。
const probeLimit = 1000
type MaterialRepo struct{}
func NewMaterialRepo() *MaterialRepo {
return &MaterialRepo{}
}
// Exists checks ownership without decoding optional legacy material fields.
func (r *MaterialRepo) Exists(organizationID, id int64) (bool, error) {
var exists bool
err := database.DB.QueryRow(`SELECT EXISTS(SELECT 1 FROM materials WHERE organization_id = ? AND id = ?)`, organizationID, id).Scan(&exists)
return exists, err
}
func (r *MaterialRepo) IsActiveSubcategory(categoryID, organizationID int64) (bool, error) {
query := `SELECT EXISTS(
SELECT 1 FROM material_categories
WHERE id = ? AND organization_id = ? AND parent_id IS NOT NULL AND status = 'active'
)`
var exists bool
err := database.DB.QueryRow(query, categoryID, organizationID).Scan(&exists)
return exists, err
}
// IsActiveUnit 判断某个基本单位是否属于当前公司且处于启用状态。物料引用的单位
// 必须落在同一公司,否则 A 公司可以引用 B 公司的单位。
func (r *MaterialRepo) IsActiveUnit(unitID, organizationID int64) (bool, error) {
query := `SELECT EXISTS(SELECT 1 FROM units WHERE id = ? AND organization_id = ? AND status = 'active')`
var exists bool
err := database.DB.QueryRow(query, unitID, organizationID).Scan(&exists)
return exists, err
}
// sqlExecer 让插入语句既能直接走连接池,也能跑在事务里。
type sqlExecer interface {
Exec(query string, args ...interface{}) (sql.Result, error)
}
// insertMaterial 写入物料并回填自增主键,唯一键冲突统一翻译成
// ErrDuplicateMaterialCode。
func insertMaterial(execer sqlExecer, m *model.Material) error {
query := `INSERT INTO materials (organization_id, material_code, name, material_image, spec, manufacturer_model, category_id, base_unit_id, unit, status, version) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`
result, err := execer.Exec(query, m.OrganizationID, m.MaterialCode, m.Name, m.MaterialImage, m.Spec,
m.ManufacturerModel, m.CategoryID, m.BaseUnitID, m.Unit, m.Status, m.Version)
if err != nil {
if strings.Contains(err.Error(), "Duplicate entry") {
return ErrDuplicateMaterialCode
}
return err
}
id, err := result.LastInsertId()
if err != nil {
return err
}
m.ID = id
return nil
}
// lockCodeRule 锁住小类当前启用的编码规则行,返回 nil 表示该小类没有规则。
// FOR UPDATE 让同一个小类的并发创建串行化,避免两条记录抢到同一个流水号。
func lockCodeRule(tx *sql.Tx, organizationID, categoryID int64) (*model.CodeRule, error) {
query := `SELECT id, organization_id, category_id, prefix, sequence_length, next_sequence FROM code_rules WHERE organization_id = ? AND category_id = ? AND status = 'active' FOR UPDATE`
rule := &model.CodeRule{}
err := tx.QueryRow(query, organizationID, categoryID).Scan(&rule.ID, &rule.OrganizationID,
&rule.CategoryID, &rule.Prefix, &rule.SequenceLength, &rule.NextSequence)
if err == sql.ErrNoRows {
return nil, nil
}
if err != nil {
return nil, err
}
return rule, nil
}
// isMaterialCodeTaken 判断同组织下某个物料编码是否已被占用。
func isMaterialCodeTaken(tx *sql.Tx, organizationID int64, code string) (bool, error) {
query := `SELECT EXISTS(SELECT 1 FROM materials WHERE organization_id = ? AND material_code = ?)`
var exists bool
err := tx.QueryRow(query, organizationID, code).Scan(&exists)
return exists, err
}
// CreateWithCodeRule 在事务内插入物料,并根据提交的编码决定是否推进号段。
//
// 提交的编码是不是「规则当前的产物」决定了走哪条分支:相等说明这个编码来自自动
// 生成、用户没有改过,于是沿着号段向后探测出第一个没被占用的流水号,插入成功后
// 把 next_sequence 推进到它之后;不相等说明用户手改了编码、或者该小类根本没有
// 规则,此时原样插入、号段保持不变,撞唯一键就报编码已存在。
func (r *MaterialRepo) CreateWithCodeRule(m *model.Material, submittedCode string) error {
tx, err := BeginTx()
if err != nil {
return err
}
defer func() { _ = tx.Rollback() }()
var categoryID int64
if m.CategoryID != nil {
categoryID = *m.CategoryID
}
rule, err := lockCodeRule(tx, m.OrganizationID, categoryID)
if err != nil {
return err
}
code := submittedCode
var nextSequence int64
if rule != nil && code == rule.FormatCode(rule.NextSequence) {
sequence := rule.NextSequence
for probed := 0; probed < probeLimit; probed++ {
code = rule.FormatCode(sequence)
taken, err := isMaterialCodeTaken(tx, m.OrganizationID, code)
if err != nil {
return err
}
if !taken {
nextSequence = sequence + 1
break
}
sequence++
}
if nextSequence == 0 {
return fmt.Errorf("编码规则 %s 的可用流水号已用尽", rule.Prefix)
}
}
m.MaterialCode = code
if err := insertMaterial(tx, m); err != nil {
return err
}
if nextSequence > 0 {
if _, err := tx.Exec(`UPDATE code_rules SET next_sequence = ? WHERE id = ?`, nextSequence, rule.ID); err != nil {
return err
}
}
if err := tx.Commit(); err != nil {
return err
}
return r.GetByID(m.OrganizationID, m.ID, m)
}
// GetByID 读取本公司物料。organization_id 与 id 同时作为条件,跨公司记录读不到
// 会返回 sql.ErrNoRows,由上层映射成 404。
func (r *MaterialRepo) GetByID(organizationID, id int64, m *model.Material) error {
query := `SELECT id, organization_id, material_code, name, material_image, spec, manufacturer_model,
category_id, base_unit_id, unit, status, version, created_at, updated_at
FROM materials WHERE organization_id = ? AND id = ?`
var materialImage sql.NullString
var manufacturerModel sql.NullString
err := database.DB.QueryRow(query, organizationID, id).Scan(&m.ID, &m.OrganizationID, &m.MaterialCode, &m.Name, &materialImage,
&m.Spec, &manufacturerModel, &m.CategoryID, &m.BaseUnitID,
&m.Unit, &m.Status, &m.Version, &m.CreatedAt, &m.UpdatedAt)
if err != nil {
return err
}
if materialImage.Valid {
m.MaterialImage = &materialImage.String
}
if manufacturerModel.Valid {
m.ManufacturerModel = &manufacturerModel.String
}
return nil
}
func (r *MaterialRepo) GetByMaterialCode(organizationID int64, code string) (*model.Material, error) {
m := &model.Material{}
query := `SELECT id, organization_id, material_code, name, material_image, spec, manufacturer_model,
category_id, base_unit_id, unit, status, version, created_at, updated_at
FROM materials WHERE organization_id = ? AND material_code = ?`
var materialImage sql.NullString
var manufacturerModel sql.NullString
err := database.DB.QueryRow(query, organizationID, code).Scan(&m.ID, &m.OrganizationID, &m.MaterialCode, &m.Name, &materialImage,
&m.Spec, &manufacturerModel, &m.CategoryID, &m.BaseUnitID,
&m.Unit, &m.Status, &m.Version, &m.CreatedAt, &m.UpdatedAt)
if err != nil {
return nil, err
}
if materialImage.Valid {
m.MaterialImage = &materialImage.String
}
if manufacturerModel.Valid {
m.ManufacturerModel = &manufacturerModel.String
}
return m, nil
}
func (r *MaterialRepo) List(organizationID int64, search, status string, limit, offset int) ([]model.Material, int64, error) {
countQuery := `SELECT COUNT(*) FROM materials WHERE organization_id = ?`
countArgs := []interface{}{organizationID}
if status != "all" {
countQuery += ` AND status = ?`
countArgs = append(countArgs, status)
}
if search != "" {
countQuery += ` AND (material_code LIKE ? OR name LIKE ?)`
searchPattern := "%" + search + "%"
countArgs = append(countArgs, searchPattern, searchPattern)
}
var total int64
err := database.DB.QueryRow(countQuery, countArgs...).Scan(&total)
if err != nil {
return nil, 0, err
}
query := `SELECT id, organization_id, material_code, name, material_image, spec, manufacturer_model,
category_id, base_unit_id, unit, status, version, created_at, updated_at
FROM materials WHERE organization_id = ?`
args := []interface{}{organizationID}
if status != "all" {
query += ` AND status = ?`
args = append(args, status)
}
if search != "" {
query += ` AND (material_code LIKE ? OR name LIKE ?)`
searchPattern := "%" + search + "%"
args = append(args, searchPattern, searchPattern)
}
query += ` ORDER BY id DESC 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 materials []model.Material
for rows.Next() {
var m model.Material
var materialImage sql.NullString
var manufacturerModel sql.NullString
err := rows.Scan(&m.ID, &m.OrganizationID, &m.MaterialCode, &m.Name, &materialImage, &m.Spec,
&manufacturerModel, &m.CategoryID, &m.BaseUnitID,
&m.Unit, &m.Status, &m.Version, &m.CreatedAt, &m.UpdatedAt)
if err != nil {
return nil, 0, err
}
if materialImage.Valid {
m.MaterialImage = &materialImage.String
}
if manufacturerModel.Valid {
m.ManufacturerModel = &manufacturerModel.String
}
materials = append(materials, m)
}
return materials, total, nil
}
func (r *MaterialRepo) Update(organizationID, id int64, req *model.UpdateMaterialRequest) error {
var sets []string
var args []interface{}
if req.Name != "" {
sets = append(sets, "name = ?")
args = append(args, req.Name)
}
if req.MaterialImage != nil && *req.MaterialImage != "" {
sets = append(sets, "material_image = ?")
args = append(args, *req.MaterialImage)
}
if req.Spec != "" {
sets = append(sets, "spec = ?")
args = append(args, req.Spec)
}
if req.ManufacturerModel != nil && *req.ManufacturerModel != "" {
sets = append(sets, "manufacturer_model = ?")
args = append(args, *req.ManufacturerModel)
}
if req.CategoryID != nil {
sets = append(sets, "category_id = ?")
args = append(args, req.CategoryID)
}
if req.Unit != "" {
sets = append(sets, "unit = ?")
args = append(args, req.Unit)
}
if req.BaseUnitID != nil {
sets = append(sets, "base_unit_id = ?")
args = append(args, req.BaseUnitID)
}
if req.Status != "" {
sets = append(sets, "status = ?")
args = append(args, req.Status)
}
sets = append(sets, "version = version + 1")
if len(sets) == 0 {
return nil
}
// organization_id 与 id 同时作为条件,A 不能改到 B 的物料。
query := fmt.Sprintf("UPDATE materials SET %s WHERE organization_id = ? AND id = ?", strings.Join(sets, ", "))
args = append(args, organizationID, id)
_, err := database.DB.Exec(query, args...)
return err
}
func (r *MaterialRepo) Delete(organizationID, id int64) error {
// Check if material has related records
var count int
// Check inbound_items
database.DB.QueryRow(`SELECT COUNT(*) FROM inbound_items WHERE material_id = ?`, id).Scan(&count)
if count > 0 {
return fmt.Errorf("该物料有入库记录,无法删除")
}
// Check outbound_items
database.DB.QueryRow(`SELECT COUNT(*) FROM outbound_items WHERE material_id = ?`, id).Scan(&count)
if count > 0 {
return fmt.Errorf("该物料有出库记录,无法删除")
}
// Check stock_logs
database.DB.QueryRow(`SELECT COUNT(*) FROM stock_logs WHERE material_id = ?`, id).Scan(&count)
if count > 0 {
return fmt.Errorf("该物料有库存流水记录,无法删除")
}
_, err := database.DB.Exec(`DELETE FROM materials WHERE organization_id = ? AND id = ?`, organizationID, id)
return err
}
// obsoleteMaterialTx 在给定事务内作废物料:先确认没有非零库存,再带组织条件更新
// 状态并推进版本号。影响 0 行表示物料不存在或不属于该组织。
func obsoleteMaterialTx(tx *sql.Tx, organizationID, id int64) error {
var nonZero int
if err := tx.QueryRow(`SELECT COUNT(*) FROM inventory_balances WHERE material_id = ? AND (on_hand <> 0 OR reserved <> 0 OR locked <> 0 OR in_transit <> 0)`, id).Scan(&nonZero); err != nil {
return err
}
if nonZero > 0 {
return fmt.Errorf("该物料仍有库存,无法作废")
}
result, err := tx.Exec(`UPDATE materials SET status = 'obsolete', version = version + 1 WHERE organization_id = ? AND id = ?`, organizationID, id)
if err != nil {
return err
}
// version always increments, so zero affected rows means the material does
// not exist in this organization; re-obsoleting an already-obsolete material
// still reports 1.
affected, _ := result.RowsAffected()
if affected == 0 {
return fmt.Errorf("物料不存在")
}
return nil
}
// Obsolete 自带事务地作废物料,供直接调用;Service 需要把变更日志写进同一事务时
// 改用 ObsoleteTx。
func (r *MaterialRepo) Obsolete(organizationID, id int64) error {
tx, err := BeginTx()
if err != nil {
return err
}
if err := obsoleteMaterialTx(tx, organizationID, id); err != nil {
tx.Rollback()
return err
}
return tx.Commit()
}
// ObsoleteTx 在调用方事务内作废物料,便于和变更日志一起提交或回滚。
func (r *MaterialRepo) ObsoleteTx(tx *sql.Tx, organizationID, id int64) error {
return obsoleteMaterialTx(tx, organizationID, id)
}
// restoreMaterialTx 在给定事务内恢复物料,影响 0 行表示物料不存在或不属于该组织。
func restoreMaterialTx(tx *sql.Tx, organizationID, id int64) error {
result, err := tx.Exec(`UPDATE materials SET status = 'active', version = version + 1 WHERE organization_id = ? AND id = ?`, organizationID, id)
if err != nil {
return err
}
affected, _ := result.RowsAffected()
if affected == 0 {
return fmt.Errorf("物料不存在")
}
return nil
}
func (r *MaterialRepo) Restore(organizationID, id int64) error {
tx, err := BeginTx()
if err != nil {
return err
}
if err := restoreMaterialTx(tx, organizationID, id); err != nil {
tx.Rollback()
return err
}
return tx.Commit()
}
// RestoreTx 在调用方事务内恢复物料,便于和变更日志一起提交或回滚。
func (r *MaterialRepo) RestoreTx(tx *sql.Tx, organizationID, id int64) error {
return restoreMaterialTx(tx, organizationID, id)
}
// materialImportBatchSize 是批量插入物料时的每批行数。编码已在内存里按小类
// 顺序分配好,这里只负责把 INSERT 折叠成分批的多值语句,避免两万行逐条往返。
const materialImportBatchSize = 500
// ImportTx 在调用方事务内为整批物料分配编码并插入,任意一行失败都由调用方回滚。
// 编码按小类分组生成:组内先对 code_rules 行加锁(按 category_id 升序加锁,
// 避免两个导入事务以不同顺序持锁而死锁),再跳过库中已占用的号段。
func (r *MaterialRepo) ImportTx(tx *sql.Tx, organizationID int64, materials []*model.Material) error {
if len(materials) == 0 {
return nil
}
// 1. 按 category_id 分组,并保证加锁顺序升序。
type group struct {
categoryID int64
items []*model.Material
}
index := make(map[int64]int)
groups := make([]*group, 0)
for _, m := range materials {
var categoryID int64
if m.CategoryID != nil {
categoryID = *m.CategoryID
}
position, ok := index[categoryID]
if !ok {
position = len(groups)
index[categoryID] = position
groups = append(groups, &group{categoryID: categoryID})
}
groups[position].items = append(groups[position].items, m)
}
sort.Slice(groups, func(i, j int) bool { return groups[i].categoryID < groups[j].categoryID })
// 2. 逐组锁规则、分配编码。
for _, g := range groups {
rule, err := lockCodeRule(tx, organizationID, g.categoryID)
if err != nil {
return err
}
if rule == nil {
return fmt.Errorf("小类(ID %d)未配置启用的编码规则", g.categoryID)
}
taken, err := loadTakenMaterialCodes(tx, organizationID, rule.Prefix)
if err != nil {
return err
}
sequence := rule.NextSequence
for _, m := range g.items {
code, next, err := nextAvailableMaterialCode(rule, sequence, taken)
if err != nil {
return err
}
m.MaterialCode = code
taken[code] = struct{}{}
sequence = next
}
if _, err := tx.Exec(`UPDATE code_rules SET next_sequence = ? WHERE id = ?`, sequence, rule.ID); err != nil {
return err
}
}
// 3. 分批多值插入,每批插入后按编码回查自增 ID。
// 多值 INSERT 只返回首个自增 ID 且不保证连续,拿不到逐行 ID,
// 而后面写审计日志需要它;按批回查则把 IN 列表控制在 500 项以内。
for start := 0; start < len(materials); start += materialImportBatchSize {
end := start + materialImportBatchSize
if end > len(materials) {
end = len(materials)
}
batch := materials[start:end]
query, args := buildMaterialInsertBatch(batch)
if _, err := tx.Exec(query, args...); err != nil {
if strings.Contains(err.Error(), "Duplicate entry") {
return ErrDuplicateMaterialCode
}
return err
}
if err := fillMaterialIDs(tx, organizationID, batch); err != nil {
return err
}
}
return nil
}
// loadTakenMaterialCodes 取出本公司该前缀下已存在的全部物料编码,供内存判重。
// 前缀来自公司自己的编码规则配置,若它含 % 或 _ 只会让 LIKE 多匹配到无关编码,
// 而那些编码的格式与本规则生成的编码不同、永远不会被 FormatCode 产出,
// 因此多匹配不会漏判,无需额外转义。
func loadTakenMaterialCodes(tx *sql.Tx, organizationID int64, prefix string) (map[string]struct{}, error) {
rows, err := tx.Query(
`SELECT material_code FROM materials WHERE organization_id = ? AND material_code LIKE ?`,
organizationID, prefix+"%")
if err != nil {
return nil, err
}
defer rows.Close()
taken := make(map[string]struct{})
for rows.Next() {
var code string
if err := rows.Scan(&code); err != nil {
return nil, err
}
taken[code] = struct{}{}
}
return taken, rows.Err()
}
// nextAvailableMaterialCode 从 start 起找到第一个未占用的编码,返回编码与下一个
// 待用序号。探测上限沿用单条新增的 probeLimit,防止规则被用尽时无限循环。
func nextAvailableMaterialCode(rule *model.CodeRule, start int64, taken map[string]struct{}) (string, int64, error) {
for sequence := start; sequence < start+probeLimit; sequence++ {
code := rule.FormatCode(sequence)
if _, occupied := taken[code]; !occupied {
return code, sequence + 1, nil
}
}
return "", 0, fmt.Errorf("编码规则 %s 的可用流水号已用尽", rule.Prefix)
}
// buildMaterialInsertBatch 拼一条多值 INSERT,返回 SQL 与按顺序展开的参数。
func buildMaterialInsertBatch(batch []*model.Material) (string, []interface{}) {
var builder strings.Builder
builder.WriteString(`INSERT INTO materials (organization_id, material_code, name, material_image, spec, manufacturer_model, category_id, base_unit_id, unit, status, version) VALUES `)
args := make([]interface{}, 0, len(batch)*11)
for i, m := range batch {
if i > 0 {
builder.WriteString(", ")
}
builder.WriteString("(?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)")
args = append(args, m.OrganizationID, m.MaterialCode, m.Name, m.MaterialImage, m.Spec,
m.ManufacturerModel, m.CategoryID, m.BaseUnitID, m.Unit, m.Status, m.Version)
}
return builder.String(), args
}
// fillMaterialIDs 按物料编码回查自增 ID 并回填。多值插入拿不到逐行 ID,
// 而后续写 material_logs 需要它。
func fillMaterialIDs(tx *sql.Tx, organizationID int64, materials []*model.Material) error {
codes := make([]interface{}, 0, len(materials))
for _, m := range materials {
codes = append(codes, m.MaterialCode)
}
placeholders := strings.TrimSuffix(strings.Repeat("?,", len(codes)), ",")
query := `SELECT id, material_code FROM materials WHERE organization_id = ? AND material_code IN (` + placeholders + `)`
args := append([]interface{}{organizationID}, codes...)
rows, err := tx.Query(query, args...)
if err != nil {
return err
}
defer rows.Close()
ids := make(map[string]int64, len(materials))
for rows.Next() {
var id int64
var code string
if err := rows.Scan(&id, &code); err != nil {
return err
}
ids[code] = id
}
if err := rows.Err(); err != nil {
return err
}
for _, m := range materials {
id, ok := ids[m.MaterialCode]
if !ok {
return fmt.Errorf("物料 %s 插入后未能回查到 ID", m.MaterialCode)
}
m.ID = id
}
return nil
}