dygou_go/internal/seckill/scheduler.go

package seckill

import (
	"bufio"
	"errors"
	"fmt"
	"math/rand"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"time"

	"github.com/golang/freetype/truetype"

	"douyu_go/internal/client"
	"douyu_go/internal/constants"
	"douyu_go/internal/image"
	"douyu_go/internal/logger"
	"douyu_go/internal/types"
	"douyu_go/internal/utils"
)

const (
	warmupBeforeTrigger = 2 * time.Second
	warmupMaxAge        = 30 * time.Second
	timeCalibrateRounds = 3
	timeCalibrateWait   = 1200 * time.Millisecond
	resyncBeforeTrigger = 10 * time.Second
	seckillMaxDuration  = 75 * time.Second
)

type roundOutcome struct {
	successAccounts map[string]bool
	stoppedAccounts map[string]string
	stoppedProducts map[string]string
	attempts        int
}

// Scheduler 抢购调度器。
type Scheduler struct {
	client *client.DouyuClient
	paths  types.AppPaths
	font   *truetype.Font

	concurrency int // 抢购并发数
	attempts    int // 单号重试次数

	accountsMu sync.RWMutex
	accounts   []types.Account

	resultsMu sync.RWMutex
	results   map[string]string

	runMu   sync.Mutex
	running bool

	scheduleMu     sync.Mutex
	schedule       *time.Timer
	warmupSchedule *time.Timer
	resyncSchedule *time.Timer

	warmupsMu sync.RWMutex
	warmups   map[string]types.WarmupData
}

// New 创建抢购调度器。
func New(cli *client.DouyuClient, paths types.AppPaths, font *truetype.Font, concurrency, attempts int) *Scheduler {
	if concurrency <= 0 {
		concurrency = constants.DefaultConcurrency
	}
	if attempts <= 0 {
		attempts = constants.DefaultAttempts
	}
	return &Scheduler{
		client:      cli,
		paths:       paths,
		font:        font,
		concurrency: concurrency,
		attempts:    attempts,
		results:     make(map[string]string),
	}
}

// UpdateSettings 运行时更新并发数和重试次数。
func (s *Scheduler) UpdateSettings(concurrency, attempts int) {
	if concurrency > 0 {
		s.concurrency = concurrency
	}
	if attempts > 0 {
		s.attempts = attempts
	}
}

// LoadAccounts 从账号文件加载账号。
func (s *Scheduler) LoadAccounts() {
	path := s.paths.AccountFile
	if path == "" {
		path = constants.DefaultAccountFile
	}

	// 文件不存在时自动创建目录和空文件，方便用户直接粘贴 Cookie
	if _, err := os.Stat(path); os.IsNotExist(err) {
		if dir := filepath.Dir(path); dir != "" && dir != "." {
			if mkErr := os.MkdirAll(dir, 0755); mkErr != nil {
				logger.LogEvent("账号", "warn", "账号目录创建失败: %s: %v", dir, mkErr)
			}
		}
		hint := "# 格式: 账号名----密码----Cookie\n# 每行一个账号，粘贴后保存即可\n"
		if wrErr := os.WriteFile(path, []byte(hint), 0644); wrErr != nil {
			logger.LogEvent("账号", "warn", "账号文件创建失败: %s: %v", path, wrErr)
		} else {
			logger.LogEvent("账号", "info", "已创建空账号文件: %s，请粘贴 Cookie 后重新加载", path)
		}
	}

	f, err := os.Open(path)
	if err != nil {
		logger.LogEvent("账号", "error", "读取账号文件失败: %s: %v", path, err)
		s.accountsMu.Lock()
		s.accounts = nil
		s.accountsMu.Unlock()
		return
	}
	defer f.Close()

	var list []types.Account
	scanner := bufio.NewScanner(f)
	scanner.Buffer(make([]byte, 1024), 1024*1024)
	for scanner.Scan() {
		line := strings.TrimSpace(scanner.Text())
		if line == "" || strings.HasPrefix(line, "#") {
			continue
		}
		parts := strings.SplitN(line, "----", 3)
		if len(parts) != 3 {
			logger.LogEvent("账号", "warn", "账号行格式错误，已跳过: %s", line)
			continue
		}
		list = append(list, types.Account{
			Name:     strings.TrimSpace(parts[0]),
			Password: strings.TrimSpace(parts[1]),
			Cookie:   strings.TrimSpace(parts[2]),
		})
	}
	if err := scanner.Err(); err != nil {
		logger.LogEvent("账号", "error", "读取账号文件中断: %v", err)
	}

	s.accountsMu.Lock()
	s.accounts = list
	s.accountsMu.Unlock()
	logger.LogEvent("账号", "success", "账号加载完成: %d 个，来源=%s", len(list), path)
}

