iotlcm/worker-svr/internal/service/allocate/repository.go

package allocate

import (
	"context"
	"encoding/json"
	"fmt"
	"sort"
	"strings"

	"github.com/gogf/gf/v2/database/gdb"
	"github.com/gogf/gf/v2/encoding/gjson"
	"github.com/gogf/gf/v2/errors/gerror"
	"github.com/gogf/gf/v2/frame/g"
	"github.com/gogf/gf/v2/os/gtime"

	"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/consts"
	"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/dao"
	"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/model/do"
	"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/model/entity"
	"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/service/batchresult"
)

type Repository struct {
	db      gdb.DB
	results batchresult.Generator
}

const (
	allocateCardQueryBatchSize    = 500
	allocateCardUpdateBatchSize   = 500
	allocateDetailUpdateBatchSize = 500
)

type allocateParentAggregate struct {
	ChildCount        int         `orm:"child_count"`
	FinishedCount     int         `orm:"finished_count"`
	TotalCount        int         `orm:"total_count"`
	ProcessedCount    int         `orm:"processed_count"`
	SuccessCount      int         `orm:"success_count"`
	FailCount         int         `orm:"fail_count"`
	EarliestStartTime *gtime.Time `orm:"earliest_start_time"`
}

func NewRepository() *Repository {
	return &Repository{results: batchresult.New()}
}

func (r *Repository) database() gdb.DB {
	if r.db != nil {
		return r.db
	}
	return g.DB()
}

func (r *Repository) LoadMain(ctx context.Context, batchNo string) (*entity.BatchMain, error) {
	batchNo = strings.TrimSpace(batchNo)
	if batchNo == "" {
		return nil, gerror.New("allocate batch number is required")
	}
	main := &entity.BatchMain{}
	if err := r.database().Model(dao.BatchMain.Table()).Safe().Ctx(ctx).
		Where(dao.BatchMain.Columns().BatchNo, batchNo).
		Scan(main); err != nil {
		return nil, gerror.Wrap(err, "load allocate batch main failed")
	}
	if main.Id == 0 {
		return nil, gerror.New("allocate batch does not exist")
	}
	return main, nil
}

func (r *Repository) Load(ctx context.Context, batchNo string) (*Task, error) {
	batchNo = strings.TrimSpace(batchNo)
	if batchNo == "" {
		return nil, gerror.New("划拨批次号不能为空")
	}
	var main *entity.BatchMain
	if err := r.database().Model(dao.BatchMain.Table()).Safe().Ctx(ctx).
		Where(dao.BatchMain.Columns().BatchNo, batchNo).
		Where(dao.BatchMain.Columns().BatchType, consts.BatchTypeAllocate).
		Where(dao.BatchMain.Columns().BatchAsyncStatus, consts.BatchAsyncStatusProcessing).
		Scan(&main); err != nil {
		return nil, gerror.Wrap(err, "查询划拨批次主表失败")
	}
	if main == nil || main.Id == 0 {
		return nil, gerror.New("处理中划拨批次不存在")
	}

	var batchDetails []*entity.BatchDetail
	if err := r.database().Model(dao.BatchDetail.Table()).Safe().Ctx(ctx).
		Where(dao.BatchDetail.Columns().BatchNo, batchNo).
		Where(dao.BatchDetail.Columns().BatchType, consts.BatchTypeAllocate).
		OrderAsc(dao.BatchDetail.Columns().RowNo).
		Scan(&batchDetails); err != nil {
		return nil, gerror.Wrap(err, "查询划拨批次明细失败")
	}
	if len(batchDetails) == 0 {
		return nil, gerror.New("划拨批次没有运行中明细")
	}

	task := &Task{
		Batch: Batch{
			ID:            main.Id,
			BatchNo:       main.BatchNo,
			BatchType:     main.BatchType,
			ParentBatchNo: strings.TrimSpace(main.SourceBatchNo),
			TenantID:      main.TenantId,
			TargetDeptID:  targetDeptIDFromBatch(main),
			CreatedDept:   main.CreatedDept,
			CreatedBy:     main.CreatedBy,
			StartTime:     main.StartTime,
			BatchInfo:     main.BatchInfo,
		},
		Details: make([]*Detail, 0, len(batchDetails)),
	}
	for _, batchDetail := range batchDetails {
		if batchDetail == nil {
			continue
		}
		detail := &Detail{OriginalSnapshot: batchDetail.Snapshot,
			ID:           batchDetail.Id,
			RowNo:        batchDetail.RowNo,
			DetailStatus: batchDetail.DetailStatus,
			ICCID:        strings.TrimSpace(batchDetail.ItemId),
			Reason:       strings.TrimSpace(batchDetail.ErrorReason),
		}
		if detail.DetailStatus != consts.BatchDetailStatusPending &&
			detail.DetailStatus != consts.BatchDetailStatusRunning &&
			detail.Reason == "" {
			detail.Reason = "批次明细已处理，不能重复执行"
		}
		if batchDetail.Snapshot != nil {
			if value := firstSnapshot(batchDetail.Snapshot, "ICCID", "iccid"); value != "" {
				detail.ICCID = value
			}
			detail.MSISDN = firstSnapshot(batchDetail.Snapshot, "MSISDN", "msisdn")
			detail.IMSI = firstSnapshot(batchDetail.Snapshot, "IMSI", "imsi")
			detail.TargetDeptID = batchDetail.Snapshot.Get("targetDeptId").Int64()
			detail.TargetDeptKey = firstSnapshot(batchDetail.Snapshot, "targetDeptKey", "target_dept_key")
			detail.TargetDeptName = firstSnapshot(batchDetail.Snapshot, "targetDeptName", "target_dept_name")
			detail.Remark = firstSnapshot(batchDetail.Snapshot, "remark")
		}
		if detail.TargetDeptID <= 0 {
			detail.TargetDeptID = task.Batch.TargetDeptID
		}
		task.Details = append(task.Details, detail)
	}
	if len(task.Details) == 0 {
		return nil, gerror.New("划拨批次没有可读取明细")
	}
	if err := r.loadCards(ctx, task); err != nil {
		return nil, err
	}
	if err := r.loadTargets(ctx, task); err != nil {
		return nil, err
	}
	return task, nil
}

