1. 套餐过期前再主动同步一次流量

2. 生效套餐逻辑错误的问题

cmd/worker/main.go

@@ -346,6 +346,19 @@ func startPollingScheduler(
 		appLogger,
 	)
 
+	// 注入套餐失效前流量同步器：最小化 iot_card.Service（只需 gateway + store + redis + 流量扣减）
+	trafficSyncer := iot_card_svc.New(
+		runtime.db,
+		runtime.pollingIotCardStore,
+		nil, nil, nil, nil, nil,
+		runtime.gatewayClient,
+		appLogger,
+		nil,
+	)
+	trafficSyncer.SetRedisClient(runtime.redisClient)
+	trafficSyncer.SetDataDeductor(runtime.workerResult.Services.UsageService)
+	activationHandler.SetTrafficSyncer(trafficSyncer)
+
 	pollingScheduler := polling.NewScheduler(
 		runtime.redisClient,
 		runtime.asynqClient,

internal/polling/package_activation_handler.go

@@ -17,6 +17,12 @@ import (
 	"github.com/break/junhong_cmp_fiber/pkg/errors"
 )
 
+// TrafficSyncer 套餐失效前流量同步接口
+// 在主套餐标记为已过期之前，从 Gateway 拉取最新流量写入 DB
+type TrafficSyncer interface {
+	RefreshCardDataByID(ctx context.Context, cardID uint) error
+}
+
 // PackageActivationHandler 套餐激活检查处理器
 // 任务 19: 处理主套餐过期、加油包级联失效、待生效主套餐激活
 type PackageActivationHandler struct {
@@ -27,6 +33,7 @@ type PackageActivationHandler struct {
 	deviceSimBinding   *postgres.DeviceSimBindingStore
 	activationService  *packagepkg.ActivationService
 	stopResumeCallback packagepkg.StopResumeCallback // 停复机回调，用于套餐过期后主动停机
+	trafficSyncer      TrafficSyncer                 // 套餐失效前流量同步，可选
 	logger             *zap.Logger
 }
 
@@ -60,6 +67,48 @@ func NewPackageActivationHandler(
 	}
 }
 
+// SetTrafficSyncer 注入套餐失效前流量同步器（在 Start 前调用）
+func (h *PackageActivationHandler) SetTrafficSyncer(syncer TrafficSyncer) {
+	h.trafficSyncer = syncer
+}
+
+// syncTrafficBeforeExpiry 套餐失效前从 Gateway 同步一次最新流量
+// iot_card 直接同步；device 遍历所有绑定卡同步
+// 失败只记录警告，不阻断失效流程
+func (h *PackageActivationHandler) syncTrafficBeforeExpiry(ctx context.Context, carrierType string, carrierID uint) {
+	if h.trafficSyncer == nil {
+		return
+	}
+
+	if carrierType == constants.AssetTypeIotCard {
+		if err := h.trafficSyncer.RefreshCardDataByID(ctx, carrierID); err != nil {
+			h.logger.Warn("套餐失效前流量同步失败",
+				zap.String("carrier_type", carrierType),
+				zap.Uint("carrier_id", carrierID),
+				zap.Error(err))
+		}
+		return
+	}
+
+	if carrierType == constants.AssetTypeDevice {
+		bindings, err := h.deviceSimBinding.ListByDeviceID(ctx, carrierID)
+		if err != nil {
+			h.logger.Warn("套餐失效前查询设备绑定卡失败",
+				zap.Uint("device_id", carrierID),
+				zap.Error(err))
+			return
+		}
+		for _, b := range bindings {
+			if err := h.trafficSyncer.RefreshCardDataByID(ctx, b.IotCardID); err != nil {
+				h.logger.Warn("套餐失效前流量同步失败（设备绑定卡）",
+					zap.Uint("device_id", carrierID),
+					zap.Uint("card_id", b.IotCardID),
+					zap.Error(err))
+			}
+		}
+	}
+}
+
 // HandlePackageActivationCheck 任务 19.2-19.5: 处理套餐激活检查
 // 每 10 秒调度一次，检查过期主套餐并激活下一个待生效主套餐
 func (h *PackageActivationHandler) HandlePackageActivationCheck(ctx context.Context) error {
@@ -145,20 +194,21 @@ func (h *PackageActivationHandler) findAndActivateOrphanPackages(ctx context.Con
 			continue
 		}
 
-		// 检查该载体是否已有生效主套餐
+		// 检查该载体是否已有占位主套餐（生效中或已用完均视为占位，不允许激活待生效套餐）
 		var activeCount int64
 		var countErr error
+		occupiedStatuses := []int{constants.PackageUsageStatusActive, constants.PackageUsageStatusDepleted}
 		if key.carrierType == "iot_card" {
 			countErr = h.db.WithContext(ctx).
 				Model(&model.PackageUsage{}).
-				Where("status = ?", constants.PackageUsageStatusActive).
+				Where("status IN ?", occupiedStatuses).
 				Where("master_usage_id IS NULL").
 				Where("iot_card_id = ?", key.carrierID).
 				Count(&activeCount).Error
 		} else {
 			countErr = h.db.WithContext(ctx).
 				Model(&model.PackageUsage{}).
-				Where("status = ?", constants.PackageUsageStatusActive).
+				Where("status IN ?", occupiedStatuses).
 				Where("master_usage_id IS NULL").
 				Where("device_id = ?", key.carrierID).
 				Count(&activeCount).Error
@@ -171,7 +221,7 @@ func (h *PackageActivationHandler) findAndActivateOrphanPackages(ctx context.Con
 			continue
 		}
 
-		// 已有生效套餐，跳过
+		// 已有生效或已用完（占位）套餐，跳过
 		if activeCount > 0 {
 			continue
 		}
@@ -219,10 +269,15 @@ func (h *PackageActivationHandler) findExpiredMainPackages(ctx context.Context)
 }
 
 // processExpiredPackage 处理单个过期套餐
-// triggerStopAfterExpiry 在事务提交后才调用，避免 goroutine 在 TX 提交前读到脏数据
+// 流程：先同步最新流量 → 事务内标记过期/失效/激活下一个 → 事务提交后触发停机
 func (h *PackageActivationHandler) processExpiredPackage(ctx context.Context, pkg *model.PackageUsage) error {
 	carrierType, carrierID := h.getCarrierInfo(pkg)
 
+	// 失效前先从 Gateway 同步一次最新流量，确保 data_usage_mb 精确
+	if carrierType != "" && carrierID > 0 {
+		h.syncTrafficBeforeExpiry(ctx, carrierType, carrierID)
+	}
+
 	err := h.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
 		// 任务 19.3: 更新过期主套餐状态为 Expired (status=3)
 		if err := tx.Model(pkg).Update("status", constants.PackageUsageStatusExpired).Error; err != nil {

internal/service/iot_card/service.go

@@ -1494,6 +1494,19 @@ func (s *Service) buildCardNotFoundFailedItems(iccids []string) []dto.CardSeries
 	return items
 }
 
+// RefreshCardDataByID 通过卡 ID 从 Gateway 同步卡数据
+// 内部查询 ICCID 后委托给 RefreshCardDataFromGateway，供套餐失效前流量同步使用
+func (s *Service) RefreshCardDataByID(ctx context.Context, cardID uint) error {
+	card, err := s.iotCardStore.GetByID(ctx, cardID)
+	if err != nil {
+		if err == gorm.ErrRecordNotFound {
+			return nil
+		}
+		return err
+	}
+	return s.RefreshCardDataFromGateway(ctx, card.ICCID)
+}
+
 // RefreshCardDataFromGateway 从 Gateway 完整同步卡数据
 // 调用网关查询网络状态、实名状态、本月流量，并写回数据库
 // 流量采用增量计算（与轮询逻辑一致）：increment = 当前网关读数 - 上次网关读数

internal/service/package/activation_service.go

@@ -92,19 +92,20 @@ func (s *ActivationService) ActivateByRealname(ctx context.Context, carrierType
 
 			// 检查是否是主套餐
 			if usage.MasterUsageID == nil {
-				// 主套餐：需要检查是否有已激活的主套餐
+				// 主套餐：需要检查是否有已激活或已用完（仍在有效期）的主套餐
 				var activeMain model.PackageUsage
-				err := tx.Where("status = ?", constants.PackageUsageStatusActive).
+				err := tx.Where("status IN ?", []int{constants.PackageUsageStatusActive, constants.PackageUsageStatusDepleted}).
 					Where("master_usage_id IS NULL").
 					Where(carrierType+"_id = ?", carrierID).
 					Order("priority ASC").
 					First(&activeMain).Error
 
 				if err == nil {
-					// 已有激活的主套餐，保持排队状态
+					// 已有生效中或已用完（占位）的主套餐，保持排队状态
-					s.logger.Warn("已有激活主套餐，跳过激活",
+					s.logger.Warn("已有占位主套餐，跳过激活",
 						zap.Uint("usage_id", usage.ID),
-						zap.Uint("active_main_id", activeMain.ID))
+						zap.Uint("active_main_id", activeMain.ID),
+						zap.Int("active_main_status", activeMain.Status))
 					continue
 				}
 				if err != gorm.ErrRecordNotFound {
@@ -500,9 +501,10 @@ func (s *ActivationService) getNextPendingMainPackage(ctx context.Context, tx *g
 
 func (s *ActivationService) hasActiveMainPackage(ctx context.Context, tx *gorm.DB, carrierType string, carrierID uint) (bool, error) {
 	var count int64
+	// 生效中（1）和已用完（2）都属于"占位"状态：已用完的套餐仍在有效期内，不允许提前激活下一个主套餐
 	if err := tx.WithContext(ctx).
 		Model(&model.PackageUsage{}).
-		Where("status = ?", constants.PackageUsageStatusActive).
+		Where("status IN ?", []int{constants.PackageUsageStatusActive, constants.PackageUsageStatusDepleted}).
 		Where("master_usage_id IS NULL").
 		Where(carrierType+"_id = ?", carrierID).
 		Count(&count).Error; err != nil {