// AccountCount 返回当前账号数。
func (s *Scheduler) AccountCount() int {
	s.accountsMu.RLock()
	defer s.accountsMu.RUnlock()
	return len(s.accounts)
}

// AccountSnapshot 复制一份账号列表。
func (s *Scheduler) AccountSnapshot() []types.Account {
	s.accountsMu.RLock()
	defer s.accountsMu.RUnlock()
	list := make([]types.Account, len(s.accounts))
	copy(list, s.accounts)
	return list
}

// AccountCookieByName 按账号名查找 Cookie。
func (s *Scheduler) AccountCookieByName(name string) (string, bool) {
	s.accountsMu.RLock()
	defer s.accountsMu.RUnlock()
	for _, account := range s.accounts {
		if account.Name == name {
			return account.Cookie, true
		}
	}
	return "", false
}

// Results 返回当前抢购结果快照。
func (s *Scheduler) Results() map[string]string {
	s.resultsMu.RLock()
	defer s.resultsMu.RUnlock()
	out := make(map[string]string, len(s.results))
	for k, v := range s.results {
		out[k] = v
	}
	return out
}

// ClearResults 清空抢购结果。
func (s *Scheduler) ClearResults() {
	s.resultsMu.Lock()
	defer s.resultsMu.Unlock()
	s.results = make(map[string]string)
}

func (s *Scheduler) setResult(key, value string) {
	s.resultsMu.Lock()
	defer s.resultsMu.Unlock()
	s.results[key] = value
}

// TryStartRun 尝试开始抢购，返回 false 表示已有任务在运行。
func (s *Scheduler) TryStartRun() bool {
	s.runMu.Lock()
	defer s.runMu.Unlock()
	if s.running {
		return false
	}
	s.running = true
	return true
}

func (s *Scheduler) finishRun() {
	s.runMu.Lock()
	defer s.runMu.Unlock()
	s.running = false
}

