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) // 直播计费:每 60s 扫描进行中的 live_session 按码率扣减 _, _ = s.cron.AddFunc("@every 60s", DefaultBillingService.chargeLiveSessions) // 余额快照:每 30min 将 Redis 权威余额回写 MySQL _, _ = s.cron.AddFunc("@every 30m", DefaultBillingService.flushBalanceSnapshot) // SIM 卡用量:每 1h 从运营商同步用量 _, _ = s.cron.AddFunc("@every 1h", DefaultBillingService.syncSimUsage) // 设备心跳:清理未再收到真实 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) } }