junhong_cmp_fiber/pkg/constants/carrier_polling.go

package constants

import "fmt"

// 卡轮询调度模式（受控系统配置 polling.scheduler.mode），全局整体切换，不按账户灰度。
const (
	PollingSchedulerModeLegacy  = "legacy"  // 旧分片调度（Asynq）
	PollingSchedulerModeCarrier = "carrier" // 按运营商账户调度（Redis 延时队列 + NSQ）
)

// 卡任务检查类型（新调度内部与配置使用的短名）。
const (
	CheckTypeCardStatus = "card_status"
	CheckTypeCarddata   = "carddata"
	CheckTypeRealname   = "realname"
	CheckTypePackage    = "package"
)

// AllCheckTypes 返回全部检查类型（默认步骤顺序）。返回新切片。
func AllCheckTypes() []string {
	return []string{CheckTypeCardStatus, CheckTypeCarddata, CheckTypeRealname, CheckTypePackage}
}

// CheckTypeFromTaskType 把旧任务类型（polling:xxx）转为检查类型短名；不认识返回空串。
func CheckTypeFromTaskType(taskType string) string {
	switch taskType {
	case TaskTypePollingCardStatus:
		return CheckTypeCardStatus
	case TaskTypePollingCarddata:
		return CheckTypeCarddata
	case TaskTypePollingRealname:
		return CheckTypeRealname
	case TaskTypePollingPackage:
		return CheckTypePackage
	default:
		return ""
	}
}

// TaskTypeFromCheckType 把检查类型短名转为旧任务类型（polling:xxx）；不认识返回空串。
func TaskTypeFromCheckType(checkType string) string {
	switch checkType {
	case CheckTypeCardStatus:
		return TaskTypePollingCardStatus
	case CheckTypeCarddata:
		return TaskTypePollingCarddata
	case CheckTypeRealname:
		return TaskTypePollingRealname
	case CheckTypePackage:
		return TaskTypePollingPackage
	default:
		return ""
	}
}

// CheckTypeName 返回检查类型中文名称。
func CheckTypeName(checkType string) string {
	switch checkType {
	case CheckTypeCardStatus:
		return "卡状态检查"
	case CheckTypeCarddata:
		return "流量检查"
	case CheckTypeRealname:
		return "实名检查"
	case CheckTypePackage:
		return "套餐检查"
	case CheckTypeStopResume:
		return "停复机"
	case PollingAlertCheckTypeAll:
		return "全部"
	default:
		return checkType
	}
}

// CheckTypeCallsGateway 判断检查类型是否调用运营商网关（套餐检查只按规则判断停复机，不调网关）。
func CheckTypeCallsGateway(checkType string) bool {
	return checkType == CheckTypeCardStatus || checkType == CheckTypeCarddata || checkType == CheckTypeRealname
}

// 请求检查场景（tb_polling_check_scene.scene，代码预置）。
const (
	PollingCheckSceneOrderPaid           = "order_paid"
	PollingCheckScenePackageActivated    = "package_activated"
	PollingCheckSceneStopResumeSucceeded = "stop_resume_succeeded"
	PollingCheckSceneDeviceOperation     = "device_operation"
	PollingCheckSceneAutoRenewalPurchase = "auto_renewal_purchase"
	PollingCheckSceneClientAssetDetail   = "client_asset_detail"
	PollingCheckSceneAdminAssetRealtime  = "admin_asset_realtime"
	PollingCheckSceneRealnameLink        = "realname_link"
)

// 失败告警级别与状态（tb_polling_failure_alert）。
const (
	PollingAlertLevelCard    = "card"
	PollingAlertLevelCarrier = "carrier"
	// PollingAlertCheckTypeAll 是账户级告警的检查类型。
	PollingAlertCheckTypeAll = "all"

	PollingAlertStatusOpen            = 1 // 未处理
	PollingAlertStatusRecovered       = 2 // 已恢复（首次成功自动转入）
	PollingAlertStatusPollingDisabled = 3 // 已关闭轮询
	PollingAlertStatusHandled         = 4 // 已处理（人工标记）
)

// PollingAlertStatusName 返回失败告警状态名称。
func PollingAlertStatusName(status int) string {
	switch status {
	case PollingAlertStatusOpen:
		return "未处理"
	case PollingAlertStatusRecovered:
		return "已恢复"
	case PollingAlertStatusPollingDisabled:
		return "已关闭轮询"
	case PollingAlertStatusHandled:
		return "已处理"
	default:
		return ""
	}
}

// 失败类别中不计入连续失败的取值（限流等待超时与上游频率限制）。
const (
	// PollingFailureRateLimited 本账户限流等待超时（请求未发出）。
	PollingFailureRateLimited = "rate_limited"
)

