iotlcm/worker-svr/internal/service/renewchange/change.go

package renewchange

import (
	"context"
	"fmt"
	"strings"
	"time"

	"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"
	rc "git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/repository/renewchange"
	"git.5iot.com/iotlcm-public/iotlcm-worker-svr/internal/service/packagecustomer"
)

// ChangeExecutor executes a confirmed batch_type=102 change as one business
// transaction. The transaction includes old subscription cleanup, the new
// order/subscriptions, card reset, detail success, and batch success.
type ChangeExecutor struct {
	repo    *Repository
	nowFunc NowFunc
}

func NewChangeExecutor(repo *Repository) *ChangeExecutor {
	if repo == nil {
		repo = NewRepository()
	}
	return &ChangeExecutor{repo: repo, nowFunc: timeNow}
}

func (c *ChangeExecutor) SetNowFunc(now NowFunc) { c.nowFunc = now }

func (c *ChangeExecutor) Execute(ctx context.Context, batchNo string) error {
	batchNo = strings.TrimSpace(batchNo)
	if batchNo == "" {
		return gerror.New("change batch number is empty")
	}
	main, err := c.repo.LoadProcessingBatch(ctx, batchNo, consts.BatchTypeChange)
	if err != nil {
		return err
	}
	details, err := c.repo.LoadRunningDetails(ctx, batchNo, consts.BatchTypeChange)
	if err != nil {
		return gerror.Wrap(err, "load change details failed")
	}
	if len(details) == 0 {
		return c.finalizeFailure(ctx, main, "change batch has no runnable details")
	}
	contract, err := ParseExecutionContract(main)
	if err != nil {
		return c.finalizeFailure(ctx, main, err.Error())
	}
	if contract.BatchType != consts.BatchTypeChange {
		return c.finalizeFailure(ctx, main, "change execution contract has invalid batch type")
	}

	now := pnow(c.nowFunc)
	orderSN, skipped, err := c.apply(ctx, main, contract, details, now)
	if err != nil {
		return c.finalizeFailure(ctx, main, err.Error())
	}
	if skipped {
		return nil
	}
	packagecustomer.PublishSyncForDeptsSafe(ctx, main.TenantId, main.CreatedBy, []int64{contract.CustomerAgentID}, "change", main.BatchNo)

	rows, snapshotErr := c.repo.ResolveResultSnapshot(ctx, main.BatchNo, main.BatchType)
	if snapshotErr == nil {
		_ = c.repo.WriteResultFileSafe(ctx, main.BatchNo, main.BatchType, main.TenantId, main.CreatedDept, main.CreatedBy, rows)
	}
	g.Log().Infof(ctx, "change batch success, batchNo:%s, orderSn:%s, tenant:%s", main.BatchNo, orderSN, main.TenantId)
	return nil
}

