xiaolajiaoyyds
6f71df4849
多平台账号批量注册系统，支持 OpenAI / Grok / Kiro / Gemini 平台的全自动注册。 Features: - 用户名/密码注册登录（首个用户自动管理员） - JWT 认证 + bcrypt 密码加密 - 积分系统与多用户管理 - HuggingFace Space 弹性 Worker 节点管理 - Cloudflare Worker 负载均衡 - 实时日志推送（SSE/WebSocket） - 管理后台（用户管理、系统设置、代理池、卡池） Tech Stack: Go 1.25 / Gin / GORM / PostgreSQL / Vue 3 / Vite / TailwindCSS
1171 lines
36 KiB
Go
1171 lines
36 KiB
Go
package service
|
||
|
||
import (
|
||
"context"
|
||
"encoding/json"
|
||
"fmt"
|
||
"strconv"
|
||
"strings"
|
||
"sync"
|
||
"sync/atomic"
|
||
"time"
|
||
|
||
"github.com/rs/zerolog/log"
|
||
"github.com/xiaolajiaoyyds/regplatformm/internal/model"
|
||
"github.com/xiaolajiaoyyds/regplatformm/internal/worker"
|
||
"gorm.io/gorm"
|
||
)
|
||
|
||
// taskKey 运行中任务的唯一标识
|
||
type taskKey struct {
|
||
UserID uint
|
||
Platform string
|
||
}
|
||
|
||
// RunningTask 运行中的任务
|
||
type RunningTask struct {
|
||
TaskID uint
|
||
UserID uint
|
||
Platform string
|
||
Target int
|
||
Threads int
|
||
SuccessCount atomic.Int64
|
||
FailCount atomic.Int64
|
||
ClaimedCount atomic.Int64 // 已认领的注册名额（防止并发超发）
|
||
Stopping atomic.Bool // 用户已请求停止（允许立即启动新任务）
|
||
Cancel context.CancelFunc
|
||
LogCh chan string
|
||
StatusCh chan struct{} // 状态变更通知（成功/失败计数变化时触发）
|
||
Done chan struct{}
|
||
StartedAt time.Time // 任务启动时间（用于耗时统计）
|
||
subscribers []chan string
|
||
subMu sync.RWMutex
|
||
logBuffer []string // 日志缓冲区（供刷新页面后回放）
|
||
logBufMu sync.RWMutex
|
||
}
|
||
|
||
// NotifyStatus 通知前端状态已变更（非阻塞）
|
||
func (rt *RunningTask) NotifyStatus() {
|
||
select {
|
||
case rt.StatusCh <- struct{}{}:
|
||
default: // 已有待处理的通知，跳过
|
||
}
|
||
}
|
||
|
||
// Subscribe 订阅日志
|
||
func (rt *RunningTask) Subscribe() <-chan string {
|
||
ch := make(chan string, 200)
|
||
rt.subMu.Lock()
|
||
rt.subscribers = append(rt.subscribers, ch)
|
||
rt.subMu.Unlock()
|
||
return ch
|
||
}
|
||
|
||
// Unsubscribe 取消订阅
|
||
func (rt *RunningTask) Unsubscribe(ch <-chan string) {
|
||
rt.subMu.Lock()
|
||
defer rt.subMu.Unlock()
|
||
for i, sub := range rt.subscribers {
|
||
if sub == ch {
|
||
rt.subscribers = append(rt.subscribers[:i], rt.subscribers[i+1:]...)
|
||
close(sub)
|
||
return
|
||
}
|
||
}
|
||
}
|
||
|
||
// IsDone 任务是否完成
|
||
func (rt *RunningTask) IsDone() bool {
|
||
select {
|
||
case <-rt.Done:
|
||
return true
|
||
default:
|
||
return false
|
||
}
|
||
}
|
||
|
||
// Log 写入日志
|
||
func (rt *RunningTask) Log(format string, args ...interface{}) {
|
||
msg := fmt.Sprintf(format, args...)
|
||
// 防止往已关闭的 LogCh 写入导致 panic
|
||
defer func() { recover() }()
|
||
select {
|
||
case rt.LogCh <- msg:
|
||
default:
|
||
// 通道满了丢弃
|
||
}
|
||
}
|
||
|
||
// isInternalBrowserLog 判断 [~] 前缀的浏览器日志是否包含敏感/内部信息
|
||
// 返回 true 表示该日志不应展示给用户（URL、Token、IP、内部选择器等）
|
||
func isInternalBrowserLog(msg string) bool {
|
||
// 完整 URL（泄漏内部服务地址）
|
||
if strings.Contains(msg, "https://") || strings.Contains(msg, "http://") {
|
||
return true
|
||
}
|
||
// Token / 密钥值
|
||
if strings.Contains(msg, "Token:") || strings.Contains(msg, "token=") ||
|
||
strings.Contains(msg, "xsrf") || strings.Contains(msg, "cookie") {
|
||
return true
|
||
}
|
||
// OIDC 客户端凭据 / 设备码
|
||
if strings.Contains(msg, "clientId") || strings.Contains(msg, "user_code=") ||
|
||
strings.Contains(msg, "客户端已注册") {
|
||
return true
|
||
}
|
||
// IP / 地理位置
|
||
if strings.Contains(msg, "IP 地理位置") || strings.Contains(msg, "tz=") {
|
||
return true
|
||
}
|
||
// 内部页面状态（含 URL 片段）
|
||
if strings.Contains(msg, "页面:") || strings.Contains(msg, "URL:") ||
|
||
strings.Contains(msg, "注册链接:") || strings.Contains(msg, "页面已自动跳转") {
|
||
return true
|
||
}
|
||
// 邮箱明文 + 页面地址
|
||
if strings.Contains(msg, "填写邮箱:") {
|
||
return true
|
||
}
|
||
// Playwright CSS 选择器细节
|
||
if strings.Contains(msg, "点击按钮 (") || strings.Contains(msg, "尝试点击按钮") {
|
||
return true
|
||
}
|
||
// OIDC 内部步骤
|
||
if strings.Contains(msg, "OIDC Step") || strings.Contains(msg, "OIDC:") {
|
||
return true
|
||
}
|
||
// Playwright 内部状态
|
||
if strings.Contains(msg, "networkidle") {
|
||
return true
|
||
}
|
||
return false
|
||
}
|
||
|
||
// shouldShowToUser 判断日志是否应推送到前端
|
||
// 只保留用户关心的结果/进度/状态信息，过滤掉注册流程内部细节
|
||
func shouldShowToUser(msg string) bool {
|
||
// 远程浏览器服务的流式日志（Kiro/Gemini 注册过程）
|
||
// 需脱敏：隐藏 token/URL/cookie/内部技术细节，只展示进度
|
||
if strings.Contains(msg, "[~]") {
|
||
return !isInternalBrowserLog(msg)
|
||
}
|
||
// 成功结果（只保留 [✓] 最终汇总行，过滤掉中间步骤的 "注册成功" / "远程注册成功"）
|
||
if strings.Contains(msg, "[✓]") {
|
||
return true
|
||
}
|
||
// 分隔线
|
||
if strings.Contains(msg, "════") {
|
||
return true
|
||
}
|
||
// 任务汇总
|
||
if strings.Contains(msg, "任务完成") || strings.Contains(msg, "任务已加入") {
|
||
return true
|
||
}
|
||
// 统计信息（仅匹配任务汇总行，如 "成功: 5 失败: 3"，不匹配 "注册失败: xxx"）
|
||
if strings.Contains(msg, "成功:") && strings.Contains(msg, "失败:") {
|
||
return true
|
||
}
|
||
// 耗时/预计（不含宽泛的"等待"，避免排队轮询日志刷屏）
|
||
if strings.Contains(msg, "耗时") || strings.Contains(msg, "预计") {
|
||
return true
|
||
}
|
||
// 排队状态（仅匹配明确的排队反馈，不匹配内部 pollMailCode 等待日志）
|
||
if strings.Contains(msg, "排队等待完成") || strings.Contains(msg, "排队中") {
|
||
return true
|
||
}
|
||
// 进度指示
|
||
if strings.Contains(msg, "第 ") || strings.Contains(msg, "进度") {
|
||
return true
|
||
}
|
||
// 用户操作反馈
|
||
if strings.Contains(msg, "手动停止") || strings.Contains(msg, "强制停止") {
|
||
return true
|
||
}
|
||
return false
|
||
}
|
||
|
||
// broadcastLogs 广播日志到所有订阅者，同时缓存到 logBuffer
|
||
// 连续相同消息自动去重，避免大量并发注册时前端 DOM 爆炸
|
||
func (rt *RunningTask) broadcastLogs() {
|
||
var (
|
||
lastRaw string // 上一条原始消息（不含时间戳）
|
||
dupCount int // 连续重复计数
|
||
)
|
||
|
||
// sendOne 发送单条日志到 buffer + 所有 subscriber
|
||
// 日志始终写入 logBuffer（后端调试用），仅用户关心的日志推送到前端
|
||
sendOne := func(line string) {
|
||
rt.logBufMu.Lock()
|
||
rt.logBuffer = append(rt.logBuffer, line)
|
||
if len(rt.logBuffer) > 2000 {
|
||
rt.logBuffer = rt.logBuffer[len(rt.logBuffer)-2000:]
|
||
}
|
||
rt.logBufMu.Unlock()
|
||
|
||
if !shouldShowToUser(line) {
|
||
return
|
||
}
|
||
|
||
rt.subMu.RLock()
|
||
for _, sub := range rt.subscribers {
|
||
select {
|
||
case sub <- line:
|
||
default:
|
||
}
|
||
}
|
||
rt.subMu.RUnlock()
|
||
}
|
||
|
||
// flushDup 输出累积的重复计数
|
||
flushDup := func() {
|
||
if dupCount > 0 {
|
||
sendOne(fmt.Sprintf("[%s] [~] ↑ 重复 ×%d", time.Now().Format("15:04:05"), dupCount))
|
||
dupCount = 0
|
||
}
|
||
}
|
||
|
||
for msg := range rt.LogCh {
|
||
// 连续相同消息去重
|
||
if msg == lastRaw {
|
||
dupCount++
|
||
// 每累积 100 条刷一次，避免长时间无输出
|
||
if dupCount%100 == 0 {
|
||
flushDup()
|
||
}
|
||
continue
|
||
}
|
||
flushDup()
|
||
lastRaw = msg
|
||
sendOne(fmt.Sprintf("[%s] %s", time.Now().Format("15:04:05"), msg))
|
||
}
|
||
flushDup()
|
||
}
|
||
|
||
// GetLogBuffer 获取已缓存的历史日志（刷新页面后回放用）
|
||
func (rt *RunningTask) GetLogBuffer() []string {
|
||
rt.logBufMu.RLock()
|
||
defer rt.logBufMu.RUnlock()
|
||
buf := make([]string, len(rt.logBuffer))
|
||
copy(buf, rt.logBuffer)
|
||
return buf
|
||
}
|
||
|
||
// TaskEngine 任务引擎
|
||
type TaskEngine struct {
|
||
db *gorm.DB
|
||
proxyPool *ProxyPool
|
||
cardPool *CardPool
|
||
creditSvc *CreditService
|
||
settingSvc *SettingService
|
||
running map[taskKey]*RunningTask
|
||
mu sync.RWMutex
|
||
activeCount atomic.Int32 // 全局活跃任务计数
|
||
platformCount map[string]*atomic.Int32 // 按平台活跃任务计数
|
||
scheduler *QueueScheduler // 队列调度器（启动后通过 SetScheduler 注入）
|
||
dispatcher *Dispatcher // 全局 Round-Robin 调度器（所有任务共享 Worker Pool）
|
||
workerCount atomic.Int32 // 当前运行中的 Pool Worker 数
|
||
targetWorkers atomic.Int32 // 目标 Pool Worker 数（从配置动态读取，热生效）
|
||
configCache map[string]*cachedConfig // 按平台缓存的 ScanConfig 结果
|
||
configCacheMu sync.RWMutex
|
||
}
|
||
|
||
// cachedConfig 平台扫描配置缓存
|
||
type cachedConfig struct {
|
||
config worker.Config
|
||
scannedAt time.Time
|
||
}
|
||
|
||
// NewTaskEngine 创建任务引擎
|
||
func NewTaskEngine(db *gorm.DB, proxyPool *ProxyPool, cardPool *CardPool, creditSvc *CreditService, settingSvc *SettingService) *TaskEngine {
|
||
// 注意：running/stopping 任务的清理和退款由 QueueScheduler.RecoverOnBoot() 统一处理
|
||
// 此处不再重复清理，避免抢在退款之前把状态改掉
|
||
|
||
// 初始化按平台并发计数器
|
||
pc := make(map[string]*atomic.Int32)
|
||
for name := range worker.Registry {
|
||
pc[name] = &atomic.Int32{}
|
||
}
|
||
|
||
e := &TaskEngine{
|
||
db: db,
|
||
proxyPool: proxyPool,
|
||
cardPool: cardPool,
|
||
creditSvc: creditSvc,
|
||
settingSvc: settingSvc,
|
||
running: make(map[taskKey]*RunningTask),
|
||
platformCount: pc,
|
||
dispatcher: NewDispatcher(),
|
||
configCache: make(map[string]*cachedConfig),
|
||
}
|
||
e.startWorkerPool()
|
||
go e.startConfigRefresher()
|
||
return e
|
||
}
|
||
|
||
// SetScheduler 注入队列调度器（解决循环依赖）
|
||
func (e *TaskEngine) SetScheduler(s *QueueScheduler) {
|
||
e.scheduler = s
|
||
}
|
||
|
||
// countActiveUsers 统计当前有多少个不同用户在运行任务
|
||
func (e *TaskEngine) countActiveUsers() int {
|
||
e.mu.RLock()
|
||
defer e.mu.RUnlock()
|
||
users := make(map[uint]struct{})
|
||
for _, rt := range e.running {
|
||
if !rt.IsDone() {
|
||
users[rt.UserID] = struct{}{}
|
||
}
|
||
}
|
||
return len(users)
|
||
}
|
||
|
||
// userHasRunningTask 检查该用户是否已有运行中任务（已占槽位）
|
||
func (e *TaskEngine) userHasRunningTask(userID uint) bool {
|
||
e.mu.RLock()
|
||
defer e.mu.RUnlock()
|
||
for _, rt := range e.running {
|
||
if rt.UserID == userID && !rt.IsDone() {
|
||
return true
|
||
}
|
||
}
|
||
return false
|
||
}
|
||
|
||
// countActiveUsersByPlatform 统计指定平台当前有多少个不同用户在运行任务
|
||
func (e *TaskEngine) countActiveUsersByPlatform(platform string) int {
|
||
e.mu.RLock()
|
||
defer e.mu.RUnlock()
|
||
users := make(map[uint]struct{})
|
||
for _, rt := range e.running {
|
||
if rt.Platform == platform && !rt.IsDone() {
|
||
users[rt.UserID] = struct{}{}
|
||
}
|
||
}
|
||
return len(users)
|
||
}
|
||
|
||
// computeMaxInflight 计算指定平台的单任务最大并发数（统一算法，供启动时 + RefreshLimits 复用）
|
||
// 自适应策略：单人 80% 容量，多人均分，上限不超过平台容量的 90%
|
||
func (e *TaskEngine) computeMaxInflight(platform string) int32 {
|
||
defaultCap := 50
|
||
if platform == "kiro" || platform == "gemini" {
|
||
defaultCap = 15
|
||
}
|
||
maxConcurrent := e.settingSvc.GetInt(platform+"_max_concurrent", defaultCap)
|
||
if maxConcurrent <= 0 {
|
||
maxConcurrent = defaultCap
|
||
}
|
||
|
||
var activeCount int64
|
||
e.db.Model(&model.Task{}).Where("platform = ? AND status = ?", platform, "running").Count(&activeCount)
|
||
|
||
var maxParallel int
|
||
if activeCount <= 1 {
|
||
maxParallel = maxConcurrent * 4 / 5
|
||
} else {
|
||
maxParallel = maxConcurrent / int(activeCount)
|
||
}
|
||
|
||
// 上限：不超过平台总容量的 90%（预留缓冲给突发/新任务进入）
|
||
parallelCap := maxConcurrent * 9 / 10
|
||
if parallelCap < 5 {
|
||
parallelCap = 5
|
||
}
|
||
if maxParallel > parallelCap {
|
||
maxParallel = parallelCap
|
||
}
|
||
if maxParallel < 3 {
|
||
maxParallel = 3
|
||
}
|
||
return int32(maxParallel)
|
||
}
|
||
|
||
// userHasRunningTaskOnPlatform 检查该用户在指定平台是否已有运行中任务
|
||
func (e *TaskEngine) userHasRunningTaskOnPlatform(userID uint, platform string) bool {
|
||
key := taskKey{userID, platform}
|
||
e.mu.RLock()
|
||
defer e.mu.RUnlock()
|
||
rt, ok := e.running[key]
|
||
return ok && !rt.IsDone()
|
||
}
|
||
|
||
// EstimateWaitTime 估算排队用户的预计等待时间（秒）
|
||
// 基于该平台最近已完成任务的平均耗时 + 当前运行中任务的剩余时间
|
||
func (e *TaskEngine) EstimateWaitTime(queuePosition int, platform string) int {
|
||
// 查询该平台最近 50 个已完成任务的平均耗时
|
||
var avgDuration float64
|
||
e.db.Model(&model.Task{}).
|
||
Where("platform = ? AND status IN ? AND started_at IS NOT NULL AND stopped_at IS NOT NULL", platform, []string{"completed", "stopped"}).
|
||
Order("stopped_at DESC").Limit(50).
|
||
Select("COALESCE(AVG(EXTRACT(EPOCH FROM (stopped_at - started_at))), 120)").
|
||
Scan(&avgDuration)
|
||
if avgDuration < 10 {
|
||
avgDuration = 120 // 兜底 2 分钟
|
||
}
|
||
|
||
// 找该平台当前运行中任务的最短剩余时间
|
||
e.mu.RLock()
|
||
minRemaining := avgDuration
|
||
for _, rt := range e.running {
|
||
if rt.IsDone() || rt.Platform != platform {
|
||
continue
|
||
}
|
||
elapsed := time.Since(rt.StartedAt).Seconds()
|
||
remaining := avgDuration - elapsed
|
||
if remaining < 0 {
|
||
remaining = 30 // 已超时的任务给 30 秒缓冲
|
||
}
|
||
if remaining < minRemaining {
|
||
minRemaining = remaining
|
||
}
|
||
}
|
||
e.mu.RUnlock()
|
||
|
||
// 预计等待 = 最短剩余时间 × ceil(排队位置 / 槽位数)
|
||
maxUsers, _ := strconv.Atoi(e.settingSvc.Get(platform+"_max_concurrent_users", "0"))
|
||
if maxUsers <= 0 {
|
||
maxUsers, _ = strconv.Atoi(e.settingSvc.Get("max_concurrent_users", "1"))
|
||
}
|
||
if maxUsers <= 0 {
|
||
maxUsers = 1
|
||
}
|
||
rounds := (queuePosition + maxUsers - 1) / maxUsers
|
||
return int(minRemaining) * rounds
|
||
}
|
||
|
||
// formatWaitTime 格式化等待秒数为可读字符串
|
||
func formatWaitTime(sec int) string {
|
||
if sec < 60 {
|
||
return fmt.Sprintf("约 %d 秒", sec)
|
||
}
|
||
min := sec / 60
|
||
s := sec % 60
|
||
if s > 0 {
|
||
return fmt.Sprintf("约 %d 分 %d 秒", min, s)
|
||
}
|
||
return fmt.Sprintf("约 %d 分钟", min)
|
||
}
|
||
|
||
// StartTask 启动任务
|
||
func (e *TaskEngine) StartTask(userID, taskID uint, platform string, target, threads int) (bool, string) {
|
||
key := taskKey{userID, platform}
|
||
|
||
w, ok := worker.Get(platform)
|
||
if !ok {
|
||
return false, fmt.Sprintf("不支持的平台: %s", platform)
|
||
}
|
||
|
||
// ── 按平台用户槽位并发控制 ──
|
||
maxUsers, _ := strconv.Atoi(e.settingSvc.Get(platform+"_max_concurrent_users", "0"))
|
||
if maxUsers <= 0 {
|
||
// 向后兼容：平台专属未设置时回退到全局设置
|
||
maxUsers, _ = strconv.Atoi(e.settingSvc.Get("max_concurrent_users", "0"))
|
||
}
|
||
if maxUsers > 0 && !e.userHasRunningTaskOnPlatform(userID, platform) {
|
||
activeUsers := e.countActiveUsersByPlatform(platform)
|
||
if activeUsers >= maxUsers {
|
||
e.db.Model(&model.Task{}).Where("id = ? AND status != ?", taskID, "queued").Update("status", "queued")
|
||
// 计算该平台排队位置和预计等待时间
|
||
var queuePos int64
|
||
e.db.Model(&model.Task{}).
|
||
Where("status = ? AND platform = ? AND created_at < (SELECT created_at FROM tasks WHERE id = ?)", "queued", platform, taskID).
|
||
Count(&queuePos)
|
||
queuePos++ // 自己也算一个位置
|
||
waitSec := e.EstimateWaitTime(int(queuePos), platform)
|
||
log.Info().Uint("task_id", taskID).Uint("user_id", userID).Str("platform", platform).
|
||
Int("active_users", activeUsers).Int("limit", maxUsers).Int64("queue_pos", queuePos).
|
||
Msg("平台用户槽位已满，任务加入队列")
|
||
return true, fmt.Sprintf("排队中，前方 %d 位用户，预计等待 %s", queuePos, formatWaitTime(waitSec))
|
||
}
|
||
}
|
||
|
||
// ── 按平台并发控制 ──
|
||
platformLimitKey := platform + "_max_concurrent"
|
||
platformLimit, _ := strconv.Atoi(e.settingSvc.Get(platformLimitKey, "0"))
|
||
if platformLimit <= 0 {
|
||
// 浏览器类平台（kiro/gemini）并发能力远低于 HTTP 类（openai/grok），兜底值需区分
|
||
switch platform {
|
||
case "kiro", "gemini":
|
||
platformLimit = 8
|
||
default:
|
||
platformLimit = 100
|
||
}
|
||
}
|
||
pc := e.getPlatformCount(platform)
|
||
newPlatformActive := pc.Add(1)
|
||
if int(newPlatformActive) > platformLimit {
|
||
pc.Add(-1)
|
||
e.db.Model(&model.Task{}).Where("id = ? AND status != ?", taskID, "queued").Update("status", "queued")
|
||
log.Info().Uint("task_id", taskID).Str("platform", platform).
|
||
Int("active", int(newPlatformActive-1)).Int("limit", platformLimit).
|
||
Msg("平台并发已满，任务加入队列")
|
||
return true, "已加入队列"
|
||
}
|
||
|
||
// ── 全局并发控制（安全上限）──
|
||
globalLimit, _ := strconv.Atoi(e.settingSvc.Get("max_concurrent_tasks", "100"))
|
||
if globalLimit <= 0 {
|
||
globalLimit = 100
|
||
}
|
||
newActive := e.activeCount.Add(1)
|
||
if int(newActive) > globalLimit {
|
||
e.activeCount.Add(-1)
|
||
pc.Add(-1)
|
||
e.db.Model(&model.Task{}).Where("id = ? AND status != ?", taskID, "queued").Update("status", "queued")
|
||
log.Info().Uint("task_id", taskID).Int("active", int(newActive-1)).Int("limit", globalLimit).
|
||
Msg("全局并发已满，任务加入队列")
|
||
return true, "已加入队列"
|
||
}
|
||
|
||
ctx, cancel := context.WithCancel(context.Background())
|
||
rt := &RunningTask{
|
||
TaskID: taskID,
|
||
UserID: userID,
|
||
Platform: platform,
|
||
Target: target,
|
||
Threads: threads,
|
||
Cancel: cancel,
|
||
LogCh: make(chan string, 500),
|
||
StatusCh: make(chan struct{}, 1),
|
||
Done: make(chan struct{}),
|
||
StartedAt: time.Now(),
|
||
}
|
||
|
||
// 原子检查并设置，防止 TOCTOU 竞态
|
||
e.mu.Lock()
|
||
if existing, ok := e.running[key]; ok && !existing.IsDone() {
|
||
e.mu.Unlock()
|
||
cancel() // 清理刚创建的 context
|
||
e.activeCount.Add(-1) // 回滚全局计数
|
||
pc.Add(-1) // 回滚平台计数
|
||
return false, "该平台已有运行中的任务"
|
||
}
|
||
e.running[key] = rt
|
||
e.mu.Unlock()
|
||
|
||
// 更新任务状态为 running
|
||
now := time.Now()
|
||
e.db.Model(&model.Task{}).Where("id = ?", taskID).Updates(map[string]interface{}{
|
||
"status": "running",
|
||
"started_at": &now,
|
||
})
|
||
|
||
// 启动日志广播
|
||
go rt.broadcastLogs()
|
||
|
||
log.Info().Uint("task_id", taskID).Str("platform", platform).Int("target", target).Msg("任务已启动")
|
||
|
||
// 异步执行重量级操作（ScanConfig + 调度 + 监控），避免阻塞调用方
|
||
go e.executeStartedTask(ctx, cancel, rt, key, w, pc, taskID, userID, platform, target)
|
||
|
||
return true, "任务已启动"
|
||
}
|
||
|
||
// executeStartedTask 执行已通过并发检查的任务（ScanConfig + 调度 + 监控）
|
||
// 在 goroutine 中运行，避免阻塞 StartTask 的调用方
|
||
func (e *TaskEngine) executeStartedTask(
|
||
ctx context.Context, cancel context.CancelFunc,
|
||
rt *RunningTask, key taskKey, w worker.Worker, pc *atomic.Int32,
|
||
taskID, userID uint, platform string, target int,
|
||
) {
|
||
// 获取共享基础配置（ScanConfig 只调一次，所有 Pool Worker 共享结果）
|
||
baseCfg := worker.Config{}
|
||
e.injectSettings(baseCfg)
|
||
|
||
// 优先使用缓存配置（后台每 5 分钟自动刷新）
|
||
scannedCfg, cached := e.getCachedConfig(platform)
|
||
if !cached {
|
||
rt.Log("[*] 正在获取页面配置（直连失败时会走浏览器过盾，约需 30-60 秒）...")
|
||
var scanErr error
|
||
scannedCfg, scanErr = w.ScanConfig(ctx, e.proxyPool.GetNext(), baseCfg)
|
||
if scanErr != nil {
|
||
rt.Log("[-] 配置获取失败: %s", scanErr)
|
||
cancel()
|
||
close(rt.Done)
|
||
close(rt.LogCh) // 停止 broadcastLogs goroutine
|
||
stopNow := time.Now()
|
||
e.db.Model(&model.Task{}).Where("id = ?", taskID).Updates(map[string]interface{}{
|
||
"status": "stopped",
|
||
"stopped_at": &stopNow,
|
||
})
|
||
// 退还所有预扣积分
|
||
var task model.Task
|
||
if dbErr := e.db.First(&task, taskID).Error; dbErr == nil && task.CreditsReserved > 0 {
|
||
e.creditSvc.RefundCredits(e.db, userID, taskID, task.CreditsReserved)
|
||
}
|
||
e.mu.Lock()
|
||
delete(e.running, key)
|
||
e.mu.Unlock()
|
||
e.creditSvc.CleanupTask(taskID)
|
||
e.activeCount.Add(-1)
|
||
pc.Add(-1)
|
||
if e.scheduler != nil {
|
||
e.scheduler.TryDispatch()
|
||
}
|
||
return
|
||
}
|
||
// 扫描成功，更新缓存供后续任务使用
|
||
e.updateConfigCache(platform, scannedCfg)
|
||
} else {
|
||
rt.Log("[*] 使用缓存配置（后台每 5 分钟自动刷新）")
|
||
}
|
||
// 合并扫描配置到基础配置
|
||
for k, v := range scannedCfg {
|
||
if baseCfg[k] == "" {
|
||
baseCfg[k] = v
|
||
}
|
||
}
|
||
// 将扫描到的动态配置回写数据库，避免 site_key / action_id 过期
|
||
for _, k := range []string{"site_key", "action_id"} {
|
||
if v := scannedCfg[k]; v != "" {
|
||
dbKey := platform + "_" + k
|
||
if setErr := e.settingSvc.Set(dbKey, v); setErr != nil {
|
||
rt.Log("[!] 配置持久化失败 (%s): %s", dbKey, setErr)
|
||
} else {
|
||
rt.Log("[*] 动态配置已更新: %s = %s", dbKey, v)
|
||
}
|
||
}
|
||
}
|
||
|
||
// 注入用户指定的代理（任务创建时保存在 task.proxy_url）
|
||
{
|
||
var task model.Task
|
||
if err := e.db.Select("proxy_url").First(&task, taskID).Error; err == nil && task.ProxyURL != "" {
|
||
baseCfg["user_proxy"] = task.ProxyURL
|
||
rt.Log("[*] 使用用户指定代理")
|
||
}
|
||
}
|
||
|
||
// 解析多节点 URL 配置（逗号分隔 → 轮询分配到各 Pool Worker）
|
||
multiNodeKeys := []string{"kiro_reg_url", "gemini_reg_url", "openai_reg_url", "grok_reg_url", "camoufox_reg_url", "turnstile_solver_url", "cf_bypass_solver_url"}
|
||
nodeURLs := make(map[string][]string)
|
||
for _, nk := range multiNodeKeys {
|
||
if val := baseCfg[nk]; val != "" && strings.Contains(val, ",") {
|
||
parts := strings.Split(val, ",")
|
||
trimmed := make([]string, 0, len(parts))
|
||
for _, p := range parts {
|
||
if t := strings.TrimSpace(p); t != "" {
|
||
trimmed = append(trimmed, t)
|
||
}
|
||
}
|
||
if len(trimmed) > 1 {
|
||
nodeURLs[nk] = trimmed
|
||
rt.Log("[*] %s → %d 个节点轮询", nk, len(trimmed))
|
||
}
|
||
}
|
||
}
|
||
|
||
// 计算单任务最大并发数（与 RefreshLimits 相同的自适应算法，避免前 10 秒使用过时值）
|
||
maxParallel := int(e.computeMaxInflight(platform))
|
||
|
||
// 构建调度条目并加入全局 Dispatcher（Pool Worker 会自动拾取并执行）
|
||
entry := &DispatchEntry{
|
||
Key: key,
|
||
RT: rt,
|
||
Worker: w,
|
||
Ctx: ctx,
|
||
Cancel: cancel,
|
||
Config: baseCfg,
|
||
NodeURLs: nodeURLs,
|
||
Platform: platform,
|
||
done: make(chan struct{}),
|
||
}
|
||
entry.maxInflight.Store(int32(maxParallel))
|
||
e.dispatcher.Add(entry)
|
||
// 新任务加入后立即刷新所有任务的 maxInflight（不等 10 秒定时器，公平分配即时生效）
|
||
e.dispatcher.RefreshLimits(func(p string) int32 {
|
||
return e.computeMaxInflight(p)
|
||
})
|
||
|
||
rt.Log("════════════════════════════════════")
|
||
rt.Log("[*] 配置扫描完成，任务已加入调度池（当前 %d 个 Pool Worker，单任务并发上限 %d）", e.workerCount.Load(), maxParallel)
|
||
rt.Log("════════════════════════════════════")
|
||
|
||
// 监控 goroutine：等待调度条目完成后清理资源
|
||
go func() {
|
||
<-entry.done
|
||
|
||
success := int(rt.SuccessCount.Load())
|
||
fail := int(rt.FailCount.Load())
|
||
elapsed := time.Since(rt.StartedAt).Round(time.Second)
|
||
rt.Log("════════════════════════════════════")
|
||
rt.Log("[✓] 任务完成 成功: %d 失败: %d 耗时: %s", success, fail, elapsed)
|
||
rt.Log("════════════════════════════════════")
|
||
|
||
// 从调度器移除（停止分配新名额给此任务）
|
||
e.dispatcher.Remove(key)
|
||
// 立即刷新剩余任务的 maxInflight（任务退出后释放的容量即时分配给剩余任务）
|
||
e.dispatcher.RefreshLimits(func(p string) int32 {
|
||
return e.computeMaxInflight(p)
|
||
})
|
||
|
||
// 更新任务状态（区分用户停止 vs 自然完成）
|
||
stopNow := time.Now()
|
||
finalStatus := "completed"
|
||
if ctx.Err() != nil {
|
||
// context 被 cancel，检查是否用户主动停止（非达标取消）
|
||
total := rt.SuccessCount.Load() + rt.FailCount.Load()
|
||
if total < int64(target) {
|
||
finalStatus = "stopped"
|
||
}
|
||
}
|
||
e.db.Model(&model.Task{}).Where("id = ?", taskID).Updates(map[string]interface{}{
|
||
"status": finalStatus,
|
||
"stopped_at": &stopNow,
|
||
})
|
||
|
||
// 退还未消费积分（必须在 close(rt.Done) 之前，确保前端收到 complete 时退款已入库）
|
||
// 使用 DB 中的 success_count 作为退款依据（内存计数器可能因 DB 写入失败而偏高）
|
||
var task model.Task
|
||
if err := e.db.First(&task, taskID).Error; err == nil {
|
||
refund := task.CreditsReserved - task.SuccessCount
|
||
if refund > 0 {
|
||
e.creditSvc.RefundCredits(e.db, userID, taskID, refund)
|
||
}
|
||
}
|
||
|
||
// 关闭 Done channel → 触发 SSE complete 事件（此时退款已到账）
|
||
close(rt.Done)
|
||
close(rt.LogCh)
|
||
|
||
// 清理运行中任务 + 退款锁内存
|
||
e.mu.Lock()
|
||
delete(e.running, key)
|
||
e.mu.Unlock()
|
||
e.creditSvc.CleanupTask(taskID)
|
||
|
||
// 释放全局并发槽位 + 平台槽位，并尝试调度下一个排队任务
|
||
e.activeCount.Add(-1)
|
||
pc.Add(-1)
|
||
if e.scheduler != nil {
|
||
e.scheduler.TryDispatch()
|
||
}
|
||
}()
|
||
}
|
||
|
||
// StopTask 停止任务
|
||
func (e *TaskEngine) StopTask(userID uint, platform string) bool {
|
||
key := taskKey{userID, platform}
|
||
e.mu.RLock()
|
||
rt, ok := e.running[key]
|
||
e.mu.RUnlock()
|
||
if !ok || rt.IsDone() {
|
||
return false
|
||
}
|
||
|
||
rt.Log("[!] 用户手动停止任务")
|
||
rt.Stopping.Store(true)
|
||
rt.Cancel()
|
||
e.dispatcher.ForceComplete(key) // 立即标记完成，不等 inflight worker
|
||
|
||
// 更新状态
|
||
e.db.Model(&model.Task{}).Where("id = ?", rt.TaskID).Update("status", "stopping")
|
||
return true
|
||
}
|
||
|
||
// GetStatus 获取运行中任务的状态
|
||
func (e *TaskEngine) GetStatus(userID uint, platform string) *RunningTask {
|
||
key := taskKey{userID, platform}
|
||
e.mu.RLock()
|
||
defer e.mu.RUnlock()
|
||
rt, ok := e.running[key]
|
||
if !ok || rt.IsDone() || rt.Stopping.Load() {
|
||
return nil
|
||
}
|
||
return rt
|
||
}
|
||
|
||
// FindRunningTaskByID 通过 taskID 查找运行中任务
|
||
func (e *TaskEngine) FindRunningTaskByID(taskID uint) *RunningTask {
|
||
e.mu.RLock()
|
||
defer e.mu.RUnlock()
|
||
for _, rt := range e.running {
|
||
if rt.TaskID == taskID {
|
||
return rt
|
||
}
|
||
}
|
||
return nil
|
||
}
|
||
|
||
// RunningTaskInfo 运行中任务的安全快照（不暴露 channel/cancel 等内部字段）
|
||
type RunningTaskInfo struct {
|
||
TaskID uint `json:"task_id"`
|
||
UserID uint `json:"user_id"`
|
||
Platform string `json:"platform"`
|
||
Target int `json:"target"`
|
||
Threads int `json:"threads"`
|
||
SuccessCount int64 `json:"success_count"`
|
||
FailCount int64 `json:"fail_count"`
|
||
StartedAt time.Time `json:"started_at"`
|
||
Stopping bool `json:"stopping"`
|
||
}
|
||
|
||
// ListAllRunning 返回所有运行中任务的快照（管理员用）
|
||
func (e *TaskEngine) ListAllRunning() []RunningTaskInfo {
|
||
e.mu.RLock()
|
||
defer e.mu.RUnlock()
|
||
result := make([]RunningTaskInfo, 0, len(e.running))
|
||
for _, rt := range e.running {
|
||
if rt.IsDone() {
|
||
continue
|
||
}
|
||
result = append(result, RunningTaskInfo{
|
||
TaskID: rt.TaskID,
|
||
UserID: rt.UserID,
|
||
Platform: rt.Platform,
|
||
Target: rt.Target,
|
||
Threads: rt.Threads,
|
||
SuccessCount: rt.SuccessCount.Load(),
|
||
FailCount: rt.FailCount.Load(),
|
||
StartedAt: rt.StartedAt,
|
||
Stopping: rt.Stopping.Load(),
|
||
})
|
||
}
|
||
return result
|
||
}
|
||
|
||
// AdminStopTask 管理员强制停止任务（通过 taskID 查找）
|
||
func (e *TaskEngine) AdminStopTask(taskID uint) bool {
|
||
e.mu.RLock()
|
||
var found *RunningTask
|
||
var foundKey taskKey
|
||
for k, rt := range e.running {
|
||
if rt.TaskID == taskID && !rt.IsDone() {
|
||
found = rt
|
||
foundKey = k
|
||
break
|
||
}
|
||
}
|
||
e.mu.RUnlock()
|
||
|
||
if found == nil {
|
||
return false
|
||
}
|
||
|
||
log.Info().Uint("task_id", taskID).Msg("管理员强制停止任务")
|
||
found.Log("[!] 管理员强制停止任务")
|
||
found.Stopping.Store(true)
|
||
found.Cancel()
|
||
e.dispatcher.ForceComplete(foundKey) // 立即标记完成，不等 inflight worker
|
||
|
||
e.db.Model(&model.Task{}).Where("id = ?", taskID).Update("status", "stopping")
|
||
return true
|
||
}
|
||
|
||
// getPlatformCount 获取平台并发计数器（惰性初始化）
|
||
func (e *TaskEngine) getPlatformCount(platform string) *atomic.Int32 {
|
||
e.mu.RLock()
|
||
if pc, ok := e.platformCount[platform]; ok {
|
||
e.mu.RUnlock()
|
||
return pc
|
||
}
|
||
e.mu.RUnlock()
|
||
|
||
e.mu.Lock()
|
||
defer e.mu.Unlock()
|
||
if pc, ok := e.platformCount[platform]; ok {
|
||
return pc
|
||
}
|
||
pc := &atomic.Int32{}
|
||
e.platformCount[platform] = pc
|
||
return pc
|
||
}
|
||
|
||
// saveResult 保存注册结果
|
||
func (e *TaskEngine) saveResult(taskID, userID uint, platform, email string, credential map[string]interface{}) {
|
||
credJSON, err := json.Marshal(credential)
|
||
if err != nil {
|
||
log.Error().Err(err).Uint("task_id", taskID).Msg("credential 序列化失败")
|
||
}
|
||
result := model.TaskResult{
|
||
TaskID: taskID,
|
||
UserID: userID,
|
||
Platform: platform,
|
||
Email: email,
|
||
CredentialData: credJSON,
|
||
}
|
||
// Grok 向后兼容：提取 auth_token（兼容旧数据 sso_token）
|
||
if platform == "grok" {
|
||
if token, ok := credential["auth_token"].(string); ok {
|
||
result.SSOToken = token
|
||
} else if token, ok := credential["sso_token"].(string); ok {
|
||
result.SSOToken = token
|
||
}
|
||
}
|
||
if err := e.db.Create(&result).Error; err != nil {
|
||
log.Error().Err(err).Uint("task_id", taskID).Str("email", email).Msg("保存注册结果失败")
|
||
}
|
||
}
|
||
|
||
// injectSettings 将系统设置注入到 Worker 配置中（不覆盖已有值）
|
||
func (e *TaskEngine) injectSettings(cfg worker.Config) {
|
||
// 需要注入的系统设置键列表
|
||
keys := []string{
|
||
"yydsmail_api_key",
|
||
"yydsmail_base_url",
|
||
"email_provider_priority",
|
||
"turnstile_solver_url",
|
||
"turnstile_solver_proxy",
|
||
"cf_bypass_solver_url",
|
||
"capsolver_key",
|
||
"yescaptcha_key",
|
||
"grok_action_id",
|
||
"grok_site_key",
|
||
"camoufox_reg_url",
|
||
"openai_reg_url",
|
||
"kiro_reg_url",
|
||
"grok_reg_url",
|
||
"grok_email_providers",
|
||
"openai_email_providers",
|
||
"kiro_email_providers",
|
||
"kiro_proxy",
|
||
"gemini_reg_url",
|
||
"gemini_email_providers",
|
||
"gemini_proxy",
|
||
"grok_proxy",
|
||
"openai_proxy",
|
||
"stripe_pk",
|
||
}
|
||
for _, k := range keys {
|
||
if cfg[k] == "" {
|
||
if v := e.settingSvc.Get(k, ""); v != "" {
|
||
cfg[k] = v
|
||
}
|
||
}
|
||
}
|
||
}
|
||
|
||
// startWorkerPool 启动 Worker Pool 并开启后台动态伸缩
|
||
// 每 10 秒检查 hf_instance_count × workers_per_instance，自动扩容/缩容
|
||
func (e *TaskEngine) startWorkerPool() {
|
||
poolSize := e.calcPoolSize()
|
||
e.targetWorkers.Store(int32(poolSize))
|
||
e.workerCount.Store(int32(poolSize))
|
||
for i := 0; i < poolSize; i++ {
|
||
go e.runPoolWorker()
|
||
}
|
||
go e.poolResizer()
|
||
log.Info().Int("pool_size", poolSize).Msg("Worker Pool 已启动")
|
||
}
|
||
|
||
// calcPoolSize 根据 HF 实例数 × 每实例并发数 计算 Pool 大小
|
||
func (e *TaskEngine) calcPoolSize() int {
|
||
instances, _ := strconv.Atoi(e.settingSvc.Get("hf_instance_count", "20"))
|
||
if instances <= 0 {
|
||
instances = 20
|
||
}
|
||
perInstance, _ := strconv.Atoi(e.settingSvc.Get("workers_per_instance", "10"))
|
||
if perInstance <= 0 {
|
||
perInstance = 10
|
||
}
|
||
return instances * perInstance
|
||
}
|
||
|
||
// poolResizer 后台动态伸缩 Worker Pool + 刷新任务并发上限
|
||
// 每 10 秒读取配置，扩容立即生效，缩容等 Worker 干完手头活后退出
|
||
// 同时刷新所有活跃任务的 maxInflight，使 SyncPoolSize 的动态参数热生效
|
||
func (e *TaskEngine) poolResizer() {
|
||
ticker := time.NewTicker(10 * time.Second)
|
||
defer ticker.Stop()
|
||
for range ticker.C {
|
||
desired := int32(e.calcPoolSize())
|
||
e.targetWorkers.Store(desired)
|
||
|
||
current := e.workerCount.Load()
|
||
if desired > current {
|
||
// 扩容：立即补充差额 Worker
|
||
diff := desired - current
|
||
for i := int32(0); i < diff; i++ {
|
||
e.workerCount.Add(1)
|
||
go e.runPoolWorker()
|
||
}
|
||
log.Info().Int32("old", current).Int32("new", desired).Int32("spawned", diff).Msg("Worker Pool 扩容")
|
||
} else if desired < current {
|
||
// 缩容：多余 Worker 完成当前任务后自动退出
|
||
log.Info().Int32("current", current).Int32("target", desired).Msg("Worker Pool 缩容中")
|
||
e.dispatcher.Wake() // 唤醒空闲 Worker 让它们检查退出条件
|
||
}
|
||
|
||
// 动态刷新活跃任务的 maxInflight（复用 computeMaxInflight 统一算法）
|
||
e.dispatcher.RefreshLimits(func(p string) int32 {
|
||
return e.computeMaxInflight(p)
|
||
})
|
||
}
|
||
}
|
||
|
||
// runPoolWorker 单个 Pool Worker 的主循环
|
||
// 不断从 Dispatcher 获取下一个注册任务并执行，空闲时阻塞等待
|
||
func (e *TaskEngine) runPoolWorker() {
|
||
defer e.workerCount.Add(-1)
|
||
for {
|
||
// 缩容检查：超过目标数的 Worker 主动退出
|
||
if e.workerCount.Load() > e.targetWorkers.Load() {
|
||
return
|
||
}
|
||
job, ok := e.dispatcher.Next(e.proxyPool)
|
||
if !ok {
|
||
return // Dispatcher 已关闭（服务关停）
|
||
}
|
||
e.executeRegJob(job)
|
||
}
|
||
}
|
||
|
||
// executeRegJob 执行单次注册任务（含重试逻辑）
|
||
// 由 Pool Worker 调用，完成后通知 DispatchEntry 检查任务是否全部结束
|
||
func (e *TaskEngine) executeRegJob(job *RegJob) {
|
||
entry := job.Entry
|
||
rt := entry.RT
|
||
|
||
defer func() {
|
||
entry.inflight.Add(-1)
|
||
entry.checkCompletion()
|
||
e.dispatcher.Wake() // 释放并发槽位，唤醒等待中的 Worker
|
||
}()
|
||
|
||
const maxRetries = 3
|
||
|
||
// openai_team: 从卡池分配卡并注入配置
|
||
var cardID uint
|
||
if rt.Platform == "openai_team" && e.cardPool != nil {
|
||
card := e.cardPool.GetNext()
|
||
if card == nil {
|
||
rt.Log("[-] 卡池无可用卡，跳过")
|
||
rt.ClaimedCount.Add(-1)
|
||
rt.FailCount.Add(1)
|
||
rt.NotifyStatus()
|
||
e.db.Model(&model.Task{}).Where("id = ?", rt.TaskID).
|
||
Update("fail_count", gorm.Expr("fail_count + 1"))
|
||
e.dispatcher.Wake()
|
||
return
|
||
}
|
||
cardID = card.ID
|
||
job.Cfg["card_number"] = card.CardNumber
|
||
job.Cfg["exp_month"] = strconv.Itoa(card.ExpMonth)
|
||
job.Cfg["exp_year"] = strconv.Itoa(card.ExpYear)
|
||
job.Cfg["cvc"] = card.CVC
|
||
job.Cfg["billing_name"] = card.BillingName
|
||
job.Cfg["billing_email"] = card.BillingEmail
|
||
job.Cfg["billing_country"] = card.BillingCountry
|
||
job.Cfg["billing_city"] = card.BillingCity
|
||
job.Cfg["billing_line1"] = card.BillingLine1
|
||
job.Cfg["billing_zip"] = card.BillingZip
|
||
}
|
||
|
||
for retryN := 1; retryN <= maxRetries; retryN++ {
|
||
// 用户已停止 → 立即退出，不再尝试注册
|
||
if entry.Ctx.Err() != nil {
|
||
break
|
||
}
|
||
var succeeded, failed bool
|
||
entry.Worker.RegisterOne(entry.Ctx, worker.RegisterOpts{
|
||
Proxy: job.Proxy,
|
||
Config: job.Cfg,
|
||
LogCh: rt.LogCh,
|
||
OnSuccess: func(email string, credential map[string]interface{}) {
|
||
succeeded = true
|
||
newCount := rt.SuccessCount.Add(1)
|
||
rt.NotifyStatus()
|
||
// 实时推送成功账号到前端日志（序号+邮箱）
|
||
rt.Log("[✓] #%d %s", newCount, email)
|
||
e.saveResult(rt.TaskID, rt.UserID, rt.Platform, email, credential)
|
||
// 原子递增防止并发 worker 互相覆盖
|
||
e.db.Model(&model.Task{}).Where("id = ?", rt.TaskID).
|
||
Update("success_count", gorm.Expr("success_count + 1"))
|
||
// 卡池生命周期：标记卡已使用
|
||
if cardID > 0 && e.cardPool != nil {
|
||
e.cardPool.MarkUsed(cardID)
|
||
}
|
||
if newCount >= int64(rt.Target) {
|
||
entry.Cancel()
|
||
}
|
||
},
|
||
OnFail: func() {
|
||
failed = true
|
||
},
|
||
})
|
||
if succeeded || !failed {
|
||
break // 成功或 ctx 已取消，退出重试
|
||
}
|
||
if retryN < maxRetries {
|
||
// 中间重试不推送前端，静默重试
|
||
} else {
|
||
rt.Log("[-] 已重试 %d 次，计入失败", maxRetries)
|
||
rt.ClaimedCount.Add(-1) // 释放名额，让 Pool Worker 可以重新认领
|
||
// 卡池生命周期：标记卡失败
|
||
if cardID > 0 && e.cardPool != nil {
|
||
e.cardPool.MarkFailed(cardID)
|
||
}
|
||
newCount := rt.FailCount.Add(1)
|
||
rt.NotifyStatus()
|
||
// 原子递增防止并发 worker 互相覆盖
|
||
e.db.Model(&model.Task{}).Where("id = ?", rt.TaskID).
|
||
Update("fail_count", gorm.Expr("fail_count + 1"))
|
||
if newCount >= int64(rt.Target) {
|
||
entry.Cancel()
|
||
}
|
||
e.dispatcher.Wake() // 唤醒等待中的 Worker，重新认领释放的名额
|
||
}
|
||
}
|
||
}
|
||
|
||
// ─── ScanConfig 全局缓存 ───
|
||
|
||
// startConfigRefresher 后台定时刷新所有平台的扫描配置
|
||
// 首次延迟 10 秒（等服务就绪），之后每 5 分钟刷新一次
|
||
func (e *TaskEngine) startConfigRefresher() {
|
||
time.Sleep(10 * time.Second)
|
||
e.refreshAllConfigs()
|
||
|
||
ticker := time.NewTicker(5 * time.Minute)
|
||
defer ticker.Stop()
|
||
for range ticker.C {
|
||
e.refreshAllConfigs()
|
||
}
|
||
}
|
||
|
||
// refreshAllConfigs 刷新所有已注册平台的配置缓存
|
||
func (e *TaskEngine) refreshAllConfigs() {
|
||
for name, w := range worker.Registry {
|
||
ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
|
||
baseCfg := worker.Config{}
|
||
e.injectSettings(baseCfg)
|
||
|
||
scanned, err := w.ScanConfig(ctx, e.proxyPool.GetNext(), baseCfg)
|
||
cancel()
|
||
if err != nil {
|
||
log.Warn().Str("platform", name).Err(err).Msg("配置缓存刷新失败（保留旧缓存）")
|
||
continue
|
||
}
|
||
|
||
e.updateConfigCache(name, scanned)
|
||
log.Info().Str("platform", name).Msg("配置缓存已刷新")
|
||
}
|
||
}
|
||
|
||
// getCachedConfig 获取平台的缓存配置（10 分钟内有效）
|
||
func (e *TaskEngine) getCachedConfig(platform string) (worker.Config, bool) {
|
||
e.configCacheMu.RLock()
|
||
defer e.configCacheMu.RUnlock()
|
||
cc, ok := e.configCache[platform]
|
||
if !ok || time.Since(cc.scannedAt) > 10*time.Minute {
|
||
return nil, false
|
||
}
|
||
// 返回副本，避免并发修改
|
||
cp := make(worker.Config, len(cc.config))
|
||
for k, v := range cc.config {
|
||
cp[k] = v
|
||
}
|
||
return cp, true
|
||
}
|
||
|
||
// updateConfigCache 更新平台配置缓存
|
||
func (e *TaskEngine) updateConfigCache(platform string, cfg worker.Config) {
|
||
e.configCacheMu.Lock()
|
||
defer e.configCacheMu.Unlock()
|
||
e.configCache[platform] = &cachedConfig{
|
||
config: cfg,
|
||
scannedAt: time.Now(),
|
||
}
|
||
}
|