iotlcm/admin-svr/server/internal/logic/iot/simcardbatch/renew/repository.go

package renew

import (
	"context"
	"strings"

	"git.5iot.com/iotlcm-public/iotlcm-admin-svr/internal/consts"
	"git.5iot.com/iotlcm-public/iotlcm-admin-svr/internal/dao"
	"git.5iot.com/iotlcm-public/iotlcm-admin-svr/internal/model/entity"

	"github.com/gogf/gf/v2/database/gdb"
	"github.com/gogf/gf/v2/errors/gerror"
)

type Repository struct{}

func NewRepository() *Repository {
	return &Repository{}
}

func (r *Repository) GetSourceBatch(ctx context.Context, batchNo string) (*entity.BatchMain, error) {
	batch := &entity.BatchMain{}
	columns := dao.BatchMain.Columns()
	err := dao.BatchMain.Ctx(ctx).
		Fields(
			columns.Id,
			columns.BatchNo,
			columns.BatchType,
			columns.BatchStatus,
			columns.RunResult,
			columns.TenantId,
			columns.DeptId,
			columns.CreatedDept,
			columns.CreatedBy,
		).
		Where(columns.BatchNo, strings.TrimSpace(batchNo)).
		Scan(batch)
	if err != nil {
		return nil, err
	}
	if batch.Id == 0 {
		return nil, nil
	}
	return batch, nil
}

func (r *Repository) GetRenewBatchRows(ctx context.Context, batchNo string, batchType int) ([]*Row, error) {
	return r.getRenewBatchRows(ctx, nil, batchNo, batchType)
}

func (r *Repository) GetRenewBatchRowsTx(ctx context.Context, tx gdb.TX, batchNo string, batchType int) ([]*Row, error) {
	return r.getRenewBatchRows(ctx, tx, batchNo, batchType)
}

func (r *Repository) getRenewBatchRows(ctx context.Context, tx gdb.TX, batchNo string, batchType int) ([]*Row, error) {
	var details []*entity.BatchDetail
	columns := dao.BatchDetail.Columns()
	if batchType == 0 {
		batchType = consts.BatchTypeRenew
	}
	model := dao.BatchDetail.Ctx(ctx)
	if tx != nil {
		model = tx.Ctx(ctx).Model(dao.BatchDetail.Table())
	}
	err := model.
		Fields(columns.RowNo, columns.ItemId, columns.Snapshot).
		Where(columns.BatchNo, strings.TrimSpace(batchNo)).
		Where(columns.BatchType, batchType).
		Where(columns.DetailStatus, consts.BatchDetailStatusSuccess).
		OrderAsc(columns.RowNo).
		Scan(&details)
	if err != nil {
		return nil, err
	}
	return detailsToRows(details), nil
}

func (r *Repository) GetSourceBatchRows(ctx context.Context, batchNo string) ([]*Row, error) {
	var details []*entity.BatchDetail
	columns := dao.BatchDetail.Columns()
	err := dao.BatchDetail.Ctx(ctx).
		Fields(columns.RowNo, columns.ItemId, columns.Snapshot).
		Where(columns.BatchNo, strings.TrimSpace(batchNo)).
		Where(columns.BatchType, consts.BatchTypeOutbound).
		WhereIn(columns.DetailStatus, []int{consts.BatchDetailStatusSuccess, consts.BatchAsyncStatusSuccess}).
		OrderAsc(columns.RowNo).
		Scan(&details)
	if err != nil {
		return nil, err
	}

	return detailsToRows(details), nil
}