func targetDeptIDFromBatch(main *entity.BatchMain) int64 {
	if main == nil {
		return 0
	}
	if main.TargetDeptId > 0 {
		return main.TargetDeptId
	}
	if main.BatchInfo != nil {
		return main.BatchInfo.Get("targetDeptId").Int64()
	}
	return 0
}

func firstSnapshot(snapshot *gjson.Json, keys ...string) string {
	if snapshot == nil {
		return ""
	}
	for _, key := range keys {
		if value := strings.TrimSpace(snapshot.Get(key).String()); value != "" {
			return value
		}
	}
	return ""
}

func (r *Repository) loadCards(ctx context.Context, task *Task) error {
	iccids := make([]string, 0, len(task.Details))
	for _, detail := range task.Details {
		if detail != nil && strings.TrimSpace(detail.ICCID) != "" {
			iccids = append(iccids, strings.TrimSpace(detail.ICCID))
		}
	}
	iccids = uniqueStrings(iccids)
	if len(iccids) == 0 {
		return nil
	}
	columns := dao.TSimCardInfo.Columns()
	cards := make([]*entity.TSimCardInfo, 0, len(iccids))
	for start := 0; start < len(iccids); start += allocateCardQueryBatchSize {
		end := min(start+allocateCardQueryBatchSize, len(iccids))
		var chunk []*entity.TSimCardInfo
		if err := r.database().Model(dao.TSimCardInfo.Table()).Safe().Ctx(ctx).
			Where(columns.TenantId, task.Batch.TenantID).
			WhereIn(columns.Iccid, iccids[start:end]).
			Scan(&chunk); err != nil {
			return gerror.Wrapf(err, "查询SIM卡档案失败，范围:%d-%d", start+1, end)
		}
		cards = append(cards, chunk...)
	}
	byICCID := make(map[string]*entity.TSimCardInfo, len(cards))
	for _, card := range cards {
		if card != nil {
			byICCID[strings.TrimSpace(card.Iccid)] = card
		}
	}
	for _, detail := range task.Details {
		if detail == nil {
			continue
		}
		card := byICCID[strings.TrimSpace(detail.ICCID)]
		if card == nil {
			continue
		}
		detail.Card = &Card{
			ID:                card.Id,
			ICCID:             strings.TrimSpace(card.Iccid),
			MSISDN:            strings.TrimSpace(card.Msisdn),
			IMSI:              strings.TrimSpace(card.Imsi),
			DeptID:            card.DeptId,
			AllocationBatchNo: strings.TrimSpace(card.AllocationBatchNo),
			PackageStatus:     card.PackageStatus,
			BossCardStatus:    card.BossCardStatus,
			FlowPkgEndAt:      card.FlowPkgEndAt,
		}
	}
	return nil
}

func (r *Repository) loadTargets(ctx context.Context, task *Task) error {
	task.TargetsByID = make(map[int64]*Target)
	task.TargetsByKey = make(map[string]*Target)
	task.TargetsByName = make(map[string][]*Target)
	if task.Batch.CreatedDept <= 0 {
		return nil
	}
	var depts []*entity.SysDept
	columns := dao.SysDept.Columns()
	if err := r.database().Model(dao.SysDept.Table()).Safe().Ctx(ctx).
		Fields(columns.DeptId, columns.ParentId, columns.DeptKey, columns.DeptName).
		Where(columns.TenantId, task.Batch.TenantID).
		Where(columns.ParentId, task.Batch.CreatedDept).
		Where(columns.Status, "0").
		Where(columns.DeletedAt, nil).
		Scan(&depts); err != nil {
		return gerror.Wrap(err, "查询划拨目标客户失败")
	}
	for _, dept := range depts {
		if dept == nil || dept.DeptId <= 0 {
			continue
		}
		target := &Target{
			DeptID:   dept.DeptId,
			ParentID: dept.ParentId,
			DeptKey:  strings.TrimSpace(dept.DeptKey),
			DeptName: strings.TrimSpace(dept.DeptName),
		}
		task.TargetsByID[target.DeptID] = target
		if target.DeptKey != "" {
			task.TargetsByKey[target.DeptKey] = target
		}
		if target.DeptName != "" {
			task.TargetsByName[target.DeptName] = append(task.TargetsByName[target.DeptName], target)
		}
	}
	return nil
}