// RunNow 立即执行抢购任务，持续循环直到超时或全部成功。
func (s *Scheduler) RunNow(accountProducts map[string]string) {
	defer s.finishRun()
	s.ClearResults()

	allAttempts := s.buildAttempts(accountProducts)
	if len(allAttempts) == 0 {
		logger.LogEvent("抢购", "warn", "没有可执行的抢购任务")
		return
	}

	successAccounts := make(map[string]bool)
	stoppedAccounts := make(map[string]string)
	stoppedProducts := make(map[string]string)
	totalAccounts := uniqueAttemptAccounts(allAttempts)
	totalAttempts := 0
	startTime := time.Now()
	round := 0

	logger.LogEvent("抢购", "info", "开始抢购: %d 个账号，窗口=%v，前段高频后段降速", totalAccounts, seckillMaxDuration)

	for {
		elapsed := time.Since(startTime)
		if elapsed >= seckillMaxDuration {
			logger.LogEvent("抢购", "info", "已达到抢购窗口 %v，停止抢购", seckillMaxDuration)
			break
		}

		round++
		remaining := seckillMaxDuration - elapsed

		pendingAttempts := s.filterPendingAttempts(allAttempts, stoppedAccounts, stoppedProducts)
		if len(pendingAttempts) == 0 {
			logger.LogEvent("抢购", "info", "所有账号都已成功或遇到终止条件，停止抢购")
			break
		}

		roundAttempts := s.pickRoundAttempts(pendingAttempts, round)
		if len(roundAttempts) == 0 {
			logger.LogEvent("抢购", "info", "第%d轮无可执行账号，停止抢购", round)
			break
		}

		logger.LogEvent("抢购", "info", "第%d轮: 待抢购 %d 个账号，本轮执行=%d，剩余时间=%v",
			round, uniqueAttemptAccounts(pendingAttempts), len(roundAttempts), utils.RoundDuration(remaining))

		outcome := s.executeRound(roundAttempts)
		totalAttempts += outcome.attempts

		for account := range outcome.successAccounts {
			successAccounts[account] = true
			stoppedAccounts[account] = "已抢到"
		}
		for account, reason := range outcome.stoppedAccounts {
			if stoppedAccounts[account] == "" {
				stoppedAccounts[account] = reason
			}
		}
		for product, reason := range outcome.stoppedProducts {
			if stoppedProducts[product] == "" {
				stoppedProducts[product] = reason
				logger.LogEvent("抢购", "warn", "商品停止重试: %s，原因=%s", product, reason)
			}
		}

		if len(outcome.successAccounts) > 0 {
			logger.LogEvent("抢购", "success", "第%d轮成功: %d 个账号", round, len(outcome.successAccounts))
		}
		if len(stoppedAccounts) >= totalAccounts {
			logger.LogEvent("抢购", "info", "所有账号都已成功或停止重试，停止抢购")
			break
		}

		retryInterval := seckillRetryInterval(time.Since(startTime))
		if retryInterval > 0 && time.Since(startTime)+retryInterval < seckillMaxDuration {
			time.Sleep(jitterDuration(retryInterval))
		}
	}

	logger.LogEvent("抢购", "info", "抢购任务结束: 成功=%d/%d，停止=%d，账号尝试=%d，耗时=%v",
		len(successAccounts), totalAccounts, len(stoppedAccounts)-len(successAccounts), totalAttempts, utils.RoundDuration(time.Since(startTime)))
}

// filterPendingAttempts 过滤掉已终止账号和已确认无库存的商品。
func (s *Scheduler) filterPendingAttempts(allAttempts []types.ExchangeAttempt, stoppedAccounts, stoppedProducts map[string]string) []types.ExchangeAttempt {
	var pending []types.ExchangeAttempt
	for _, attempt := range allAttempts {
		if stoppedAccounts[attempt.Account.Name] != "" {
			continue
		}
		if stoppedProducts[attempt.Product.CommodityID] != "" {
			continue
		}
		pending = append(pending, attempt)
	}
	return pending
}

// pickRoundAttempts 每轮每个账号最多执行一次，避免同账号并发过猛触发频控。
func (s *Scheduler) pickRoundAttempts(attempts []types.ExchangeAttempt, round int) []types.ExchangeAttempt {
	if len(attempts) == 0 {
		return nil
	}
	preferredIndex := 1
	if s.attempts > 0 {
		preferredIndex = ((round - 1) % s.attempts) + 1
	}

	byAccount := make(map[string]types.ExchangeAttempt)
	fallback := make(map[string]types.ExchangeAttempt)
	order := make([]string, 0, len(attempts))
	for _, attempt := range attempts {
		name := attempt.Account.Name
		if _, ok := fallback[name]; !ok {
			fallback[name] = attempt
			order = append(order, name)
		}
		if attempt.Index == preferredIndex {
			byAccount[name] = attempt
		}
	}

	out := make([]types.ExchangeAttempt, 0, len(order))
	for _, name := range order {
		if attempt, ok := byAccount[name]; ok {
			out = append(out, attempt)
		} else {
			out = append(out, fallback[name])
		}
	}
	return out
}

