Files

1151 lines
34 KiB
Go

package packagelifecycle
import (
"context"
"fmt"
"math"
"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/logic/packagecustomer"
"git.5iot.com/iotlcm-public/iotlcm-admin-svr/internal/model"
"git.5iot.com/iotlcm-public/iotlcm-admin-svr/internal/model/do"
"git.5iot.com/iotlcm-public/iotlcm-admin-svr/internal/model/entity"
"git.5iot.com/iotlcm-public/iotlcm-admin-svr/internal/service"
"github.com/gogf/gf/v2/database/gdb"
"github.com/gogf/gf/v2/errors/gerror"
"github.com/gogf/gf/v2/frame/g"
"github.com/gogf/gf/v2/os/gtime"
"github.com/gogf/gf/v2/util/gconv"
)
const (
syncPageSize = 200
packageLifecycleSystemUID = 0
packageTypeCard = 101
packageTypePool = 102
packageTypeCardTopup = 103
packageTypePoolTopup = 104
)
type sPackageLifecycle struct{}
type syncStats struct {
Cards int
Activated int
Expired int
CardUpdated int
CardTopupExpired int
CardTopupFlowUpdated int
PoolTopupExpired int
PoolTopupFlowUpdated int
SkipCount int
ErrorCount int
}
type cardSyncResult struct {
Activated int
Expired int
CardUpdated bool
Skipped bool
TenantId string
DeptId int64
}
func init() {
service.RegisterPackageLifecycle(New())
}
func New() *sPackageLifecycle {
return &sPackageLifecycle{}
}
func (s *sPackageLifecycle) Sync(ctx context.Context) {
if ctx == nil {
ctx = context.Background()
}
now := gtime.Now()
stats := syncStats{}
for {
if err := ctx.Err(); err != nil {
g.Log().Warningf(ctx, "PackageLifecycle.Sync canceled, err:%+v", err)
break
}
iccids, err := s.queryDueIccids(ctx, now, syncPageSize)
if err != nil {
stats.ErrorCount++
g.Log().Errorf(ctx, "PackageLifecycle.Sync query due iccids failed, err:%+v", err)
break
}
if len(iccids) == 0 {
break
}
progressed := false
for _, iccid := range iccids {
result, err := s.syncCard(ctx, strings.TrimSpace(iccid), now)
if err != nil {
stats.ErrorCount++
g.Log().Errorf(ctx, "PackageLifecycle.Sync card failed, iccid:%s, err:%+v", iccid, err)
continue
}
stats.Cards++
stats.Activated += result.Activated
stats.Expired += result.Expired
if result.CardUpdated {
stats.CardUpdated++
}
if result.Activated > 0 || result.Expired > 0 || result.CardUpdated {
progressed = true
}
if result.CardUpdated && result.DeptId > 0 {
packagecustomer.PublishSyncSafe(ctx, model.PackageCustomerSyncMessage{
TenantId: result.TenantId,
DeptId: result.DeptId,
UserId: packageLifecycleSystemUID,
Trigger: "package_lifecycle",
SourceBatchNo: iccid,
})
}
if result.Skipped {
stats.SkipCount++
}
}
if !progressed {
g.Log().Warningf(ctx, "PackageLifecycle.Sync stopped because current due page made no progress, size:%d", len(iccids))
break
}
}
cardTopupExpired, cardTopupFlowUpdated, err := s.expireCardTopupPackages(ctx, now)
if err != nil {
stats.ErrorCount++
g.Log().Errorf(ctx, "PackageLifecycle.Sync card topup failed, err:%+v", err)
} else {
stats.CardTopupExpired += cardTopupExpired
stats.CardTopupFlowUpdated += cardTopupFlowUpdated
}
poolTopupExpired, poolTopupFlowUpdated, err := s.expirePoolTopupPackages(ctx, now)
if err != nil {
stats.ErrorCount++
g.Log().Errorf(ctx, "PackageLifecycle.Sync pool topup failed, err:%+v", err)
} else {
stats.PoolTopupExpired += poolTopupExpired
stats.PoolTopupFlowUpdated += poolTopupFlowUpdated
}
g.Log().Infof(ctx,
"PackageLifecycle.Sync end, cards:%d, activated:%d, expired:%d, cardUpdated:%d, cardTopupExpired:%d, cardTopupFlowUpdated:%d, poolTopupExpired:%d, poolTopupFlowUpdated:%d, skipped:%d, errors:%d",
stats.Cards,
stats.Activated,
stats.Expired,
stats.CardUpdated,
stats.CardTopupExpired,
stats.CardTopupFlowUpdated,
stats.PoolTopupExpired,
stats.PoolTopupFlowUpdated,
stats.SkipCount,
stats.ErrorCount,
)
}
func (s *sPackageLifecycle) queryDueIccids(ctx context.Context, now *gtime.Time, limit int) ([]string, error) {
if limit <= 0 {
limit = syncPageSize
}
var rows []struct {
Iccid string `orm:"iccid"`
}
c := dao.TSimCardBindPkg.Columns()
table := dao.TSimCardBindPkg.Table()
pendingDueCondition := fmt.Sprintf(
"(%s = ? AND (%s IS NULL OR %s <= ?) AND NOT EXISTS (SELECT 1 FROM %s b2 WHERE b2.%s = %s.%s AND b2.%s IN (?, ?) AND b2.%s = ? AND (b2.%s IS NULL OR b2.%s > ?)))",
c.EffectiveStatus,
c.EffectiveTime,
c.EffectiveTime,
table,
c.Iccid,
table,
c.Iccid,
c.PackageType,
c.EffectiveStatus,
c.ExpectedExpireTime,
c.ExpectedExpireTime,
)
effectiveExpiredCondition := fmt.Sprintf(
"(%s = ? AND %s IS NOT NULL AND %s <= ?)",
c.EffectiveStatus,
c.ExpectedExpireTime,
c.ExpectedExpireTime,
)
err := dao.TSimCardBindPkg.Ctx(ctx).
Fields(c.Iccid).
WhereIn(c.PackageType, []int{packageTypeCard, packageTypePool}).
WhereIn(c.EffectiveStatus, []int{consts.EffectiveStatusPending, consts.EffectiveStatusEffective}).
Where(
fmt.Sprintf("%s OR %s", pendingDueCondition, effectiveExpiredCondition),
consts.EffectiveStatusPending,
now,
packageTypeCard,
packageTypePool,
consts.EffectiveStatusEffective,
now,
consts.EffectiveStatusEffective,
now,
).
Group(c.Iccid).
OrderAsc(c.Iccid).
Limit(limit).
Scan(&rows)
if err != nil {
return nil, err
}
iccids := make([]string, 0, len(rows))
for _, row := range rows {
if iccid := strings.TrimSpace(row.Iccid); iccid != "" {
iccids = append(iccids, iccid)
}
}
return iccids, nil
}
func (s *sPackageLifecycle) syncCard(ctx context.Context, iccid string, now *gtime.Time) (cardSyncResult, error) {
if iccid == "" {
return cardSyncResult{Skipped: true}, nil
}
result := cardSyncResult{}
err := g.DB().Transaction(ctx, func(ctx context.Context, tx gdb.TX) error {
card, err := s.lockCard(ctx, tx, iccid)
if err != nil {
return err
}
if card == nil || card.Id == 0 {
result.Skipped = true
return nil
}
result.TenantId = card.TenantId
result.DeptId = card.DeptId
binds, err := s.lockActiveBinds(ctx, tx, iccid)
if err != nil {
return err
}
if len(binds) == 0 {
result.Skipped = true
return nil
}
changed := false
expiredBinds := make([]*entity.TSimCardBindPkg, 0, len(binds))
for {
current := currentEffectiveBind(binds)
if current != nil && expireDue(current.ExpectedExpireTime, now) {
if err = s.updateBindStatus(ctx, tx, current, consts.EffectiveStatusExpired, current.ExpectedExpireTime, now); err != nil {
return err
}
current.EffectiveStatus = consts.EffectiveStatusExpired
current.ActualExpireTime = naturalExpireTime(current.ExpectedExpireTime, now)
current.UpdatedAt = now
current.UpdatedBy = packageLifecycleSystemUID
expiredBinds = append(expiredBinds, current)
result.Expired++
changed = true
continue
}
if current == nil {
next := nextDuePendingBind(binds, now)
if next == nil {
break
}
if expireDue(next.ExpectedExpireTime, now) {
if err = s.updateBindStatus(ctx, tx, next, consts.EffectiveStatusExpired, next.ExpectedExpireTime, now); err != nil {
return err
}
next.EffectiveStatus = consts.EffectiveStatusExpired
next.ActualExpireTime = naturalExpireTime(next.ExpectedExpireTime, now)
next.UpdatedAt = now
next.UpdatedBy = packageLifecycleSystemUID
expiredBinds = append(expiredBinds, next)
result.Expired++
changed = true
continue
}
if err = s.updateBindStatus(ctx, tx, next, consts.EffectiveStatusEffective, nil, now); err != nil {
return err
}
next.EffectiveStatus = consts.EffectiveStatusEffective
next.UpdatedAt = now
next.UpdatedBy = packageLifecycleSystemUID
result.Activated++
changed = true
}
break
}
if len(expiredBinds) > 0 {
if err = s.upsertExpiredLogs(ctx, tx, expiredBinds, now); err != nil {
return err
}
}
target := chooseCardPackageFact(binds)
displayExpireAt := latestExpectedExpireTime(binds)
if target == nil && !changed {
result.Skipped = true
return nil
}
updated, err := s.updateCardFact(ctx, tx, card, target, displayExpireAt, now)
if err != nil {
return err
}
result.CardUpdated = updated
if !changed && !updated {
result.Skipped = true
}
return nil
})
if err != nil {
return cardSyncResult{}, err
}
return result, nil
}
func (s *sPackageLifecycle) lockCard(ctx context.Context, tx gdb.TX, iccid string) (*entity.TSimCardInfo, error) {
var card *entity.TSimCardInfo
err := tx.Ctx(ctx).Model(dao.TSimCardInfo.Table()).
Where(dao.TSimCardInfo.Columns().Iccid, iccid).
LockUpdate().
Scan(&card)
if err != nil {
return nil, gerror.Wrap(err, "锁定卡信息失败")
}
return card, nil
}
func (s *sPackageLifecycle) lockActiveBinds(ctx context.Context, tx gdb.TX, iccid string) ([]*entity.TSimCardBindPkg, error) {
var binds []*entity.TSimCardBindPkg
err := tx.Ctx(ctx).Model(dao.TSimCardBindPkg.Table()).
Where(dao.TSimCardBindPkg.Columns().Iccid, iccid).
WhereIn(dao.TSimCardBindPkg.Columns().PackageType, []int{packageTypeCard, packageTypePool}).
WhereIn(dao.TSimCardBindPkg.Columns().EffectiveStatus, []int{consts.EffectiveStatusPending, consts.EffectiveStatusEffective}).
OrderAsc(dao.TSimCardBindPkg.Columns().EffectiveStatus).
OrderAsc(dao.TSimCardBindPkg.Columns().EffectiveTime).
OrderAsc(dao.TSimCardBindPkg.Columns().Id).
LockUpdate().
Scan(&binds)
if err != nil {
return nil, gerror.Wrap(err, "锁定套餐订购记录失败")
}
return binds, nil
}
func (s *sPackageLifecycle) updateBindStatus(ctx context.Context, tx gdb.TX, bind *entity.TSimCardBindPkg, status int, actualExpireAt *gtime.Time, now *gtime.Time) error {
if bind == nil || bind.Id <= 0 {
return gerror.New("套餐订购记录为空")
}
data := g.Map{
dao.TSimCardBindPkg.Columns().EffectiveStatus: status,
dao.TSimCardBindPkg.Columns().UpdatedBy: packageLifecycleSystemUID,
dao.TSimCardBindPkg.Columns().UpdatedAt: now,
}
if status == consts.EffectiveStatusExpired {
data[dao.TSimCardBindPkg.Columns().ActualExpireTime] = naturalExpireTime(actualExpireAt, now)
}
res, err := tx.Ctx(ctx).Model(dao.TSimCardBindPkg.Table()).
Where(dao.TSimCardBindPkg.Columns().Id, bind.Id).
Where(dao.TSimCardBindPkg.Columns().EffectiveStatus, bind.EffectiveStatus).
Data(data).
Update()
if err != nil {
return gerror.Wrap(err, "更新套餐订购状态失败")
}
affected, err := res.RowsAffected()
if err != nil {
return gerror.Wrap(err, "读取套餐订购状态更新数量失败")
}
if affected != 1 {
return gerror.New("套餐订购状态已变化,请稍后重试")
}
return nil
}
func (s *sPackageLifecycle) upsertExpiredLogs(ctx context.Context, tx gdb.TX, binds []*entity.TSimCardBindPkg, now *gtime.Time) error {
rows := make([]do.TSimBindPackageLogs, 0, len(binds))
for _, bind := range binds {
if !isLifecycleBind(bind) || strings.TrimSpace(bind.SubscriptionNo) == "" {
continue
}
actualExpireAt := naturalExpireTime(bind.ActualExpireTime, now)
rows = append(rows, do.TSimBindPackageLogs{
SubscriptionNo: bind.SubscriptionNo,
BatchNo: bind.BatchNo,
Iccid: bind.Iccid,
Msisdn: bind.Msisdn,
DeptId: bind.DeptId,
PackageType: bind.PackageType,
PackageCode: bind.PackageCode,
PreviousPackageCode: bind.PreviousPackageCode,
BusinessType: bind.BusinessType,
EffectiveStatus: consts.EffectiveStatusExpired,
HandleTime: bind.HandleTime,
EffectiveTime: bind.EffectiveTime,
ExpectedExpireTime: bind.ExpectedExpireTime,
ActualExpireTime: actualExpireAt,
DurationMonths: bind.DurationMonths,
GiftDurationMonths: bind.GiftDurationMonths,
OrderCode: bind.OrderCode,
Remark: appendRemark(bind.Remark, "packageLifecycle=natural_expire"),
TenantId: bind.TenantId,
CreatedDept: bind.CreatedDept,
CreatedBy: bind.CreatedBy,
CreatedAt: bind.CreatedAt,
UpdatedBy: packageLifecycleSystemUID,
UpdatedAt: now,
})
}
if len(rows) == 0 {
return nil
}
_, err := tx.Ctx(ctx).Model(dao.TSimBindPackageLogs.Table()).
Data(rows).
OnConflict(dao.TSimBindPackageLogs.Columns().SubscriptionNo).
Save()
return gerror.Wrap(err, "写入自然到期套餐历史失败")
}
func (s *sPackageLifecycle) expireCardTopupPackages(ctx context.Context, now *gtime.Time) (expired int, cardFlowUpdated int, err error) {
for {
ids, err := s.queryDueCardTopupIds(ctx, now, syncPageSize)
if err != nil {
return expired, cardFlowUpdated, err
}
if len(ids) == 0 {
return expired, cardFlowUpdated, nil
}
expiredInTx := 0
updatedInTx := 0
err = g.DB().Transaction(ctx, func(ctx context.Context, tx gdb.TX) error {
binds, err := s.lockDueCardTopupBinds(ctx, tx, ids, now)
if err != nil {
return err
}
if len(binds) == 0 {
return nil
}
affectedIccids := make(map[string]struct{})
expiredBinds := make([]*entity.TSimCardBindPkg, 0, len(binds))
for _, bind := range binds {
if err = s.updateBindStatus(ctx, tx, bind, consts.EffectiveStatusExpired, bind.ExpectedExpireTime, now); err != nil {
return err
}
bind.EffectiveStatus = consts.EffectiveStatusExpired
bind.ActualExpireTime = naturalExpireTime(bind.ExpectedExpireTime, now)
bind.UpdatedAt = now
bind.UpdatedBy = packageLifecycleSystemUID
expiredBinds = append(expiredBinds, bind)
if iccid := strings.TrimSpace(bind.Iccid); iccid != "" {
affectedIccids[iccid] = struct{}{}
}
expiredInTx++
}
if err = s.upsertExpiredLogs(ctx, tx, expiredBinds, now); err != nil {
return err
}
for iccid := range affectedIccids {
updated, err := s.recalculateCardTotalFlow(ctx, tx, iccid, now)
if err != nil {
return err
}
if updated {
updatedInTx++
}
}
return nil
})
if err != nil {
return expired, cardFlowUpdated, err
}
expired += expiredInTx
cardFlowUpdated += updatedInTx
if expiredInTx == 0 {
g.Log().Warningf(ctx, "PackageLifecycle.Sync card topup stopped because current due page made no progress, size:%d", len(ids))
return expired, cardFlowUpdated, nil
}
}
}
func (s *sPackageLifecycle) queryDueCardTopupIds(ctx context.Context, now *gtime.Time, limit int) ([]int64, error) {
if limit <= 0 {
limit = syncPageSize
}
var rows []struct {
Id int64 `orm:"id"`
}
c := dao.TSimCardBindPkg.Columns()
err := dao.TSimCardBindPkg.Ctx(ctx).
Fields(c.Id).
Where(c.PackageType, packageTypeCardTopup).
WhereIn(c.EffectiveStatus, []int{consts.EffectiveStatusPending, consts.EffectiveStatusEffective}).
Where(fmt.Sprintf("%s IS NOT NULL AND %s <= ?", c.ExpectedExpireTime, c.ExpectedExpireTime), now).
OrderAsc(c.Id).
Limit(limit).
Scan(&rows)
if err != nil {
return nil, err
}
ids := make([]int64, 0, len(rows))
for _, row := range rows {
if row.Id > 0 {
ids = append(ids, row.Id)
}
}
return ids, nil
}
func (s *sPackageLifecycle) lockDueCardTopupBinds(ctx context.Context, tx gdb.TX, ids []int64, now *gtime.Time) ([]*entity.TSimCardBindPkg, error) {
if len(ids) == 0 {
return nil, nil
}
var binds []*entity.TSimCardBindPkg
c := dao.TSimCardBindPkg.Columns()
err := tx.Ctx(ctx).Model(dao.TSimCardBindPkg.Table()).
WhereIn(c.Id, ids).
Where(c.PackageType, packageTypeCardTopup).
WhereIn(c.EffectiveStatus, []int{consts.EffectiveStatusPending, consts.EffectiveStatusEffective}).
Where(fmt.Sprintf("%s IS NOT NULL AND %s <= ?", c.ExpectedExpireTime, c.ExpectedExpireTime), now).
OrderAsc(c.Id).
LockUpdate().
Scan(&binds)
if err != nil {
return nil, gerror.Wrap(err, "lock due card topup records failed")
}
return binds, nil
}
func (s *sPackageLifecycle) recalculateCardTotalFlow(ctx context.Context, tx gdb.TX, iccid string, now *gtime.Time) (bool, error) {
card, err := s.lockCard(ctx, tx, iccid)
if err != nil {
return false, err
}
if card == nil || card.Id <= 0 {
return false, nil
}
totalFlowMb := float64(0)
if strings.TrimSpace(card.CurrentPackageCode) != "" {
pkg, err := s.loadPackage(ctx, tx, card.TenantId, card.CurrentPackageCode)
if err != nil {
return false, err
}
totalFlowMb = pkg.FlowMb
}
topupFlowMb, err := s.activeCardTopupFlow(ctx, tx, card.Iccid, card.TenantId, now)
if err != nil {
return false, err
}
totalFlowMb += topupFlowMb
if floatEquals(card.PackageTotalFlowMb, totalFlowMb) {
return false, nil
}
res, err := tx.Ctx(ctx).Model(dao.TSimCardInfo.Table()).
Where(dao.TSimCardInfo.Columns().Id, card.Id).
Data(g.Map{
dao.TSimCardInfo.Columns().PackageTotalFlowMb: totalFlowMb,
dao.TSimCardInfo.Columns().UpdatedBy: packageLifecycleSystemUID,
dao.TSimCardInfo.Columns().UpdatedAt: now,
}).
Update()
if err != nil {
return false, gerror.Wrap(err, "sync card topup total flow failed")
}
affected, err := res.RowsAffected()
if err != nil {
return false, gerror.Wrap(err, "read card topup total flow update count failed")
}
return affected > 0, nil
}
type poolTopupKey struct {
TenantId string
DeptId int64
DeptIdRaw string
PoolNo string
}
func (s *sPackageLifecycle) expirePoolTopupPackages(ctx context.Context, now *gtime.Time) (expired int, poolFlowUpdated int, err error) {
for {
ids, err := s.queryDuePoolTopupIds(ctx, now, syncPageSize)
if err != nil {
return expired, poolFlowUpdated, err
}
if len(ids) == 0 {
return expired, poolFlowUpdated, nil
}
expiredInTx := 0
updatedInTx := 0
err = g.DB().Transaction(ctx, func(ctx context.Context, tx gdb.TX) error {
binds, err := s.lockDuePoolTopupBinds(ctx, tx, ids, now)
if err != nil {
return err
}
if len(binds) == 0 {
return nil
}
affectedPools := make(map[poolTopupKey]struct{})
for _, bind := range binds {
if err = s.updatePoolBindStatus(ctx, tx, bind, consts.EffectiveStatusExpired, bind.ExpectedExpireTime, now); err != nil {
return err
}
bind.EffectiveStatus = consts.EffectiveStatusExpired
bind.ActualExpireTime = naturalExpireTime(bind.ExpectedExpireTime, now)
bind.UpdatedAt = now
bind.UpdatedBy = packageLifecycleSystemUID
if key, ok := poolTopupKeyFromBind(bind); ok {
affectedPools[key] = struct{}{}
}
expiredInTx++
}
for key := range affectedPools {
updated, err := s.recalculatePoolTopupFlow(ctx, tx, key, now)
if err != nil {
return err
}
if updated {
updatedInTx++
}
}
return nil
})
if err != nil {
return expired, poolFlowUpdated, err
}
expired += expiredInTx
poolFlowUpdated += updatedInTx
if expiredInTx == 0 {
g.Log().Warningf(ctx, "PackageLifecycle.Sync pool topup stopped because current due page made no progress, size:%d", len(ids))
return expired, poolFlowUpdated, nil
}
}
}
func (s *sPackageLifecycle) queryDuePoolTopupIds(ctx context.Context, now *gtime.Time, limit int) ([]int64, error) {
if limit <= 0 {
limit = syncPageSize
}
var rows []struct {
Id int64 `orm:"id"`
}
c := dao.TSimPoolBindPackage.Columns()
err := dao.TSimPoolBindPackage.Ctx(ctx).
Fields(c.Id).
Where(c.PackageType, packageTypePoolTopup).
WhereIn(c.EffectiveStatus, []int{consts.EffectiveStatusPending, consts.EffectiveStatusEffective}).
Where(fmt.Sprintf("%s IS NOT NULL AND %s <= ?", c.ExpectedExpireTime, c.ExpectedExpireTime), now).
OrderAsc(c.Id).
Limit(limit).
Scan(&rows)
if err != nil {
return nil, err
}
ids := make([]int64, 0, len(rows))
for _, row := range rows {
if row.Id > 0 {
ids = append(ids, row.Id)
}
}
return ids, nil
}
func (s *sPackageLifecycle) lockDuePoolTopupBinds(ctx context.Context, tx gdb.TX, ids []int64, now *gtime.Time) ([]*entity.TSimPoolBindPackage, error) {
if len(ids) == 0 {
return nil, nil
}
var binds []*entity.TSimPoolBindPackage
c := dao.TSimPoolBindPackage.Columns()
err := tx.Ctx(ctx).Model(dao.TSimPoolBindPackage.Table()).
WhereIn(c.Id, ids).
Where(c.PackageType, packageTypePoolTopup).
WhereIn(c.EffectiveStatus, []int{consts.EffectiveStatusPending, consts.EffectiveStatusEffective}).
Where(fmt.Sprintf("%s IS NOT NULL AND %s <= ?", c.ExpectedExpireTime, c.ExpectedExpireTime), now).
OrderAsc(c.Id).
LockUpdate().
Scan(&binds)
if err != nil {
return nil, gerror.Wrap(err, "lock due pool topup records failed")
}
return binds, nil
}
func (s *sPackageLifecycle) updatePoolBindStatus(ctx context.Context, tx gdb.TX, bind *entity.TSimPoolBindPackage, status int, actualExpireAt *gtime.Time, now *gtime.Time) error {
if bind == nil || bind.Id <= 0 {
return gerror.New("pool package subscription is empty")
}
c := dao.TSimPoolBindPackage.Columns()
data := g.Map{
c.EffectiveStatus: status,
c.UpdatedBy: packageLifecycleSystemUID,
c.UpdatedAt: now,
}
if status == consts.EffectiveStatusExpired {
data[c.ActualExpireTime] = naturalExpireTime(actualExpireAt, now)
}
res, err := tx.Ctx(ctx).Model(dao.TSimPoolBindPackage.Table()).
Where(c.Id, bind.Id).
Where(c.EffectiveStatus, bind.EffectiveStatus).
Data(data).
Update()
if err != nil {
return gerror.Wrap(err, "update pool topup status failed")
}
affected, err := res.RowsAffected()
if err != nil {
return gerror.Wrap(err, "read pool topup status update count failed")
}
if affected != 1 {
return gerror.New("pool topup status changed, please retry")
}
return nil
}
func poolTopupKeyFromBind(bind *entity.TSimPoolBindPackage) (poolTopupKey, bool) {
if bind == nil {
return poolTopupKey{}, false
}
poolNo := strings.TrimSpace(bind.PoolNo)
deptIdRaw := strings.TrimSpace(bind.DeptId)
if poolNo == "" {
return poolTopupKey{}, false
}
return poolTopupKey{
TenantId: strings.TrimSpace(bind.TenantId),
DeptId: gconv.Int64(deptIdRaw),
DeptIdRaw: deptIdRaw,
PoolNo: poolNo,
}, true
}
func (s *sPackageLifecycle) recalculatePoolTopupFlow(ctx context.Context, tx gdb.TX, key poolTopupKey, now *gtime.Time) (bool, error) {
pool, err := s.lockFlowPool(ctx, tx, key)
if err != nil {
return false, err
}
if pool == nil || pool.Id <= 0 {
return false, nil
}
activeTopupFlowMb, err := s.activePoolTopupFlow(ctx, tx, key, now)
if err != nil {
return false, err
}
baseFlowMb := pool.PoolPackageFlowMb
if pool.AddOnFlowMb > 0 && pool.TotalFlowMb > 0 && pool.TotalFlowMb < pool.PoolPackageFlowMb {
baseFlowMb = pool.TotalFlowMb
}
if baseFlowMb < 0 {
baseFlowMb = 0
}
totalFlowMb := baseFlowMb + activeTopupFlowMb
remainFlowMb := totalFlowMb - pool.UsedFlowMb
poolStatus := 1
if remainFlowMb > 0 {
poolStatus = 0
}
if floatEquals(pool.PoolPackageFlowMb, baseFlowMb) &&
floatEquals(pool.AddOnFlowMb, activeTopupFlowMb) &&
floatEquals(pool.TotalFlowMb, totalFlowMb) &&
floatEquals(pool.RemainFlowMb, remainFlowMb) &&
pool.PoolStatus == poolStatus {
return false, nil
}
c := dao.TSimFlowPool.Columns()
res, err := tx.Ctx(ctx).Model(dao.TSimFlowPool.Table()).
Where(c.Id, pool.Id).
Data(g.Map{
c.PoolPackageFlowMb: baseFlowMb,
c.AddOnFlowMb: activeTopupFlowMb,
c.TotalFlowMb: totalFlowMb,
c.RemainFlowMb: remainFlowMb,
c.PoolStatus: poolStatus,
c.UpdatedBy: packageLifecycleSystemUID,
c.UpdatedAt: now,
}).
Update()
if err != nil {
return false, gerror.Wrap(err, "sync pool topup total flow failed")
}
affected, err := res.RowsAffected()
if err != nil {
return false, gerror.Wrap(err, "read pool topup total flow update count failed")
}
return affected > 0, nil
}
func (s *sPackageLifecycle) lockFlowPool(ctx context.Context, tx gdb.TX, key poolTopupKey) (*entity.TSimFlowPool, error) {
if key.PoolNo == "" {
return nil, nil
}
var pool *entity.TSimFlowPool
c := dao.TSimFlowPool.Columns()
mod := tx.Ctx(ctx).Model(dao.TSimFlowPool.Table()).
Where(c.FlowPoolCode, key.PoolNo)
if key.DeptId > 0 {
mod = mod.Where(c.DeptId, key.DeptId)
}
if key.TenantId != "" {
mod = mod.Where(c.TenantId, key.TenantId)
}
err := mod.OrderDesc(c.Id).LockUpdate().Scan(&pool)
if err != nil {
return nil, gerror.Wrap(err, "lock flow pool failed")
}
return pool, nil
}
func (s *sPackageLifecycle) activePoolTopupFlow(ctx context.Context, tx gdb.TX, key poolTopupKey, now *gtime.Time) (float64, error) {
if key.PoolNo == "" {
return 0, nil
}
bindTable := dao.TSimPoolBindPackage.Table()
packageTable := dao.TSimFlowPackageInfo.Table()
bc := dao.TSimPoolBindPackage.Columns()
pc := dao.TSimFlowPackageInfo.Columns()
var row struct {
FlowMb float64 `orm:"flow_mb"`
}
mod := tx.Ctx(ctx).Model(bindTable).
LeftJoin(packageTable, fmt.Sprintf("%s.%s = %s.%s AND %s.%s = %s.%s", bindTable, bc.PackageCode, packageTable, pc.PackageCode, bindTable, bc.TenantId, packageTable, pc.TenantId)).
Where(bindTable+"."+bc.PoolNo, key.PoolNo).
Where(bindTable+"."+bc.PackageType, packageTypePoolTopup).
Where(bindTable+"."+bc.EffectiveStatus, consts.EffectiveStatusEffective).
Where(fmt.Sprintf("(%s.%s IS NULL OR %s.%s <= ?)", bindTable, bc.EffectiveTime, bindTable, bc.EffectiveTime), now).
Where(fmt.Sprintf("(%s.%s IS NULL OR %s.%s > ?)", bindTable, bc.ExpectedExpireTime, bindTable, bc.ExpectedExpireTime), now)
if key.DeptIdRaw != "" {
mod = mod.Where(bindTable+"."+bc.DeptId, key.DeptIdRaw)
}
if key.TenantId != "" {
mod = mod.Where(bindTable+"."+bc.TenantId, key.TenantId)
}
err := mod.Fields(fmt.Sprintf("IFNULL(SUM(%s.%s), 0) AS flow_mb", packageTable, pc.FlowMb)).
Scan(&row)
if err != nil {
return 0, gerror.Wrap(err, "sum active pool topup flow failed")
}
return row.FlowMb, nil
}
func (s *sPackageLifecycle) updateCardFact(ctx context.Context, tx gdb.TX, card *entity.TSimCardInfo, bind *entity.TSimCardBindPkg, displayExpireAt *gtime.Time, now *gtime.Time) (bool, error) {
if card == nil || card.Id <= 0 {
return false, gerror.New("card info is empty")
}
data := g.Map{
dao.TSimCardInfo.Columns().UpdatedBy: packageLifecycleSystemUID,
dao.TSimCardInfo.Columns().UpdatedAt: now,
}
if bind == nil {
data[dao.TSimCardInfo.Columns().CurrentPackageCode] = ""
data[dao.TSimCardInfo.Columns().PackageStatus] = consts.PackageStatusNone
data[dao.TSimCardInfo.Columns().FlowPkgBeginAt] = nil
data[dao.TSimCardInfo.Columns().FlowPkgEndAt] = nil
data[dao.TSimCardInfo.Columns().PackageTotalFlowMb] = 0
data[dao.TSimCardInfo.Columns().UsedFlowMb] = 0
} else {
pkg, err := s.loadPackage(ctx, tx, card.TenantId, bind.PackageCode)
if err != nil {
return false, err
}
topupFlowMb, err := s.activeCardTopupFlow(ctx, tx, card.Iccid, card.TenantId, now)
if err != nil {
return false, err
}
data[dao.TSimCardInfo.Columns().CurrentPackageCode] = strings.TrimSpace(bind.PackageCode)
data[dao.TSimCardInfo.Columns().PackageStatus] = packageStatusFromEffective(bind.EffectiveStatus)
data[dao.TSimCardInfo.Columns().FlowPkgBeginAt] = bind.EffectiveTime
data[dao.TSimCardInfo.Columns().FlowPkgEndAt] = chooseGTime(displayExpireAt, bind.ExpectedExpireTime)
data[dao.TSimCardInfo.Columns().PackageTotalFlowMb] = pkg.FlowMb + topupFlowMb
if cardPackagePeriodChanged(card, bind) {
data[dao.TSimCardInfo.Columns().UsedFlowMb] = 0
}
}
if cardPackageFactMatches(card, bind, data) {
return false, nil
}
res, err := tx.Ctx(ctx).Model(dao.TSimCardInfo.Table()).
Where(dao.TSimCardInfo.Columns().Id, card.Id).
Data(data).
Update()
if err != nil {
return false, gerror.Wrap(err, "sync card package fact failed")
}
affected, err := res.RowsAffected()
if err != nil {
return false, gerror.Wrap(err, "read card package fact update count failed")
}
if affected != 1 {
return false, gerror.New("card package fact changed, please retry")
}
return true, nil
}
func (s *sPackageLifecycle) loadPackage(ctx context.Context, tx gdb.TX, tenantId string, packageCode string) (*entity.TSimFlowPackageInfo, error) {
packageCode = strings.TrimSpace(packageCode)
if tenantId == "" || packageCode == "" {
return nil, gerror.New("套餐编码不能为空")
}
pkg := &entity.TSimFlowPackageInfo{}
err := tx.Ctx(ctx).Model(dao.TSimFlowPackageInfo.Table()).
Fields(
dao.TSimFlowPackageInfo.Columns().PackageCode,
dao.TSimFlowPackageInfo.Columns().FlowMb,
).
Where(dao.TSimFlowPackageInfo.Columns().TenantId, tenantId).
Where(dao.TSimFlowPackageInfo.Columns().PackageCode, packageCode).
Scan(pkg)
if err != nil {
return nil, gerror.Wrap(err, "查询续费目标套餐失败")
}
if strings.TrimSpace(pkg.PackageCode) == "" {
return nil, gerror.New("续费目标套餐不存在,无法同步卡套餐信息")
}
return pkg, nil
}
func (s *sPackageLifecycle) activeCardTopupFlow(ctx context.Context, tx gdb.TX, iccid string, tenantId string, now *gtime.Time) (float64, error) {
iccid = strings.TrimSpace(iccid)
tenantId = strings.TrimSpace(tenantId)
if iccid == "" || tenantId == "" {
return 0, nil
}
bindTable := dao.TSimCardBindPkg.Table()
packageTable := dao.TSimFlowPackageInfo.Table()
bc := dao.TSimCardBindPkg.Columns()
pc := dao.TSimFlowPackageInfo.Columns()
var row struct {
FlowMb float64 `orm:"flow_mb"`
}
err := tx.Ctx(ctx).Model(bindTable).
LeftJoin(packageTable, fmt.Sprintf("%s.%s = %s.%s AND %s.%s = %s.%s", bindTable, bc.PackageCode, packageTable, pc.PackageCode, bindTable, bc.TenantId, packageTable, pc.TenantId)).
Where(bindTable+"."+bc.Iccid, iccid).
Where(bindTable+"."+bc.TenantId, tenantId).
Where(bindTable+"."+bc.PackageType, packageTypeCardTopup).
Where(bindTable+"."+bc.EffectiveStatus, consts.EffectiveStatusEffective).
Where(fmt.Sprintf("(%s.%s IS NULL OR %s.%s <= ?)", bindTable, bc.EffectiveTime, bindTable, bc.EffectiveTime), now).
Where(fmt.Sprintf("(%s.%s IS NULL OR %s.%s > ?)", bindTable, bc.ExpectedExpireTime, bindTable, bc.ExpectedExpireTime), now).
Fields(fmt.Sprintf("IFNULL(SUM(%s.%s), 0) AS flow_mb", packageTable, pc.FlowMb)).
Scan(&row)
if err != nil {
return 0, gerror.Wrap(err, "sum active card topup flow failed")
}
return row.FlowMb, nil
}
func cardPackageFactMatches(card *entity.TSimCardInfo, bind *entity.TSimCardBindPkg, data g.Map) bool {
if card == nil {
return false
}
if bind == nil {
return strings.TrimSpace(card.CurrentPackageCode) == "" &&
card.PackageStatus == consts.PackageStatusNone &&
card.FlowPkgBeginAt == nil &&
card.FlowPkgEndAt == nil &&
floatEquals(card.PackageTotalFlowMb, 0) &&
floatEquals(card.UsedFlowMb, 0)
}
packageTotalFlowMb, ok := data[dao.TSimCardInfo.Columns().PackageTotalFlowMb].(float64)
if !ok {
return false
}
flowPkgEndAt, ok := data[dao.TSimCardInfo.Columns().FlowPkgEndAt].(*gtime.Time)
if !ok {
return false
}
return strings.TrimSpace(card.CurrentPackageCode) == strings.TrimSpace(bind.PackageCode) &&
card.PackageStatus == packageStatusFromEffective(bind.EffectiveStatus) &&
compareGTime(card.FlowPkgBeginAt, bind.EffectiveTime) == 0 &&
compareGTime(card.FlowPkgEndAt, flowPkgEndAt) == 0 &&
floatEquals(card.PackageTotalFlowMb, packageTotalFlowMb)
}
func latestExpectedExpireTime(binds []*entity.TSimCardBindPkg) *gtime.Time {
var latest *gtime.Time
for _, bind := range binds {
if bind == nil || bind.ExpectedExpireTime == nil {
continue
}
if latest == nil || bind.ExpectedExpireTime.After(latest) {
latest = bind.ExpectedExpireTime
}
}
return latest
}
func chooseGTime(primary, fallback *gtime.Time) *gtime.Time {
if primary != nil {
return primary
}
return fallback
}
func cardPackagePeriodChanged(card *entity.TSimCardInfo, bind *entity.TSimCardBindPkg) bool {
if card == nil || bind == nil {
return false
}
return strings.TrimSpace(card.CurrentPackageCode) != strings.TrimSpace(bind.PackageCode) ||
compareGTime(card.FlowPkgBeginAt, bind.EffectiveTime) != 0
}
func floatEquals(a, b float64) bool {
return math.Abs(a-b) < 0.000001
}
func currentEffectiveBind(binds []*entity.TSimCardBindPkg) *entity.TSimCardBindPkg {
for _, bind := range binds {
if isLifecycleBind(bind) && bind.EffectiveStatus == consts.EffectiveStatusEffective {
return bind
}
}
return nil
}
func nextDuePendingBind(binds []*entity.TSimCardBindPkg, now *gtime.Time) *entity.TSimCardBindPkg {
for _, bind := range binds {
if !isLifecycleBind(bind) || bind.EffectiveStatus != consts.EffectiveStatusPending {
continue
}
if effectiveDue(bind.EffectiveTime, now) {
return bind
}
}
return nil
}
func chooseCardPackageFact(binds []*entity.TSimCardBindPkg) *entity.TSimCardBindPkg {
var pending *entity.TSimCardBindPkg
var expired *entity.TSimCardBindPkg
for _, bind := range binds {
if !isLifecycleBind(bind) {
continue
}
switch bind.EffectiveStatus {
case consts.EffectiveStatusEffective:
return bind
case consts.EffectiveStatusPending:
if pending == nil || bindBefore(bind, pending) {
pending = bind
}
case consts.EffectiveStatusExpired:
if expired == nil || bindAfter(bind, expired) {
expired = bind
}
}
}
if pending != nil {
return pending
}
return expired
}
func isLifecycleBind(bind *entity.TSimCardBindPkg) bool {
return bind != nil && (bind.PackageType == packageTypeCard || bind.PackageType == packageTypePool)
}
func bindBefore(a, b *entity.TSimCardBindPkg) bool {
if a == nil {
return false
}
if b == nil {
return true
}
if compareGTime(a.EffectiveTime, b.EffectiveTime) < 0 {
return true
}
if compareGTime(a.EffectiveTime, b.EffectiveTime) > 0 {
return false
}
return a.Id < b.Id
}
func bindAfter(a, b *entity.TSimCardBindPkg) bool {
if a == nil {
return false
}
if b == nil {
return true
}
if compareGTime(a.ExpectedExpireTime, b.ExpectedExpireTime) > 0 {
return true
}
if compareGTime(a.ExpectedExpireTime, b.ExpectedExpireTime) < 0 {
return false
}
return a.Id > b.Id
}
func compareGTime(a, b *gtime.Time) int {
switch {
case a == nil && b == nil:
return 0
case a == nil:
return -1
case b == nil:
return 1
case a.Before(b):
return -1
case a.After(b):
return 1
default:
return 0
}
}
func effectiveDue(beginAt, now *gtime.Time) bool {
return beginAt == nil || !beginAt.After(now)
}
func expireDue(expireAt, now *gtime.Time) bool {
return expireAt != nil && !expireAt.After(now)
}
func naturalExpireTime(expireAt, now *gtime.Time) *gtime.Time {
if expireAt != nil {
return expireAt
}
return now
}
func packageStatusFromEffective(status int) int {
switch status {
case consts.EffectiveStatusPending:
return consts.PackageStatusPending
case consts.EffectiveStatusEffective:
return consts.PackageStatusEffective
case consts.EffectiveStatusExpired:
return consts.PackageStatusExpired
default:
return consts.PackageStatusNone
}
}
func appendRemark(remark string, extra string) string {
remark = strings.TrimSpace(remark)
extra = strings.TrimSpace(extra)
if extra == "" {
return remark
}
if remark == "" {
return extra
}
return remark + "; " + extra
}