func (r *Repository) Commit(ctx context.Context, task *Task) error {
	if task == nil {
		return gerror.New("allocate task is nil")
	}

	if _, failCount := countResult(task); failCount > 0 {
		markAtomicFailure(task, "划拨批次存在失败明细，整批未执行")
	}
	successCount, failCount := countResult(task)
	if failCount > 0 {
		failCount = countValidationFailures(task)
	}
	autoSplitChild := strings.TrimSpace(task.Batch.ParentBatchNo) != ""
	var artifact batchresult.Artifact
	var artifactErr error
	if !autoSplitChild {
		artifact, artifactErr = r.createResultFile(ctx, task)
	}
	if artifactErr != nil && !autoSplitChild {
		g.Log().Warningf(ctx, "allocate result file generation failed, batchNo:%s, err:%+v", task.Batch.BatchNo, artifactErr)
	}

	now := gtime.Now()
	err := r.database().Transaction(ctx, func(ctx context.Context, tx gdb.TX) error {
		if successCount > 0 && !autoSplitChild {
			if err := r.updateCardOwners(ctx, tx, task, now); err != nil {
				return err
			}
		}
		if err := r.updateDetails(ctx, tx, task, now); err != nil {
			return err
		}

		var ossID, downloadID int64
		if !autoSplitChild && artifactErr == nil {
			var err error
			ossID, downloadID, err = r.persistArtifact(ctx, tx, task, artifact, successCount, failCount, now)
			if err != nil {
				return err
			}
		}
		return r.finalizeBatch(ctx, tx, task, successCount, failCount, ossID, downloadID, now, !autoSplitChild && artifactErr != nil)
	})
	if err != nil && artifactErr == nil && artifact.Path != "" {
		batchresult.RemoveLocalFile(artifact)
	}
	if err != nil {
		return gerror.Wrap(err, "commit allocate batch failed")
	}
	if autoSplitChild {
		if err := r.AggregateParent(ctx, task.Batch.ParentBatchNo); err != nil {
			return gerror.Wrap(err, "aggregate allocate parent batch failed")
		}
	}
	return nil
}

func (r *Repository) AggregateParent(ctx context.Context, parentBatchNo string) error {
	parentBatchNo = strings.TrimSpace(parentBatchNo)
	if parentBatchNo == "" {
		return gerror.New("allocate parent batch number is required")
	}
	release := acquireAllocateParentLock(parentBatchNo)
	defer release()

	maxRetries := allocateDeadlockMaxRetries(ctx)
	for attempt := 0; ; attempt++ {
		err := r.aggregateParentOnce(ctx, parentBatchNo)
		if err == nil || !isAllocateDeadlock(err) || attempt >= maxRetries {
			return err
		}
		if err := waitAllocateDeadlockRetry(ctx, attempt); err != nil {
			return err
		}
		g.Log().Warningf(ctx, "allocate parent aggregation deadlock, retrying batchNo:%s attempt:%d/%d", parentBatchNo, attempt+1, maxRetries)
	}
}