// executeRound 执行一轮抢购，并返回成功和终止重试的账号。
func (s *Scheduler) executeRound(attempts []types.ExchangeAttempt) roundOutcome {
	outcome := roundOutcome{
		successAccounts: make(map[string]bool),
		stoppedAccounts: make(map[string]string),
		stoppedProducts: make(map[string]string),
	}
	var mu sync.Mutex

	limit := make(chan struct{}, s.concurrency)
	var wg sync.WaitGroup

	for _, attempt := range attempts {
		wg.Add(1)
		go func(attempt types.ExchangeAttempt) {
			defer wg.Done()
			limit <- struct{}{}
			defer func() { <-limit }()

			result := s.runAttempt(attempt)
			key := fmt.Sprintf("%s_第%d包", result.Account, result.Attempt)
			s.setResult(key, result.Message)

			mu.Lock()
			outcome.attempts++
			mu.Unlock()

			if result.OK {
				logger.LogEvent("抢购", "success", "成功: %s %s 第%d包 %s", result.Account, result.Product, result.Attempt, result.Message)
				go image.SaveSuccessImage(attempt.Account, attempt.Product, result.RoleName, s.font, s.paths.SuccessDir)

				mu.Lock()
				outcome.successAccounts[attempt.Account.Name] = true
				mu.Unlock()
			} else {
				logger.LogEvent("抢购", "info", "尝试: %s %s 第%d包 %s", result.Account, result.Product, result.Attempt, result.Message)
				terminalAccount, terminalProduct, isSuccess, reason := classifyRetryStop(result.Message)
				if terminalAccount || terminalProduct {
					mu.Lock()
					if terminalAccount {
						outcome.stoppedAccounts[attempt.Account.Name] = reason
					}
					if terminalProduct {
						outcome.stoppedProducts[attempt.Product.CommodityID] = reason
					}
					// 兑换上限也算成功
					if isSuccess {
						outcome.successAccounts[attempt.Account.Name] = true
						logger.LogEvent("抢购", "success", "成功（兑换上限）: %s %s 第%d包", result.Account, result.Product, result.Attempt)
						go image.SaveSuccessImage(attempt.Account, attempt.Product, result.RoleName, s.font, s.paths.SuccessDir)
					}
					mu.Unlock()
				}
			}
		}(attempt)
	}

	wg.Wait()
	return outcome
}

func seckillRetryInterval(elapsed time.Duration) time.Duration {
	switch {
	case elapsed < 8*time.Second:
		return 180 * time.Millisecond
	case elapsed < 25*time.Second:
		return 350 * time.Millisecond
	default:
		return 900 * time.Millisecond
	}
}

func jitterDuration(base time.Duration) time.Duration {
	if base <= 0 {
		return 0
	}
	// 加一点抖动，避免所有机器严格同频请求。
	jitter := time.Duration(rand.Int63n(int64(base / 3)))
	return base + jitter
}

func classifyRetryStop(message string) (account bool, product bool, success bool, reason string) {
	switch {
	case strings.Contains(message, "商品本次活动库存已耗完"):
		return true, true, false, "商品库存已耗完"
	case strings.Contains(message, "您已达本次活动兑换上限"):
		return true, false, true, "已抢到（兑换上限）"
	case strings.Contains(message, "未绑定账号"):
		return true, false, false, "账号未绑定角色"
	case strings.Contains(message, "需要开通精英手册"):
		return true, false, false, "账号不满足精英手册条件"
	default:
		return false, false, false, ""
	}
}

func uniqueAttemptAccounts(attempts []types.ExchangeAttempt) int {
	seen := make(map[string]struct{}, len(attempts))
	for _, attempt := range attempts {
		seen[attempt.Account.Name] = struct{}{}
	}
	return len(seen)
}

func (s *Scheduler) buildAttempts(accountProducts map[string]string) []types.ExchangeAttempt {
	accounts := s.AccountSnapshot()
	var attempts []types.ExchangeAttempt
	for _, account := range accounts {
		commodityID := accountProducts[account.Name]
		if commodityID == "" {
			continue
		}
		product := types.Product{CommodityID: commodityID, CommodityName: commodityID}
		for i := 1; i <= s.attempts; i++ {
			attempts = append(attempts, types.ExchangeAttempt{
				Account: account,
				Product: product,
				Index:   i,
			})
		}
	}
	return attempts
}

func warmupKey(attempt types.ExchangeAttempt) string {
	return fmt.Sprintf("%s|%s|%d", attempt.Account.Name, attempt.Product.CommodityID, attempt.Index)
}

func (s *Scheduler) setWarmup(key string, data types.WarmupData) {
	s.warmupsMu.Lock()
	defer s.warmupsMu.Unlock()
	if s.warmups == nil {
		s.warmups = make(map[string]types.WarmupData)
	}
	s.warmups[key] = data
}

