marker/shuffle-cracked
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

added cron support

Browse Source
This commit is contained in:
yashsinghcodes
2025-07-14 20:33:50 +05:30
parent ff4a63daea
commit b2dfe8d8f9
3 changed files with 54 additions and 20 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/go-app/go.mod
+3
View File
@@ -74,6 +74,7 @@ require (
github.com/felixge/httpsnoop v1.0.4 // indirect
github.com/frikky/schemaless v0.0.17 // indirect
github.com/fxamacker/cbor/v2 v2.7.0 // indirect
github.com/go-co-op/gocron v1.37.0 // indirect
github.com/go-git/gcfg v1.5.1-0.20230307220236-3a3c6141e376 // indirect
github.com/go-jose/go-jose/v4 v4.0.5 // indirect
github.com/go-logr/logr v1.4.2 // indirect
@@ -118,6 +119,7 @@ require (
github.com/pjbgf/sha1cd v0.3.2 // indirect
github.com/pkg/errors v0.9.1 // indirect
github.com/planetscale/vtprotobuf v0.6.1-0.20240319094008-0393e58bdf10 // indirect
github.com/robfig/cron/v3 v3.0.1 // indirect
github.com/sashabaranov/go-openai v1.40.1 // indirect
github.com/sendgrid/rest v2.6.9+incompatible // indirect
github.com/sendgrid/sendgrid-go v3.16.1+incompatible // indirect
@@ -141,6 +143,7 @@ require (
go.opentelemetry.io/otel/sdk/metric v1.36.0 // indirect
go.opentelemetry.io/otel/trace v1.36.0 // indirect
go.opentelemetry.io/proto/otlp v1.5.0 // indirect
go.uber.org/atomic v1.9.0 // indirect
go4.org v0.0.0-20230225012048-214862532bf5 // indirect
golang.org/x/net v0.40.0 // indirect
golang.org/x/oauth2 v0.30.0 // indirect
backend/go-app/main.go
+17 -6
View File
@@ -4034,6 +4034,7 @@ func runInitEs(ctx context.Context) {
// FIXME: This should ONLY run on one backend instance
schedules, err := shuffle.GetAllSchedules(ctx, "ALL")
log.Printf("Schedules %s", schedules)
if err != nil {
log.Printf("[WARNING] Failed getting schedules during service init: %s", err)
} else {
@@ -4078,14 +4079,23 @@ func runInitEs(ctx context.Context) {
//log.Printf("Schedule: %#v", schedule)
//log.Printf("Schedule time: every %d seconds", schedule.Seconds)
jobret, err := newscheduler.Every(schedule.Seconds).Seconds().NotImmediately().Run(job(schedule))
if err != nil {
log.Printf("[ERROR] Failed to start schedule for workflow %s: %s", schedule.WorkflowId, err)
if schedule.Seconds == 0 && len(schedule.Frequency) > 0 {
cronJob, err := CronScheduler.Cron(schedule.Frequency).Do(job(schedule))
if err != nil {
log.Printf("[ERROR] Failed to start schedule for workflow %s: %s", schedule.WorkflowId, err)
} else {
log.Printf("[DEBUG] Successfully started schedule for workflow %s", schedule.WorkflowId)
}
cronJobs[schedule.Id] = cronJob
} else {
log.Printf("[DEBUG] Successfully started schedule for workflow %s", schedule.WorkflowId)
jobret, err := newscheduler.Every(schedule.Seconds).Seconds().NotImmediately().Run(job(schedule))
if err != nil {
log.Printf("[ERROR] Failed to start schedule for workflow %s: %s", schedule.WorkflowId, err)
} else {
log.Printf("[DEBUG] Successfully started schedule for workflow %s", schedule.WorkflowId)
}
scheduledJobs[schedule.Id] = jobret
}
scheduledJobs[schedule.Id] = jobret
}
}
@@ -5090,6 +5100,7 @@ func handleAppZipUpload(resp http.ResponseWriter, request *http.Request) {
func initHandlers() {
var err error
ctx := context.Background()
CronScheduler.StartAsync()
log.Printf("[DEBUG] Starting Shuffle backend - initializing database connection")
//requestCache = cache.New(5*time.Minute, 10*time.Minute)
backend/go-app/walkoff.go
+34 -14
View File
@@ -26,6 +26,7 @@ import (
uuid "github.com/satori/go.uuid"
newscheduler "github.com/carlescere/scheduler"
"github.com/go-co-op/gocron"
"github.com/frikky/kin-openapi/openapi3"
"github.com/go-git/go-billy/v5"
"github.com/go-git/go-billy/v5/memfs"
@@ -42,17 +43,22 @@ var baseEnvironment = "onprem"
var cloudname = "cloud"
var scheduledJobs = map[string]*newscheduler.Job{}
var cronJobs = map[string]*gocron.Job{}
var scheduledOrgs = map[string]*newscheduler.Job{}
var CronScheduler = gocron.NewScheduler(time.UTC)
// Frequency = cronjob OR minutes between execution
func createSchedule(ctx context.Context, scheduleId, workflowId, name, startNode, frequency, orgId string, body []byte) error {
var err error
testSplit := strings.Split(frequency, "*")
cronJob := ""
isCron := false
newfrequency := 0
if len(testSplit) > 5 {
cronJob = frequency
isCron = true
} else {
newfrequency, err = strconv.Atoi(frequency)
if err != nil {
@@ -65,12 +71,7 @@ func createSchedule(ctx context.Context, scheduleId, workflowId, name, startNode
//} else if int(newfrequency) <
}
// Reverse. Can't handle CRON, only numbers
if len(cronJob) > 0 {
return errors.New("cronJob isn't formatted correctly")
}
if newfrequency < 1 {
if newfrequency < 1 && !isCron {
return errors.New("Frequency has to be more than 0")
}
@@ -96,17 +97,27 @@ func createSchedule(ctx context.Context, scheduleId, workflowId, name, startNode
}
}
log.Printf("[INFO] Starting frequency for execution: %d", newfrequency)
log.Printf("[INFO] Starting frequency for execution: %s", frequency)
//jobret, err := newscheduler.Every(newfrequency).Seconds().NotImmediately().Run(job)
jobret, err := newscheduler.Every(newfrequency).Seconds().Run(job)
if err != nil {
log.Printf("Failed to schedule workflow: %s", err)
return err
if isCron {
cronJob, err := CronScheduler.Cron(cronJob).Do(job)
if err != nil {
log.Printf("[ERROR] Failed to start schedule with cron(%s): %s", cronJob, err)
}
cronJobs[scheduleId] = cronJob
} else {
//jobret, err := newscheduler.Every(newfrequency).Seconds().NotImmediately().Run(job)
jobret, err := newscheduler.Every(newfrequency).Seconds().Run(job)
if err != nil {
log.Printf("Failed to schedule workflow: %s", err)
return err
}
scheduledJobs[scheduleId] = jobret
}
//scheduledJobs = append(scheduledJobs, jobret)
scheduledJobs[scheduleId] = jobret
// Doesn't need running/not running. If stopped, we just delete it.
timeNow := int64(time.Now().Unix())
@@ -117,6 +128,7 @@ func createSchedule(ctx context.Context, scheduleId, workflowId, name, startNode
Argument: string(body),
WrappedArgument: bodyWrapper,
Seconds: newfrequency,
Frequency: frequency,
CreationTime: timeNow,
LastModificationtime: timeNow,
LastRuntime: timeNow,
@@ -1541,7 +1553,15 @@ func deleteSchedule(ctx context.Context, id string) error {
value.Lock()
} else {
// FIXME - allow it to kind of stop anyway?
return errors.New("Can't find the schedule.")
if j, ok := cronJobs[id]; ok {
err := CronScheduler.RemoveByID(j)
if err != nil {
log.Printf("[ERROR] Failed to remove the scheduler %s", err)
return err
}
} else {
return errors.New("Can't find the schedule.")
}
}
}