func (r *Repository) aggregateParentOnce(ctx context.Context, parentBatchNo string) error {
	parentBatchNo = strings.TrimSpace(parentBatchNo)
	if parentBatchNo == "" {
		return gerror.New("allocate parent batch number is required")
	}
	now := gtime.Now()
	var committedTask *Task
	var parentArtifact batchresult.Artifact
	parentArtifactReady := false
	err := r.database().Transaction(ctx, func(ctx context.Context, tx gdb.TX) error {
		parent := &entity.BatchMain{}
		mainColumns := dao.BatchMain.Columns()
		if err := tx.Ctx(ctx).Model(dao.BatchMain.Table()).Where(mainColumns.BatchNo, parentBatchNo).Where(mainColumns.BatchType, consts.BatchTypeAllocate).LockUpdate().Scan(parent); err != nil {
			return gerror.Wrap(err, "load allocate parent batch failed")
		}
		if parent.Id == 0 {
			return gerror.New("allocate parent batch does not exist")
		}
		if !isAllocateParent(parent) {
			return gerror.New("allocate parent batch metadata is invalid")
		}
		if parent.BatchAsyncStatus == consts.BatchAsyncStatusSuccess || parent.BatchAsyncStatus == consts.BatchAsyncStatusFailed || parent.BatchAsyncStatus == consts.BatchAsyncStatusPartialSuccess {
			return nil
		}
		childColumns := dao.BatchMain.Columns()
		aggregate := &allocateParentAggregate{}
		finishedStatus := fmt.Sprintf("%s IN (%d, %d, %d)", childColumns.BatchAsyncStatus, consts.BatchAsyncStatusSuccess, consts.BatchAsyncStatusFailed, consts.BatchAsyncStatusPartialSuccess)
		fields := fmt.Sprintf("COUNT(*) AS child_count, COALESCE(SUM(CASE WHEN %s THEN 1 ELSE 0 END), 0) AS finished_count, COALESCE(SUM(%s), 0) AS total_count, COALESCE(SUM(CASE WHEN %s THEN %s ELSE 0 END), 0) AS processed_count, COALESCE(SUM(%s), 0) AS success_count, COALESCE(SUM(CASE WHEN %s THEN %s - %s ELSE 0 END), 0) AS fail_count, MIN(%s) AS earliest_start_time", finishedStatus, childColumns.TotalCount, finishedStatus, childColumns.TotalCount, childColumns.SuccessCount, finishedStatus, childColumns.TotalCount, childColumns.SuccessCount, childColumns.StartTime)
		if err := tx.Ctx(ctx).Model(dao.BatchMain.Table()).Fields(fields).Where(childColumns.BatchType, consts.BatchTypeAllocate).Where(childColumns.SourceBatchNo, parentBatchNo).Scan(aggregate); err != nil {
			return gerror.Wrap(err, "aggregate allocate child batches failed")
		}
		info := mergeAllocateParentInfo(parent.BatchInfo, aggregate)
		splitCompleted := parent.BatchInfo != nil && parent.BatchInfo.Get("splitCompleted").Bool()
		allChildrenFinished := splitCompleted && aggregate.ChildCount > 0 && aggregate.ChildCount == aggregate.FinishedCount
		if !allChildrenFinished {
			startTime := resolveAllocateParentStartTime(parent.StartTime, aggregate.EarliestStartTime, now)
			infoData := info.Map()
			infoData["successCount"] = 0
			infoData["failCount"] = 0
			data := do.BatchMain{StartTime: startTime, TotalCount: aggregate.TotalCount, ValidCount: 0, SuccessCount: 0, FailCount: 0, BatchInfo: gjson.New(infoData), BatchAsyncStatus: consts.BatchAsyncStatusProcessing, RunResult: consts.BatchRunResultRunning, UpdatedBy: parent.CreatedBy, UpdatedAt: now}
			_, err := tx.Ctx(ctx).Model(dao.BatchMain.Table()).Where(mainColumns.Id, parent.Id).WhereIn(mainColumns.BatchAsyncStatus, []int{consts.BatchAsyncStatusPending, consts.BatchAsyncStatusProcessing}).Data(data).Update()
			if err != nil {
				return gerror.Wrap(err, "update allocate parent progress failed")
			}
			return nil
		}
		task, err := r.loadParentTaskTx(ctx, tx, parent)
		if err != nil {
			return err
		}
		if len(task.Details) != aggregate.TotalCount {
			return gerror.Newf("allocate parent detail count mismatch, expected:%d actual:%d", aggregate.TotalCount, len(task.Details))
		}
		if aggregate.FailCount > 0 {
			markAtomicFailure(task, "至少一个划拨子批次校验失败，整批未执行")
		} else {
			validateTask(task)
			if _, failures := countResult(task); failures > 0 {
				markAtomicFailure(task, "划拨明细复核失败，整批未执行")
			}
		}
		successCount, failCount := normalizeAtomicParentResult(task)
		if successCount > 0 {
			if err := r.updateCardOwners(ctx, tx, task, now); err != nil {
				return err
			}
		}
		if err := r.updateParentDetails(ctx, tx, task, now); err != nil {
			return err
		}
		artifact, artifactErr := r.createResultFile(ctx, task)
		if artifactErr == nil {
			parentArtifact, parentArtifactReady = artifact, true
		}
		if artifactErr != nil {
			g.Log().Warningf(ctx, "allocate parent result file generation failed, batchNo:%s, err:%+v", parentBatchNo, artifactErr)
		}
		var ossID, downloadID int64
		if artifactErr == nil {
			ossID, downloadID, err = r.persistArtifact(ctx, tx, task, artifact, successCount, failCount, now)
			if err != nil {
				batchresult.RemoveLocalFile(artifact)
				return err
			}
		}
		status, runResult, failReason := parentBatchResult(successCount, failCount)
		if failCount > 0 {
			failReason = coalesce(task.FailureReason, failReason)
		}
		finalInfo := info.Map()
		finalInfo["processedCount"] = len(task.Details)
		finalInfo["successCount"] = successCount
		finalInfo["failCount"] = failCount
		startTime := resolveAllocateParentStartTime(parent.StartTime, aggregate.EarliestStartTime, now)
		data := do.BatchMain{StartTime: startTime, BatchAsyncStatus: status, RunResult: runResult, TotalCount: len(task.Details), ValidCount: len(task.Details) - failCount, SuccessCount: successCount, FailCount: failCount, FailReason: failReason, BatchInfo: mergeBatchInfo(gjson.New(finalInfo), ossID, downloadID, artifactErr != nil), EndTime: now, UpdatedBy: parent.CreatedBy, UpdatedAt: now}
		result, err := tx.Ctx(ctx).Model(dao.BatchMain.Table()).Where(mainColumns.Id, parent.Id).WhereIn(mainColumns.BatchAsyncStatus, []int{consts.BatchAsyncStatusPending, consts.BatchAsyncStatusProcessing}).Data(data).Update()
		if err != nil {
			if artifactErr == nil {
				batchresult.RemoveLocalFile(artifact)
			}
			return gerror.Wrap(err, "finalize allocate parent batch failed")
		}
		affected, err := result.RowsAffected()
		if err != nil {
			if artifactErr == nil {
				batchresult.RemoveLocalFile(artifact)
			}
			return err
		}
		if affected != 1 {
			if artifactErr == nil {
				batchresult.RemoveLocalFile(artifact)
			}
			return gerror.New("allocate parent batch is no longer processing")
		}
		if successCount > 0 {
			committedTask = task
		}
		return nil
	})
	if err != nil {
		if parentArtifactReady {
			batchresult.RemoveLocalFile(parentArtifact)
		}
		return err
	}
	if committedTask != nil {
		publishPackageCustomerSync(ctx, committedTask)
	}
	return nil
}

func resolveAllocateParentStartTime(parentStartTime, earliestChildStartTime, fallback *gtime.Time) *gtime.Time {
	if parentStartTime != nil && !parentStartTime.IsZero() {
		return parentStartTime
	}
	if earliestChildStartTime != nil && !earliestChildStartTime.IsZero() {
		return earliestChildStartTime
	}
	return fallback
}