func (s *Scheduler) warmupFor(attempt types.ExchangeAttempt) (types.WarmupData, bool) {
	s.warmupsMu.RLock()
	data, ok := s.warmups[warmupKey(attempt)]
	s.warmupsMu.RUnlock()
	if !ok || data.CreatedAt.IsZero() || time.Since(data.CreatedAt) > warmupMaxAge {
		return types.WarmupData{}, false
	}
	return data, true
}

func (s *Scheduler) clearWarmups() {
	s.warmupsMu.Lock()
	defer s.warmupsMu.Unlock()
	s.warmups = make(map[string]types.WarmupData)
}

func (s *Scheduler) warmupAttempts(attempts []types.ExchangeAttempt) {
	if len(attempts) == 0 {
		return
	}
	s.clearWarmups()
	logger.LogEvent("预热", "info", "开始预热: %d 个请求", len(attempts))

	limit := make(chan struct{}, s.concurrency)
	var wg sync.WaitGroup
	for _, attempt := range attempts {
		wg.Add(1)
		go func(attempt types.ExchangeAttempt) {
			defer wg.Done()
			limit <- struct{}{}
			defer func() { <-limit }()

			start := time.Now()
			csrf, err := s.client.GenerateCSRF(attempt.Account.Cookie)
			if err != nil {
				logger.LogEvent("预热", "warn", "CSRF预热失败: 账号=%s 商品=%s err=%v", attempt.Account.Name, attempt.Product.CommodityName, err)
				return
			}
			roleName, err := s.client.RoleName(attempt.Account.Cookie)
			if err != nil {
				logger.LogEvent("预热", "warn", "角色预热失败: 账号=%s err=%v", attempt.Account.Name, err)
			}
			s.setWarmup(warmupKey(attempt), types.WarmupData{
				CSRF:      csrf,
				RoleName:  roleName,
				CreatedAt: time.Now(),
			})
			logger.LogEvent("预热", "success", "预热完成: 账号=%s 商品=%s 耗时=%s", attempt.Account.Name, attempt.Product.CommodityName, utils.RoundDuration(time.Since(start)))
		}(attempt)
	}
	wg.Wait()
	logger.LogEvent("预热", "info", "预热任务结束")
}

func (s *Scheduler) calibrateDouyuTime(label string) types.TimeCalibration {
	calibration, err := s.client.BestServerTimeCalibration(timeCalibrateRounds, timeCalibrateWait)
	if err != nil {
		logger.LogEvent("时间", "warn", "%s校准降级: %v", label, err)
	}
	logger.LogEvent("时间", "success", "%s校准完成: 斗鱼估算=%s 偏移=%dms 误差<=%dms RTT=%dms 采样=%d 来源=%s",
		label,
		calibration.EstimatedServerNow().In(constants.CST).Format("2006-01-02 15:04:05.000"),
		calibration.Offset.Milliseconds(),
		calibration.ErrorBound.Milliseconds(),
		calibration.RTT.Milliseconds(),
		calibration.Samples,
		calibration.Source)
	return calibration
}

func (s *Scheduler) replaceTimerLocked(timer **time.Timer, d time.Duration, fn func()) {
	if *timer != nil {
		(*timer).Stop()
	}
	*timer = time.AfterFunc(d, fn)
}

func (s *Scheduler) stopSchedulesLocked() {
	if s.schedule != nil {
		s.schedule.Stop()
		s.schedule = nil
	}
	if s.warmupSchedule != nil {
		s.warmupSchedule.Stop()
		s.warmupSchedule = nil
	}
	if s.resyncSchedule != nil {
		s.resyncSchedule.Stop()
		s.resyncSchedule = nil
	}
}

