diff --git a/scheduler.go b/scheduler.go index be761a9..913c46f 100644 --- a/scheduler.go +++ b/scheduler.go @@ -5,12 +5,16 @@ import ( "database/sql" "log" "time" + + "github.com/robfig/cron/v3" ) // Scheduler data structure type Scheduler struct { - db *sql.DB - listeners Listeners + db *sql.DB + listeners Listeners + cron *cron.Cron + cronEntries map[string]cron.EntryID } // Listeners has attached event listeners @@ -24,14 +28,19 @@ type Event struct { ID uint Name string Payload string + Cron string } // NewScheduler creates a new scheduler func NewScheduler(db *sql.DB, listeners Listeners) Scheduler { + return Scheduler{ - db: db, - listeners: listeners, + db: db, + listeners: listeners, + cron: cron.New(), + cronEntries: map[string]cron.EntryID{}, } + } // AddListener adds the listener function to Listeners @@ -78,7 +87,7 @@ func (s Scheduler) CheckEventsInInterval(ctx context.Context, duration time.Dura // checkDueEvents checks and returns due events func (s Scheduler) checkDueEvents() []Event { events := []Event{} - rows, err := s.db.Query(`SELECT "id", "name", "payload" FROM "public"."jobs" WHERE "runAt" < $1`, time.Now()) + rows, err := s.db.Query(`SELECT "id", "name", "payload" FROM "public"."jobs" WHERE "runAt" < $1 AND "cron"='-'`, time.Now()) if err != nil { log.Print("💀 error: ", err) return nil @@ -99,3 +108,62 @@ func (s Scheduler) Schedule(event string, payload string, runAt time.Time) { log.Print("schedule insert error: ", err) } } + +// ScheduleCron schedules a cron job +func (s Scheduler) ScheduleCron(event string, payload string, cron string) { + log.Print("🚀 Scheduling event ", event, " with cron string ", cron) + entryID, ok := s.cronEntries[event] + if ok { + s.cron.Remove(entryID) + _, err := s.db.Exec(`UPDATE "public"."jobs" SET "cron" = $1 , "payload" = $2 WHERE "name" = $3 AND "cron" != '-'`, cron, payload, event) + if err != nil { + log.Print("schedule cron update error: ", err) + } + } else { + _, err := s.db.Exec(`INSERT INTO "public"."jobs" ("name", "payload", "runAt", "cron") VALUES ($1, $2, $3, $4)`, event, payload, time.Now(), cron) + if err != nil { + log.Print("schedule cron insert error: ", err) + } + } + + eventFn, ok := s.listeners[event] + if ok { + entryID, err := s.cron.AddFunc(cron, func() { eventFn(payload) }) + s.cronEntries[event] = entryID + if err != nil { + log.Print("💀 error: ", err) + } + } +} + +// attachCronJobs attaches cron jobs +func (s Scheduler) attachCronJobs() { + log.Printf("Attaching cron jobs") + rows, err := s.db.Query(`SELECT "id", "name", "payload", "cron" FROM "public"."jobs" WHERE "cron"!='-'`) + if err != nil { + log.Print("💀 error: ", err) + } + for rows.Next() { + evt := Event{} + rows.Scan(&evt.ID, &evt.Name, &evt.Payload, &evt.Cron) + eventFn, ok := s.listeners[evt.Name] + if ok { + entryID, err := s.cron.AddFunc(evt.Cron, func() { eventFn(evt.Payload) }) + s.cronEntries[evt.Name] = entryID + + if err != nil { + log.Print("💀 error: ", err) + } + } + } +} + +// StartCron starts cron job +func (s Scheduler) StartCron() func() { + s.attachCronJobs() + s.cron.Start() + + return func() { + s.cron.Stop() + } +}