func mergeAllocateParentInfo(info *gjson.Json, aggregate *allocateParentAggregate) *gjson.Json {
	data := make(map[string]any)
	if info != nil {
		for key, value := range info.Map() {
			data[key] = value
		}
	}
	completed := 0
	totalChunks := 0
	if aggregate != nil {
		completed = aggregate.FinishedCount
	}
	if info != nil {
		totalChunks = info.Get("chunkCount").Int()
	}
	if totalChunks == 0 && aggregate != nil {
		totalChunks = aggregate.ChildCount
	}
	data["completedChunkCount"] = completed
	data["chunkCount"] = totalChunks
	if aggregate != nil {
		data["processedCount"] = aggregate.ProcessedCount
		data["successCount"] = aggregate.SuccessCount
		data["failCount"] = aggregate.FailCount
	}
	return gjson.New(data)
}

func parentBatchResult(successCount, failCount int) (int, int, string) {
	if failCount > 0 || successCount == 0 {
		return consts.BatchAsyncStatusFailed, consts.BatchRunResultFailed, "allocate child batches failed"
	}
	return consts.BatchAsyncStatusSuccess, consts.BatchRunResultSuccess, ""
}

func normalizeAtomicParentResult(task *Task) (int, int) {
	successCount, failCount := countResult(task)
	if failCount == 0 {
		return successCount, 0
	}
	if task == nil {
		return 0, 0
	}
	task.FailureReason = coalesce(task.FailureReason, "划拨批次存在失败明细，整批未执行")
	markAtomicFailure(task, task.FailureReason)
	return 0, countValidationFailures(task)
}

func (r *Repository) updateDetails(ctx context.Context, tx gdb.TX, task *Task, now *gtime.Time) error {
	details := make([]*Detail, 0, len(task.Details))
	for _, detail := range task.Details {
		if detail == nil ||
			(detail.DetailStatus != consts.BatchDetailStatusPending &&
				detail.DetailStatus != consts.BatchDetailStatusRunning) {
			continue
		}
		details = append(details, detail)
	}
	for start := 0; start < len(details); start += allocateDetailUpdateBatchSize {
		end := min(start+allocateDetailUpdateBatchSize, len(details))
		if err := r.updateDetailChunk(ctx, tx, task, details[start:end], now); err != nil {
			return gerror.Wrapf(err, "批量更新划拨明细失败，范围:%d-%d", start+1, end)
		}
	}
	return nil
}

func (r *Repository) updateParentDetails(ctx context.Context, tx gdb.TX, task *Task, now *gtime.Time) error {
	detailsByBatch := make(map[string][]*Detail)
	for _, detail := range task.Details {
		if detail == nil {
			continue
		}
		batchNo := strings.TrimSpace(detail.BatchNo)
		if batchNo == "" {
			return gerror.New("allocate parent detail is missing child batch number")
		}
		detailsByBatch[batchNo] = append(detailsByBatch[batchNo], detail)
	}
	batchNos := make([]string, 0, len(detailsByBatch))
	for batchNo := range detailsByBatch {
		batchNos = append(batchNos, batchNo)
	}
	sort.Strings(batchNos)
	for _, batchNo := range batchNos {
		details := detailsByBatch[batchNo]
		sort.SliceStable(details, func(i, j int) bool {
			if details[i].ID != details[j].ID {
				return details[i].ID < details[j].ID
			}
			return details[i].RowNo < details[j].RowNo
		})
		batchTask := *task
		batchTask.Batch.BatchNo = batchNo
		for start := 0; start < len(details); start += allocateDetailUpdateBatchSize {
			end := min(start+allocateDetailUpdateBatchSize, len(details))
			if err := r.updateDetailChunk(ctx, tx, &batchTask, details[start:end], now); err != nil {
				return gerror.Wrapf(err, "update allocate child details failed, batchNo:%s, range:%d-%d", batchNo, start+1, end)
			}
		}
	}
	return nil
}

func (r *Repository) updateDetailChunk(ctx context.Context, tx gdb.TX, task *Task, details []*Detail, now *gtime.Time) error {
	if len(details) == 0 {
		return nil
	}
	statement, args := buildDetailUpdateStatement(task, details, now)
	result, err := tx.Ctx(ctx).Exec(statement, args...)
	if err != nil {
		return err
	}
	affected, err := result.RowsAffected()
	if err != nil {
		return err
	}
	if affected != int64(len(details)) {
		// MySQL 默认返回实际修改行数。子批次已写入相同结果时可能为零，
		// 必须在同一事务核对最终结果，不能据此放过缺失或越界明细。
		ids := make([]int64, 0, len(details))
		for _, detail := range details {
			ids = append(ids, detail.ID)
		}
		var persisted []*entity.BatchDetail
		columns := dao.BatchDetail.Columns()
		if err := tx.Ctx(ctx).Model(dao.BatchDetail.Table()).
			Where(columns.BatchNo, task.Batch.BatchNo).
			Where(columns.BatchType, consts.BatchTypeAllocate).
			Where(columns.TenantId, task.Batch.TenantID).
			WhereIn(columns.Id, ids).Scan(&persisted); err != nil {
			return gerror.Wrap(err, "核对划拨明细更新结果失败")
		}
		if err := validatePersistedDetailResults(task, details, persisted); err != nil {
			return gerror.Wrapf(err, "划拨明细更新未确认，预期:%d，实际修改:%d", len(details), affected)
		}
	}
	return nil
}