// apply is deliberately the only success transaction. A duplicate delivery
// that reaches this method after a completed transaction is safely ignored.
func (c *ChangeExecutor) apply(ctx context.Context, main *rc.BatchMainRow, contract *ExecutionContract, details []*rc.BatchDetailRow, now time.Time) (orderSN string, skipped bool, err error) {
	iccids := uniqueIccidsFromDetails(details)
	if len(iccids) != len(details) {
		return "", false, gerror.New("change batch contains duplicate or empty ICCID")
	}
	if contract.Quantity != len(iccids) {
		return "", false, gerror.Newf("change contract quantity mismatch: contract=%d actual=%d", contract.Quantity, len(iccids))
	}

	err = c.repo.DB().Transaction(ctx, func(ctx context.Context, tx gdb.TX) error {
		lockedMain, lockErr := c.repo.Tables().LockChangeBatch(ctx, tx, main.BatchNo)
		if lockErr != nil {
			return lockErr
		}
		if lockedMain.BatchAsyncStatus != consts.BatchAsyncStatusProcessing {
			skipped = true
			return nil
		}
		lockedDetails, lockErr := c.repo.Tables().LockChangeDetails(ctx, tx, main.BatchNo)
		if lockErr != nil {
			return lockErr
		}
		if err = validateRunningChangeDetails(lockedDetails, details); err != nil {
			return err
		}

		cards, lockErr := c.repo.Tables().LockCardsByIccids(ctx, tx, iccids)
		if lockErr != nil {
			return lockErr
		}
		if len(cards) != len(iccids) {
			return gerror.New("some change cards no longer exist")
		}
		bindRows, lockErr := c.repo.Tables().LockChangeBinds(ctx, tx, iccids)
		if lockErr != nil {
			return lockErr
		}
		currentBinds, futureBinds, lockErr := resolveCurrentSubscriptions(bindRows, iccids)
		if lockErr != nil {
			return lockErr
		}

		oldPackageCode := readChangeOldPackageCode(main.BatchInfo)
		if oldPackageCode == "" {
			return gerror.New("change contract old package is missing")
		}
		newPackage, loadErr := c.repo.Tables().LoadChangePackage(ctx, tx, contract.TenantID, contract.PackageCode)
		if loadErr != nil {
			return loadErr
		}
		oldPackage, loadErr := c.repo.Tables().LoadChangePackage(ctx, tx, contract.TenantID, oldPackageCode)
		if loadErr != nil {
			return loadErr
		}
		if newPackage.Enabled != 1 || oldPackage.Enabled != 1 {
			return gerror.New("change package is disabled")
		}
		if newPackage.PackageType != oldPackage.PackageType {
			return gerror.New("new package type differs from current package")
		}
		if !containsPackageCode(oldPackage.ChangeablePackageCodes, contract.PackageCode) {
			return gerror.New("new package is not in current package changeable range")
		}
		durationOptions, durationErr := changeDurationOptions(newPackage.RenewableDurationConfig)
		if durationErr != nil {
			return gerror.Wrap(durationErr, "load target package change duration options failed")
		}
		if !rc.ContainsInt(durationOptions, contract.DurationMonths) {
			return gerror.Newf("change duration %d months is not allowed for target package", contract.DurationMonths)
		}
		customer, loadErr := c.repo.Tables().LoadChangeDept(ctx, tx, contract.TenantID, contract.CustomerAgentID)
		if loadErr != nil {
			return loadErr
		}
		if customer.DeptId != contract.SellerAgentID && customer.ParentId != contract.SellerAgentID {
			return gerror.New("change customer is outside the operator scope")
		}

		sourceBatchNo := contract.SourceBatchNo
		for _, iccid := range iccids {
			card := cards[iccid]
			current := currentBinds[iccid]
			var expectedMSISDN, expectedIMSI string
			if detail := findDetailByICCID(lockedDetails, iccid); detail != nil && detail.Snapshot != nil {
				expectedMSISDN = strings.TrimSpace(detail.Snapshot.Get("msisdn").String())
				expectedIMSI = strings.TrimSpace(detail.Snapshot.Get("imsi").String())
			}
			if err = validateChangeCard(card, current, contract, iccid, oldPackageCode, sourceBatchNo, expectedMSISDN, expectedIMSI); err != nil {
				return gerror.Wrapf(err, "validate change card failed, iccid:%s", iccid)
			}
			for _, future := range futureBinds[iccid] {
				if future == nil || future.TenantId != contract.TenantID || future.DeptId != contract.CustomerAgentID {
					return gerror.Newf("future change subscription is inconsistent: %s", iccid)
				}
			}
		}

		currentList := make([]*rc.SimCardBindPkg, 0, len(iccids))
		currentOrderCodes := make([]string, 0, len(iccids))
		currentOrderQuantities := make(map[string]int)
		for _, iccid := range iccids {
			currentList = append(currentList, currentBinds[iccid])
			if orderCode := strings.TrimSpace(currentBinds[iccid].OrderCode); orderCode != "" {
				currentOrderCodes = append(currentOrderCodes, orderCode)
				currentOrderQuantities[orderCode]++
			}
		}
		if err = c.repo.Tables().ArchiveChangeBinds(ctx, tx, currentList, consts.EffectiveStatusTerminated, gtime.NewFromTime(now), main.BatchNo, contract.OperatorUserID); err != nil {
			return err
		}
		currentSubscriptionNos := subscriptionNos(currentList)
		if affected, deleteErr := c.repo.Tables().DeleteChangeBinds(ctx, tx, currentSubscriptionNos); deleteErr != nil {
			return deleteErr
		} else if int(affected) != len(currentSubscriptionNos) {
			return gerror.New("current change subscriptions changed concurrently")
		}
		if err = c.repo.Tables().UpdateChangeOrdersAfterRemoval(ctx, tx, currentOrderCodes, main.BatchNo, "lastUnsubscribe", currentOrderQuantities, gtime.NewFromTime(now), contract.OperatorUserID); err != nil {
			return err
		}

		futureList := flattenFutureBinds(futureBinds, iccids)
		futureOrderCodes := make([]string, 0, len(futureList))
		futureOrderQuantities := make(map[string]int)
		for _, bind := range futureList {
			if orderCode := strings.TrimSpace(bind.OrderCode); orderCode != "" {
				futureOrderCodes = append(futureOrderCodes, orderCode)
				futureOrderQuantities[orderCode]++
			}
		}
		if err = c.repo.Tables().ArchiveChangeBinds(ctx, tx, futureList, consts.EffectiveStatusVoided, gtime.NewFromTime(now), main.BatchNo, contract.OperatorUserID); err != nil {
			return err
		}
		futureSubscriptionNos := subscriptionNos(futureList)
		if affected, deleteErr := c.repo.Tables().DeleteChangeBinds(ctx, tx, futureSubscriptionNos); deleteErr != nil {
			return deleteErr
		} else if int(affected) != len(futureSubscriptionNos) {
			return gerror.New("future change subscriptions changed concurrently")
		}
		if err = c.repo.Tables().UpdateChangeOrdersAfterRemoval(ctx, tx, futureOrderCodes, main.BatchNo, "lastChange", futureOrderQuantities, gtime.NewFromTime(now), contract.OperatorUserID); err != nil {
			return err
		}

		beginAt, resolveErr := resolveChangeBeginAt(contract, now)
		if resolveErr != nil {
			return resolveErr
		}
		endAt := contract.ServiceEnd(beginAt)
		packageStatus := calculateChangePackageStatus(now, beginAt, endAt)
		effectiveStatus := calculateChangeEffectiveStatus(now, beginAt, endAt)
		orderSN, err = rc.NextOrderSN(now)
		if err != nil {
			return err
		}
		order, buildErr := BuildOrderDO(contract, orderSN, now)
		if buildErr != nil {
			return buildErr
		}
		orderID, insertErr := c.repo.Tables().InsertContractOrder(ctx, tx, order)
		if insertErr != nil {
			return gerror.Wrap(insertErr, "insert change order failed")
		}
		item, buildErr := BuildOrderItemDO(contract, OrderProductContract{ID: newPackage.Id, Code: newPackage.PackageCode, Name: newPackage.PackageName, Type: newPackage.PackageType}, orderID, orderSN, beginAt, endAt, now)
		if buildErr != nil {
			return buildErr
		}
		if err = c.repo.Tables().InsertContractOrderItem(ctx, tx, item); err != nil {
			return gerror.Wrap(err, "insert change order item failed")
		}

		snapshots := make(map[int]*gjson.Json, len(lockedDetails))
		for _, detail := range lockedDetails {
			iccid := strings.TrimSpace(detail.ItemId)
			card := cards[iccid]
			newSubscriptionNo := rc.GenerateSubscriptionNo("SUB")
			newBind := buildChangeBind(contract, newPackage, card, main.BatchNo, oldPackageCode, orderSN, newSubscriptionNo, effectiveStatus, beginAt, endAt, now)
			if err = c.repo.Tables().InsertBind(ctx, tx, newBind); err != nil {
				return gerror.Wrap(err, "insert change subscription failed")
			}
			if err = c.repo.Tables().UpdateChangeCard(ctx, tx, iccid, newPackage.PackageCode, packageStatus, newPackage.FlowMb, gtime.NewFromTime(beginAt), gtime.NewFromTime(endAt), contract.OperatorUserID); err != nil {
				return err
			}
			snapshots[detail.RowNo] = buildChangeDetailSnapshot(detail, card, currentBinds[iccid], newSubscriptionNo, futureBinds[iccid], contract, oldPackageCode, orderSN, beginAt, endAt)
		}
		if err = c.repo.Tables().MarkChangeDetailsSuccess(ctx, tx, main.BatchNo, lockedDetails, snapshots, contract.OperatorUserID); err != nil {
			return err
		}
		info := mergeBatchInfo(main.BatchInfo, map[string]any{
			"orderSn":            orderSN,
			"oldPackageCode":     oldPackageCode,
			"newPackageCode":     newPackage.PackageCode,
			"newPackageName":     newPackage.PackageName,
			"durationMonths":     contract.DurationMonths,
			"giftDurationMonths": contract.GiftDurationMonths,
			"effectiveMode":      contract.EffectiveMode,
			"packageBeginTime":   formatTime(gtime.NewFromTime(beginAt)),
			"packageEndTime":     formatTime(gtime.NewFromTime(endAt)),
			"amount":             contract.TotalAmount(),
			"changeTime":         formatTime(gtime.NewFromTime(now)),
		})
		return c.repo.UpdateBatchMainInfo(ctx, tx, main.BatchNo, main.BatchType, buildChangeSuccessBatchMain(main, contract, gjson.New(info), orderSN, lockedMain.StartTime, now, len(lockedDetails)))
	})
	return orderSN, skipped, err
}

