junhong_cmp_fiber/pkg/constants/stop_resume_queue.go

package constants

import (
	"fmt"
	"strings"
)

// 停复机指令队列（add-stop-resume-command-queue）。

// 停复机请求评估场景：随请求评估与指令保存，执行前重新判断沿用同一场景。
const (
	StopResumeScenePolling          = "polling"           // 套餐步骤（周期评估）
	StopResumeSceneFlowObserved     = "flow_observed"     // 流量增加观测
	StopResumeSceneNetworkObserved  = "network_observed"  // 网络状态变化观测
	StopResumeSceneRealnameObserved = "realname_observed" // 实名变化观测（含人工实名）
	StopResumeSceneActivation       = "activation"        // 套餐激活
	StopResumeSceneReset            = "reset"             // 流量重置
	StopResumeSceneOrder            = "order"             // 订单支付后套餐生效
	StopResumeSceneRefund           = "refund"            // 退款后资产无可用主套餐
	StopResumeScenePackageExpired   = "package_expired"   // 套餐到期且无其他有效主套餐
	StopResumeSceneRealnameControl  = "realname_control"  // 运营商账户实名管控修改
	StopResumeSceneUsage            = "usage"             // 流量扣减后
)

// StopResumeSceneAllowsSuspendedStop 判断场景集合中是否含套餐失效类场景：暂停轮询的卡在这些场景下照常停机。
func StopResumeSceneAllowsSuspendedStop(scenes []string) bool {
	for _, scene := range scenes {
		if scene == StopResumeScenePackageExpired || scene == StopResumeSceneRefund {
			return true
		}
	}
	return false
}

// JoinStopResumeScenes 把场景集合编码为逗号分隔串（去重、去空）。
func JoinStopResumeScenes(scenes []string) string {
	seen := map[string]bool{}
	parts := make([]string, 0, len(scenes))
	for _, scene := range scenes {
		scene = strings.TrimSpace(scene)
		if scene == "" || seen[scene] {
			continue
		}
		seen[scene] = true
		parts = append(parts, scene)
	}
	return strings.Join(parts, ",")
}

// SplitStopResumeScenes 解析逗号分隔的场景串。
func SplitStopResumeScenes(raw string) []string {
	if strings.TrimSpace(raw) == "" {
		return nil
	}
	parts := strings.Split(raw, ",")
	scenes := make([]string, 0, len(parts))
	for _, part := range parts {
		if part = strings.TrimSpace(part); part != "" {
			scenes = append(scenes, part)
		}
	}
	return scenes
}

// CheckTypeStopResume 是失败告警表中停复机方向的检查类型。
const CheckTypeStopResume = "stop_resume"

// 停复机指令失败类别（不来自上游错误分类的部分）。
const (
	StopResumeFailureUnknown = "unknown" // 结果未知（超时、连接失败、可能已受理）
)

// 停复机指令默认参数（账户未配置时使用）。
const (
	DefaultStopResumeMinIntervalSeconds  = 600 // 单卡最小重试间隔：上游卡状态变更每十分钟只允许调用一次
	DefaultStopResumeConfirmDelaySeconds = 60  // 结果未知后的确认延迟
	DefaultStopResumeInflightLimit       = 10  // 账户停复机执行上限
	DefaultStopResumeMaxInFlight         = 10  // 每进程停复机消费者 max_in_flight
)

// 停复机指令队列受控系统配置键（module = polling）。
const (
	SystemConfigStopResumeQueueEnabled   = "polling.stop_resume.queue_enabled"
	SystemConfigStopResumeUnknownAlert   = "polling.stop_resume.unknown_alert_threshold"
	SystemConfigStopResumeBackoff        = "polling.stop_resume.backoff_seconds"
	SystemConfigStopResumeAlertFailCount = "polling.stop_resume.alert_fail_count"
	// SystemConfigStopResumeStatusGrace 停复机成功后的状态保护窗口（秒）：窗口内查到操作前的状态视为上游滞后，不回写。0 表示不保护。
	SystemConfigStopResumeStatusGrace = "polling.stop_resume.status_grace_seconds"
)

// ========================================
// 停复机指令队列 Redis Key
// ========================================