func (s *Scheduler) runAttempt(attempt types.ExchangeAttempt) types.ExchangeResult {
	result := types.ExchangeResult{
		Account: attempt.Account.Name,
		Product: attempt.Product.CommodityName,
		Attempt: attempt.Index,
	}

	logger.LogEvent("抢购", "info", "执行兑换链路: 账号=%s 商品=%s 第%d包", attempt.Account.Name, attempt.Product.CommodityName, attempt.Index)
	var roleName string
	csrf := ""
	if warmup, ok := s.warmupFor(attempt); ok {
		csrf = warmup.CSRF
		roleName = warmup.RoleName
		logger.LogEvent("抢购", "info", "使用预热CSRF: 账号=%s 商品=%s 预热距今=%s", attempt.Account.Name, attempt.Product.CommodityName, utils.RoundDuration(time.Since(warmup.CreatedAt)))
	}
	if csrf == "" {
		var err error
		csrf, err = s.client.GenerateCSRF(attempt.Account.Cookie)
		if err != nil {
			result.Message = "失败|生成CSRF失败: " + err.Error()
			return result
		}
		logger.LogEvent("抢购", "info", "CSRF已生成，本次兑换链路复用同一个 token: 账号=%s 商品=%s", attempt.Account.Name, attempt.Product.CommodityName)
	}

	limit, err := s.client.BatchExchangeLimit(attempt.Account.Cookie, attempt.Product.CommodityID, &csrf)
	if err != nil {
		result.Message = "失败|批量限制检查失败: " + err.Error()
		return result
	}
	if limit <= 0 {
		result.Message = "失败|当前商品限制兑换"
		return result
	}
	logger.LogEvent("抢购", "info", "批量限制通过: 账号=%s 商品=%s limit=%d", attempt.Account.Name, attempt.Product.CommodityName, limit)

	if err := s.client.PreExchange(attempt.Account.Cookie, attempt.Product.CommodityID, &csrf); err != nil {
		result.Message = "失败|预兑换失败: " + err.Error()
		return result
	}

	exchangeInfo, err := s.client.Exchange(attempt.Account.Cookie, attempt.Product.CommodityID, &csrf)
	if err != nil {
		result.Message = "失败|" + err.Error()
		return result
	}

	if roleName == "" {
		roleName, err = s.client.RoleName(attempt.Account.Cookie)
		if err != nil {
			logger.LogEvent("抢购", "warn", "%s 角色名查询失败: %v", attempt.Account.Name, err)
		}
	}

	result.OK = true
	result.RoleName = roleName
	if roleName != "" {
		result.Message = "成功|" + roleName + "|" + exchangeInfo
	} else {
		result.Message = "成功|" + exchangeInfo
	}
	return result
}

// ScheduleRun 注册一次定时抢购。
func (s *Scheduler) ScheduleRun(req types.SeckillRequest) (map[string]any, error) {
	if len(req.AccountProducts) == 0 {
		return nil, errors.New("请选择账号和商品")
	}
	if strings.TrimSpace(req.TargetTime) == "" {
		return nil, errors.New("请设置目标时间")
	}

	targetLocal, err := time.ParseInLocation("2006-01-02 15:04:05", req.TargetTime, constants.CST)
	if err != nil {
		return nil, fmt.Errorf("时间格式错误: %w", err)
	}
	targetTime := targetLocal.UTC()

	calibration := s.calibrateDouyuTime("定时")
	serverTime := calibration.EstimatedServerNow()

	advance := time.Duration(req.Advance) * time.Millisecond
	triggerTime := targetTime.Add(-advance)
	remain := triggerTime.Sub(serverTime)
	if remain <= 0 {
		return nil, errors.New("目标时间已过")
	}

	ap := utils.CloneStringMap(req.AccountProducts)
	attempts := s.buildAttempts(ap)
	s.clearWarmups()
	s.scheduleMu.Lock()
	s.stopSchedulesLocked()
	s.scheduleMu.Unlock()

	s.scheduleTimers(ap, attempts, targetTime, triggerTime, calibration, false)

	resyncRemain := triggerTime.Sub(serverTime) - resyncBeforeTrigger
	if resyncRemain < 0 {
		resyncRemain = 0
	}

	s.scheduleMu.Lock()
	s.replaceTimerLocked(&s.resyncSchedule, resyncRemain, func() {
		newCalibration := s.calibrateDouyuTime("临近")
		s.scheduleTimers(ap, attempts, targetTime, triggerTime, newCalibration, true)
	})
	s.scheduleMu.Unlock()

	warmupRemain := remain - warmupBeforeTrigger
	if warmupRemain < 0 {
		warmupRemain = 0
	}

	logger.LogEvent("定时", "success", "定时已设置: 目标=%s 触发=%s 提前=%dms 斗鱼估算=%s 本地偏移=%dms 误差<=%dms RTT=%dms 预热=%s 复校=%s",
		targetTime.In(constants.CST).Format("2006-01-02 15:04:05.000"),
		triggerTime.In(constants.CST).Format("2006-01-02 15:04:05.000"),
		req.Advance,
		serverTime.In(constants.CST).Format("2006-01-02 15:04:05.000"),
		calibration.Offset.Milliseconds(),
		calibration.ErrorBound.Milliseconds(),
		calibration.RTT.Milliseconds(),
		time.Now().Add(warmupRemain).In(constants.CST).Format("15:04:05.000"),
		time.Now().Add(resyncRemain).In(constants.CST).Format("15:04:05.000"),
	)

	return map[string]any{
		"success":      true,
		"targetTime":   targetTime.In(constants.CST).Format("2006-01-02 15:04:05"),
		"triggerTime":  triggerTime.In(constants.CST).Format("2006-01-02 15:04:05.000"),
		"advance":      req.Advance,
		"remainSecond": int(remain.Seconds()),
		"serverTime":   serverTime.In(constants.CST).Format("2006-01-02 15:04:05.000"),
		"offsetMs":     calibration.Offset.Milliseconds(),
		"errorBoundMs": calibration.ErrorBound.Milliseconds(),
		"rttMs":        calibration.RTT.Milliseconds(),
	}, nil
}