func buildChangeBind(contract *ExecutionContract, pkg *rc.FlowPackageInfo, card *rc.SimCardInfo, batchNo, oldPackageCode, orderSN, subscriptionNo string, effectiveStatus int, beginAt, endAt, now time.Time) *rc.SimCardBindPkg {
	return &rc.SimCardBindPkg{
		BatchNo:             batchNo,
		Iccid:               card.Iccid,
		Msisdn:              card.Msisdn,
		DeptId:              contract.CustomerAgentID,
		PackageType:         pkg.PackageType,
		PackageCode:         pkg.PackageCode,
		PreviousPackageCode: oldPackageCode,
		BusinessType:        consts.PackageBusinessTypeChange,
		EffectiveStatus:     effectiveStatus,
		HandleTime:          gtime.NewFromTime(now),
		EffectiveTime:       gtime.NewFromTime(beginAt),
		ExpectedExpireTime:  gtime.NewFromTime(endAt),
		DurationMonths:      contract.DurationMonths,
		GiftDurationMonths:  contract.GiftDurationMonths,
		OrderCode:           orderSN,
		SubscriptionNo:      subscriptionNo,
		Remark:              "卡变更套餐自动生成",
		TenantId:            contract.TenantID,
		CreatedDept:         contract.CreatedDeptID,
		CreatedBy:           contract.OperatorUserID,
		CreatedAt:           gtime.NewFromTime(now),
		UpdatedBy:           contract.OperatorUserID,
		UpdatedAt:           gtime.NewFromTime(now),
	}
}

