lucky_bot/magic_admin/scheduler/red_packet_scheduler.go

package scheduler

import (
	"fmt"
	model "go_server/model/biz_modules/app"
	"go_server/service/app"
	"sync"
	"time"

	"github.com/robfig/cron/v3"
	"go_server/base/core"
	"gorm.io/gorm"
)

// RedPacketScheduler 红包定时任务调度器
type RedPacketScheduler struct {
	cron      *cron.Cron
	db        *gorm.DB
	jobs      map[int64]cron.EntryID // configId -> entryID
	jobsMutex sync.RWMutex
	service   *app.TgRedPacketSendService
}

var (
	scheduler *RedPacketScheduler
	once      sync.Once
)

// GetScheduler 获取调度器单例
func GetScheduler(db *gorm.DB) *RedPacketScheduler {
	once.Do(func() {
		scheduler = &RedPacketScheduler{
			cron:    cron.New(cron.WithSeconds()), // 支持秒级
			db:      db,
			jobs:    make(map[int64]cron.EntryID),
			service: &app.TgRedPacketSendService{},
		}
	})
	return scheduler
}

// Start 启动调度器
func (s *RedPacketScheduler) Start() error {
	core.Log.Info("🕐 启动红包定时任务调度器...")

	// 加载所有启用的定时配置
	if err := s.LoadAllConfigs(); err != nil {
		return fmt.Errorf("加载定时配置失败: %v", err)
	}

	// 启动 cron
	s.cron.Start()

	// 启动配置监听器（每分钟检查一次配置变化）
	go s.watchConfigChanges()

	core.Log.Info("✅ 红包定时任务调度器启动成功")
	return nil
}

// Stop 停止调度器
func (s *RedPacketScheduler) Stop() {
	core.Log.Info("🛑 停止红包定时任务调度器...")
	s.cron.Stop()
}

// LoadAllConfigs 加载所有启用的定时配置
func (s *RedPacketScheduler) LoadAllConfigs() error {
	var configs []model.TgRedPacketConfig

	// 查询所有启用的定时红包配置
	err := s.db.Where("config_type = ? AND status = ?", 1, 1).Find(&configs).Error
	if err != nil {
		return err
	}

	core.Log.Infof("📋 加载到 %d 个定时红包配置", len(configs))

	// 添加到调度器
	for _, config := range configs {
		if err := s.AddJob(&config); err != nil {
			core.Log.Errorf("添加定时任务失败 [ID:%d]: %v", config.Id, err)
			continue
		}
	}

	return nil
}

// AddJob 添加定时任务
func (s *RedPacketScheduler) AddJob(config *model.TgRedPacketConfig) error {
	s.jobsMutex.Lock()
	defer s.jobsMutex.Unlock()

	// 检查是否已存在
	if _, exists := s.jobs[config.Id]; exists {
		s.RemoveJobLocked(config.Id)
	}

	// 验证 Cron 表达式
	if config.CronExpr == "" {
		return fmt.Errorf("Cron 表达式为空")
	}

	// 添加任务
	entryID, err := s.cron.AddFunc(config.CronExpr, func() {
		s.executeJob(config.Id)
	})

	if err != nil {
		return fmt.Errorf("添加 Cron 任务失败: %v", err)
	}

	s.jobs[config.Id] = entryID

	core.Log.Infof("✅ 添加定时任务 [ID:%d, Name:%s, Cron:%s]",
		config.Id, config.ConfigName, config.CronExpr)

	return nil
}

// RemoveJob 移除定时任务
func (s *RedPacketScheduler) RemoveJob(configId int64) {
	s.jobsMutex.Lock()
	defer s.jobsMutex.Unlock()
	s.RemoveJobLocked(configId)
}

// RemoveJobLocked 移除定时任务（需要提前加锁）
func (s *RedPacketScheduler) RemoveJobLocked(configId int64) {
	if entryID, exists := s.jobs[configId]; exists {
		s.cron.Remove(entryID)
		delete(s.jobs, configId)
		core.Log.Infof("🗑️ 移除定时任务 [ID:%d]", configId)
	}
}

// executeJob 执行定时任务
func (s *RedPacketScheduler) executeJob(configId int64) {
	core.Log.Infof("⏰ 执行定时红包任务 [ID:%d]", configId)

	// 查询配置
	var config model.TgRedPacketConfig
	if err := s.db.Where("id = ?", configId).First(&config).Error; err != nil {
		core.Log.Errorf("查询配置失败 [ID:%d]: %v", configId, err)
		return
	}

	// 检查状态
	if config.Status != 1 {
		core.Log.Warnf("配置已禁用，跳过执行 [ID:%d]", configId)
		return
	}

	// 检查时间范围
	now := time.Now()
	if config.StartDate != nil && now.Unix() < *config.StartDate {
		core.Log.Infof("未到开始时间，跳过执行 [ID:%d]", configId)
		return
	}
	if config.EndDate != nil && now.Unix() > *config.EndDate {
		core.Log.Infof("已过结束时间，停止任务 [ID:%d]", configId)
		s.RemoveJob(configId)
		return
	}

	// 调用发送服务
	s.service.SetDbAlias("app")
	packetNo, err := s.service.ExecuteSendRedPacket(&config)
	if err != nil {
		core.Log.Errorf("发送红包失败 [ID:%d]: %v", configId, err)
		return
	}

	// 更新执行统计
	nowUnix := now.Unix()
	updates := map[string]interface{}{
		"last_exec_time": nowUnix,
		"exec_count":     config.ExecCount + 1,
	}

	// 计算下次执行时间（仅用于显示）
	schedule, _ := cron.ParseStandard(config.CronExpr)
	if schedule != nil {
		nextTime := schedule.Next(now)
		nextUnix := nextTime.Unix()
		updates["next_exec_time"] = nextUnix
	}

	s.db.Model(&model.TgRedPacketConfig{}).Where("id = ?", configId).Updates(updates)

	core.Log.Infof("✅ 定时红包发送成功 [ID:%d, PacketNo:%s]", configId, packetNo)
}

// watchConfigChanges 监听配置变化
func (s *RedPacketScheduler) watchConfigChanges() {
	ticker := time.NewTicker(1 * time.Minute)
	defer ticker.Stop()

	for range ticker.C {
		s.reloadConfigs()
	}
}

// reloadConfigs 重新加载配置
func (s *RedPacketScheduler) reloadConfigs() {
	var configs []model.TgRedPacketConfig

	// 查询所有启用的定时配置
	err := s.db.Where("config_type = ? AND status = ?", 1, 1).Find(&configs).Error
	if err != nil {
		core.Log.Errorf("重新加载配置失败: %v", err)
		return
	}

	s.jobsMutex.Lock()
	defer s.jobsMutex.Unlock()

	// 构建当前应该存在的任务 ID 集合
	activeIds := make(map[int64]bool)
	for _, config := range configs {
		activeIds[config.Id] = true

		// 如果任务不存在，添加
		if _, exists := s.jobs[config.Id]; !exists {
			s.jobsMutex.Unlock()
			s.AddJob(&config)
			s.jobsMutex.Lock()
		}
	}

	// 移除不再需要的任务
	for id := range s.jobs {
		if !activeIds[id] {
			s.RemoveJobLocked(id)
		}
	}
}