func (s *Scheduler) scheduleTimers(accountProducts map[string]string, attempts []types.ExchangeAttempt, targetTime, triggerTime time.Time, calibration types.TimeCalibration, resync bool) {
	serverTime := calibration.EstimatedServerNow()
	remain := triggerTime.Sub(serverTime)
	if remain < 0 {
		remain = 0
	}
	warmupRemain := remain - warmupBeforeTrigger
	if warmupRemain < 0 {
		warmupRemain = 0
	}

	s.scheduleMu.Lock()
	defer s.scheduleMu.Unlock()
	s.replaceTimerLocked(&s.warmupSchedule, warmupRemain, func() {
		logger.LogEvent("预热", "success", "预热触发: 本地=%s", time.Now().In(constants.CST).Format("15:04:05.000"))
		s.warmupAttempts(attempts)
	})
	s.replaceTimerLocked(&s.schedule, remain, func() {
		now := time.Now()
		estimatedServer := now.Add(calibration.Offset)
		logger.LogEvent("定时", "success", "定时触发: 本地=%s 斗鱼估算=%s 偏移=%dms 误差<=%dms",
			now.In(constants.CST).Format("15:04:05.000"),
			estimatedServer.In(constants.CST).Format("15:04:05.000"),
			calibration.Offset.Milliseconds(),
			calibration.ErrorBound.Milliseconds())
		if !s.TryStartRun() {
			logger.LogEvent("抢购", "warn", "已有抢购任务正在运行，本次定时触发已跳过")
			return
		}
		s.RunNow(accountProducts)
	})
	if resync {
		logger.LogEvent("定时", "success", "临近复校已应用: 目标=%s 触发=%s 斗鱼估算=%s 本地偏移=%dms 误差<=%dms RTT=%dms 预热=%s",
			targetTime.In(constants.CST).Format("2006-01-02 15:04:05.000"),
			triggerTime.In(constants.CST).Format("2006-01-02 15:04:05.000"),
			serverTime.In(constants.CST).Format("2006-01-02 15:04:05.000"),
			calibration.Offset.Milliseconds(),
			calibration.ErrorBound.Milliseconds(),
			calibration.RTT.Milliseconds(),
			time.Now().Add(warmupRemain).In(constants.CST).Format("15:04:05.000"))
	}
}

// CancelSchedule 停止尚未触发的定时任务。
func (s *Scheduler) CancelSchedule() bool {
	s.scheduleMu.Lock()
	defer s.scheduleMu.Unlock()
	stopped := false
	if s.schedule != nil {
		stopped = s.schedule.Stop()
	}
	s.stopSchedulesLocked()
	s.clearWarmups()
	return stopped
}