laic-backend/service/scheduler.go

package service

import (
	"sync"
	"time"

	"github.com/robfig/cron/v3"

	"laic-backend/common"
	"laic-backend/logger"
	"laic-backend/model"
)

// Scheduler 后台定时任务调度器：固定周期计费任务 + 任务计划 cron 自动执行
type Scheduler struct {
	cron    *cron.Cron
	mu      sync.Mutex
	entries map[string]cron.EntryID // taskID -> cron entry
}

var DefaultScheduler = &Scheduler{}

// Start 启动调度器：注册固定周期任务，加载并注册启用的 cron 任务计划
func (s *Scheduler) Start() {
	s.cron = cron.New()
	s.entries = make(map[string]cron.EntryID)

	// 直播租约与会话对账：清理过期租约并处理超时/无人观看会话
	interval := common.AppConf.Live.ReconcileIntervalSeconds
	if interval <= 0 {
		interval = 5
	}
	_, _ = s.cron.AddFunc("@every "+(time.Duration(interval)*time.Second).String(), DefaultLiveService.ReconcileLeases)
	_, _ = s.cron.AddFunc("@every "+(time.Duration(interval)*time.Second).String(), DefaultLiveService.ReconcileCloudSessions)
	// 直播计费：按配置周期扫描进行中的 live_session 按码率扣减
	billingInterval := common.AppConf.Live.BillingIntervalSeconds
	if billingInterval <= 0 {
		billingInterval = 60
	}
	_, _ = s.cron.AddFunc("@every "+(time.Duration(billingInterval)*time.Second).String(), DefaultBillingService.chargeLiveSessions)
	// 余额快照：每 30min 将 Redis 权威余额回写 MySQL
	_, _ = s.cron.AddFunc("@every 30m", DefaultBillingService.flushBalanceSnapshot)
	// SIM 卡用量：每 1h 从运营商同步用量
	_, _ = s.cron.AddFunc("@every 1h", DefaultBillingService.syncSimUsage)
	_, _ = s.cron.AddFunc("@every 30s", func() {
		DefaultBillingService.CloseExpiredOrders()
	})
	// 设备心跳：清理未再收到真实 MQTT 消息的在线设备
	_, _ = s.cron.AddFunc("@every "+heartbeatScanInterval().String(), expireDeviceHeartbeats)

	s.reloadTaskPlans()
	s.cron.Start()
	logger.INFO("scheduler started")
}

// reloadTaskPlans 从 DB 加载启用中的 cron 任务计划并注册
func (s *Scheduler) reloadTaskPlans() {
	var tasks []model.TaskPlan
	if err := common.DB.Where("schedule_type = ? AND status = ?", "cron", "enabled").Find(&tasks).Error; err != nil {
		logger.ERROR("加载 cron 任务计划失败", err)
		return
	}
	for i := range tasks {
		s.upsertTask(&tasks[i])
	}
}

// upsertTask 注册或更新单个任务计划的 cron 条目
func (s *Scheduler) upsertTask(task *model.TaskPlan) {
	s.mu.Lock()
	defer s.mu.Unlock()
	if old, ok := s.entries[task.ID]; ok {
		s.cron.Remove(old)
		delete(s.entries, task.ID)
	}
	if task.ScheduleType != "cron" || task.Status != "enabled" || task.ScheduleCron == "" {
		return
	}
	id, err := s.cron.AddFunc(task.ScheduleCron, func() { s.executeTask(task.ID) })
	if err != nil {
		logger.ERROR("注册 cron 任务计划失败: "+task.ID, err)
		return
	}
	s.entries[task.ID] = id
}

// SyncTask 任务创建/更新后同步调度（按 ID 重新加载）
func (s *Scheduler) SyncTask(taskID string) {
	var task model.TaskPlan
	if err := common.DB.First(&task, "id = ?", taskID).Error; err != nil {
		s.RemoveTask(taskID)
		return
	}
	s.upsertTask(&task)
}

// RemoveTask 移除任务计划的 cron 条目
func (s *Scheduler) RemoveTask(taskID string) {
	s.mu.Lock()
	defer s.mu.Unlock()
	if old, ok := s.entries[taskID]; ok {
		s.cron.Remove(old)
		delete(s.entries, taskID)
	}
}

// executeTask 定时触发任务执行（admin 视角，绕过用户过滤）
func (s *Scheduler) executeTask(taskID string) {
	if _, busiErr := DefaultTaskService.Execute(0, true, taskID); busiErr != nil {
		logger.WARN("定时执行任务失败:", taskID, busiErr.Msg)
	}
}