From f3146af838b744309e1d93deed6074589c283f39 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=96=87=E7=9D=BF=20=E5=BC=A0?= Date: Sat, 10 Oct 2026 15:13:18 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E7=89=A9=E6=96=99=E6=89=B9=E9=87=8F?= =?UTF-8?q?=E5=85=A5=E5=BA=93=E7=9A=84=E7=BC=96=E7=A0=81=E5=88=86=E9=85=8D?= =?UTF-8?q?=E4=B8=8E=E4=BA=8B=E5=8A=A1=E6=8F=92=E5=85=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- internal/repository/material.go | 178 ++++++++++++++++++++ internal/repository/material_import_test.go | 177 +++++++++++++++++++ internal/repository/material_log.go | 20 +++ 3 files changed, 375 insertions(+) create mode 100644 internal/repository/material_import_test.go diff --git a/internal/repository/material.go b/internal/repository/material.go index e7a5df9d..41577c3c 100644 --- a/internal/repository/material.go +++ b/internal/repository/material.go @@ -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 +} diff --git a/internal/repository/material_import_test.go b/internal/repository/material_import_test.go new file mode 100644 index 00000000..f9bb7ae4 --- /dev/null +++ b/internal/repository/material_import_test.go @@ -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) + } +} diff --git a/internal/repository/material_log.go b/internal/repository/material_log.go index 35dcb46a..c4506c38 100644 --- a/internal/repository/material_log.go +++ b/internal/repository/material_log.go @@ -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