func (r *Repository) BatchDetailItemIDs(ctx context.Context, tx gdb.TX, batchNo string, batchType int, statuses []int) ([]string, error) {
	batchNo = strings.TrimSpace(batchNo)
	if batchNo == "" {
		return nil, nil
	}
	var details []*entity.BatchDetail
	model := tx.Ctx(ctx).Model(dao.BatchDetail.Table()).
		Fields(dao.BatchDetail.Columns().ItemId).
		Where(dao.BatchDetail.Columns().BatchNo, batchNo)
	if batchType > 0 {
		model = model.Where(dao.BatchDetail.Columns().BatchType, batchType)
	}
	if len(statuses) > 0 {
		model = model.WhereIn(dao.BatchDetail.Columns().DetailStatus, statuses)
	}
	if err := model.Scan(&details); err != nil {
		return nil, err
	}
	ids := make([]string, 0, len(details))
	for _, detail := range details {
		if detail == nil {
			continue
		}
		if itemID := strings.TrimSpace(detail.ItemId); itemID != "" {
			ids = append(ids, itemID)
		}
	}
	return uniqueStrings(ids), nil
}

func (r *Repository) UnfinishedRenewBatchNosByIccids(ctx context.Context, tx gdb.TX, tenantID string, iccids []string, currentBatchNo string) (map[string]string, error) {
	result := make(map[string]string)
	iccids = uniqueStrings(iccids)
	if len(iccids) == 0 {
		return result, nil
	}

	currentBatchID, err := r.batchIDByNo(ctx, tx, currentBatchNo)
	if err != nil {
		return nil, err
	}

	type conflictRow struct {
		ItemId  string `orm:"item_id"`
		BatchNo string `orm:"batch_no"`
	}

	bd := "bd"
	bm := "bm"
	detailColumns := dao.BatchDetail.Columns()
	mainColumns := dao.BatchMain.Columns()
	model := tx.Ctx(ctx).Model(dao.BatchDetail.Table()+" "+bd).
		InnerJoin(dao.BatchMain.Table()+" "+bm, bm+"."+mainColumns.BatchNo+" = "+bd+"."+detailColumns.BatchNo).
		Fields(bd+"."+detailColumns.ItemId, bm+"."+mainColumns.BatchNo).
		Where(bd+"."+detailColumns.BatchType, consts.BatchTypeRenew).
		Where(bm+"."+mainColumns.BatchType, consts.BatchTypeRenew).
		Where(bm+"."+mainColumns.BatchStatus, consts.BatchStatusNormal).
		WhereIn(bm+"."+mainColumns.BatchAsyncStatus, []int{
			consts.BatchAsyncStatusCreated,
			consts.BatchAsyncStatusPending,
			consts.BatchAsyncStatusProcessing,
		}).
		WhereIn(bm+"."+mainColumns.RunResult, []int{
			consts.BatchRunResultPending,
			consts.BatchRunResultRunning,
		}).
		WhereIn(bd+"."+detailColumns.DetailStatus, []int{
			consts.BatchDetailStatusPending,
			consts.BatchDetailStatusRunning,
		}).
		WhereIn(bd+"."+detailColumns.ItemId, iccids).
		OrderAsc(bm + "." + mainColumns.Id)
	if strings.TrimSpace(tenantID) != "" {
		model = model.Where(bm+"."+mainColumns.TenantId, strings.TrimSpace(tenantID))
	}
	if currentBatchID > 0 {
		// When duplicate formal renew batches already exist, the earliest batch
		// wins and later batches fail. This prevents two batches from blocking
		// each other forever.
		model = model.Where(bm+"."+mainColumns.Id+" < ?", currentBatchID)
	} else if currentBatchNo = strings.TrimSpace(currentBatchNo); currentBatchNo != "" {
		model = model.Where(bm+"."+mainColumns.BatchNo+" <> ?", currentBatchNo)
	}

	var rows []conflictRow
	if err := model.Scan(&rows); err != nil {
		return nil, err
	}
	for _, row := range rows {
		itemID := strings.TrimSpace(row.ItemId)
		if itemID == "" {
			continue
		}
		if _, exists := result[itemID]; !exists {
			result[itemID] = strings.TrimSpace(row.BatchNo)
		}
	}
	return result, nil
}

