package app import ( "context" "log/slog" "math/rand" "time" "github.com/robfig/cron/v3" "cryptoHermes/internal/worker" ) type Scheduler struct { cron *cron.Cron collector *worker.Collector coinglass *worker.CoinglassCollector log *slog.Logger rootCtx context.Context coinglassLiqHeatMapCron string coinglassStartupJitter time.Duration } func NewScheduler(c *worker.Collector, log *slog.Logger) *Scheduler { return &Scheduler{ cron: cron.New(), collector: c, log: log, } } func (s *Scheduler) WithCoinglass(c *worker.CoinglassCollector, liqHeatMapCron string, startupJitter time.Duration) *Scheduler { s.coinglass = c s.coinglassLiqHeatMapCron = liqHeatMapCron s.coinglassStartupJitter = startupJitter return s } func (s *Scheduler) Start(ctx context.Context) { s.rootCtx = ctx if s.collector != nil { s.add("*/15 * * * *", "klines_15m", func(c context.Context) { _ = s.collector.CollectKlines(c, "15m") }) s.add("0 * * * *", "klines_1h", func(c context.Context) { _ = s.collector.CollectKlines(c, "1h") }) s.add("0 */4 * * *", "klines_4h", func(c context.Context) { _ = s.collector.CollectKlines(c, "4h") }) s.add("5 0 * * *", "klines_1d", func(c context.Context) { _ = s.collector.CollectKlines(c, "1d") }) s.add("10 0 * * 1", "klines_1w", func(c context.Context) { _ = s.collector.CollectKlines(c, "1w") }) s.add("*/15 * * * *", "funding", func(c context.Context) { _ = s.collector.CollectFunding(c) }) s.add("*/15 * * * *", "oi", func(c context.Context) { _ = s.collector.CollectOpenInterest(c) }) s.add("*/15 * * * *", "ls", func(c context.Context) { _ = s.collector.CollectLongShortRatio(c) }) s.add("*/15 * * * *", "taker", func(c context.Context) { _ = s.collector.CollectTakerVolume(c) }) } if s.coinglass != nil { spec := s.coinglassLiqHeatMapCron if spec == "" { spec = "*/5 * * * *" } s.add(spec, "coinglass_liq_heatmap", func(c context.Context) { _ = s.coinglass.CollectLiqHeatMap(c) }) s.startCoinglassWarmup(ctx) } s.cron.Start() s.log.Info("scheduler_started") } func (s *Scheduler) add(spec, name string, fn func(context.Context)) { _, err := s.cron.AddFunc(spec, func() { s.log.Info("cron_tick", "job", name) fn(s.rootCtx) }) if err != nil { s.log.Error("cron_add_failed", "job", name, "err", err) } } func (s *Scheduler) startCoinglassWarmup(ctx context.Context) { go func() { if s.coinglassStartupJitter > 0 { jitter := time.Duration(rand.Int63n(int64(s.coinglassStartupJitter))) select { case <-ctx.Done(): return case <-time.After(jitter): } } s.log.Info("cron_tick", "job", "coinglass_liq_heatmap_warmup") _ = s.coinglass.CollectLiqHeatMap(ctx) }() } func (s *Scheduler) Stop() { s.log.Info("scheduler_stopping") <-s.cron.Stop().Done() }