300 lines
10 KiB
Go
300 lines
10 KiB
Go
package repository
|
||
|
||
import (
|
||
"database/sql"
|
||
"time"
|
||
|
||
"github.com/wmms/wmms/internal/model"
|
||
"github.com/wmms/wmms/pkg/database"
|
||
)
|
||
|
||
// orderQuerier 同时被 *sql.DB 与 *sql.Tx 满足，让单据头表加明细的读取逻辑能在
|
||
// 事务内外复用：事务内用 FOR UPDATE 锁定当前公司的原单据，事务外只做普通读取。
|
||
type orderQuerier interface {
|
||
Query(query string, args ...interface{}) (*sql.Rows, error)
|
||
QueryRow(query string, args ...interface{}) *sql.Row
|
||
}
|
||
|
||
type InboundRepo struct{}
|
||
|
||
func NewInboundRepo() *InboundRepo {
|
||
return &InboundRepo{}
|
||
}
|
||
|
||
func (r *InboundRepo) Create(order *model.InboundOrder) error {
|
||
return r.CreateTx(nil, order)
|
||
}
|
||
|
||
func (r *InboundRepo) CreateTx(tx *sql.Tx, order *model.InboundOrder) error {
|
||
shouldCommit := false
|
||
if tx == nil {
|
||
var err error
|
||
tx, err = database.DB.Begin()
|
||
if err != nil {
|
||
return err
|
||
}
|
||
shouldCommit = true
|
||
}
|
||
if shouldCommit {
|
||
defer tx.Rollback()
|
||
}
|
||
|
||
// 三个库存维度由服务端解析后显式落库；表上这三列 NOT NULL 无默认值，缺一即报错，
|
||
// 不会静默写进默认公司。
|
||
query := `INSERT INTO inbound_orders (organization_id, warehouse_id, location_id, inbound_date, operator, handler, supplier, remark) VALUES (?, ?, ?, ?, ?, ?, ?, ?)`
|
||
result, err := tx.Exec(query, order.OrganizationID, order.WarehouseID, order.LocationID, order.InboundDate, order.Operator, order.Handler, order.Supplier, order.Remark)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
orderID, err := result.LastInsertId()
|
||
if err != nil {
|
||
return err
|
||
}
|
||
order.ID = orderID
|
||
order.InboundNo = model.OrderNo("IN", orderID)
|
||
if _, err = tx.Exec(`UPDATE inbound_orders SET inbound_no = ? WHERE id = ?`, order.InboundNo, orderID); err != nil {
|
||
return err
|
||
}
|
||
|
||
for i := range order.Items {
|
||
item := &order.Items[i]
|
||
item.InboundOrderID = orderID
|
||
query := `INSERT INTO inbound_items (inbound_order_id, material_id, quantity) VALUES (?, ?, ?)`
|
||
result, err := tx.Exec(query, orderID, item.MaterialID, item.Quantity)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
itemID, err := result.LastInsertId()
|
||
if err != nil {
|
||
return err
|
||
}
|
||
item.ID = itemID
|
||
}
|
||
|
||
if shouldCommit {
|
||
return tx.Commit()
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// loadInboundItems 读取单据明细，并 JOIN 头表与物料表把明细限制在同一组织内：
|
||
// 头表已按组织过滤，明细物料也必须属于同一公司，避免读到跨公司的物料行。
|
||
func loadInboundItems(q orderQuerier, orderID int64) ([]model.InboundItem, error) {
|
||
itemsQuery := `SELECT ii.id, ii.inbound_order_id, ii.material_id, ii.quantity, m.material_code, m.name, m.status
|
||
FROM inbound_items ii
|
||
JOIN inbound_orders io ON io.id = ii.inbound_order_id
|
||
JOIN materials m ON m.id = ii.material_id AND m.organization_id = io.organization_id
|
||
WHERE ii.inbound_order_id = ?`
|
||
rows, err := q.Query(itemsQuery, orderID)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
defer rows.Close()
|
||
|
||
var items []model.InboundItem
|
||
for rows.Next() {
|
||
var item model.InboundItem
|
||
if err := rows.Scan(&item.ID, &item.InboundOrderID, &item.MaterialID, &item.Quantity, &item.MaterialCode, &item.MaterialName, &item.MaterialStatus); err != nil {
|
||
return nil, err
|
||
}
|
||
items = append(items, item)
|
||
}
|
||
return items, rows.Err()
|
||
}
|
||
|
||
// getByID 按公司读取单据头表加明细。lock=true 时用 FOR UPDATE 锁定当前公司的原单据，
|
||
// 供更新/删除事务在同一把锁下冲销原库存。别的公司的单据在这里命中 sql.ErrNoRows。
|
||
func (r *InboundRepo) getByID(q orderQuerier, organizationID, id int64, lock bool) (*model.InboundOrder, error) {
|
||
order := &model.InboundOrder{}
|
||
query := `SELECT id, COALESCE(inbound_no, CONCAT('IN-', LPAD(id, 8, '0'))), COALESCE(operator, ''), inbound_date, handler, supplier, COALESCE(remark, ''), organization_id, warehouse_id, location_id, created_at, updated_at FROM inbound_orders WHERE id = ? AND organization_id = ?`
|
||
if lock {
|
||
query += ` FOR UPDATE`
|
||
}
|
||
err := q.QueryRow(query, id, organizationID).Scan(&order.ID, &order.InboundNo, &order.Operator, &order.InboundDate, &order.Handler, &order.Supplier, &order.Remark, &order.OrganizationID, &order.WarehouseID, &order.LocationID, &order.CreatedAt, &order.UpdatedAt)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
order.Items, err = loadInboundItems(q, id)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
|
||
// 状态不落库：inbound_orders 没有 status 列，统一在这里给出。引入状态机时改为真实读写。
|
||
order.Status = "completed"
|
||
return order, nil
|
||
}
|
||
|
||
func (r *InboundRepo) GetByID(organizationID, id int64) (*model.InboundOrder, error) {
|
||
return r.getByID(database.DB, organizationID, id, false)
|
||
}
|
||
|
||
// GetByIDForUpdateTx 在事务内锁定当前公司的原单据，别的公司的单据命中 ErrNoRows。
|
||
func (r *InboundRepo) GetByIDForUpdateTx(tx *sql.Tx, organizationID, id int64) (*model.InboundOrder, error) {
|
||
return r.getByID(tx, organizationID, id, true)
|
||
}
|
||
|
||
func (r *InboundRepo) List(organizationID int64, search string, limit, offset int) ([]model.InboundOrder, int64, error) {
|
||
// COUNT 和主查询是两条独立 SQL，组织条件与筛选条件必须成对出现，否则分页总数会和
|
||
// 实际行数对不上。
|
||
countQuery := `SELECT COUNT(*) FROM inbound_orders WHERE organization_id = ?`
|
||
countArgs := []interface{}{organizationID}
|
||
|
||
if search != "" {
|
||
countQuery += ` AND (handler LIKE ? OR supplier 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, COALESCE(inbound_no, CONCAT('IN-', LPAD(id, 8, '0'))), COALESCE(operator, ''), inbound_date, handler, supplier, COALESCE(remark, ''), organization_id, warehouse_id, location_id, created_at, updated_at FROM inbound_orders WHERE organization_id = ?`
|
||
args := []interface{}{organizationID}
|
||
|
||
if search != "" {
|
||
query += ` AND (handler LIKE ? OR supplier 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 orders []model.InboundOrder
|
||
for rows.Next() {
|
||
var order model.InboundOrder
|
||
err := rows.Scan(&order.ID, &order.InboundNo, &order.Operator, &order.InboundDate, &order.Handler, &order.Supplier, &order.Remark, &order.OrganizationID, &order.WarehouseID, &order.LocationID, &order.CreatedAt, &order.UpdatedAt)
|
||
if err != nil {
|
||
return nil, 0, err
|
||
}
|
||
// 同上：状态在查询层给出，列表里一并补上。
|
||
order.Status = "completed"
|
||
orders = append(orders, order)
|
||
}
|
||
return orders, total, nil
|
||
}
|
||
|
||
// GetByMaterialID 返回某物料在本公司的入库历史。头表按组织过滤，明细 JOIN 的物料也
|
||
// 必须属于同一公司，公司 A 不能凭物料 ID 读到公司 B 的入库记录。
|
||
func (r *InboundRepo) GetByMaterialID(organizationID, materialID int64) ([]model.InboundRecord, error) {
|
||
query := `SELECT ii.id, ii.inbound_order_id, io.handler, io.supplier, ii.quantity, ii.created_at
|
||
FROM inbound_items ii
|
||
JOIN inbound_orders io ON ii.inbound_order_id = io.id
|
||
JOIN materials m ON ii.material_id = m.id AND m.organization_id = io.organization_id
|
||
WHERE ii.material_id = ? AND io.organization_id = ?
|
||
ORDER BY ii.id DESC`
|
||
|
||
rows, err := database.DB.Query(query, materialID, organizationID)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
defer rows.Close()
|
||
|
||
var records []model.InboundRecord
|
||
for rows.Next() {
|
||
var rec model.InboundRecord
|
||
err := rows.Scan(&rec.ID, &rec.InboundID, &rec.Handler, &rec.Supplier, &rec.Quantity, &rec.CreatedAt)
|
||
if err != nil {
|
||
return nil, err
|
||
}
|
||
records = append(records, rec)
|
||
}
|
||
return records, nil
|
||
}
|
||
|
||
func (r *InboundRepo) GetInboundDate(organizationID, id int64) (time.Time, error) {
|
||
var inboundDate time.Time
|
||
query := `SELECT inbound_date FROM inbound_orders WHERE id = ? AND organization_id = ?`
|
||
err := database.DB.QueryRow(query, id, organizationID).Scan(&inboundDate)
|
||
return inboundDate, err
|
||
}
|
||
|
||
func (r *InboundRepo) Delete(organizationID, id int64) error {
|
||
return r.DeleteTx(nil, organizationID, id)
|
||
}
|
||
|
||
func (r *InboundRepo) DeleteTx(tx *sql.Tx, organizationID, id int64) error {
|
||
if tx == nil {
|
||
var err error
|
||
tx, err = database.DB.Begin()
|
||
if err != nil {
|
||
return err
|
||
}
|
||
defer tx.Rollback()
|
||
_, err = tx.Exec(`DELETE FROM inbound_orders WHERE organization_id = ? AND id = ?`, organizationID, id)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
return tx.Commit()
|
||
}
|
||
_, err := tx.Exec(`DELETE FROM inbound_orders WHERE organization_id = ? AND id = ?`, organizationID, id)
|
||
return err
|
||
}
|
||
|
||
func (r *InboundRepo) Update(organizationID, id int64, req *model.UpdateInboundRequest) error {
|
||
return r.UpdateTx(nil, organizationID, id, req)
|
||
}
|
||
|
||
func (r *InboundRepo) UpdateTx(tx *sql.Tx, organizationID, id int64, req *model.UpdateInboundRequest) error {
|
||
shouldCommit := false
|
||
if tx == nil {
|
||
var err error
|
||
tx, err = database.DB.Begin()
|
||
if err != nil {
|
||
return err
|
||
}
|
||
shouldCommit = true
|
||
}
|
||
if shouldCommit {
|
||
defer tx.Rollback()
|
||
}
|
||
|
||
// 先确认目标单据属于本公司：跨公司的单据在这里直接返回 ErrNoRows，后面的明细
|
||
// 删除/插入绝不会碰到别家公司的单据（明细表没有组织列，只能靠头表归属收口）。
|
||
var existsID int64
|
||
err := tx.QueryRow(`SELECT id FROM inbound_orders WHERE organization_id = ? AND id = ?`, organizationID, id).Scan(&existsID)
|
||
if err == sql.ErrNoRows {
|
||
return sql.ErrNoRows
|
||
}
|
||
if err != nil {
|
||
return err
|
||
}
|
||
|
||
// Update order basic fields，写入带组织条件，跨公司的单据不会在这里被改动。
|
||
query := `UPDATE inbound_orders SET handler = ?, supplier = ?, remark = ? WHERE organization_id = ? AND id = ?`
|
||
_, err = tx.Exec(query, req.Handler, req.Supplier, req.Remark, organizationID, id)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
|
||
// Delete old items
|
||
_, err = tx.Exec(`DELETE FROM inbound_items WHERE inbound_order_id = ?`, id)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
|
||
// Insert new items
|
||
for i := range req.Items {
|
||
item := &req.Items[i]
|
||
query := `INSERT INTO inbound_items (inbound_order_id, material_id, quantity) VALUES (?, ?, ?)`
|
||
_, err = tx.Exec(query, id, item.MaterialID, item.Quantity)
|
||
if err != nil {
|
||
return err
|
||
}
|
||
}
|
||
|
||
if shouldCommit {
|
||
return tx.Commit()
|
||
}
|
||
return nil
|
||
}
|