func buildChangeSuccessBatchMain(main *rc.BatchMainRow, contract *ExecutionContract, info *gjson.Json, orderSN string, startTime *gtime.Time, now time.Time, validCount int) *rc.BatchMainRow {
	return &rc.BatchMainRow{
		BatchType:        main.BatchType,
		BatchAsyncStatus: consts.BatchAsyncStatusSuccess,
		RunResult:        consts.BatchRunResultSuccess,
		TotalCount:       main.TotalCount,
		ValidCount:       validCount,
		SuccessCount:     validCount,
		FailCount:        0,
		BatchInfo:        info,
		OrderSn:          orderSN,
		StartTime:        startTime,
		EndTime:          gtime.NewFromTime(now),
		TargetDeptId:     contract.CustomerAgentID,
		UpdatedBy:        contract.OperatorUserID,
	}
}

func (c *ChangeExecutor) finalizeFailure(ctx context.Context, main *rc.BatchMainRow, reason string) error {
	now := pnow(c.nowFunc)
	err := c.repo.DB().Transaction(ctx, func(ctx context.Context, tx gdb.TX) error {
		if err := c.repo.MarkAllDetailsFailed(ctx, tx, main.BatchNo, main.BatchType, reason, main.CreatedBy); err != nil {
			return err
		}
		return c.repo.FinalizeBatchMain(ctx, tx, main.BatchNo, main.BatchType, &rc.BatchMainRow{
			BatchAsyncStatus: consts.BatchAsyncStatusFailed,
			RunResult:        consts.BatchRunResultFailed,
			FailReason:       reason,
			TotalCount:       main.TotalCount,
			FailCount:        main.TotalCount,
			EndTime:          gtime.NewFromTime(now),
			UpdatedBy:        main.CreatedBy,
		})
	})
	if err != nil {
		return gerror.Wrap(err, "finalize change failure failed")
	}
	g.Log().Warningf(ctx, "change batch failed, batchNo:%s, reason:%s, tenant:%s", main.BatchNo, reason, main.TenantId)
	return nil
}

func validateRunningChangeDetails(locked, loaded []*rc.BatchDetailRow) error {
	if len(locked) != len(loaded) {
		return gerror.New("change detail count changed during execution")
	}
	for _, detail := range locked {
		if detail == nil || detail.DetailStatus != consts.BatchDetailStatusRunning {
			return gerror.New("change detail is no longer running")
		}
	}
	return nil
}

func validateChangeCard(card *rc.SimCardInfo, bind *rc.SimCardBindPkg, contract *ExecutionContract, iccid, oldPackageCode, sourceBatchNo, expectedMSISDN, expectedIMSI string) error {
	if contract == nil {
		return gerror.New("change execution contract is missing")
	}
	return validateCurrentSubscription(card, bind, subscriptionValidation{
		ICCID:          iccid,
		MSISDN:         expectedMSISDN,
		IMSI:           expectedIMSI,
		TenantID:       contract.TenantID,
		CustomerDeptID: contract.CustomerAgentID,
		PackageCode:    oldPackageCode,
		SourceBatchNo:  sourceBatchNo,
	})
}