func (r *Repository) batchIDByNo(ctx context.Context, tx gdb.TX, batchNo string) (int64, error) {
	batchNo = strings.TrimSpace(batchNo)
	if batchNo == "" {
		return 0, nil
	}
	batch := &entity.BatchMain{}
	if err := tx.Ctx(ctx).Model(dao.BatchMain.Table()).
		Fields(dao.BatchMain.Columns().Id).
		Where(dao.BatchMain.Columns().BatchNo, batchNo).
		Limit(1).
		Scan(batch); err != nil {
		return 0, err
	}
	return batch.Id, nil
}

func detailsToRows(details []*entity.BatchDetail) []*Row {
	rows := make([]*Row, 0, len(details))
	for _, detail := range details {
		if detail == nil {
			continue
		}
		row := &Row{RowNo: detail.RowNo, Iccid: strings.TrimSpace(detail.ItemId)}
		if detail.Snapshot != nil {
			if row.Iccid == "" {
				row.Iccid = strings.TrimSpace(detail.Snapshot.Get("iccid").String())
			}
			if row.Iccid == "" {
				row.Iccid = strings.TrimSpace(detail.Snapshot.Get("ICCID").String())
			}
			row.Msisdn = strings.TrimSpace(detail.Snapshot.Get("msisdn").String())
			if row.Msisdn == "" {
				row.Msisdn = strings.TrimSpace(detail.Snapshot.Get("MSISDN").String())
			}
			row.Imsi = strings.TrimSpace(detail.Snapshot.Get("imsi").String())
			if row.Imsi == "" {
				row.Imsi = strings.TrimSpace(detail.Snapshot.Get("IMSI").String())
			}
		}
		rows = append(rows, row)
	}
	return rows
}

func (r *Repository) LockCardsByIccids(ctx context.Context, tx gdb.TX, iccids []string) (map[string]*entity.TSimCardInfo, error) {
	result := make(map[string]*entity.TSimCardInfo)
	if len(iccids) == 0 {
		return result, nil
	}

	var list []*entity.TSimCardInfo
	if err := tx.Ctx(ctx).Model(dao.TSimCardInfo.Table()).
		WhereIn(dao.TSimCardInfo.Columns().Iccid, uniqueStrings(iccids)).
		LockUpdate().
		Scan(&list); err != nil {
		return nil, err
	}
	for _, item := range list {
		if item != nil && strings.TrimSpace(item.Iccid) != "" {
			result[strings.TrimSpace(item.Iccid)] = item
		}
	}
	return result, nil
}

func (r *Repository) LockLatestBindsByIccids(ctx context.Context, tx gdb.TX, iccids []string) (map[string]*entity.TSimCardBindPkg, error) {
	result := make(map[string]*entity.TSimCardBindPkg)
	if len(iccids) == 0 {
		return result, nil
	}

	var list []*entity.TSimCardBindPkg
	if err := tx.Ctx(ctx).Model(dao.TSimCardBindPkg.Table()).
		WhereIn(dao.TSimCardBindPkg.Columns().Iccid, uniqueStrings(iccids)).
		WhereIn(dao.TSimCardBindPkg.Columns().PackageType, []int{101, 102}).
		WhereIn(dao.TSimCardBindPkg.Columns().EffectiveStatus, []int{
			consts.EffectiveStatusPending,
			consts.EffectiveStatusEffective,
			consts.EffectiveStatusExpired,
		}).
		OrderDesc(dao.TSimCardBindPkg.Columns().ExpectedExpireTime).
		OrderDesc(dao.TSimCardBindPkg.Columns().Id).
		LockUpdate().
		Scan(&list); err != nil {
		return nil, err
	}
	effectiveCount := make(map[string]int)
	for _, item := range list {
		if item == nil || strings.TrimSpace(item.Iccid) == "" {
			continue
		}
		iccid := strings.TrimSpace(item.Iccid)
		if item.EffectiveStatus == consts.EffectiveStatusEffective {
			effectiveCount[iccid]++
			if effectiveCount[iccid] > 1 {
				return nil, gerror.Newf("同一张卡存在多个生效中的套餐订购记录：%s", iccid)
			}
		}
		if _, exists := result[iccid]; exists {
			continue
		}
		result[iccid] = item
	}
	return result, nil
}