dygou_go/internal/seckill/schedule_timer.go

package seckill

import (
	"errors"
	"fmt"
	"math"
	"sync/atomic"
	"time"

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

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

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

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

	advance := dynamicAdvance(calibration)
	triggerTime := targetTime.Add(-advance)
	remain := triggerTime.Sub(serverTime)
	if remain <= 0 {
		return types.ScheduleResult{}, errors.New("目标时间已过")
	}

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

	s.scheduleTimers(ap, targetTime, triggerTime, calibration, req.Attempts, req.DelayMs, false)

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

	s.scheduleMu.Lock()
	s.replaceTimerLocked(&s.resyncSchedule, resyncRemain, func() {
		if s.IsRunning() {
			logger.LogEvent("定时", "info", "临近复校时抢购已开始，不再改写触发时刻")
			return
		}
		newCalibration := s.calibrateDouyuTime("临近")
		if s.IsRunning() {
			logger.LogEvent("定时", "info", "临近复校完成时抢购已开始，保持当前任务")
			return
		}
		newTriggerTime := targetTime.Add(-dynamicAdvance(newCalibration))
		s.scheduleTimers(ap, targetTime, newTriggerTime, newCalibration, req.Attempts, req.DelayMs, 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"),
		advance.Milliseconds(),
		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"),
	)

	remainSecond := int(math.Ceil(triggerTime.Sub(serverTime).Seconds()))
	if remainSecond < 0 {
		remainSecond = 0
	}

	return types.ScheduleResult{
		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"),
		RemainSecond: remainSecond,
		OffsetMs:     calibration.Offset.Milliseconds(),
		ErrorBoundMs: calibration.ErrorBound.Milliseconds(),
		RttMs:        calibration.RTT.Milliseconds(),
	}, nil
}

func (s *Scheduler) scheduleTimers(accountProducts map[string]string, targetTime, triggerTime time.Time, calibration types.TimeCalibration, attempts, delayMs int, resync bool) {
	serverTime := calibration.EstimatedServerNow()
	remain := triggerTime.Sub(serverTime)
	if remain < 0 {
		remain = 0
	}
	s.scheduleMu.Lock()
	defer s.scheduleMu.Unlock()
	if !resync {
		warmupRemain := remain - warmupBeforeTrigger
		if warmupRemain < 0 {
			warmupRemain = 0
		}
		gen := atomic.AddInt64(&s.warmupGen, 1)
		s.replaceTimerLocked(&s.warmupSchedule, warmupRemain, func() {
			if atomic.LoadInt64(&s.warmupGen) != gen {
				return
			}
			if s.clientSnapshot().Activity.IsXiaodian {
				return
			}
			logger.LogEvent("预热", "success", "预热触发: 本地=%s", time.Now().In(constants.CST).Format("15:04:05.000"))
			warmupTasks := s.buildTasks(accountProducts)
			s.warmupAttempts(warmupTasks)
		})
	}
	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
		}
		if s.clientSnapshot().Activity.IsXiaodian {
			s.RunXiaodian(accountProducts, attempts, delayMs, "")
			return
		}
		s.RunNow(accountProducts)
	})
	if resync {
		logger.LogEvent("定时", "success", "临近复校已应用: 目标=%s 触发=%s 斗鱼估算=%s 偏移=%dms 误差<=%dms RTT=%dms",
			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())
	}
}

// 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
}

func (s *Scheduler) calibrateDouyuTime(label string) types.TimeCalibration {
	cli := s.clientSnapshot()
	calibration, err := cli.BestServerTimeCalibration(timeCalibrateRounds, timeCalibrateWait)
	if err != nil {
		logger.LogEvent("时间", "warn", "%s校准降级: %v", label, err)
	}
	if !calibration.Fallback {
		s.recordCalibrationOffset(calibration.Offset)
	}
	smoothed := s.smoothedOffset()
	if smoothed == 0 && calibration.Fallback {
		smoothed = calibration.Offset
	}
	calibration.Offset = smoothed
	logger.SetDouyuOffset(smoothed)
	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"),
		smoothed.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
	}
}