// 按运营商账户调度的受控系统配置键（module = polling）。
const (
	SystemConfigPollingSchedulerMode         = "polling.scheduler.mode"
	SystemConfigPollingStepOrder             = "polling.scheduler.step_order"
	SystemConfigPollingStepTimeout           = "polling.scheduler.step_timeout_seconds"
	SystemConfigPollingTaskTimeoutMargin     = "polling.scheduler.task_timeout_margin_seconds"
	SystemConfigPollingSkipPackageAfterFlow  = "polling.scheduler.skip_package_after_flow"
	SystemConfigPollingFailureBackoff        = "polling.failure.backoff_seconds"
	SystemConfigPollingAlertFailCount        = "polling.failure.alert_fail_count"
	SystemConfigPollingAlertFailDuration     = "polling.failure.alert_fail_duration_seconds"
	SystemConfigPollingImmediateAlert        = "polling.failure.immediate_alert_categories"
	SystemConfigPollingAccountAlertWindow    = "polling.failure.account_alert_window_seconds"
	SystemConfigPollingAccountAlertThreshold = "polling.failure.account_alert_threshold"
)

// ========================================
// 按运营商账户调度的 Redis Key（前缀与旧分片调度不冲突）
// ========================================

// RedisCarrierPriorityQueueKey 账户优先队列（ZSET，member=card_id，score=可执行时间毫秒）。
func RedisCarrierPriorityQueueKey(carrierID uint) string {
	return fmt.Sprintf("polling:carrier:%d:priority", carrierID)
}

// RedisCarrierQueueKey 账户普通队列（ZSET，member=card_id，score=该卡最早到期时间毫秒）。
func RedisCarrierQueueKey(carrierID uint) string {
	return fmt.Sprintf("polling:carrier:%d:queue", carrierID)
}

// RedisCarrierRunningKey 账户执行中计数（ZSET，member=派发 token，score=租约到期毫秒），只用于执行上限与监控。
func RedisCarrierRunningKey(carrierID uint) string {
	return fmt.Sprintf("polling:carrier:%d:running", carrierID)
}

// RedisCarrierDispatchLockKey 账户分发锁：同一账户每轮只由一个分发器实例领取。
func RedisCarrierDispatchLockKey(carrierID uint) string {
	return fmt.Sprintf("polling:carrier:%d:dispatch_lock", carrierID)
}

// RedisCarrierExpiredClaimsKey 账户处理中标记到期次数（Hash，field=分钟，value=次数），供监控。
func RedisCarrierExpiredClaimsKey(carrierID uint) string {
	return fmt.Sprintf("polling:carrier:%d:expired_claims", carrierID)
}

// RedisCardDueKey 卡到期信息（Hash）：各检查类型 due:{type}（毫秒）、fail:{type}、first_fail:{type}、
// err:{type}、ok:{type}（最近成功毫秒）、confirm:{type}（待确认延迟序列）、carrier（所属账户）、priority（优先标记）。
func RedisCardDueKey(cardID uint) string {
	return fmt.Sprintf("polling:card:%d:due", cardID)
}

// RedisCardClaimKey 卡任务处理中标记，值=派发 token，TTL=租约。
func RedisCardClaimKey(cardID uint) string {
	return fmt.Sprintf("polling:card:%d:claim", cardID)
}

// RedisCardExecKey 卡任务执行权，值=派发 token + 消费者实例，TTL=续期间隔×2。
func RedisCardExecKey(cardID uint) string {
	return fmt.Sprintf("polling:card:%d:exec", cardID)
}

// RedisCarrierSchedulerHeartbeatKey 分发器心跳（Hash，field=实例，value=毫秒）。
const RedisCarrierSchedulerHeartbeatKey = "polling:carrier:dispatcher:heartbeat"

// RedisCarrierSchedulerInitStateKey 新调度初始化状态（String：running / done），初始化完成前分发器不领取。
const RedisCarrierSchedulerInitStateKey = "polling:carrier:init_state"

// RedisCarrierAccountAlertKey 账户级告警窗口计数（ZSET，member=告警ID，score=毫秒）。
func RedisCarrierAccountAlertKey(carrierID uint) string {
	return fmt.Sprintf("polling:carrier:%d:alert_window", carrierID)
}

// NSQ topic 与 channel。
const (
	NSQChannelWorker = "worker"
)

// NSQCardTopic 返回账户卡任务 topic。
func NSQCardTopic(carrierID uint) string {
	return fmt.Sprintf("polling.card.%d", carrierID)
}