feat: 物料批量入库的编码分配与事务插入
This commit is contained in:
文睿 张
parent
4960a88497
commit
f3146af838
|
|
@ -4,6 +4,7 @@ import (
|
|||
"database/sql"
|
||||
"errors"
|
||||
"fmt"
|
||||
"sort"
|
||||
"strings"
|
||||
|
||||
"github.com/wmms/wmms/internal/model"
|
||||
|
|
@ -417,3 +418,180 @@ func (r *MaterialRepo) Restore(organizationID, id int64) error {
|
|||
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
|
||||
}
|
||||
|
|
|
|||
177
internal/repository/material_import_test.go
Normal file
177
internal/repository/material_import_test.go
Normal file
|
|
@ -0,0 +1,177 @@
|
|||
package repository
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"regexp"
|
||||
"strings"
|
||||
"testing"
|
||||
|
||||
"github.com/DATA-DOG/go-sqlmock"
|
||||
"github.com/wmms/wmms/internal/model"
|
||||
"github.com/wmms/wmms/pkg/database"
|
||||
)
|
||||
|
||||
const importLockRuleQuery = `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`
|
||||
|
||||
const insertMaterialBatchStmt = `INSERT INTO materials (organization_id, material_code, name, material_image, spec, manufacturer_model, category_id, base_unit_id, unit, status, version) VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?), (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)`
|
||||
|
||||
func newImportMock(t *testing.T) (sqlmock.Sqlmock, func()) {
|
||||
t.Helper()
|
||||
db, mock, err := sqlmock.New()
|
||||
if err != nil {
|
||||
t.Fatalf("open sql mock: %v", err)
|
||||
}
|
||||
previous := database.DB
|
||||
database.DB = db
|
||||
return mock, func() {
|
||||
database.DB = previous
|
||||
db.Close()
|
||||
}
|
||||
}
|
||||
|
||||
func materialCategoryPtr(id int64) *int64 { return &id }
|
||||
|
||||
func TestMaterialRepoImportTxAllocatesCodesAndInsertsOneByOne(t *testing.T) {
|
||||
mock, cleanup := newImportMock(t)
|
||||
defer cleanup()
|
||||
|
||||
mock.ExpectBegin()
|
||||
mock.ExpectQuery(regexp.QuoteMeta(importLockRuleQuery)).WithArgs(int64(1), int64(12)).
|
||||
WillReturnRows(sqlmock.NewRows([]string{"id", "organization_id", "category_id", "prefix", "sequence_length", "next_sequence"}).
|
||||
AddRow(5, 1, 12, "A01", 4, 201))
|
||||
// 已占用编码探测：A010201 已被手工录入占用，应跳到 A010202
|
||||
mock.ExpectQuery(regexp.QuoteMeta(`SELECT material_code FROM materials WHERE organization_id = ? AND material_code LIKE ?`)).
|
||||
WithArgs(int64(1), "A01%").
|
||||
WillReturnRows(sqlmock.NewRows([]string{"material_code"}).AddRow("A010201"))
|
||||
mock.ExpectExec(regexp.QuoteMeta(`UPDATE code_rules SET next_sequence = ? WHERE id = ?`)).
|
||||
WithArgs(int64(204), int64(5)).WillReturnResult(sqlmock.NewResult(0, 1))
|
||||
// 两行折叠成一条多值 INSERT
|
||||
mock.ExpectExec(regexp.QuoteMeta(insertMaterialBatchStmt)).
|
||||
WithArgs(
|
||||
int64(1), "A010202", "螺栓", nil, "", nil, int64(12), nil, "个", "active", 1,
|
||||
int64(1), "A010203", "螺母", nil, "", nil, int64(12), nil, "个", "active", 1,
|
||||
).
|
||||
WillReturnResult(sqlmock.NewResult(31, 2))
|
||||
mock.ExpectQuery(regexp.QuoteMeta(`SELECT id, material_code FROM materials WHERE organization_id = ? AND material_code IN (?,?)`)).
|
||||
WithArgs(int64(1), "A010202", "A010203").
|
||||
WillReturnRows(sqlmock.NewRows([]string{"id", "material_code"}).
|
||||
AddRow(31, "A010202").AddRow(32, "A010203"))
|
||||
|
||||
tx, err := database.DB.Begin()
|
||||
if err != nil {
|
||||
t.Fatalf("begin: %v", err)
|
||||
}
|
||||
categoryID := int64(12)
|
||||
materials := []*model.Material{
|
||||
{OrganizationID: 1, Name: "螺栓", CategoryID: &categoryID, Unit: "个", Status: "active", Version: 1},
|
||||
{OrganizationID: 1, Name: "螺母", CategoryID: &categoryID, Unit: "个", Status: "active", Version: 1},
|
||||
}
|
||||
if err := NewMaterialRepo().ImportTx(tx, 1, materials); err != nil {
|
||||
t.Fatalf("import: %v", err)
|
||||
}
|
||||
if materials[0].MaterialCode != "A010202" || materials[0].ID != 31 {
|
||||
t.Fatalf("first = %+v", materials[0])
|
||||
}
|
||||
if materials[1].MaterialCode != "A010203" || materials[1].ID != 32 {
|
||||
t.Fatalf("second = %+v", materials[1])
|
||||
}
|
||||
if err := mock.ExpectationsWereMet(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestMaterialRepoImportTxReportsMissingCodeRule(t *testing.T) {
|
||||
mock, cleanup := newImportMock(t)
|
||||
defer cleanup()
|
||||
|
||||
mock.ExpectBegin()
|
||||
mock.ExpectQuery(regexp.QuoteMeta(importLockRuleQuery)).WithArgs(int64(1), int64(12)).
|
||||
WillReturnRows(sqlmock.NewRows([]string{"id", "organization_id", "category_id", "prefix", "sequence_length", "next_sequence"}))
|
||||
|
||||
tx, err := database.DB.Begin()
|
||||
if err != nil {
|
||||
t.Fatalf("begin: %v", err)
|
||||
}
|
||||
materials := []*model.Material{
|
||||
{OrganizationID: 1, Name: "螺栓", CategoryID: materialCategoryPtr(12), Unit: "个", Status: "active", Version: 1},
|
||||
}
|
||||
err = NewMaterialRepo().ImportTx(tx, 1, materials)
|
||||
if err == nil || !strings.Contains(err.Error(), "编码规则") {
|
||||
t.Fatalf("err = %v, want 编码规则错误", err)
|
||||
}
|
||||
if err := mock.ExpectationsWereMet(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
func TestMaterialRepoImportTxBatchesInsertsOverFiveHundredRows(t *testing.T) {
|
||||
// 501 行 → 两批插入（500 + 1）。插入语句与回查参数很长，这里用子串松匹配，
|
||||
// 只断言"分了两批"以及编码序号按批连续。
|
||||
mock, cleanup := newImportMock(t)
|
||||
defer cleanup()
|
||||
|
||||
mock.ExpectBegin()
|
||||
mock.ExpectQuery(regexp.QuoteMeta(importLockRuleQuery)).WithArgs(int64(1), int64(12)).
|
||||
WillReturnRows(sqlmock.NewRows([]string{"id", "organization_id", "category_id", "prefix", "sequence_length", "next_sequence"}).
|
||||
AddRow(5, 1, 12, "A01", 4, 1))
|
||||
mock.ExpectQuery(`SELECT material_code FROM materials`).
|
||||
WillReturnRows(sqlmock.NewRows([]string{"material_code"}))
|
||||
mock.ExpectExec(`UPDATE code_rules SET next_sequence`).WillReturnResult(sqlmock.NewResult(0, 1))
|
||||
mock.ExpectExec(`INSERT INTO materials`).WillReturnResult(sqlmock.NewResult(1000, 500))
|
||||
mock.ExpectQuery(`SELECT id, material_code FROM materials`).WillReturnRows(idRowsFor(1, 500))
|
||||
mock.ExpectExec(`INSERT INTO materials`).WillReturnResult(sqlmock.NewResult(1500, 1))
|
||||
mock.ExpectQuery(`SELECT id, material_code FROM materials`).WillReturnRows(idRowsFor(501, 501))
|
||||
|
||||
tx, err := database.DB.Begin()
|
||||
if err != nil {
|
||||
t.Fatalf("begin: %v", err)
|
||||
}
|
||||
categoryID := int64(12)
|
||||
materials := make([]*model.Material, 0, 501)
|
||||
for i := 0; i < 501; i++ {
|
||||
materials = append(materials, &model.Material{
|
||||
OrganizationID: 1, Name: "物料", CategoryID: &categoryID, Unit: "个", Status: "active", Version: 1,
|
||||
})
|
||||
}
|
||||
if err := NewMaterialRepo().ImportTx(tx, 1, materials); err != nil {
|
||||
t.Fatalf("import: %v", err)
|
||||
}
|
||||
if materials[0].MaterialCode != "A010001" || materials[500].MaterialCode != "A010501" {
|
||||
t.Fatalf("codes = %s ... %s", materials[0].MaterialCode, materials[500].MaterialCode)
|
||||
}
|
||||
if err := mock.ExpectationsWereMet(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
||||
// idRowsFor 造出回查语句的返回行：序号 [from, to] 对应的物料 ID 与编码。
|
||||
func idRowsFor(from, to int) *sqlmock.Rows {
|
||||
rows := sqlmock.NewRows([]string{"id", "material_code"})
|
||||
for i := from; i <= to; i++ {
|
||||
rows.AddRow(int64(i), fmt.Sprintf("A01%04d", i))
|
||||
}
|
||||
return rows
|
||||
}
|
||||
|
||||
func TestMaterialLogRepoCreateBatchTx(t *testing.T) {
|
||||
mock, cleanup := newImportMock(t)
|
||||
defer cleanup()
|
||||
|
||||
mock.ExpectBegin()
|
||||
mock.ExpectExec(`INSERT INTO material_logs`).WillReturnResult(sqlmock.NewResult(1, 2))
|
||||
|
||||
tx, err := database.DB.Begin()
|
||||
if err != nil {
|
||||
t.Fatalf("begin: %v", err)
|
||||
}
|
||||
logs := []*model.MaterialLog{
|
||||
{OperationType: "CREATE", MaterialID: 31, Operator: "tester", Description: "批量导入"},
|
||||
{OperationType: "CREATE", MaterialID: 32, Operator: "tester", Description: "批量导入"},
|
||||
}
|
||||
if err := NewMaterialLogRepo().CreateBatchTx(tx, logs); err != nil {
|
||||
t.Fatalf("create batch: %v", err)
|
||||
}
|
||||
if err := mock.ExpectationsWereMet(); err != nil {
|
||||
t.Fatal(err)
|
||||
}
|
||||
}
|
||||
|
|
@ -127,6 +127,26 @@ func (r *MaterialLogRepo) List(organizationID int64, operationType, materialId s
|
|||
return logs, total, nil
|
||||
}
|
||||
|
||||
// CreateBatchTx 在同一事务里批量写审计日志。导入场景一条物料一条 CREATE 日志，
|
||||
// 逐条往返太慢，这里折叠成多值 INSERT。
|
||||
func (r *MaterialLogRepo) CreateBatchTx(tx *sql.Tx, logs []*model.MaterialLog) error {
|
||||
if len(logs) == 0 {
|
||||
return nil
|
||||
}
|
||||
var builder strings.Builder
|
||||
builder.WriteString(`INSERT INTO material_logs (operation_type, material_id, old_value, new_value, operator, description) VALUES `)
|
||||
args := make([]interface{}, 0, len(logs)*6)
|
||||
for i, log := range logs {
|
||||
if i > 0 {
|
||||
builder.WriteString(", ")
|
||||
}
|
||||
builder.WriteString("(?, ?, ?, ?, ?, ?)")
|
||||
args = append(args, log.OperationType, log.MaterialID, log.OldValue, log.NewValue, log.Operator, log.Description)
|
||||
}
|
||||
_, err := tx.Exec(builder.String(), args...)
|
||||
return err
|
||||
}
|
||||
|
||||
// GetByMaterialID 只返回属于当前公司的该物料日志；物料不属于本公司时结果为空。
|
||||
func (r *MaterialLogRepo) GetByMaterialID(organizationID, materialID int64) ([]model.MaterialLog, error) {
|
||||
query := `SELECT ml.id, ml.operation_type, ml.material_id, ml.old_value, ml.new_value, ml.operator, ml.description, ml.created_at
|
||||
|
|
|
|||
Loading…
Reference in New Issue
Block a user