// RedisCarrierStopResumeQueueKey 账户停复机指令队列（ZSET，member=card_id，score=可执行时间毫秒）。
func RedisCarrierStopResumeQueueKey(carrierID uint) string {
	return fmt.Sprintf("polling:carrier:%d:stop_resume", carrierID)
}

// RedisCarrierStopResumeRunningKey 账户停复机执行中计数（ZSET，member=token，score=租约到期毫秒）。
func RedisCarrierStopResumeRunningKey(carrierID uint) string {
	return fmt.Sprintf("polling:carrier:%d:sr_running", carrierID)
}

// RedisCardStopResumeCmdKey 卡停复机指令（Hash：action、reason、scenes、source、carrier、ver、fail、ffail、err、pending、unknown、at）。
func RedisCardStopResumeCmdKey(cardID uint) string {
	return fmt.Sprintf("polling:card:%d:stop_resume_cmd", cardID)
}

// RedisCardStopResumeClaimKey 停复机指令处理中标记，值=派发 token，TTL=租约。
func RedisCardStopResumeClaimKey(cardID uint) string {
	return fmt.Sprintf("polling:card:%d:sr_claim", cardID)
}

// RedisCardStopResumeExecKey 停复机指令执行权，值=token + 实例。
func RedisCardStopResumeExecKey(cardID uint) string {
	return fmt.Sprintf("polling:card:%d:sr_exec", cardID)
}

// RedisCardStopResumeLastCallKey 本卡最近一次停复机网关调用时间（毫秒，停机与复机合计）。
func RedisCardStopResumeLastCallKey(cardID uint) string {
	return fmt.Sprintf("polling:card:%d:sr_last_call", cardID)
}

// RedisDeviceEvalQueueKey 设备判断队列（ZSET，member=device_id，score=可执行时间毫秒）。
const RedisDeviceEvalQueueKey = "polling:device_eval:queue"

// RedisDeviceEvalRunningKey 设备判断执行中计数（ZSET，member=token）。
const RedisDeviceEvalRunningKey = "polling:device_eval:running"

// RedisDeviceEvalDispatchLockKey 设备判断分发锁。
const RedisDeviceEvalDispatchLockKey = "polling:device_eval:dispatch_lock"

// RedisDeviceEvalVersionKey 设备判断版本号（每次请求 INCR）。
func RedisDeviceEvalVersionKey(deviceID uint) string {
	return fmt.Sprintf("polling:device:%d:eval_ver", deviceID)
}

// RedisDeviceEvalScenesKey 设备待判断场景集合（Set）。
func RedisDeviceEvalScenesKey(deviceID uint) string {
	return fmt.Sprintf("polling:device:%d:eval_scenes", deviceID)
}

// RedisDeviceEvalClaimKey 设备判断处理中标记，值=token。
func RedisDeviceEvalClaimKey(deviceID uint) string {
	return fmt.Sprintf("polling:device:%d:eval_claim", deviceID)
}

// RedisDeviceEvalLockKey 设备判断锁（执行权），同一设备同一时刻只有一个判断在执行。
func RedisDeviceEvalLockKey(deviceID uint) string {
	return fmt.Sprintf("polling:device:%d:eval_lock", deviceID)
}

// NSQStopResumeTopic 返回账户停复机指令 topic。
func NSQStopResumeTopic(carrierID uint) string {
	return fmt.Sprintf("polling.stop_resume.%d", carrierID)
}

// NSQDeviceEvalTopic 设备判断 topic。
const NSQDeviceEvalTopic = "polling.device_eval"

// 停复机指令执行计数字段（按 10 分钟分桶，供监控汇总近 1 小时）。
const (
	StopResumeStatSuccess = "success"
	StopResumeStatFailed  = "failed"
	StopResumeStatRetry   = "retry"
	StopResumeStatUnknown = "unknown"
)

// StopResumeStatBucketSeconds 是停复机执行计数的分桶时长。
const StopResumeStatBucketSeconds = 600

// RedisCarrierStopResumeStatKey 账户停复机执行计数（Hash，field=计数字段），bucket = unix 秒 / 分桶时长。
func RedisCarrierStopResumeStatKey(carrierID uint, bucket int64) string {
	return fmt.Sprintf("polling:carrier:%d:sr_stats:%d", carrierID, bucket)
}