mirror of
https://github.com/muety/wakapi.git
synced 2023-08-10 21:12:56 +03:00
chore: add random time offset to scheduled reports jobs
This commit is contained in:
parent
f3b738b250
commit
9dbc2039fc
@ -341,22 +341,27 @@ func Load(version string) *Config {
|
||||
}
|
||||
}
|
||||
|
||||
if config.Server.ListenIpV4 == "" && config.Server.ListenIpV6 == "" {
|
||||
logbuch.Fatal("either of listen_ipv4 or listen_ipv6 must be set")
|
||||
}
|
||||
|
||||
if config.Db.MaxConn <= 0 {
|
||||
logbuch.Fatal("you must allow at least one database connection")
|
||||
}
|
||||
|
||||
if config.Sentry.Dsn != "" {
|
||||
logbuch.Info("enabling sentry integration")
|
||||
initSentry(config.Sentry, config.IsDev())
|
||||
}
|
||||
|
||||
// some validation checks
|
||||
if config.Server.ListenIpV4 == "" && config.Server.ListenIpV6 == "" {
|
||||
logbuch.Fatal("either of listen_ipv4 or listen_ipv6 must be set")
|
||||
}
|
||||
if config.Db.MaxConn <= 0 {
|
||||
logbuch.Fatal("you must allow at least one database connection")
|
||||
}
|
||||
if config.Mail.Provider != "" && findString(config.Mail.Provider, emailProviders, "") == "" {
|
||||
logbuch.Fatal("unknown mail provider '%s'", config.Mail.Provider)
|
||||
}
|
||||
if _, err := time.Parse("15:04", config.App.GetWeeklyReportTime()); err != nil {
|
||||
logbuch.Fatal("invalid interval set for report_time_weekly")
|
||||
}
|
||||
if _, err := time.Parse("15:04", config.App.AggregationTime); err != nil {
|
||||
logbuch.Fatal("invalid interval set for aggregation_time")
|
||||
}
|
||||
|
||||
Set(config)
|
||||
return Get()
|
||||
|
@ -6,31 +6,40 @@ import (
|
||||
"github.com/leandro-lugaresi/hub"
|
||||
"github.com/muety/wakapi/config"
|
||||
"github.com/muety/wakapi/models"
|
||||
"math/rand"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
var reportLock = sync.Mutex{}
|
||||
|
||||
// range for random offset to add / subtract when scheduling a new job
|
||||
// to avoid all mails being sent at once, but distributed over 2*offsetIntervalMin minutes
|
||||
const offsetIntervalMin = 15
|
||||
|
||||
type ReportService struct {
|
||||
config *config.Config
|
||||
eventBus *hub.Hub
|
||||
summaryService ISummaryService
|
||||
userService IUserService
|
||||
mailService IMailService
|
||||
schedulersWeekly map[string]*gocron.Scheduler // user id -> scheduler
|
||||
config *config.Config
|
||||
eventBus *hub.Hub
|
||||
summaryService ISummaryService
|
||||
userService IUserService
|
||||
mailService IMailService
|
||||
scheduler *gocron.Scheduler
|
||||
rand *rand.Rand
|
||||
}
|
||||
|
||||
func NewReportService(summaryService ISummaryService, userService IUserService, mailService IMailService) *ReportService {
|
||||
srv := &ReportService{
|
||||
config: config.Get(),
|
||||
eventBus: config.EventBus(),
|
||||
summaryService: summaryService,
|
||||
userService: userService,
|
||||
mailService: mailService,
|
||||
schedulersWeekly: map[string]*gocron.Scheduler{},
|
||||
config: config.Get(),
|
||||
eventBus: config.EventBus(),
|
||||
summaryService: summaryService,
|
||||
userService: userService,
|
||||
mailService: mailService,
|
||||
scheduler: gocron.NewScheduler(time.Local),
|
||||
rand: rand.New(rand.NewSource(time.Now().Unix())),
|
||||
}
|
||||
|
||||
srv.scheduler.StartAsync()
|
||||
|
||||
sub := srv.eventBus.Subscribe(0, config.EventUserUpdate)
|
||||
go func(sub *hub.Subscription) {
|
||||
for m := range sub.Receiver {
|
||||
@ -62,24 +71,24 @@ func (srv *ReportService) SyncSchedule(u *models.User) bool {
|
||||
defer reportLock.Unlock()
|
||||
|
||||
// unschedule
|
||||
if s, ok := srv.schedulersWeekly[u.ID]; ok && !u.ReportsWeekly {
|
||||
s.Stop()
|
||||
s.Clear()
|
||||
delete(srv.schedulersWeekly, u.ID)
|
||||
if !u.ReportsWeekly {
|
||||
_ = srv.scheduler.RemoveByTag(u.ID)
|
||||
return false
|
||||
}
|
||||
|
||||
// schedule
|
||||
if _, ok := srv.schedulersWeekly[u.ID]; !ok && u.ReportsWeekly {
|
||||
s := gocron.NewScheduler(u.TZ())
|
||||
s.
|
||||
if j := srv.getJobByTag(u.ID); j == nil && u.ReportsWeekly {
|
||||
t, _ := time.ParseInLocation("15:04", srv.config.App.GetWeeklyReportTime(), u.TZ())
|
||||
t = t.Add(time.Duration(srv.rand.Intn(offsetIntervalMin)*srv.rand.Intn(2)) * time.Minute)
|
||||
if _, err := srv.scheduler.
|
||||
Every(1).
|
||||
Week().
|
||||
Weekday(srv.config.App.GetWeeklyReportDay()).
|
||||
At(srv.config.App.GetWeeklyReportTime()).
|
||||
Do(srv.Run, u, 7*24*time.Hour)
|
||||
s.StartAsync()
|
||||
srv.schedulersWeekly[u.ID] = s
|
||||
At(t).
|
||||
Tag(u.ID).
|
||||
Do(srv.Run, u, 7*24*time.Hour); err != nil {
|
||||
config.Log().Error("failed to schedule report job for user '%s' – %v", u.ID, err)
|
||||
}
|
||||
}
|
||||
|
||||
return u.ReportsWeekly
|
||||
@ -119,3 +128,14 @@ func (srv *ReportService) Run(user *models.User, duration time.Duration) error {
|
||||
logbuch.Info("sent report to user '%s'", user.ID)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (srv *ReportService) getJobByTag(tag string) *gocron.Job {
|
||||
for _, j := range srv.scheduler.Jobs() {
|
||||
for _, t := range j.Tags() {
|
||||
if t == tag {
|
||||
return j
|
||||
}
|
||||
}
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
Loading…
Reference in New Issue
Block a user