func findDetailByICCID(details []*rc.BatchDetailRow, iccid string) *rc.BatchDetailRow {
	for _, detail := range details {
		if detail != nil && strings.TrimSpace(detail.ItemId) == strings.TrimSpace(iccid) {
			return detail
		}
	}
	return nil
}

func readChangeOldPackageCode(info *gjson.Json) string {
	return firstString(info, "oldPackageCode", "changePrepareSnapshot.oldPackageCode")
}

func containsPackageCode(values *gjson.Json, target string) bool {
	if values == nil || values.IsNil() {
		return false
	}
	for _, value := range values.Interfaces() {
		if strings.TrimSpace(gconvString(value)) == strings.TrimSpace(target) {
			return true
		}
	}
	return false
}

func resolveChangeBeginAt(contract *ExecutionContract, now time.Time) (time.Time, error) {
	if contract == nil || contract.PackageBeginTime == nil {
		return time.Time{}, gerror.New("change package begin time is missing")
	}
	switch strings.TrimSpace(contract.EffectiveMode) {
	case "immediate":
		return now, nil
	case "next_month":
		begin := contract.PackageBeginTime.Time
		if begin.Day() != 1 || begin.Hour() != 0 || begin.Minute() != 0 || begin.Second() != 0 {
			return time.Time{}, gerror.New("change next month begin time is invalid")
		}
		return begin, nil
	default:
		return time.Time{}, gerror.New("change effective mode is unsupported")
	}
}

func calculateChangePackageStatus(now, begin, end time.Time) int {
	if begin.After(now) {
		return consts.PackageStatusPending
	}
	if !end.After(now) {
		return consts.PackageStatusExpired
	}
	return consts.PackageStatusEffective
}

func calculateChangeEffectiveStatus(now, begin, end time.Time) int {
	if begin.After(now) {
		return consts.EffectiveStatusPending
	}
	if !end.After(now) {
		return consts.EffectiveStatusExpired
	}
	return consts.EffectiveStatusEffective
}

func subscriptionNos(rows []*rc.SimCardBindPkg) []string {
	result := make([]string, 0, len(rows))
	for _, row := range rows {
		if row != nil {
			result = append(result, strings.TrimSpace(row.SubscriptionNo))
		}
	}
	return result
}

func flattenFutureBinds(future map[string][]*rc.SimCardBindPkg, iccids []string) []*rc.SimCardBindPkg {
	result := make([]*rc.SimCardBindPkg, 0)
	for _, iccid := range iccids {
		result = append(result, future[iccid]...)
	}
	return result
}

func buildChangeDetailSnapshot(detail *rc.BatchDetailRow, card *rc.SimCardInfo, oldBind *rc.SimCardBindPkg, newSubscriptionNo string, voided []*rc.SimCardBindPkg, contract *ExecutionContract, oldPackageCode, orderSN string, beginAt, endAt time.Time) *gjson.Json {
	data := map[string]any{}
	if detail != nil && detail.Snapshot != nil {
		for key, value := range detail.Snapshot.Map() {
			data[key] = value
		}
	}
	data["ICCID"] = card.Iccid
	data["MSISDN"] = card.Msisdn
	data["IMSI"] = card.Imsi
	data["iccid"] = card.Iccid
	data["msisdn"] = card.Msisdn
	data["imsi"] = card.Imsi
	data["oldPackageCode"] = oldPackageCode
	data["newPackageCode"] = contract.PackageCode
	data["newPackageName"] = contract.PackageName
	data["oldSubscriptionNo"] = oldBind.SubscriptionNo
	data["oldOrderSn"] = oldBind.OrderCode
	data["newSubscriptionNo"] = newSubscriptionNo
	data["newOrderSn"] = orderSN
	data["durationMonths"] = contract.DurationMonths
	data["giftDurationMonths"] = contract.GiftDurationMonths
	data["effectiveMode"] = contract.EffectiveMode
	data["changeUnitPrice"] = contract.UnitPrice
	data["amount"] = roundAmount(contract.UnitPrice * float64(contract.DurationMonths))
	data["newEffectiveTime"] = formatTime(gtime.NewFromTime(beginAt))
	data["newExpectedExpireTime"] = formatTime(gtime.NewFromTime(endAt))
	data["oldActualExpireTime"] = formatTime(gtime.Now())
	voidedNos := make([]string, 0, len(voided))
	for _, row := range voided {
		if row != nil {
			voidedNos = append(voidedNos, row.SubscriptionNo)
		}
	}
	data["voidedSubscriptionNos"] = voidedNos
	return gjson.New(data)
}

func gconvString(value any) string {
	if value == nil {
		return ""
	}
	return strings.TrimSpace(fmt.Sprint(value))
}