func validatePersistedDetailResults(task *Task, details []*Detail, persisted []*entity.BatchDetail) error {
	if len(persisted) != len(details) {
		return gerror.New("划拨明细缺失或不在当前批次租户范围内")
	}
	byID := make(map[int64]*entity.BatchDetail, len(persisted))
	for _, row := range persisted {
		if row == nil || row.BatchNo != task.Batch.BatchNo || row.BatchType != consts.BatchTypeAllocate || row.TenantId != task.Batch.TenantID {
			return gerror.New("划拨明细不在当前批次租户范围内")
		}
		byID[row.Id] = row
	}
	for _, detail := range details {
		row := byID[detail.ID]
		status := persistedAllocationDetailStatus(task, detail)
		if row == nil || row.DetailStatus != status || row.ErrorReason != detail.Reason || row.UpdatedBy != task.Batch.CreatedBy || row.Snapshot == nil || !sameDetailSnapshot(row.Snapshot, detail.Snapshot(task.Batch)) {
			return gerror.Newf("划拨明细最终结果不一致，明细ID:%d", detail.ID)
		}
		delete(byID, detail.ID)
	}
	return nil
}

func sameDetailSnapshot(actual, expected *gjson.Json) bool {
	if actual == nil || expected == nil {
		return actual == expected
	}
	actualJSON, actualErr := json.Marshal(actual.Map())
	expectedJSON, expectedErr := json.Marshal(expected.Map())
	return actualErr == nil && expectedErr == nil && string(actualJSON) == string(expectedJSON)
}

func buildDetailUpdateStatement(task *Task, details []*Detail, now *gtime.Time) (string, []any) {
	columns := dao.BatchDetail.Columns()
	var builder strings.Builder
	fmt.Fprintf(&builder, "UPDATE %s SET %s = CASE %s ", dao.BatchDetail.Table(), columns.DetailStatus, columns.Id)
	args := make([]any, 0, len(details)*7+9)
	for _, detail := range details {
		status := persistedAllocationDetailStatus(task, detail)
		builder.WriteString("WHEN ? THEN ? ")
		args = append(args, detail.ID, status)
	}
	fmt.Fprintf(&builder, "ELSE %s END, %s = CASE %s ", columns.DetailStatus, columns.ErrorReason, columns.Id)
	for _, detail := range details {
		builder.WriteString("WHEN ? THEN ? ")
		args = append(args, detail.ID, detail.Reason)
	}
	fmt.Fprintf(&builder, "ELSE %s END, %s = CASE %s ", columns.ErrorReason, columns.Snapshot, columns.Id)
	for _, detail := range details {
		builder.WriteString("WHEN ? THEN ? ")
		args = append(args, detail.ID, detail.Snapshot(task.Batch).String())
	}
	fmt.Fprintf(&builder, "ELSE %s END, %s = ?, %s = ? ", columns.Snapshot, columns.UpdatedBy, columns.UpdatedAt)
	args = append(args, task.Batch.CreatedBy, now)
	fmt.Fprintf(&builder, "WHERE %s = ? AND %s = ? AND %s = ? AND %s IN (?, ?, ?, ?, ?) AND %s IN (%s)",
		columns.BatchNo,
		columns.BatchType,
		columns.TenantId,
		columns.DetailStatus,
		columns.Id,
		strings.TrimSuffix(strings.Repeat("?,", len(details)), ","),
	)
	args = append(args,
		task.Batch.BatchNo,
		consts.BatchTypeAllocate,
		task.Batch.TenantID,
		consts.BatchDetailStatusPending,
		consts.BatchDetailStatusRunning,
		consts.BatchDetailStatusSuccess,
		consts.BatchDetailStatusFailed,
		consts.BatchDetailStatusNotExecuted,
	)
	for _, detail := range details {
		args = append(args, detail.ID)
	}
	return builder.String(), args
}

func (r *Repository) updateCardOwners(ctx context.Context, tx gdb.TX, task *Task, now *gtime.Time) error {
	groups := make(map[int64][]int64)
	totalCards := 0
	for _, detail := range task.Details {
		if detail == nil || !detail.Success() || detail.Card == nil || detail.Target == nil {
			continue
		}
		groups[detail.Target.DeptID] = append(groups[detail.Target.DeptID], detail.Card.ID)
		totalCards++
	}
	successCount, _ := countResult(task)
	if totalCards != successCount {
		return gerror.Newf("划拨成功明细缺少卡或目标客户快照，预期:%d，实际:%d", successCount, totalCards)
	}
	for _, cardIDs := range groups {
		sort.Slice(cardIDs, func(i, j int) bool { return cardIDs[i] < cardIDs[j] })
	}
	targetDeptIDs := make([]int64, 0, len(groups))
	for targetDeptID := range groups {
		targetDeptIDs = append(targetDeptIDs, targetDeptID)
	}
	sort.Slice(targetDeptIDs, func(i, j int) bool { return targetDeptIDs[i] < targetDeptIDs[j] })
	for _, targetDeptID := range targetDeptIDs {
		cardIDs := groups[targetDeptID]
		for start := 0; start < len(cardIDs); start += allocateCardUpdateBatchSize {
			end := min(start+allocateCardUpdateBatchSize, len(cardIDs))
			ids := cardIDs[start:end]
			result, err := tx.Ctx(ctx).Model(dao.TSimCardInfo.Table()).Safe().
				Where(dao.TSimCardInfo.Columns().TenantId, task.Batch.TenantID).
				Where(dao.TSimCardInfo.Columns().DeptId, task.Batch.CreatedDept).
				WhereIn(dao.TSimCardInfo.Columns().Id, ids).
				Data(do.TSimCardInfo{
					DeptId:            targetDeptID,
					OutboundBatchNo:   task.Batch.BatchNo,
					AllocationBatchNo: task.Batch.BatchNo,
					UpdatedBy:         task.Batch.CreatedBy,
					UpdatedAt:         now,
				}).Update()
			if err != nil {
				return gerror.Wrapf(err, "批量更新SIM卡划拨归属失败，目标客户:%d，范围:%d-%d", targetDeptID, start+1, end)
			}
			affected, err := result.RowsAffected()
			if err != nil {
				return err
			}
			if affected != int64(len(ids)) {
				return gerror.Newf("卡归属已变化，划拨批次整批回滚，目标客户:%d，预期:%d，实际:%d", targetDeptID, len(ids), affected)
			}
		}
	}
	return nil
}

