57 lines
1.5 KiB
Go
57 lines
1.5 KiB
Go
package jobs
|
|
|
|
import (
|
|
"cybertron/configs"
|
|
"cybertron/internal/client/elastic"
|
|
"cybertron/pkg/log"
|
|
"cybertron/pkg/utils"
|
|
"github.com/go-co-op/gocron/v2"
|
|
"go.uber.org/zap"
|
|
"time"
|
|
)
|
|
|
|
type JobScheduler struct {
|
|
logger *log.Logger
|
|
scheduler gocron.Scheduler
|
|
elasticSearchClient *elastic.ElasticSearchClient
|
|
config configs.JobSchedulerConfig
|
|
}
|
|
|
|
func NewJobScheduler(logger *log.Logger, elasticSearchClient *elastic.ElasticSearchClient, jonSchedulerConfig *configs.JobSchedulerConfig) *JobScheduler {
|
|
s, err := gocron.NewScheduler()
|
|
if err != nil {
|
|
logger.Error("Failed to start scheduler", zap.Error(err))
|
|
}
|
|
|
|
return &JobScheduler{
|
|
logger: logger,
|
|
scheduler: s,
|
|
elasticSearchClient: elasticSearchClient,
|
|
config: *jonSchedulerConfig,
|
|
}
|
|
}
|
|
|
|
func (s *JobScheduler) ScheduleEsCleanUpJob() {
|
|
s.scheduler.NewJob(gocron.DurationJob(2*time.Hour), gocron.NewTask(esCleanUpJob, s))
|
|
}
|
|
func esCleanUpJob(s *JobScheduler) {
|
|
|
|
now := time.Now().Unix()
|
|
// Subtract 7 days (7 days * 24 hours * 60 minutes * 60 seconds)
|
|
retentionDays := s.config.GetMaxRetentionDays()
|
|
|
|
if retentionDays <= 0 {
|
|
retentionDays = 7
|
|
}
|
|
sevenDaysAgo := now - (int64(retentionDays) * 24 * 60 * 60)
|
|
rangeQuery := utils.CreateRangeQueryForLteString("created_at", sevenDaysAgo)
|
|
search_query := utils.CreateSearchQuery(rangeQuery)
|
|
es_query := utils.CreateEsQuery(search_query)
|
|
s.elasticSearchClient.DeleteDocuments(es_query)
|
|
|
|
}
|
|
func (s *JobScheduler) Start() {
|
|
s.logger.Info("Starting job scheduler")
|
|
s.scheduler.Start()
|
|
}
|