junhong_cmp_fiber/internal/polling/carrierqueue/sr_dispatch.go

package carrierqueue

import (
	"context"
	"encoding/json"
	"strconv"
	"time"

	"go.uber.org/zap"

	"github.com/break/junhong_cmp_fiber/pkg/constants"
)

// StopResumeMessage 是停复机指令 NSQ 消息体。
type StopResumeMessage struct {
	CardID       uint   `json:"card_id"`
	CarrierID    uint   `json:"carrier_id"`
	Token        string `json:"token"`
	Version      int64  `json:"version"`
	Action       string `json:"action"`
	Pending      bool   `json:"pending"`
	DispatchedAt int64  `json:"dispatched_at"`
}

// DeviceEvalMessage 是设备判断 NSQ 消息体。
type DeviceEvalMessage struct {
	DeviceID     uint     `json:"device_id"`
	Token        string   `json:"token"`
	Version      string   `json:"version"`
	Scenes       []string `json:"scenes"`
	DispatchedAt int64    `json:"dispatched_at"`
}

// deviceEvalLimit 是全局同时执行的设备判断上限：判断只读数据库与写指令，毫秒级。
const deviceEvalLimit = 50

// stopResumeLease 是停复机指令与设备判断的租约：单次执行只调一次网关，远小于该值。
func (d *Dispatcher) stopResumeLease() time.Duration {
	return 2 * time.Minute
}

// dispatchStopResume 在卡任务之前领取账户停复机指令：领取数 = min(停复机所在限流组本秒剩余, 执行上限 − 执行中)。
func (d *Dispatcher) dispatchStopResume(ctx context.Context, carrierID uint) {
	params := d.ParamsFor(carrierID)
	limit := params.SRInflightLimit
	processes := d.cfg.SubscriberProcesses
	if processes <= 0 {
		processes = 1
	}
	if consume := params.SRMaxInFlight * processes; consume > 0 && limit > consume {
		limit = consume
	}
	budget := -1
	if d.budgets != nil {
		ops := []string{constants.GatewayOpStop, constants.GatewayOpStart}
		if groups, _, limited := d.budgets.Budget(ctx, carrierID, ops); limited {
			for _, remaining := range groups {
				if budget < 0 || remaining < budget {
					budget = remaining
				}
			}
		}
	}
	if limit <= 0 || budget == 0 {
		return
	}
	claimed, expired, err := d.store.ClaimStopResume(ctx, SRClaimInput{
		CarrierID: carrierID, Lease: d.stopResumeLease(), InflightLimit: limit, Budget: budget, MinInterval: params.SRMinInterval,
		TokenPrefix: d.instanceID + ":sr:" + strconv.FormatInt(time.Now().UnixNano(), 36),
	})
	if err != nil {
		d.logger.Warn("运营商账户停复机指令领取失败", zap.Uint("carrier_id", carrierID), zap.Error(err))
		return
	}
	if expired > 0 {
		d.logger.Warn("停复机指令处理中标记到期（执行崩溃或超时）", zap.Uint("carrier_id", carrierID), zap.Int64("count", expired),
			zap.String("metric", "stop_resume.claim_expired"))
	}
	topic := constants.NSQStopResumeTopic(carrierID)
	for _, item := range claimed {
		body, _ := json.Marshal(StopResumeMessage{CardID: item.CardID, CarrierID: carrierID, Token: item.Token, Version: item.Version,
			Action: item.Action, Pending: item.Pending, DispatchedAt: time.Now().UnixMilli()})
		if err := d.publisher.Publish(topic, body); err != nil {
			releaseCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 2*time.Second)
			if releaseErr := d.store.ReleaseStopResume(releaseCtx, item.CardID, carrierID, item.Token); releaseErr != nil {
				d.logger.Warn("发布失败后释放停复机指令失败", zap.Uint("card_id", item.CardID), zap.Error(releaseErr))
			}
			cancel()
			d.logger.Warn("发布停复机指令失败，已释放", zap.Uint("card_id", item.CardID), zap.String("topic", topic), zap.Error(err))
		}
	}
}

// dispatchDeviceEvals 领取到期设备判断并发布；全局分发锁保证每轮只有一个实例领取。
func (d *Dispatcher) dispatchDeviceEvals(ctx context.Context, lockTTL time.Duration) {
	ok, err := d.store.redis.SetNX(ctx, constants.RedisDeviceEvalDispatchLockKey, d.instanceID, lockTTL).Result()
	if err != nil || !ok {
		return
	}
	claimed, expired, err := d.store.ClaimDeviceEvals(ctx, d.stopResumeLease(), deviceEvalLimit,
		d.instanceID+":dev:"+strconv.FormatInt(time.Now().UnixNano(), 36))
	if err != nil {
		d.logger.Warn("设备判断领取失败", zap.Error(err))
		return
	}
	if expired > 0 {
		d.logger.Warn("设备判断处理中标记到期（执行崩溃或超时）", zap.Int64("count", expired), zap.String("metric", "stop_resume.device_eval_expired"))
	}
	for _, item := range claimed {
		body, _ := json.Marshal(DeviceEvalMessage{DeviceID: item.DeviceID, Token: item.Token, Version: item.Version, Scenes: item.Scenes,
			DispatchedAt: time.Now().UnixMilli()})
		if err := d.publisher.Publish(constants.NSQDeviceEvalTopic, body); err != nil {
			releaseCtx, cancel := context.WithTimeout(context.WithoutCancel(ctx), 2*time.Second)
			if releaseErr := d.store.ReleaseDeviceEval(releaseCtx, item.DeviceID, item.Token); releaseErr != nil {
				d.logger.Warn("发布失败后释放设备判断失败", zap.Uint("device_id", item.DeviceID), zap.Error(releaseErr))
			}
			cancel()
			d.logger.Warn("发布设备判断失败，已释放", zap.Uint("device_id", item.DeviceID), zap.Error(err))
		}
	}
}