MiFaZhan
473 lines
15 KiB
Go
473 lines
15 KiB
Go
package data
|
||
|
||
import (
|
||
"context"
|
||
"sort"
|
||
"strconv"
|
||
"strings"
|
||
"sync"
|
||
|
||
"github.com/gogf/gf/v2/errors/gerror"
|
||
|
||
"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/batch"
|
||
"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/consts"
|
||
"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/model"
|
||
"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/model/entity"
|
||
)
|
||
|
||
type executor struct {
|
||
repository Repository
|
||
upstream Upstream
|
||
channelSlots sync.Map
|
||
}
|
||
|
||
func (e *executor) ClaimsProcessingReentry() bool { return true }
|
||
func (e *executor) CanExecuteProcessing(ctx context.Context, main *entity.BatchMain) (bool, error) {
|
||
repo, durable := e.repository.(SubBatchRepository)
|
||
_, async := e.upstream.(AsyncUpstream)
|
||
if !durable || !async {
|
||
return false, nil
|
||
}
|
||
return repo.HasDurableParent(ctx, strings.TrimSpace(main.BatchNo))
|
||
}
|
||
func (e *executor) ProtectExecutionFailure(ctx context.Context, batchNo string, batchType int, reason string) (bool, error) {
|
||
repo, durable := e.repository.(SubBatchRepository)
|
||
_, async := e.upstream.(AsyncUpstream)
|
||
if !durable || !async {
|
||
return false, nil
|
||
}
|
||
return true, repo.ProtectExecutionFailure(ctx, batchNo, reason)
|
||
}
|
||
|
||
func NewExecutor(repository Repository, upstream Upstream) Executor {
|
||
return &executor{repository: repository, upstream: upstream}
|
||
}
|
||
|
||
func (e *executor) Execute(ctx context.Context, batchNo string) error {
|
||
if durable, ok := e.repository.(SubBatchRepository); ok {
|
||
if _, ok := e.upstream.(AsyncUpstream); ok {
|
||
// Earlier child intents were durably frozen: a re-delivered parent cannot re-split running rows.
|
||
return batch.Retryable(durable.ResumeOrPrepare(ctx, strings.TrimSpace(batchNo), func(prepareCtx context.Context) error {
|
||
return e.executeFresh(prepareCtx, batchNo)
|
||
}))
|
||
}
|
||
}
|
||
return e.executeFresh(ctx, batchNo)
|
||
}
|
||
|
||
func (e *executor) executeFresh(ctx context.Context, batchNo string) error {
|
||
task, err := e.repository.Load(ctx, strings.TrimSpace(batchNo))
|
||
if err != nil {
|
||
return gerror.Wrap(err, "load data service batch")
|
||
}
|
||
if task.Operation.BatchType == consts.BatchTypeDataCut {
|
||
return e.executeTrafficCutoff(ctx, task)
|
||
}
|
||
if !validateTask(task) {
|
||
if err = e.repository.SaveValidationFailure(ctx, task); err != nil {
|
||
return gerror.Wrap(err, "save data service validation failure")
|
||
}
|
||
if _, durable := e.repository.(SubBatchRepository); durable {
|
||
if _, async := e.upstream.(AsyncUpstream); async {
|
||
return nil
|
||
}
|
||
}
|
||
return gerror.Wrap(e.repository.Finalize(ctx, task), "finalize data service validation failure")
|
||
}
|
||
|
||
return e.executeValidated(ctx, task)
|
||
}
|
||
|
||
func (e *executor) executeTrafficCutoff(ctx context.Context, task *Task) error {
|
||
if !validateTrafficCutoffTask(task) {
|
||
if err := e.repository.SaveValidationFailure(ctx, task); err != nil {
|
||
return gerror.Wrap(err, "save traffic cutoff validation failure")
|
||
}
|
||
if _, durable := e.repository.(SubBatchRepository); durable {
|
||
if _, async := e.upstream.(AsyncUpstream); async {
|
||
return nil
|
||
}
|
||
}
|
||
return gerror.Wrap(e.repository.Finalize(ctx, task), "finalize traffic cutoff validation failure")
|
||
}
|
||
if err := e.refreshDataServiceState(ctx, task); err != nil {
|
||
return gerror.Wrap(err, "refresh data service state")
|
||
}
|
||
if err := e.saveInvalidTrafficCutoffRows(ctx, task); err != nil {
|
||
return err
|
||
}
|
||
if _, durable := e.repository.(SubBatchRepository); durable {
|
||
if _, async := e.upstream.(AsyncUpstream); async {
|
||
// 本地disabled不能证明前序开通已终结；持久化路径统一经过卡级提交屏障。
|
||
return e.executeValidated(ctx, task)
|
||
}
|
||
}
|
||
if err := e.saveIdempotentCutoffSkips(ctx, task); err != nil {
|
||
return err
|
||
}
|
||
return e.executeValidated(ctx, task)
|
||
}
|
||
|
||
func (e *executor) executeValidated(ctx context.Context, task *Task) error {
|
||
if durable, ok := e.repository.(SubBatchRepository); ok {
|
||
if async, ok := e.upstream.(AsyncUpstream); ok {
|
||
seen := map[string]bool{}
|
||
for _, detail := range task.Details {
|
||
if detail == nil || detail.Reason != "" || (detail.DetailStatus != consts.BatchDetailStatusPending && detail.DetailStatus != consts.BatchDetailStatusRunning) {
|
||
continue
|
||
}
|
||
if seen[detail.ICCID] {
|
||
resolved, err := e.repository.SaveResult(ctx, task, detail, Outcome{Message: "重复ICCID，未提交上游"})
|
||
if err != nil {
|
||
return err
|
||
}
|
||
detail.ApplyOutcome(resolved)
|
||
continue
|
||
}
|
||
seen[detail.ICCID] = true
|
||
}
|
||
return e.prepareAndSubmit(ctx, task, durable, async)
|
||
}
|
||
}
|
||
groups := groupByChanKey(task.Operation, task.Details)
|
||
keys := make([]string, 0, len(groups))
|
||
for chanKey := range groups {
|
||
keys = append(keys, chanKey)
|
||
}
|
||
sort.Strings(keys)
|
||
for _, chanKey := range keys {
|
||
if err := e.changeGroup(ctx, task, groups[chanKey]); err != nil {
|
||
return err
|
||
}
|
||
}
|
||
return gerror.Wrap(e.repository.Finalize(ctx, task), "finalize data service batch")
|
||
}
|
||
|
||
type dataServiceStateRefresher interface {
|
||
RefreshDataServiceState(ctx context.Context, task *Task) error
|
||
}
|
||
|
||
func (e *executor) refreshDataServiceState(ctx context.Context, task *Task) error {
|
||
refresher, ok := e.repository.(dataServiceStateRefresher)
|
||
if !ok {
|
||
return nil
|
||
}
|
||
return refresher.RefreshDataServiceState(ctx, task)
|
||
}
|
||
|
||
func (e *executor) saveIdempotentCutoffSkips(ctx context.Context, task *Task) error {
|
||
if task == nil || task.Operation.BatchType != consts.BatchTypeDataCut {
|
||
return nil
|
||
}
|
||
for _, detail := range task.Details {
|
||
if !skipUpstreamForIdempotentCutoff(task.Operation, detail) {
|
||
continue
|
||
}
|
||
resolved, saveErr := e.repository.SaveResult(ctx, task, detail, Outcome{
|
||
Success: true,
|
||
Message: "数据服务已关闭，未发起上游请求",
|
||
})
|
||
if saveErr != nil {
|
||
return gerror.Wrapf(saveErr, "save data service detail %d idempotent skip result", detail.ID)
|
||
}
|
||
detail.ApplyOutcome(resolved)
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func validateTask(task *Task) bool {
|
||
if task == nil || len(task.Details) == 0 {
|
||
if task != nil && task.FailureReason == "" {
|
||
task.FailureReason = "数据开停批次没有有效明细"
|
||
}
|
||
return false
|
||
}
|
||
hasFailure := false
|
||
for _, detail := range task.Details {
|
||
if detail == nil {
|
||
hasFailure = true
|
||
continue
|
||
}
|
||
validateDetail(task.Operation, detail)
|
||
if detail.Reason != "" {
|
||
hasFailure = true
|
||
}
|
||
}
|
||
if !hasFailure {
|
||
task.ValidationOK = true
|
||
for _, detail := range task.Details {
|
||
if detail != nil {
|
||
detail.ValidationPassed = true
|
||
}
|
||
}
|
||
return true
|
||
}
|
||
for _, detail := range task.Details {
|
||
if detail != nil && detail.Reason == "" {
|
||
detail.Reason = "同批次存在校验失败卡，本卡未请求上游"
|
||
}
|
||
}
|
||
task.FailureReason = "批次校验未通过，未发起上游请求"
|
||
return false
|
||
}
|
||
|
||
func validateTrafficCutoffTask(task *Task) bool {
|
||
if task == nil || len(task.Details) == 0 {
|
||
if task != nil {
|
||
task.FailureReason = "流量断网批次无有效明细"
|
||
}
|
||
return false
|
||
}
|
||
hasSuccess := false
|
||
for _, detail := range task.Details {
|
||
if detail == nil {
|
||
continue
|
||
}
|
||
validateDetail(task.Operation, detail)
|
||
detail.ValidationPassed = detail.Reason == ""
|
||
if detail.ValidationPassed {
|
||
hasSuccess = true
|
||
}
|
||
}
|
||
if hasSuccess {
|
||
task.ValidationOK = true
|
||
if failed := countFailedDetails(task); failed > 0 {
|
||
task.FailureReason = strconv.Itoa(failed) + " 张卡未通过校验"
|
||
} else {
|
||
task.FailureReason = ""
|
||
}
|
||
return true
|
||
}
|
||
task.FailureReason = "流量断网批次校验未通过，未发起上游请求"
|
||
return false
|
||
}
|
||
|
||
func countFailedDetails(task *Task) int {
|
||
if task == nil {
|
||
return 0
|
||
}
|
||
count := 0
|
||
for _, detail := range task.Details {
|
||
if detail == nil || detail.Reason != "" {
|
||
count++
|
||
}
|
||
}
|
||
return count
|
||
}
|
||
|
||
func validateDetail(_ Operation, detail *Detail) {
|
||
if detail.Reason != "" {
|
||
return
|
||
}
|
||
if detail.Card == nil {
|
||
detail.Reason = "ICCID不存在或不属于当前租户"
|
||
return
|
||
}
|
||
detail.MSISDN = strings.TrimSpace(detail.Card.MSISDN)
|
||
detail.IMSI = strings.TrimSpace(detail.Card.IMSI)
|
||
detail.ChanKey = strings.TrimSpace(detail.Card.ChanKey)
|
||
detail.StatusBefore = detail.Card.DataServiceStatus
|
||
detail.StatusAfter = detail.Card.DataServiceStatus
|
||
|
||
if !detail.Authorized {
|
||
detail.Reason = "ICCID不在当前操作员授权范围内"
|
||
return
|
||
}
|
||
var reasons []string
|
||
if detail.ChanKey == "" {
|
||
reasons = append(reasons, "ICCID缺少通道标识")
|
||
}
|
||
if detail.InputMSISDN != "" && detail.InputMSISDN != detail.MSISDN {
|
||
reasons = append(reasons, "MSISDN与卡档案不一致")
|
||
}
|
||
if detail.InputIMSI != "" && detail.InputIMSI != detail.IMSI {
|
||
reasons = append(reasons, "IMSI与卡档案不一致")
|
||
}
|
||
detail.Reason = strings.Join(reasons, "；")
|
||
}
|
||
|
||
func groupByChanKey(operation Operation, details []*Detail) map[string][]*Detail {
|
||
groups := make(map[string][]*Detail)
|
||
for _, detail := range details {
|
||
if detail == nil || detail.Reason != "" {
|
||
continue
|
||
}
|
||
if operation.BatchType == consts.BatchTypeDataCut && (detail == nil || !detail.ValidationPassed || detail.Reason != "") {
|
||
continue
|
||
}
|
||
if skipUpstreamForIdempotentCutoff(operation, detail) {
|
||
continue
|
||
}
|
||
groups[detail.ChanKey] = append(groups[detail.ChanKey], detail)
|
||
}
|
||
return groups
|
||
}
|
||
|
||
func (e *executor) saveInvalidTrafficCutoffRows(ctx context.Context, task *Task) error {
|
||
if task == nil {
|
||
return nil
|
||
}
|
||
for _, detail := range task.Details {
|
||
if detail == nil || detail.ValidationPassed || detail.Reason == "" {
|
||
continue
|
||
}
|
||
resolved, saveErr := e.repository.SaveResult(ctx, task, detail, Outcome{
|
||
Message: detail.Reason,
|
||
})
|
||
if saveErr != nil {
|
||
return gerror.Wrapf(saveErr, "save traffic cutoff detail %d validation result", detail.ID)
|
||
}
|
||
detail.ApplyOutcome(resolved)
|
||
}
|
||
return nil
|
||
}
|
||
|
||
func skipUpstreamForIdempotentCutoff(operation Operation, detail *Detail) bool {
|
||
return operation.BatchType == consts.BatchTypeDataCut &&
|
||
detail != nil &&
|
||
detail.ValidationPassed &&
|
||
detail.StatusBefore == consts.SimDataServiceStatusDisabled &&
|
||
detail.StatusAfter == consts.SimDataServiceStatusDisabled
|
||
}
|
||
|
||
// changeGroup 把同一通道下的明细切成子批次并发调用上游，并在**每个子批次拿到结果后立即落库该子批的明细**。
|
||
//
|
||
// 不再等所有子批次跑完才统一写库：大批量（如 2 万张 = 200 个子批）执行时，执行日志可以按子批
|
||
// 实时反映进度，而不是在整个批次结束前始终全部停留在"执行中"。子批之间相互隔离——某个子批
|
||
// 落库失败只记录错误、不影响其他子批继续办理与落库。
|
||
func (e *executor) changeGroup(ctx context.Context, task *Task, details []*Detail) error {
|
||
subBatches := splitDetails(details, consts.BatchSubBatchSize)
|
||
|
||
sem := make(chan struct{}, consts.BatchSubBatchConcurrency)
|
||
var wg sync.WaitGroup
|
||
var mu sync.Mutex
|
||
var firstErr error
|
||
|
||
for _, sub := range subBatches {
|
||
wg.Add(1)
|
||
sem <- struct{}{}
|
||
go func(batch []*Detail) {
|
||
defer wg.Done()
|
||
defer func() { <-sem }()
|
||
if err := e.changeAndSaveSubBatch(ctx, task, batch); err != nil {
|
||
mu.Lock()
|
||
if firstErr == nil {
|
||
firstErr = err
|
||
}
|
||
mu.Unlock()
|
||
}
|
||
}(sub)
|
||
}
|
||
wg.Wait()
|
||
|
||
return firstErr
|
||
}
|
||
|
||
// changeAndSaveSubBatch 处理单个子批次：构建请求、调用上游，然后逐条落库本子批的明细。
|
||
// 明细结果只在所属子批次落库，天然做到"查到一个子批就落库一个子批"。
|
||
func (e *executor) changeAndSaveSubBatch(ctx context.Context, task *Task, details []*Detail) error {
|
||
for index, outcome := range e.changeSubBatch(ctx, task, details) {
|
||
detail := details[index]
|
||
resolved, saveErr := e.repository.SaveResult(ctx, task, detail, outcome)
|
||
if saveErr != nil {
|
||
return gerror.Wrapf(saveErr, "save data service detail %d result", detail.ID)
|
||
}
|
||
detail.ApplyOutcome(resolved)
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// changeSubBatch 处理单个子批次：构建请求并调用上游，返回与该子批次明细等长的 Outcome 切片。
|
||
func (e *executor) changeSubBatch(ctx context.Context, task *Task, details []*Detail) []Outcome {
|
||
input := &model.SimCardApnChangeInput{
|
||
SimKeys: make([]*model.SimKey, 0, len(details)),
|
||
ReqData: model.SimCardApnChangeInputItem{OperType: task.Operation.ApnOperType, ApnName: task.ApnName},
|
||
}
|
||
for _, detail := range details {
|
||
input.SimKeys = append(input.SimKeys, &model.SimKey{ChanKey: detail.ChanKey, ICCID: detail.ICCID, MSISDN: detail.MSISDN, IMSI: detail.IMSI})
|
||
}
|
||
response, err := e.upstream.ChangeSimApnFunction(ctx, input)
|
||
if err != nil {
|
||
outcomes := make([]Outcome, len(details))
|
||
for index := range outcomes {
|
||
outcomes[index].Message = "请求上游失败：" + err.Error()
|
||
}
|
||
return outcomes
|
||
}
|
||
return mapOutcomes(details, response)
|
||
}
|
||
|
||
// splitDetails 按固定大小把明细切成有序的子批次；空输入返回空切片。
|
||
func splitDetails(details []*Detail, size int) [][]*Detail {
|
||
if size <= 0 {
|
||
size = 1
|
||
}
|
||
batches := make([][]*Detail, 0, (len(details)+size-1)/size)
|
||
for i := 0; i < len(details); i += size {
|
||
end := i + size
|
||
if end > len(details) {
|
||
end = len(details)
|
||
}
|
||
batches = append(batches, details[i:end])
|
||
}
|
||
return batches
|
||
}
|
||
|
||
func mapOutcomes(details []*Detail, response []model.SimBaseRes[model.SimCardApnChangeOutputItem]) []Outcome {
|
||
outcomes := make([]Outcome, len(details))
|
||
used := make([]bool, len(response))
|
||
for detailIndex, detail := range details {
|
||
matched := -1
|
||
for responseIndex, item := range response {
|
||
if !used[responseIndex] && sameSim(detail, item.SimKey) {
|
||
matched = responseIndex
|
||
break
|
||
}
|
||
}
|
||
if matched < 0 && detailIndex < len(response) && !used[detailIndex] && response[detailIndex].SimKey == nil {
|
||
matched = detailIndex
|
||
}
|
||
if matched < 0 {
|
||
outcomes[detailIndex] = Outcome{Message: "上游未返回该卡处理结果"}
|
||
continue
|
||
}
|
||
used[matched] = true
|
||
item := response[matched]
|
||
message := strings.TrimSpace(item.Message)
|
||
if item.Code != 0 && message == "" {
|
||
message = "上游处理失败"
|
||
}
|
||
outcomes[detailIndex] = Outcome{Success: item.Code == 0, Code: item.Code, Message: message, OrderNum: strings.TrimSpace(item.ResData.OrderNum)}
|
||
}
|
||
return outcomes
|
||
}
|
||
|
||
func sameSim(detail *Detail, key *model.SimKey) bool {
|
||
if detail == nil || key == nil {
|
||
return false
|
||
}
|
||
if value := strings.TrimSpace(key.ICCID); value != "" {
|
||
return value == detail.ICCID
|
||
}
|
||
if value := strings.TrimSpace(key.MSISDN); value != "" {
|
||
return value == detail.MSISDN
|
||
}
|
||
if value := strings.TrimSpace(key.IMSI); value != "" {
|
||
return value == detail.IMSI
|
||
}
|
||
return false
|
||
}
|
||
|
||
func OperationByBatchType(batchType int) (Operation, error) {
|
||
switch batchType {
|
||
case consts.BatchTypeDataStart:
|
||
return Operation{BatchType: batchType, Name: "dataStart", ApnOperType: consts.SimApnOperTypeOpen, TargetStatus: consts.SimDataServiceStatusEnabled}, nil
|
||
case consts.BatchTypeDataCut:
|
||
return Operation{BatchType: batchType, Name: "trafficCutoff", ApnOperType: consts.SimApnOperTypeStop, TargetStatus: consts.SimDataServiceStatusDisabled}, nil
|
||
case consts.BatchTypeDataStop:
|
||
return Operation{BatchType: batchType, Name: "dataStop", ApnOperType: consts.SimApnOperTypeStop, TargetStatus: consts.SimDataServiceStatusDisabled}, nil
|
||
default:
|
||
return Operation{}, gerror.Newf("unsupported data service batch type %d", batchType)
|
||
}
|
||
}
|