package scheduler import ( "sync" "time" "github.com/robfig/cron/v3" "github.com/starfleetcptn/gomft/internal/db" "gorm.io/gorm" // Needed for DB interface method signature ) // --- Interfaces for Dependencies --- // JobExecutorDB defines the database methods needed by JobExecutor. type JobExecutorDB interface { First(dest interface{}, conds ...interface{}) *gorm.DB // Used to load job details GetConfigsForJob(jobID uint) ([]db.TransferConfig, error) UpdateJobStatus(job *db.Job) error CreateJobHistory(history *db.JobHistory) error } // JobExecutorCron defines the cron methods needed by JobExecutor. type JobExecutorCron interface { Entry(id cron.EntryID) cron.Entry } // JobExecutorTransferExecutor defines the transfer executor methods needed by JobExecutor. type JobExecutorTransferExecutor interface { executeConfigTransfer(job db.Job, config db.TransferConfig, history *db.JobHistory) } // JobExecutorNotifier defines the notification methods needed by JobExecutor. type JobExecutorNotifier interface { // SendNotifications is called within processConfiguration, which indirectly uses the Notifier interface // defined in transfer_executor.go. We need the same method here. SendNotifications(job *db.Job, history *db.JobHistory, config *db.TransferConfig) } // --- JobExecutor Implementation --- // JobExecutor handles the execution logic for a single job run. type JobExecutor struct { db JobExecutorDB // Use interface logger *Logger // Logger remains concrete cron JobExecutorCron // Use interface jobs map[uint]cron.EntryID // Shared map from Scheduler jobMutex *sync.Mutex // Shared mutex from Scheduler transferExecutor JobExecutorTransferExecutor // Use interface notifier JobExecutorNotifier // Use interface } // NewJobExecutor creates a new JobExecutor. func NewJobExecutor( database JobExecutorDB, // Accept interface logger *Logger, cron JobExecutorCron, // Accept interface jobsMap map[uint]cron.EntryID, jobMutex *sync.Mutex, transferExec JobExecutorTransferExecutor, // Accept interface notify JobExecutorNotifier, // Accept interface ) *JobExecutor { return &JobExecutor{ db: database, logger: logger, cron: cron, jobs: jobsMap, jobMutex: jobMutex, transferExecutor: transferExec, notifier: notify, } } // executeJob orchestrates the execution of a job by processing its configurations. func (je *JobExecutor) executeJob(jobID uint) { je.logger.LogDebug("Entering executeJob for job ID %d", jobID) defer je.logger.LogDebug("Exiting executeJob for job ID %d", jobID) je.logger.LogInfo("Starting execution of job %d", jobID) // Get job details var job db.Job // Calls interface method - need to handle the *gorm.DB return value if err := je.db.First(&job, jobID).Error; err != nil { je.logger.LogError("Error loading job %d: %v", jobID, err) return } je.logger.LogDebug("Loaded job details: %+v", job) // Get all configurations associated with this job configs, err := je.db.GetConfigsForJob(jobID) // Calls interface method if err != nil { je.logger.LogError("Error loading configurations for job %d: %v", jobID, err) return } je.logger.LogDebug("Loaded %d configurations for job %d", len(configs), jobID) if len(configs) == 0 { je.logger.LogError("Error: job %d has no associated configurations", jobID) return } // Get the ordered config IDs from the job orderedConfigIDs := job.GetConfigIDsList() je.logger.LogDebug("Ordered config IDs for job %d: %v", jobID, orderedConfigIDs) // Create a map of configs for easy lookup configMap := make(map[uint]db.TransferConfig) for _, config := range configs { configMap[config.ID] = config } // Process configurations in the specified order var orderedConfigs []db.TransferConfig // First, add configs in the order specified in the job's ConfigIDs for _, configID := range orderedConfigIDs { if config, exists := configMap[configID]; exists { orderedConfigs = append(orderedConfigs, config) delete(configMap, configID) // Remove from map to avoid duplicates } } // Add any remaining configs not in the ordered list (shouldn't happen, but just in case) for _, config := range configMap { orderedConfigs = append(orderedConfigs, config) } je.logger.LogInfo("Processing job %d with %d configurations in specified order", jobID, len(orderedConfigs)) // Log the order of execution for i, config := range orderedConfigs { je.logger.LogDebug("Execution order %d/%d: Config ID %d (%s)", i+1, len(orderedConfigs), config.ID, config.Name) } // Update job last run time startTime := time.Now() job.LastRun = &startTime if err := je.db.UpdateJobStatus(&job); err != nil { // Calls interface method je.logger.LogError("Error updating job last run time for job %d: %v", jobID, err) } // Process each configuration in the specified order for i, config := range orderedConfigs { je.processConfiguration(&job, &config, i+1, len(orderedConfigs)) } // Update next run time after execution // Need access to the shared jobs map and mutex from Scheduler je.jobMutex.Lock() entryID, exists := je.jobs[jobID] je.jobMutex.Unlock() if exists { entry := je.cron.Entry(entryID) // Calls interface method nextRun := entry.Next job.NextRun = &nextRun je.logger.LogInfo("Next run time for job %d: %v", jobID, nextRun) if err := je.db.UpdateJobStatus(&job); err != nil { // Calls interface method je.logger.LogError("Error updating job next run time for job %d: %v", jobID, err) } } } // processConfiguration processes a single configuration step within a job. func (je *JobExecutor) processConfiguration(job *db.Job, config *db.TransferConfig, index int, totalConfigs int) { je.logger.LogDebug("Processing configuration %d: %+v", config.ID, config) je.logger.LogInfo("Processing configuration %d (%d/%d) for job %d: source=%s:%s, dest=%s:%s", config.ID, index, totalConfigs, job.ID, config.SourceType, config.SourcePath, config.DestinationType, config.DestinationPath, ) // Create job history entry for this configuration history := &db.JobHistory{ JobID: job.ID, ConfigID: config.ID, StartTime: time.Now(), Status: "running", FilesTransferred: 0, BytesTransferred: 0, ErrorMessage: "", } if err := je.db.CreateJobHistory(history); err != nil { // Calls interface method je.logger.LogError("Error creating job history for job %d, config %d: %v", job.ID, config.ID, err) return } je.logger.LogDebug("Creating job history record: %+v", history) // Send webhook notification for job start // Notifier interface is used by TransferExecutor, which is called below. // We also added SendNotifications to the JobExecutorNotifier interface for completeness, // though it's primarily used within transferExecutor. je.notifier.SendNotifications(job, history, config) // Calls interface method // Execute the configuration transfer je.transferExecutor.executeConfigTransfer(*job, *config, history) // Calls interface method }