Nathan Coad
457d9395f0
All checks were successful
continuous-integration/drone/push Build is passing
33 lines
737 B
Go
33 lines
737 B
Go
package tasks
|
|
|
|
import (
|
|
"context"
|
|
"time"
|
|
"vctp/db"
|
|
)
|
|
|
|
// runAggregateJob wraps aggregation cron jobs with timeout, migration check, and circuit breaker semantics.
|
|
func (c *CronTask) runAggregateJob(ctx context.Context, jobName string, timeout time.Duration, fn func(context.Context) error) (err error) {
|
|
jobCtx := ctx
|
|
if timeout > 0 {
|
|
var cancel context.CancelFunc
|
|
jobCtx, cancel = context.WithTimeout(ctx, timeout)
|
|
defer cancel()
|
|
}
|
|
tracker := NewCronTracker(c.Database)
|
|
done, skip, err := tracker.Start(jobCtx, jobName)
|
|
if err != nil {
|
|
return err
|
|
}
|
|
if skip {
|
|
return nil
|
|
}
|
|
defer func() { done(err) }()
|
|
|
|
if err := db.CheckMigrationState(jobCtx, c.Database.DB()); err != nil {
|
|
return err
|
|
}
|
|
|
|
return fn(jobCtx)
|
|
}
|