func (r *Repository) commitLegacy(ctx context.Context, task *Task) error {
	if task == nil {
		return gerror.New("划拨任务不能为空")
	}
	now := gtime.Now()
	if err := r.database().Transaction(ctx, func(ctx context.Context, tx gdb.TX) error {
		for _, detail := range task.Details {
			if detail == nil {
				continue
			}
			if detail.Success() {
				if err := r.updateCardOwner(ctx, tx, task, detail, now); err != nil {
					return err
				}
			}
			if err := r.updateDetail(ctx, tx, task, detail, now); err != nil {
				return err
			}
		}
		return nil
	}); err != nil {
		return gerror.Wrap(err, "保存划拨明细结果失败")
	}

	successCount, failCount := countResult(task)
	artifact, artifactErr := r.createResultFile(ctx, task)
	if artifactErr != nil {
		g.Log().Warningf(ctx, "allocate result file generation failed, batchNo:%s, err:%+v", task.Batch.BatchNo, artifactErr)
	}

	err := r.database().Transaction(ctx, func(ctx context.Context, tx gdb.TX) error {
		var ossID, downloadID int64
		if artifactErr == nil {
			var err error
			ossID, downloadID, err = r.persistArtifact(ctx, tx, task, artifact, successCount, failCount, now)
			if err != nil {
				batchresult.RemoveLocalFile(artifact)
				return err
			}
		}
		return r.finalizeBatch(ctx, tx, task, successCount, failCount, ossID, downloadID, now, artifactErr != nil)
	})
	return gerror.Wrap(err, "汇总划拨批次失败")
}

func (r *Repository) updateCardOwner(ctx context.Context, tx gdb.TX, task *Task, detail *Detail, now *gtime.Time) error {
	if detail.Card == nil || detail.Target == nil {
		return gerror.New("划拨明细缺少卡或目标客户档案")
	}
	result, err := tx.Ctx(ctx).Model(dao.TSimCardInfo.Table()).Safe().
		Where(dao.TSimCardInfo.Columns().Id, detail.Card.ID).
		Where(dao.TSimCardInfo.Columns().TenantId, task.Batch.TenantID).
		Where(dao.TSimCardInfo.Columns().DeptId, task.Batch.CreatedDept).
		Data(do.TSimCardInfo{
			DeptId:            detail.Target.DeptID,
			OutboundBatchNo:   task.Batch.BatchNo,
			AllocationBatchNo: task.Batch.BatchNo,
			UpdatedBy:         task.Batch.CreatedBy,
			UpdatedAt:         now,
		}).Update()
	if err != nil {
		return gerror.Wrap(err, "更新SIM卡划拨归属失败")
	}
	affected, err := result.RowsAffected()
	if err != nil {
		return err
	}
	if affected != 1 {
		return gerror.New("卡归属已变化，划拨批次整批回滚")
	}
	return nil
}

func (r *Repository) updateDetail(ctx context.Context, tx gdb.TX, task *Task, detail *Detail, now *gtime.Time) error {
	if detail == nil ||
		(detail.DetailStatus != consts.BatchDetailStatusPending &&
			detail.DetailStatus != consts.BatchDetailStatusRunning) {
		return nil
	}
	status := persistedAllocationDetailStatus(task, detail)
	result, err := tx.Ctx(ctx).Model(dao.BatchDetail.Table()).
		Where(dao.BatchDetail.Columns().Id, detail.ID).
		Where(dao.BatchDetail.Columns().BatchNo, task.Batch.BatchNo).
		Where(dao.BatchDetail.Columns().BatchType, consts.BatchTypeAllocate).
		WhereIn(dao.BatchDetail.Columns().DetailStatus, []int{
			consts.BatchDetailStatusPending,
			consts.BatchDetailStatusRunning,
		}).
		Data(do.BatchDetail{
			DetailStatus: status,
			ErrorReason:  detail.Reason,
			Snapshot:     detail.Snapshot(task.Batch),
			UpdatedBy:    task.Batch.CreatedBy,
			UpdatedAt:    now,
		}).Update()
	if err != nil {
		return gerror.Wrap(err, "更新划拨批次明细失败")
	}
	affected, err := result.RowsAffected()
	if err != nil {
		return err
	}
	if affected != 1 {
		return gerror.Newf("划拨批次明细%d已不是运行中状态", detail.ID)
	}
	return nil
}

func (r *Repository) createResultFile(ctx context.Context, task *Task) (batchresult.Artifact, error) {
	if r.results == nil {
		return batchresult.Artifact{}, gerror.New("result generator is not configured")
	}
	return r.results.Create(ctx, batchresult.Request{
		BatchNo:   task.Batch.BatchNo,
		BatchType: task.Batch.BatchType,
		TenantID:  task.Batch.TenantID,
		Rows:      task.ResultRows(),
	})
}

func (r *Repository) persistArtifact(ctx context.Context, tx gdb.TX, task *Task, artifact batchresult.Artifact, successCount, failCount int, now *gtime.Time) (int64, int64, error) {
	ossID, err := tx.Ctx(ctx).Model(dao.SysOss.Table()).Data(do.SysOss{
		TenantId:     task.Batch.TenantID,
		FileName:     artifact.FileName,
		OriginalName: artifact.OriginalName,
		FileSuffix:   artifact.FileSuffix,
		Path:         artifact.Path,
		Url:          artifact.URL,
		CreatedDept:  task.Batch.CreatedDept,
		CreatedBy:    task.Batch.CreatedBy,
		UpdatedBy:    task.Batch.CreatedBy,
		Service:      "local",
		Md5:          artifact.MD5,
		FileSize:     artifact.Size,
	}).InsertAndGetId()
	if err != nil {
		return 0, 0, gerror.Wrap(err, "写入划拨结果文件OSS失败")
	}
	downloadID, err := tx.Ctx(ctx).Model(dao.TResourceBatchDownload.Table()).Data(do.TResourceBatchDownload{
		BatchNo:          task.Batch.BatchNo,
		TaskName:         batchresult.TaskName(task.Batch.BatchType),
		ResourceType:     consts.ResourceTypeExcel,
		ExportScene:      consts.BatchTypeAllocate,
		TotalCount:       len(task.Details),
		SuccessCount:     successCount,
		FailCount:        failCount,
		DownloadStatus:   consts.DownloadStatusGenerated,
		OssId:            ossID,
		FileName:         artifact.FileName,
		FileExt:          artifact.FileSuffix,
		FileUrl:          artifact.URL,
		FilePath:         artifact.Path,
		FileSize:         artifact.Size,
		FileMd5:          artifact.MD5,
		DownloadCount:    0,
		MaxDownloadCount: 0,
		DeptId:           task.Batch.CreatedDept,
		StartTime:        task.Batch.StartTime,
		FinishTime:       now,
		TenantId:         task.Batch.TenantID,
		CreatedDept:      task.Batch.CreatedDept,
		CreatedBy:        task.Batch.CreatedBy,
		UpdatedBy:        task.Batch.CreatedBy,
	}).InsertAndGetId()
	if err != nil {
		return 0, 0, gerror.Wrap(err, "写入划拨结果下载记录失败")
	}
	return ossID, downloadID, nil
}

func (r *Repository) finalizeBatch(ctx context.Context, tx gdb.TX, task *Task, successCount, failCount int, ossID, downloadID int64, now *gtime.Time, resultFileFailed bool) error {
	batchAsyncStatus, runResult := batchResult(successCount, failCount)
	result, err := tx.Ctx(ctx).Model(dao.BatchMain.Table()).
		Where(dao.BatchMain.Columns().BatchNo, task.Batch.BatchNo).
		Where(dao.BatchMain.Columns().BatchType, consts.BatchTypeAllocate).
		Where(dao.BatchMain.Columns().BatchAsyncStatus, consts.BatchAsyncStatusProcessing).
		Data(do.BatchMain{
			BatchAsyncStatus: batchAsyncStatus,
			RunResult:        runResult,
			TotalCount:       len(task.Details),
			ValidCount:       len(task.Details) - failCount,
			SuccessCount:     successCount,
			FailCount:        failCount,
			FailReason:       task.FailureReason,
			BatchInfo:        mergeBatchInfo(task.Batch.BatchInfo, ossID, downloadID, resultFileFailed),
			EndTime:          now,
			UpdatedBy:        task.Batch.CreatedBy,
		}).Update()
	if err != nil {
		return gerror.Wrap(err, "更新划拨批次主表失败")
	}
	affected, err := result.RowsAffected()
	if err != nil {
		return err
	}
	if affected != 1 {
		return gerror.New("划拨批次已不是处理中状态")
	}
	return nil
}

func mergeBatchInfo(info *gjson.Json, ossID, downloadID int64, resultFileFailed bool) *gjson.Json {
	data := make(map[string]any)
	if info != nil {
		for key, value := range info.Map() {
			data[key] = value
		}
	}
	if ossID > 0 {
		data["resultFileOssId"] = ossID
	}
	if resultFileFailed {
		data["resultFile"] = map[string]any{"status": "failed"}
	} else if downloadID > 0 {
		data["resultDownloadId"] = downloadID
		data["resultFile"] = map[string]any{"status": "generated"}
	} else {
		data["resultFile"] = map[string]any{"status": "skipped"}
	}
	return gjson.New(data)
}

func batchResult(successCount, failCount int) (int, int) {
	if failCount > 0 || successCount == 0 {
		return consts.BatchAsyncStatusFailed, consts.BatchRunResultFailed
	}
	return consts.BatchAsyncStatusSuccess, consts.BatchRunResultSuccess
}

func uniqueStrings(values []string) []string {
	seen := make(map[string]struct{}, len(values))
	result := make([]string, 0, len(values))
	for _, value := range values {
		value = strings.TrimSpace(value)
		if value == "" {
			continue
		}
		if _, exists := seen[value]; exists {
			continue
		}
		seen[value] = struct{}{}
		result = append(result, value)
	}
	return result
}