From d6fa0c16039e68bb7214e7c46e9830a1365486cd Mon Sep 17 00:00:00 2001 From: StarFleetCPTN Date: Thu, 13 Mar 2025 20:04:13 -0700 Subject: [PATCH] feat: Implement multi-configuration support for jobs - Add support for multiple transfer configurations per job, allowing users to select one or more configurations. - Update job creation and editing forms to handle multiple configuration selections with checkboxes. - Enhance job processing logic to iterate through all associated configurations during execution. - Introduce database migrations to add necessary fields for storing multiple configuration IDs. - Update dashboard and history views to display configuration details for jobs. - Refactor related templates and handlers to accommodate the new multi-configuration functionality. --- components/dashboard.templ | 3 +- components/history.templ | 25 +- components/job_form.templ | 160 ++-- components/jobs.templ | 16 +- internal/db/db.go | 85 ++ .../db/migrations/add_multi_config_support.go | 49 ++ internal/db/migrations/migrations.go | 1 + internal/handlers/jobs_handler.go | 155 ++++ internal/scheduler/scheduler.go | 811 +++++++++--------- internal/web/handlers/dashboard_handlers.go | 80 +- internal/web/handlers/job_handlers.go | 197 +++-- 11 files changed, 1054 insertions(+), 528 deletions(-) create mode 100644 internal/db/migrations/add_multi_config_support.go create mode 100644 internal/handlers/jobs_handler.go diff --git a/components/dashboard.templ b/components/dashboard.templ index 3323111..16a5a22 100644 --- a/components/dashboard.templ +++ b/components/dashboard.templ @@ -12,6 +12,7 @@ type DashboardData struct { ActiveTransfers int CompletedToday int FailedTransfers int + Configs map[uint]db.TransferConfig } templ Dashboard(ctx context.Context, data DashboardData) { @@ -113,7 +114,7 @@ templ Dashboard(ctx context.Context, data DashboardData) {

- { job.Job.Config.Name } + { getConfigNameForHistory(job, data.Configs) }

diff --git a/components/history.templ b/components/history.templ index 7548bdb..60cc885 100644 --- a/components/history.templ +++ b/components/history.templ @@ -13,6 +13,7 @@ type HistoryData struct { SearchTerm string PageSize int Total int + Configs map[uint]db.TransferConfig // Map of config IDs to configs for quick lookup } // min returns the smaller of x or y @@ -23,6 +24,28 @@ func min(x, y int) int { return y } +// getConfigNameForHistory returns the appropriate name for the config used in a job history entry +func getConfigNameForHistory(history db.JobHistory, configs map[uint]db.TransferConfig) string { + // If ConfigID is set in the history record, use that to get the config name + if history.ConfigID > 0 { + if config, exists := configs[history.ConfigID]; exists { + return config.Name + } + } + + // Fallback to the Job's default Config if it exists + if history.Job.Config.ID > 0 { + return history.Job.Config.Name + } + + // If we can't determine the config name, show a default with the job name + if history.Job.Name != "" { + return fmt.Sprintf("%s (unknown config)", history.Job.Name) + } + + return "Unknown Configuration" +} + // HistoryContent renders only the content part of the history page for HTMX requests templ HistoryContent(ctx context.Context, data HistoryData) { if len(data.History) == 0 { @@ -47,7 +70,7 @@ templ HistoryContent(ctx context.Context, data HistoryData) {
-

{ history.Job.Config.Name }

+

{ getConfigNameForHistory(history, data.Configs) }

if history.Status == "completed" { Completed diff --git a/components/job_form.templ b/components/job_form.templ index d2ff542..2ce6ff9 100644 --- a/components/job_form.templ +++ b/components/job_form.templ @@ -26,6 +26,18 @@ func getJobTitle(isNew bool) string { return "Edit Job" } +// configSelected checks if a config ID is selected for a job +func configSelected(job *db.Job, configID uint) bool { + // Check if the job has the config ID in its list + for _, id := range job.GetConfigIDsList() { + if id == configID { + return true + } + } + // As a fallback, check the primary ConfigID + return job.ConfigID == configID +} + templ JobForm(ctx context.Context, data JobFormData) { @LayoutWithContext(getJobFormTitle(data.IsNew), ctx) {
@@ -47,11 +59,7 @@ templ JobForm(ctx context.Context, data JobFormData) { class="space-y-6" hx-post="/jobs" hx-target="body" - hx-boost="true" - @htmx:before-request="loading = true" - @htmx:after-request="loading = false" - @htmx:response-error="$dispatch('notification', { message: 'Failed to create job: ' + event.detail.xhr.responseText, type: 'error' })" - x-data="{ name: '', configId: '', schedule: '', enabled: true, loading: false, validate() { return this.configId && this.schedule; } }"> + hx-boost="true">
@@ -63,7 +71,6 @@ templ JobForm(ctx context.Context, data JobFormData) { type="text" name="name" id="name" - x-model="name" class="form-input pl-10 w-full rounded-lg border-secondary-300 dark:border-secondary-700 dark:bg-secondary-800 dark:text-secondary-100 focus:ring-primary-500 focus:border-primary-500" placeholder="Daily Production Backup"/>
@@ -74,23 +81,34 @@ templ JobForm(ctx context.Context, data JobFormData) {
- -
-
- -
- + +
} - + } else { +
+ No configurations available. Create one +
+ }
+

+ + Select one or more configurations to run on this schedule. +

@@ -103,7 +121,6 @@ templ JobForm(ctx context.Context, data JobFormData) { type="text" name="schedule" id="schedule" - x-model="schedule" required class="form-input pl-10 w-full rounded-lg border-secondary-300 dark:border-secondary-700 dark:bg-secondary-800 dark:text-secondary-100 focus:ring-primary-500 focus:border-primary-500" placeholder="*/15 * * * *"/> @@ -119,10 +136,10 @@ templ JobForm(ctx context.Context, data JobFormData) { - + name="enabled" + value="true" + checked + class="h-4 w-4 text-primary-600 focus:ring-primary-500 border-secondary-300 dark:border-secondary-700 rounded"/>

@@ -139,19 +156,9 @@ templ JobForm(ctx context.Context, data JobFormData) {

@@ -160,11 +167,7 @@ templ JobForm(ctx context.Context, data JobFormData) { class="space-y-6" hx-post={ fmt.Sprintf("/jobs/%d", data.Job.ID) } hx-target="body" - hx-boost="true" - @htmx:before-request="loading = true" - @htmx:after-request="loading = false" - @htmx:response-error="$dispatch('notification', { message: 'Failed to update job: ' + event.detail.xhr.responseText, type: 'error' })" - x-data={ fmt.Sprintf("{ name: '%s', configId: '%d', schedule: '%s', enabled: %v, loading: false, validate() { return this.configId && this.schedule; } }", data.Job.Name, data.Job.ConfigID, data.Job.Schedule, data.Job.Enabled) }> + hx-boost="true">
@@ -176,7 +179,7 @@ templ JobForm(ctx context.Context, data JobFormData) { type="text" name="name" id="name" - x-model="name" + value={ data.Job.Name } class="form-input pl-10 w-full rounded-lg border-secondary-300 dark:border-secondary-700 dark:bg-secondary-800 dark:text-secondary-100 focus:ring-primary-500 focus:border-primary-500" placeholder="Daily Production Backup"/>
@@ -185,25 +188,39 @@ templ JobForm(ctx context.Context, data JobFormData) { Descriptive name for this job (optional). If not provided, the config name will be used.

- +
- -
-
- -
- + +
} - + } else { +
+ No configurations available. Create one +
+ }
+

+ + Select one or more configurations to run on this schedule. +

@@ -216,7 +233,7 @@ templ JobForm(ctx context.Context, data JobFormData) { type="text" name="schedule" id="schedule" - x-model="schedule" + value={ data.Job.Schedule } required class="form-input pl-10 w-full rounded-lg border-secondary-300 dark:border-secondary-700 dark:bg-secondary-800 dark:text-secondary-100 focus:ring-primary-500 focus:border-primary-500" placeholder="*/15 * * * *"/> @@ -232,9 +249,12 @@ templ JobForm(ctx context.Context, data JobFormData) { -

@@ -251,19 +271,9 @@ templ JobForm(ctx context.Context, data JobFormData) {

diff --git a/components/jobs.templ b/components/jobs.templ index 3167d1f..aec5f3c 100644 --- a/components/jobs.templ +++ b/components/jobs.templ @@ -70,7 +70,8 @@ script triggerJobDelete(dialogId string, jobID uint, jobName string) { } type JobsData struct { - Jobs []db.Job + Jobs []db.Job + ConfigCount map[uint]int // Maps job ID to number of configs } templ Jobs(ctx context.Context, data JobsData) { @@ -334,8 +335,17 @@ templ Jobs(ctx context.Context, data JobsData) {

- - Config: { job.Config.Name } + + Configs: + + if count, ok := data.ConfigCount[job.ID]; ok && count > 1 { + { fmt.Sprintf("%d configurations", count) } + } else if job.ConfigID > 0 { + { job.Config.Name } + } else { + { "None" } + } +

diff --git a/internal/db/db.go b/internal/db/db.go index acc0f35..d1d9177 100644 --- a/internal/db/db.go +++ b/internal/db/db.go @@ -5,6 +5,8 @@ import ( "os" "os/exec" "path/filepath" + "strconv" + "strings" "time" "github.com/glebarez/sqlite" @@ -115,6 +117,7 @@ type Job struct { Name string `form:"name"` ConfigID uint `gorm:"not null" form:"config_id"` Config TransferConfig `gorm:"foreignkey:ConfigID"` + ConfigIDs string `gorm:"column:config_ids"` // Comma-separated list of config IDs Schedule string `gorm:"not null" form:"schedule"` Enabled bool `gorm:"default:true" form:"enabled"` LastRun *time.Time @@ -125,10 +128,64 @@ type Job struct { UpdatedAt time.Time } +// GetConfigIDsList returns the list of config IDs as integers +func (j *Job) GetConfigIDsList() []uint { + if j.ConfigIDs == "" { + // If ConfigIDs is empty but ConfigID is set, return that as the only ID + if j.ConfigID > 0 { + return []uint{j.ConfigID} + } + return []uint{} + } + + // Split the comma-separated string + strIDs := strings.Split(j.ConfigIDs, ",") + ids := make([]uint, 0, len(strIDs)) + + // Convert each string to uint + for _, strID := range strIDs { + if id, err := strconv.ParseUint(strings.TrimSpace(strID), 10, 32); err == nil { + ids = append(ids, uint(id)) + } + } + + return ids +} + +// SetConfigIDsList sets the config IDs from a slice of uint +func (j *Job) SetConfigIDsList(ids []uint) { + // Convert to strings + strIDs := make([]string, len(ids)) + for i, id := range ids { + strIDs[i] = strconv.FormatUint(uint64(id), 10) + } + + // Join with commas + j.ConfigIDs = strings.Join(strIDs, ",") + + // If there's at least one ID, set ConfigID to the first one for backward compatibility + if len(ids) > 0 { + j.ConfigID = ids[0] + } +} + +// GetConfigIDsAsStrings returns the list of config IDs as strings for template rendering +func (j *Job) GetConfigIDsAsStrings() []string { + ids := j.GetConfigIDsList() + strIDs := make([]string, len(ids)) + + for i, id := range ids { + strIDs[i] = fmt.Sprintf("'%d'", id) + } + + return strIDs +} + type JobHistory struct { ID uint `gorm:"primarykey"` JobID uint `gorm:"not null"` Job Job `gorm:"foreignkey:JobID"` + ConfigID uint `gorm:"default:0"` // The specific config ID this history entry is for StartTime time.Time `gorm:"not null"` EndTime *time.Time Status string `gorm:"not null"` @@ -142,6 +199,7 @@ type FileMetadata struct { ID uint `gorm:"primarykey"` JobID uint `gorm:"not null;index"` Job Job `gorm:"foreignkey:JobID"` + ConfigID uint `gorm:"default:0"` // The specific config ID this file was processed with FileName string `gorm:"not null"` OriginalPath string `gorm:"not null"` FileSize int64 `gorm:"not null"` @@ -777,3 +835,30 @@ func (db *DB) GetActiveJobs() ([]Job, error) { err := db.Preload("Config").Where("enabled = ?", true).Find(&jobs).Error return jobs, err } + +// GetConfigsForJob returns all transfer configurations associated with a job +func (db *DB) GetConfigsForJob(jobID uint) ([]TransferConfig, error) { + var job Job + if err := db.First(&job, jobID).Error; err != nil { + return nil, err + } + + // Get the list of config IDs + configIDs := job.GetConfigIDsList() + if len(configIDs) == 0 { + // If there are no IDs in the list but there is a configID, use that + if job.ConfigID > 0 { + configIDs = []uint{job.ConfigID} + } else { + return []TransferConfig{}, nil + } + } + + // Fetch all configs + var configs []TransferConfig + if err := db.Where("id IN ?", configIDs).Find(&configs).Error; err != nil { + return nil, err + } + + return configs, nil +} diff --git a/internal/db/migrations/add_multi_config_support.go b/internal/db/migrations/add_multi_config_support.go new file mode 100644 index 0000000..f65dca1 --- /dev/null +++ b/internal/db/migrations/add_multi_config_support.go @@ -0,0 +1,49 @@ +package migrations + +import ( + "github.com/go-gormigrate/gormigrate/v2" + "gorm.io/gorm" +) + +// AddMultiConfigSupport adds support for multiple configurations per job +func AddMultiConfigSupport() *gormigrate.Migration { + return &gormigrate.Migration{ + ID: "20250315_add_multi_config_support", + Migrate: func(tx *gorm.DB) error { + // Add config_ids column to jobs table + if err := tx.Exec("ALTER TABLE jobs ADD COLUMN config_ids TEXT").Error; err != nil { + return err + } + + // Add config_id column to job_histories table + if err := tx.Exec("ALTER TABLE job_histories ADD COLUMN config_id INTEGER").Error; err != nil { + return err + } + + // Add config_id column to file_metadata table + if err := tx.Exec("ALTER TABLE file_metadata ADD COLUMN config_id INTEGER").Error; err != nil { + return err + } + + // Update existing jobs to set the config_ids field to match the current config_id + if err := tx.Exec("UPDATE jobs SET config_ids = config_id WHERE config_id > 0").Error; err != nil { + return err + } + + return nil + }, + Rollback: func(tx *gorm.DB) error { + // Drop the config_id columns from job_histories and file_metadata + if err := tx.Exec("ALTER TABLE job_histories DROP COLUMN config_id").Error; err != nil { + return err + } + + if err := tx.Exec("ALTER TABLE file_metadata DROP COLUMN config_id").Error; err != nil { + return err + } + + // Drop the config_ids column from jobs + return tx.Exec("ALTER TABLE jobs DROP COLUMN config_ids").Error + }, + } +} diff --git a/internal/db/migrations/migrations.go b/internal/db/migrations/migrations.go index 055bfc6..b3ba1a2 100644 --- a/internal/db/migrations/migrations.go +++ b/internal/db/migrations/migrations.go @@ -13,6 +13,7 @@ func InitMigrations(db *gorm.DB) *gormigrate.Gormigrate { AddCloudStorageFields(), AddSkipProcessedFilesColumn(), AddMaxConcurrentTransfersColumn(), + AddMultiConfigSupport(), } return gormigrate.New(db, gormigrate.DefaultOptions, migrations) diff --git a/internal/handlers/jobs_handler.go b/internal/handlers/jobs_handler.go new file mode 100644 index 0000000..e0e73fb --- /dev/null +++ b/internal/handlers/jobs_handler.go @@ -0,0 +1,155 @@ +package handlers + +import ( + "errors" + "fmt" + "net/http" + "strconv" + + "github.com/gorilla/mux" + "gorm.io/gorm" + + "github.com/your-project/db" +) + +// handleCreateJob handles the creation of a new job +func (h *Handler) handleCreateJob(w http.ResponseWriter, r *http.Request) { + // Parse form data + if err := r.ParseForm(); err != nil { + h.Logger.Error("Error parsing form: %v", err) + http.Error(w, "Error parsing form", http.StatusBadRequest) + return + } + + // Get form values + name := r.FormValue("name") + schedule := r.FormValue("schedule") + enabled := r.FormValue("enabled") + + // Get config IDs + configIDs := r.Form["config_ids[]"] + + // Validate required fields + if len(configIDs) == 0 { + http.Error(w, "At least one configuration must be selected", http.StatusBadRequest) + return + } + + if schedule == "" { + http.Error(w, "Schedule is required", http.StatusBadRequest) + return + } + + // Parse config IDs and validate they exist + var configIDsList []uint + for _, configIDStr := range configIDs { + cID, err := strconv.ParseUint(configIDStr, 10, 32) + if err != nil { + h.Logger.Error("Error parsing config ID: %v", err) + http.Error(w, "Invalid config ID", http.StatusBadRequest) + return + } + + // Validate config exists + var config db.TransferConfig + if err := h.DB.First(&config, cID).Error; err != nil { + h.Logger.Error("Config not found: %v", err) + http.Error(w, fmt.Sprintf("Config ID %d not found", cID), http.StatusBadRequest) + return + } + + configIDsList = append(configIDsList, uint(cID)) + } + + // Create job with parsed values + job := db.Job{ + Name: name, + Schedule: schedule, + Enabled: enabled == "true", + } + + // Set config IDs + job.SetConfigIDsList(configIDsList) + + // ... existing code ... +} + +func (h *Handler) handleUpdateJob(w http.ResponseWriter, r *http.Request) { + // Parse path params + vars := mux.Vars(r) + jobID, err := strconv.ParseUint(vars["id"], 10, 32) + if err != nil { + h.Logger.Error("Error parsing job ID: %v", err) + http.Error(w, "Invalid job ID", http.StatusBadRequest) + return + } + + // Get existing job + var job db.Job + if err := h.DB.First(&job, jobID).Error; err != nil { + if errors.Is(err, gorm.ErrRecordNotFound) { + http.Error(w, "Job not found", http.StatusNotFound) + } else { + h.Logger.Error("Error getting job: %v", err) + http.Error(w, "Error getting job", http.StatusInternalServerError) + } + return + } + + // Parse form data + if err := r.ParseForm(); err != nil { + h.Logger.Error("Error parsing form: %v", err) + http.Error(w, "Error parsing form", http.StatusBadRequest) + return + } + + // Get form values + name := r.FormValue("name") + schedule := r.FormValue("schedule") + enabled := r.FormValue("enabled") + + // Get config IDs + configIDs := r.Form["config_ids[]"] + + // Validate required fields + if len(configIDs) == 0 { + http.Error(w, "At least one configuration must be selected", http.StatusBadRequest) + return + } + + if schedule == "" { + http.Error(w, "Schedule is required", http.StatusBadRequest) + return + } + + // Parse config IDs and validate they exist + var configIDsList []uint + for _, configIDStr := range configIDs { + cID, err := strconv.ParseUint(configIDStr, 10, 32) + if err != nil { + h.Logger.Error("Error parsing config ID: %v", err) + http.Error(w, "Invalid config ID", http.StatusBadRequest) + return + } + + // Validate config exists + var config db.TransferConfig + if err := h.DB.First(&config, cID).Error; err != nil { + h.Logger.Error("Config not found: %v", err) + http.Error(w, fmt.Sprintf("Config ID %d not found", cID), http.StatusBadRequest) + return + } + + configIDsList = append(configIDsList, uint(cID)) + } + + // Update job with parsed values + job.Name = name + job.Schedule = schedule + job.Enabled = enabled == "true" + + // Set config IDs + job.SetConfigIDsList(configIDsList) + + // ... existing code ... +} diff --git a/internal/scheduler/scheduler.go b/internal/scheduler/scheduler.go index e0f6508..1786eec 100644 --- a/internal/scheduler/scheduler.go +++ b/internal/scheduler/scheduler.go @@ -296,68 +296,82 @@ func (s *Scheduler) executeJob(jobID uint) { // Get job details var job db.Job - if err := s.db.Preload("Config").First(&job, jobID).Error; err != nil { + if err := s.db.First(&job, jobID).Error; err != nil { s.log.LogError("Error loading job %d: %v", jobID, err) return } - if job.Config.ID == 0 { - s.log.LogError("Error: job %d has no associated config", jobID) + // Get all configurations associated with this job + configs, err := s.db.GetConfigsForJob(jobID) + if err != nil { + s.log.LogError("Error loading configurations for job %d: %v", jobID, err) return } - // Add explicit database reload of the config to ensure we have the latest values - var config db.TransferConfig - if err := s.db.First(&config, job.Config.ID).Error; err != nil { - s.log.LogError("Error loading config %d: %v", job.Config.ID, err) + if len(configs) == 0 { + s.log.LogError("Error: job %d has no associated configurations", jobID) return } - // Replace the job's config with the freshly loaded one - job.Config = config - // Now the rest of your code will use the correct value - s.log.LogInfo("Loaded job %d with config: source=%s:%s, dest=%s:%s, skipProcessedFiles=%v, maxConcurrentTransfers=%d", - jobID, - job.Config.SourceType, - job.Config.SourcePath, - job.Config.DestinationType, - job.Config.DestinationPath, - job.Config.SkipProcessedFiles, - job.Config.MaxConcurrentTransfers, - ) - - // Create job history entry - startTime := time.Now() - history := &db.JobHistory{ - JobID: jobID, - StartTime: startTime, - Status: "running", - FilesTransferred: 0, - BytesTransferred: 0, - ErrorMessage: "", - } - if err := s.db.CreateJobHistory(history); err != nil { - s.log.LogError("Error creating job history for job %d: %v", jobID, err) - return - } + s.log.LogInfo("Loaded job %d with %d configurations", jobID, len(configs)) // Update job last run time - job.LastRun = &history.StartTime + startTime := time.Now() + job.LastRun = &startTime if err := s.db.UpdateJobStatus(&job); err != nil { s.log.LogError("Error updating job last run time for job %d: %v", jobID, err) } - // Reload the job from the database to get the latest values - if err := s.db.Preload("Config").First(&job, jobID).Error; err != nil { - s.log.LogError("Error reloading job %d: %v", jobID, err) - return + // Process each configuration in sequence + for i, config := range configs { + // Create job history entry for this configuration + history := &db.JobHistory{ + JobID: jobID, + ConfigID: config.ID, + StartTime: time.Now(), + Status: "running", + FilesTransferred: 0, + BytesTransferred: 0, + ErrorMessage: "", + } + if err := s.db.CreateJobHistory(history); err != nil { + s.log.LogError("Error creating job history for job %d, config %d: %v", jobID, config.ID, err) + continue + } + + s.log.LogInfo("Processing configuration %d (%d/%d) for job %d: source=%s:%s, dest=%s:%s", + config.ID, + i+1, + len(configs), + jobID, + config.SourceType, + config.SourcePath, + config.DestinationType, + config.DestinationPath, + ) + + // Execute the configuration transfer + s.executeConfigTransfer(job, config, history) } + // Update next run time if job is still scheduled + if entry := s.cron.Entry(s.jobs[jobID]); entry.ID != 0 { + job.NextRun = &entry.Next + if err := s.db.UpdateJobStatus(&job); err != nil { + s.log.LogError("Error updating next run time for job %d: %v", jobID, err) + } else { + s.log.LogInfo("Next run time for job %d: %s", jobID, entry.Next.Format(time.RFC3339)) + } + } +} + +// executeConfigTransfer performs the actual file transfer for a single configuration +func (s *Scheduler) executeConfigTransfer(job db.Job, config db.TransferConfig, history *db.JobHistory) { // Track files already processed in this job execution to prevent duplicates processedFiles := make(map[string]bool) // Get rclone config path - configPath := s.db.GetConfigRclonePath(&job.Config) + configPath := s.db.GetConfigRclonePath(&config) // Use lsjson to get file list and metadata in one operation instead of separate size and ls commands listArgs := []string{ @@ -368,17 +382,17 @@ func (s *Scheduler) executeJob(jobID uint) { } // Add file pattern filter if specified - if job.Config.FilePattern != "" && job.Config.FilePattern != "*" { + if config.FilePattern != "" && config.FilePattern != "*" { // Create a temporary filter file for complex patterns - filterFile, err := createRcloneFilterFile(job.Config.FilePattern) + filterFile, err := createRcloneFilterFile(config.FilePattern) if err != nil { - s.log.LogError("Error creating filter file for job %d: %v", jobID, err) + s.log.LogError("Error creating filter file for job %d, config %d: %v", job.ID, config.ID, err) history.Status = "failed" history.ErrorMessage = fmt.Sprintf("Filter Creation Error: %v", err) endTime := time.Now() history.EndTime = &endTime if err := s.db.UpdateJobHistory(history); err != nil { - s.log.LogError("Error updating job history for job %d: %v", jobID, err) + s.log.LogError("Error updating job history for job %d, config %d: %v", job.ID, config.ID, err) } return } @@ -388,19 +402,19 @@ func (s *Scheduler) executeJob(jobID uint) { // Add source path with bucket for S3-compatible storage var sourceListPath string - if job.Config.SourceType == "s3" || job.Config.SourceType == "minio" || job.Config.SourceType == "b2" { - sourceListPath = fmt.Sprintf("source_%d:%s", job.Config.ID, job.Config.SourceBucket) - if job.Config.SourcePath != "" && job.Config.SourcePath != "/" { - sourceListPath = fmt.Sprintf("source_%d:%s/%s", job.Config.ID, job.Config.SourceBucket, job.Config.SourcePath) + if config.SourceType == "s3" || config.SourceType == "minio" || config.SourceType == "b2" { + sourceListPath = fmt.Sprintf("source_%d:%s", config.ID, config.SourceBucket) + if config.SourcePath != "" && config.SourcePath != "/" { + sourceListPath = fmt.Sprintf("source_%d:%s/%s", config.ID, config.SourceBucket, config.SourcePath) } } else { - sourceListPath = fmt.Sprintf("source_%d:%s", job.Config.ID, job.Config.SourcePath) + sourceListPath = fmt.Sprintf("source_%d:%s", config.ID, config.SourcePath) } listArgs = append(listArgs, sourceListPath) // Execute lsjson command - s.log.LogInfo("Listing files with metadata for job %d: rclone %s", jobID, strings.Join(listArgs, " ")) + s.log.LogInfo("Listing files with metadata for job %d, config %d: rclone %s", job.ID, config.ID, strings.Join(listArgs, " ")) rclonePath := os.Getenv("RCLONE_PATH") if rclonePath == "" { rclonePath = "rclone" @@ -409,14 +423,14 @@ func (s *Scheduler) executeJob(jobID uint) { listOutput, listErr := listCmd.CombinedOutput() if listErr != nil { - s.log.LogError("Error listing files for job %d: %v", jobID, listErr) + s.log.LogError("Error listing files for job %d, config %d: %v", job.ID, config.ID, listErr) // s.log.Debug.Printf("Output: %s", string(listOutput)) history.Status = "failed" history.ErrorMessage = fmt.Sprintf("File Listing Error: %v\nOutput: %s", listErr, string(listOutput)) endTime := time.Now() history.EndTime = &endTime if err := s.db.UpdateJobHistory(history); err != nil { - s.log.LogError("Error updating job history for job %d: %v", jobID, err) + s.log.LogError("Error updating job history for job %d, config %d: %v", job.ID, config.ID, err) } return } @@ -424,13 +438,13 @@ func (s *Scheduler) executeJob(jobID uint) { // Parse JSON output to get file information var fileEntries []map[string]interface{} if err := json.Unmarshal(listOutput, &fileEntries); err != nil { - s.log.LogError("Error parsing file list JSON for job %d: %v", jobID, err) + s.log.LogError("Error parsing file list JSON for job %d, config %d: %v", job.ID, config.ID, err) history.Status = "failed" history.ErrorMessage = fmt.Sprintf("JSON Parsing Error: %v", err) endTime := time.Now() history.EndTime = &endTime if err := s.db.UpdateJobHistory(history); err != nil { - s.log.LogError("Error updating job history for job %d: %v", jobID, err) + s.log.LogError("Error updating job history for job %d, config %d: %v", job.ID, config.ID, err) } return } @@ -453,400 +467,399 @@ func (s *Scheduler) executeJob(jobID uint) { } } - s.log.LogInfo("Found %d files totaling %d bytes to transfer for job %d", len(files), totalSize, jobID) + s.log.LogInfo("Found %d files totaling %d bytes to transfer for job %d, config %d", len(files), totalSize, job.ID, config.ID) // Update history with size information history.BytesTransferred = totalSize if len(files) == 0 { - s.log.LogInfo("No files to transfer for job %d", jobID) + s.log.LogInfo("No files to transfer for job %d, config %d", job.ID, config.ID) history.Status = "completed" history.ErrorMessage = "" history.FilesTransferred = 0 - } else { - var transferErrors []string - filesTransferred := 0 - - // Use mutex for thread-safe access to shared variables - var mutex sync.Mutex - - // Determine number of concurrent transfers - maxConcurrent := job.Config.MaxConcurrentTransfers - if maxConcurrent < 1 { - maxConcurrent = 1 // Default to 1 if not set + endTime := time.Now() + history.EndTime = &endTime + if err := s.db.UpdateJobHistory(history); err != nil { + s.log.LogError("Error updating job history for job %d, config %d: %v", job.ID, config.ID, err) } - s.log.LogInfo("Using %d concurrent transfers for job %d", maxConcurrent, jobID) + return + } - // Create wait group for concurrent processing - var wg sync.WaitGroup + var transferErrors []string + filesTransferred := 0 - // Create channel to limit concurrency - concurrencySemaphore := make(chan struct{}, maxConcurrent) + // Use mutex for thread-safe access to shared variables + var mutex sync.Mutex - // Process each file individually - for _, fileEntry := range files { - fileName, ok := fileEntry["Path"].(string) - if !ok || fileName == "" { - continue - } + // Determine number of concurrent transfers + maxConcurrent := config.MaxConcurrentTransfers + if maxConcurrent < 1 { + maxConcurrent = 1 // Default to 1 if not set + } + s.log.LogInfo("Using %d concurrent transfers for job %d, config %d", maxConcurrent, job.ID, config.ID) - // Skip files that have already been processed in this execution - if processedFiles[fileName] { - s.log.LogDebug("Skipping duplicate file entry: %s (already processed in this execution)", fileName) - continue - } + // Create wait group for concurrent processing + var wg sync.WaitGroup - // Extract hash from the file entry - fileHash := "" - if hash, ok := fileEntry["Hashes"].(map[string]interface{}); ok { - // Try several hash algorithms in order of preference - for _, hashType := range []string{"SHA-1", "MD5"} { - if hashValue, found := hash[hashType]; found { - if hashStr, ok := hashValue.(string); ok { - fileHash = hashStr - break - } + // Create channel to limit concurrency + concurrencySemaphore := make(chan struct{}, maxConcurrent) + + // Process each file individually + for _, fileEntry := range files { + fileName, ok := fileEntry["Path"].(string) + if !ok || fileName == "" { + continue + } + + // Skip files that have already been processed in this execution + if processedFiles[fileName] { + s.log.LogDebug("Skipping duplicate file entry: %s (already processed in this execution)", fileName) + continue + } + + // Extract hash from the file entry + fileHash := "" + if hashes, ok := fileEntry["Hashes"].(map[string]interface{}); ok { + // Try several hash algorithms in order of preference + for _, hashType := range []string{"SHA-1", "sha1", "MD5", "md5", "sha256", "crc32"} { + if hashValue, found := hashes[hashType]; found { + if hashStr, ok := hashValue.(string); ok && hashStr != "" { + s.log.LogDebug("Found hash %s: %s for file %s", hashType, hashStr, fileName) + fileHash = hashStr + break } } } + } - // Extract size from the file entry - fileSize := int64(0) - if size, ok := fileEntry["Size"].(float64); ok { - fileSize = int64(size) - } + // Log if no hash was found + if fileHash == "" { + s.log.LogDebug("No hash found for file %s. Available fields: %v", fileName, fileEntry) + } - // Skip files that have already been processed based on hash - skipFiles := job.Config.SkipProcessedFiles - if skipFiles && fileHash != "" { - alreadyProcessed, prevMetadata, err := s.hasFileBeenProcessed(jobID, fileHash) - if err == nil && alreadyProcessed { - s.log.LogDebug("File %s with hash %s was previously processed on %s with status: %s", - fileName, fileHash, prevMetadata.ProcessedTime.Format(time.RFC3339), prevMetadata.Status) + // Extract size from the file entry + fileSize := int64(0) + if size, ok := fileEntry["Size"].(float64); ok { + fileSize = int64(size) + } - // Determine if we should skip this file based on status - shouldSkip := false - if prevMetadata.Status == "processed" || - prevMetadata.Status == "archived" || - prevMetadata.Status == "deleted" || - prevMetadata.Status == "archived_and_deleted" { - shouldSkip = true - } + // Skip files that have already been processed based on hash + skipFiles := config.SkipProcessedFiles + if skipFiles && fileHash != "" { + alreadyProcessed, prevMetadata, err := s.hasFileBeenProcessed(job.ID, fileHash) + if err == nil && alreadyProcessed { + s.log.LogDebug("File %s with hash %s was previously processed on %s with status: %s", + fileName, fileHash, prevMetadata.ProcessedTime.Format(time.RFC3339), prevMetadata.Status) - if shouldSkip { - s.log.LogInfo("Skipping unchanged file %s (hash matches previous processing)", fileName) - continue - } else { - s.log.LogInfo("Re-processing file %s despite previous processing (skipProcessedFiles=%v)", fileName, skipFiles) - } - } - } - - // Also check the processing history for this specific file name - prevMetadata, histErr := s.checkFileProcessingHistory(jobID, fileName) - if histErr == nil { - s.log.LogDebug("File %s was previously processed on %s with status: %s", - fileName, prevMetadata.ProcessedTime.Format(time.RFC3339), prevMetadata.Status) - - // Determine if we should skip this file based on name+hash match + // Determine if we should skip this file based on status shouldSkip := false - if skipFiles && fileHash != "" && fileHash == prevMetadata.FileHash { - if prevMetadata.Status == "processed" || - prevMetadata.Status == "archived" || - prevMetadata.Status == "deleted" || - prevMetadata.Status == "archived_and_deleted" { - shouldSkip = true - } + if prevMetadata.Status == "processed" || + prevMetadata.Status == "archived" || + prevMetadata.Status == "deleted" || + prevMetadata.Status == "archived_and_deleted" { + shouldSkip = true } if shouldSkip { s.log.LogInfo("Skipping unchanged file %s (hash matches previous processing)", fileName) - // Skip this file and continue to the next one continue - } else if fileHash != "" && fileHash == prevMetadata.FileHash { - s.log.LogInfo("Re-processing file %s despite matching hash (skipProcessedFiles=%v)", fileName, skipFiles) + } else { + s.log.LogInfo("Re-processing file %s despite previous processing (skipProcessedFiles=%v)", fileName, skipFiles) + } + } + } + + // Also check the processing history for this specific file name + prevMetadata, histErr := s.checkFileProcessingHistory(job.ID, fileName) + if histErr == nil { + s.log.LogDebug("File %s was previously processed on %s with status: %s", + fileName, prevMetadata.ProcessedTime.Format(time.RFC3339), prevMetadata.Status) + + // Determine if we should skip this file based on name+hash match + shouldSkip := false + if skipFiles && fileHash != "" && fileHash == prevMetadata.FileHash { + if prevMetadata.Status == "processed" || + prevMetadata.Status == "archived" || + prevMetadata.Status == "deleted" || + prevMetadata.Status == "archived_and_deleted" { + shouldSkip = true } } - // Mark this file as processed for this execution before launching goroutine - // to prevent duplicate processing - processedFiles[fileName] = true - - // Add to wait group before starting goroutine - wg.Add(1) - - // Get creation time and mod time for the file metadata - createTime := time.Now() - modTime := time.Now() - if creationTimeStr, ok := fileEntry["ModTime"].(string); ok { - if t, err := time.Parse(time.RFC3339Nano, creationTimeStr); err == nil { - modTime = t - createTime = t - } + if shouldSkip { + s.log.LogInfo("Skipping unchanged file %s (hash matches previous processing)", fileName) + // Skip this file and continue to the next one + continue + } else if fileHash != "" && fileHash == prevMetadata.FileHash { + s.log.LogInfo("Re-processing file %s despite matching hash (skipProcessedFiles=%v)", fileName, skipFiles) } + } - // Capture current file information for goroutine - currentFileName := fileName - currentFileHash := fileHash - currentFileSize := fileSize - currentCreateTime := createTime - currentModTime := modTime + // Mark this file as processed for this execution before launching goroutine + // to prevent duplicate processing + processedFiles[fileName] = true - // Start goroutine for concurrent processing - go func() { - // Acquire semaphore - concurrencySemaphore <- struct{}{} - defer func() { - // Release semaphore and mark work as done - <-concurrencySemaphore - wg.Done() - }() + // Add to wait group before starting goroutine + wg.Add(1) - // Prepare moveto command for transfer - transferArgs := []string{ - "--config", configPath, - "copyto", - "--progress", - "--stats-one-line", - "--verbose", - "--stats", "1s", - } + // Get creation time and mod time for the file metadata + createTime := time.Now() + modTime := time.Now() + if creationTimeStr, ok := fileEntry["ModTime"].(string); ok { + if t, err := time.Parse(time.RFC3339Nano, creationTimeStr); err == nil { + modTime = t + createTime = t + } + } - // Source and destination paths - var sourcePath, destPath string + // Capture current file information for goroutine + currentFileName := fileName + currentFileHash := fileHash + currentFileSize := fileSize + currentCreateTime := createTime + currentModTime := modTime - // For S3, MinIO, and B2, include the bucket in the path - if job.Config.SourceType == "s3" || job.Config.SourceType == "minio" || job.Config.SourceType == "b2" { - sourcePath = fmt.Sprintf("source_%d:%s/%s", job.Config.ID, job.Config.SourceBucket, currentFileName) - if job.Config.SourcePath != "" && job.Config.SourcePath != "/" { - sourcePath = fmt.Sprintf("source_%d:%s/%s/%s", job.Config.ID, job.Config.SourceBucket, job.Config.SourcePath, currentFileName) - } - } else { - sourcePath = fmt.Sprintf("source_%d:%s/%s", job.Config.ID, job.Config.SourcePath, currentFileName) - } + // Log the file information that will be processed + s.log.LogDebug("Processing file: %s, size: %d, hash: %s", currentFileName, currentFileSize, currentFileHash) - var destFile string = currentFileName - - if job.Config.DestinationType == "s3" || job.Config.DestinationType == "minio" || job.Config.DestinationType == "b2" { - destPath = fmt.Sprintf("dest_%d:%s/%s", job.Config.ID, job.Config.DestBucket, currentFileName) - if job.Config.DestinationPath != "" && job.Config.DestinationPath != "/" { - destPath = fmt.Sprintf("dest_%d:%s/%s/%s", job.Config.ID, job.Config.DestBucket, job.Config.DestinationPath, currentFileName) - } - } else { - destPath = fmt.Sprintf("dest_%d:%s/%s", job.Config.ID, job.Config.DestinationPath, currentFileName) - } - - // Add output filename pattern if specified - if job.Config.OutputPattern != "" { - // Process the output pattern for this specific file - destFile = ProcessOutputPattern(job.Config.OutputPattern, currentFileName) - - if job.Config.DestinationType == "s3" || job.Config.DestinationType == "minio" || job.Config.DestinationType == "b2" { - destPath = fmt.Sprintf("dest_%d:%s/%s", job.Config.ID, job.Config.DestBucket, destFile) - if job.Config.DestinationPath != "" && job.Config.DestinationPath != "/" { - destPath = fmt.Sprintf("dest_%d:%s/%s/%s", job.Config.ID, job.Config.DestBucket, job.Config.DestinationPath, destFile) - } - } else { - destPath = fmt.Sprintf("dest_%d:%s/%s", job.Config.ID, job.Config.DestinationPath, destFile) - } - - s.log.LogDebug("Renaming file from %s to %s for job %d", currentFileName, destFile, jobID) - } - - // Add custom flags if specified - if job.Config.RcloneFlags != "" { - customFlags := strings.Split(job.Config.RcloneFlags, " ") - transferArgs = append(transferArgs, customFlags...) - s.log.LogDebug("Added custom flags for job %d: %v", jobID, customFlags) - } - - // Add source and destination to the command - transferArgs = append(transferArgs, sourcePath, destPath) - - // Execute transfer for this file - s.log.LogInfo("Executing rclone transfer command for job %d, file %s: rclone %s", - jobID, currentFileName, strings.Join(transferArgs, " ")) - // Get the rclone path from the environment variable or use the default path - rclonePath := os.Getenv("RCLONE_PATH") - if rclonePath == "" { - rclonePath = "rclone" - } - cmd := exec.Command(rclonePath, transferArgs...) - fileOutput, fileErr := cmd.CombinedOutput() - - // Print the output - s.log.LogDebug("Output for file %s: %s", currentFileName, string(fileOutput)) - - // Create file metadata record - fileStatus := "processed" - var fileErrorMsg string - var destPathForDB string - - // Check if file was successfully transferred - if fileErr != nil { - s.log.LogError("Error transferring file %s for job %d: %v", currentFileName, jobID, fileErr) - mutex.Lock() - transferErrors = append(transferErrors, fmt.Sprintf("File %s: %v", currentFileName, fileErr)) - mutex.Unlock() - fileStatus = "error" - fileErrorMsg = fileErr.Error() - } else { - mutex.Lock() - filesTransferred++ - mutex.Unlock() - s.log.LogInfo("Successfully transferred file %s for job %d", currentFileName, jobID) - - // Extract the actual destination path (without rclone remote prefix) - if job.Config.DestinationType == "local" { - destPathForDB = filepath.Join(job.Config.DestinationPath, destFile) - } else { - // For remote destinations, store the path format - if job.Config.DestinationType == "s3" || job.Config.DestinationType == "minio" || job.Config.DestinationType == "b2" { - if job.Config.DestinationPath != "" && job.Config.DestinationPath != "/" { - destPathForDB = fmt.Sprintf("%s/%s/%s", job.Config.DestBucket, job.Config.DestinationPath, destFile) - } else { - destPathForDB = fmt.Sprintf("%s/%s", job.Config.DestBucket, destFile) - } - } else { - destPathForDB = fmt.Sprintf("%s/%s", job.Config.DestinationPath, destFile) - } - } - - // If archiving is enabled and transfer was successful, move files to archive - if job.Config.ArchiveEnabled && job.Config.ArchivePath != "" { - s.log.LogInfo("Archiving file %s for job %d", currentFileName, jobID) - - // We don't need to move the file since we used moveto, but we can copy it to archive - archiveArgs := []string{ - "--config", configPath, - "copyto", - sourcePath, - } - - // Construct archive path with bucket if needed - var archiveDest string - if job.Config.SourceType == "s3" || job.Config.SourceType == "minio" || job.Config.SourceType == "b2" { - archiveDest = fmt.Sprintf("source_%d:%s/%s/%s", job.Config.ID, job.Config.SourceBucket, job.Config.ArchivePath, currentFileName) - } else { - archiveDest = fmt.Sprintf("source_%d:%s/%s", job.Config.ID, job.Config.ArchivePath, currentFileName) - } - - archiveArgs = append(archiveArgs, archiveDest) - - s.log.LogInfo("Executing rclone archive command for job %d, file %s: rclone %s", - jobID, currentFileName, strings.Join(archiveArgs, " ")) - // Get the rclone path from the environment variable or use the default path - rclonePath := os.Getenv("RCLONE_PATH") - if rclonePath == "" { - rclonePath = "rclone" - } - archiveCmd := exec.Command(rclonePath, archiveArgs...) - archiveOutput, archiveErr := archiveCmd.CombinedOutput() - - // Print the output - s.log.LogDebug("Output for file %s: %s", currentFileName, string(archiveOutput)) - - // Check if file was successfully transferred - if archiveErr != nil { - s.log.LogError("Warning: Error archiving file %s for job %d: %v", currentFileName, jobID, archiveErr) - mutex.Lock() - transferErrors = append(transferErrors, - fmt.Sprintf("Archive error for file %s: %v", currentFileName, archiveErr)) - mutex.Unlock() - } else { - fileStatus = "archived" - } - } - - if job.Config.DeleteAfterTransfer { - s.log.LogInfo("Deleting file %s for job %d", currentFileName, jobID) - deleteArgs := []string{ - "--config", configPath, - "deletefile", - sourcePath} - deleteCmd := exec.Command(rclonePath, deleteArgs...) - deleteOutput, deleteErr := deleteCmd.CombinedOutput() - s.log.LogDebug("Output for file %s: %s", currentFileName, string(deleteOutput)) - if deleteErr != nil { - s.log.LogError("Error deleting file %s for job %d: %v", currentFileName, jobID, deleteErr) - mutex.Lock() - transferErrors = append(transferErrors, - fmt.Sprintf("Delete error for file %s: %v", currentFileName, deleteErr)) - mutex.Unlock() - } else { - if fileStatus == "archived" { - fileStatus = "archived_and_deleted" - } else { - fileStatus = "deleted" - } - } - } - } - - // Create and save file metadata - metadata := &db.FileMetadata{ - JobID: jobID, - FileName: currentFileName, - OriginalPath: job.Config.SourcePath, - FileSize: currentFileSize, - FileHash: currentFileHash, - CreationTime: currentCreateTime, - ModTime: currentModTime, - ProcessedTime: time.Now(), - DestinationPath: destPathForDB, - Status: fileStatus, - ErrorMessage: fileErrorMsg, - } - - if err := s.db.CreateFileMetadata(metadata); err != nil { - s.log.LogError("Error creating file metadata for %s: %v", currentFileName, err) - } else { - s.log.LogDebug("Created file metadata record for %s (ID: %d)", currentFileName, metadata.ID) - } + // Start goroutine for concurrent processing + go func() { + // Acquire semaphore + concurrencySemaphore <- struct{}{} + defer func() { + // Release semaphore and mark work as done + <-concurrencySemaphore + wg.Done() }() - } - // Wait for all transfers to complete - wg.Wait() + // Prepare moveto command for transfer + transferArgs := []string{ + "--config", configPath, + "copyto", + "--progress", + "--stats-one-line", + "--verbose", + "--stats", "1s", + } - // Clean up concurrency semaphore - close(concurrencySemaphore) + // Source and destination paths + var sourcePath, destPath string - // Update job history with transfer results - history.FilesTransferred = filesTransferred + // For S3, MinIO, and B2, include the bucket in the path + if config.SourceType == "s3" || config.SourceType == "minio" || config.SourceType == "b2" { + sourcePath = fmt.Sprintf("source_%d:%s/%s", config.ID, config.SourceBucket, currentFileName) + if config.SourcePath != "" && config.SourcePath != "/" { + sourcePath = fmt.Sprintf("source_%d:%s/%s/%s", config.ID, config.SourceBucket, config.SourcePath, currentFileName) + } + } else { + sourcePath = fmt.Sprintf("source_%d:%s/%s", config.ID, config.SourcePath, currentFileName) + } - if len(transferErrors) > 0 { - history.Status = "completed_with_errors" - history.ErrorMessage = fmt.Sprintf("Transfer completed with %d errors:\n%s", - len(transferErrors), strings.Join(transferErrors, "\n")) - } + var destFile string = currentFileName + + if config.DestinationType == "s3" || config.DestinationType == "minio" || config.DestinationType == "b2" { + destPath = fmt.Sprintf("dest_%d:%s/%s", config.ID, config.DestBucket, currentFileName) + if config.DestinationPath != "" && config.DestinationPath != "/" { + destPath = fmt.Sprintf("dest_%d:%s/%s/%s", config.ID, config.DestBucket, config.DestinationPath, currentFileName) + } + } else { + destPath = fmt.Sprintf("dest_%d:%s/%s", config.ID, config.DestinationPath, currentFileName) + } + + // Add output filename pattern if specified + if config.OutputPattern != "" { + // Process the output pattern for this specific file + destFile = ProcessOutputPattern(config.OutputPattern, currentFileName) + + if config.DestinationType == "s3" || config.DestinationType == "minio" || config.DestinationType == "b2" { + destPath = fmt.Sprintf("dest_%d:%s/%s", config.ID, config.DestBucket, destFile) + if config.DestinationPath != "" && config.DestinationPath != "/" { + destPath = fmt.Sprintf("dest_%d:%s/%s/%s", config.ID, config.DestBucket, config.DestinationPath, destFile) + } + } else { + destPath = fmt.Sprintf("dest_%d:%s/%s", config.ID, config.DestinationPath, destFile) + } + + s.log.LogDebug("Renaming file from %s to %s for job %d, config %d", currentFileName, destFile, job.ID, config.ID) + } + + // Add custom flags if specified + if config.RcloneFlags != "" { + customFlags := strings.Split(config.RcloneFlags, " ") + transferArgs = append(transferArgs, customFlags...) + s.log.LogDebug("Added custom flags for job %d, config %d: %v", job.ID, config.ID, customFlags) + } + + // Add source and destination to the command + transferArgs = append(transferArgs, sourcePath, destPath) + + // Execute transfer for this file + s.log.LogInfo("Executing rclone transfer command for job %d, config %d, file %s: rclone %s", + job.ID, config.ID, currentFileName, strings.Join(transferArgs, " ")) + // Get the rclone path from the environment variable or use the default path + rclonePath := os.Getenv("RCLONE_PATH") + if rclonePath == "" { + rclonePath = "rclone" + } + cmd := exec.Command(rclonePath, transferArgs...) + fileOutput, fileErr := cmd.CombinedOutput() + + // Print the output + s.log.LogDebug("Output for file %s: %s", currentFileName, string(fileOutput)) + + // Create file metadata record + fileStatus := "processed" + var fileErrorMsg string + var destPathForDB string + + // Check if file was successfully transferred + if fileErr != nil { + s.log.LogError("Error transferring file %s for job %d, config %d: %v", currentFileName, job.ID, config.ID, fileErr) + mutex.Lock() + transferErrors = append(transferErrors, fmt.Sprintf("File %s: %v", currentFileName, fileErr)) + mutex.Unlock() + fileStatus = "error" + fileErrorMsg = fileErr.Error() + } else { + mutex.Lock() + filesTransferred++ + mutex.Unlock() + s.log.LogInfo("Successfully transferred file %s for job %d, config %d", currentFileName, job.ID, config.ID) + + // Extract the actual destination path (without rclone remote prefix) + if config.DestinationType == "local" { + destPathForDB = filepath.Join(config.DestinationPath, destFile) + } else { + // For remote destinations, store the path format + if config.DestinationType == "s3" || config.DestinationType == "minio" || config.DestinationType == "b2" { + if config.DestinationPath != "" && config.DestinationPath != "/" { + destPathForDB = fmt.Sprintf("%s/%s/%s", config.DestBucket, config.DestinationPath, destFile) + } else { + destPathForDB = fmt.Sprintf("%s/%s", config.DestBucket, destFile) + } + } else { + destPathForDB = fmt.Sprintf("%s/%s", config.DestinationPath, destFile) + } + } + + // If archiving is enabled and transfer was successful, move files to archive + if config.ArchiveEnabled && config.ArchivePath != "" { + s.log.LogInfo("Archiving file %s for job %d, config %d", currentFileName, job.ID, config.ID) + + // We don't need to move the file since we used moveto, but we can copy it to archive + archiveArgs := []string{ + "--config", configPath, + "copyto", + sourcePath, + } + + // Construct archive path with bucket if needed + var archiveDest string + if config.SourceType == "s3" || config.SourceType == "minio" || config.SourceType == "b2" { + archiveDest = fmt.Sprintf("source_%d:%s/%s/%s", config.ID, config.SourceBucket, config.ArchivePath, currentFileName) + } else { + archiveDest = fmt.Sprintf("source_%d:%s/%s", config.ID, config.ArchivePath, currentFileName) + } + + archiveArgs = append(archiveArgs, archiveDest) + + s.log.LogInfo("Executing rclone archive command for job %d, config %d, file %s: rclone %s", + job.ID, config.ID, currentFileName, strings.Join(archiveArgs, " ")) + // Get the rclone path from the environment variable or use the default path + rclonePath := os.Getenv("RCLONE_PATH") + if rclonePath == "" { + rclonePath = "rclone" + } + archiveCmd := exec.Command(rclonePath, archiveArgs...) + archiveOutput, archiveErr := archiveCmd.CombinedOutput() + + // Print the output + s.log.LogDebug("Output for file %s: %s", currentFileName, string(archiveOutput)) + + // Check if file was successfully transferred + if archiveErr != nil { + s.log.LogError("Warning: Error archiving file %s for job %d, config %d: %v", currentFileName, job.ID, config.ID, archiveErr) + mutex.Lock() + transferErrors = append(transferErrors, + fmt.Sprintf("Archive error for file %s: %v", currentFileName, archiveErr)) + mutex.Unlock() + } else { + fileStatus = "archived" + } + } + + if config.DeleteAfterTransfer { + s.log.LogInfo("Deleting file %s for job %d, config %d", currentFileName, job.ID, config.ID) + deleteArgs := []string{ + "--config", configPath, + "deletefile", + sourcePath} + deleteCmd := exec.Command(rclonePath, deleteArgs...) + deleteOutput, deleteErr := deleteCmd.CombinedOutput() + s.log.LogDebug("Output for file %s: %s", currentFileName, string(deleteOutput)) + if deleteErr != nil { + s.log.LogError("Error deleting file %s for job %d, config %d: %v", currentFileName, job.ID, config.ID, deleteErr) + mutex.Lock() + transferErrors = append(transferErrors, + fmt.Sprintf("Delete error for file %s: %v", currentFileName, deleteErr)) + mutex.Unlock() + } else { + if fileStatus == "archived" { + fileStatus = "archived_and_deleted" + } else { + fileStatus = "deleted" + } + } + } + } + + // Create and save file metadata + metadata := &db.FileMetadata{ + JobID: job.ID, + ConfigID: config.ID, + FileName: currentFileName, + OriginalPath: config.SourcePath, + FileSize: currentFileSize, + FileHash: currentFileHash, + CreationTime: currentCreateTime, + ModTime: currentModTime, + ProcessedTime: time.Now(), + DestinationPath: destPathForDB, + Status: fileStatus, + ErrorMessage: fileErrorMsg, + } + + if err := s.db.CreateFileMetadata(metadata); err != nil { + s.log.LogError("Error creating file metadata for %s: %v", currentFileName, err) + } else { + s.log.LogDebug("Created file metadata record for %s (ID: %d) with hash: %s", currentFileName, metadata.ID, currentFileHash) + } + }() + } + + // Wait for all transfers to complete + wg.Wait() + + // Clean up concurrency semaphore + close(concurrencySemaphore) + + // Update job history with transfer results + history.FilesTransferred = filesTransferred + + if len(transferErrors) > 0 { + history.Status = "completed_with_errors" + history.ErrorMessage = fmt.Sprintf("Transfer completed with %d errors:\n%s", + len(transferErrors), strings.Join(transferErrors, "\n")) + } else { + history.Status = "completed" } // Update job history with completion status and end time endTime := time.Now() history.EndTime = &endTime - if job.Config.ArchiveEnabled && job.Config.ArchivePath != "" { - if history.ErrorMessage != "" { - history.Status = "completed_with_archive_error" - } else { - history.Status = "completed" - } - } else { - history.Status = "completed" - } if err := s.db.UpdateJobHistory(history); err != nil { - s.log.LogError("Error updating job history for job %d: %v", jobID, err) - } - - // Update next run time if job is still scheduled - if entry := s.cron.Entry(s.jobs[jobID]); entry.ID != 0 { - job.NextRun = &entry.Next - if err := s.db.UpdateJobStatus(&job); err != nil { - s.log.LogError("Error updating next run time for job %d: %v", jobID, err) - } else { - s.log.LogInfo("Next run time for job %d: %s", jobID, entry.Next.Format(time.RFC3339)) - } + s.log.LogError("Error updating job history for job %d, config %d: %v", job.ID, config.ID, err) } } diff --git a/internal/web/handlers/dashboard_handlers.go b/internal/web/handlers/dashboard_handlers.go index ffd99d5..5bcbf32 100644 --- a/internal/web/handlers/dashboard_handlers.go +++ b/internal/web/handlers/dashboard_handlers.go @@ -17,7 +17,7 @@ func (h *Handlers) HandleDashboard(c *gin.Context) { // Get recent job history var recentHistory []db.JobHistory - h.DB.Order("start_time DESC").Limit(5).Find(&recentHistory) + h.DB.Preload("Job.Config").Order("start_time DESC").Limit(5).Find(&recentHistory) // Get job statistics var totalJobs int64 @@ -29,11 +29,50 @@ func (h *Handlers) HandleDashboard(c *gin.Context) { var failedJobs int64 h.DB.Model(&db.JobHistory{}).Where("status = ?", "failed").Count(&failedJobs) + // Create a map to hold all relevant config IDs + configIDs := make(map[uint]bool) + + // Collect all config IDs from recent history entries + for _, h := range recentHistory { + // Add the specific config ID used for this history entry if it exists + if h.ConfigID > 0 { + configIDs[h.ConfigID] = true + } + + // Add the job's default config ID as a fallback + if h.Job.ConfigID > 0 { + configIDs[h.Job.ConfigID] = true + } + } + + // Create a map to store all configs by their ID + configsMap := make(map[uint]db.TransferConfig) + + // Load all necessary configurations + if len(configIDs) > 0 { + var configsList []db.TransferConfig + configIDsList := make([]uint, 0, len(configIDs)) + + // Extract config IDs from the map + for id := range configIDs { + configIDsList = append(configIDsList, id) + } + + // Load all configurations in one query + if err := h.DB.Where("id IN ?", configIDsList).Find(&configsList).Error; err == nil { + // Create the lookup map + for _, config := range configsList { + configsMap[config.ID] = config + } + } + } + data := components.DashboardData{ RecentJobs: recentHistory, ActiveTransfers: int(totalJobs), CompletedToday: int(completedJobs), FailedTransfers: int(failedJobs), + Configs: configsMap, } components.Dashboard(components.CreateTemplateContext(c), data).Render(c, c.Writer) @@ -110,6 +149,44 @@ func (h *Handlers) HandleHistory(c *gin.Context) { return } + // Create a map to hold all relevant config IDs + configIDs := make(map[uint]bool) + + // Collect all config IDs from history entries + for _, h := range history { + // Add the specific config ID used for this history entry if it exists + if h.ConfigID > 0 { + configIDs[h.ConfigID] = true + } + + // Add the job's default config ID as a fallback + if h.Job.ConfigID > 0 { + configIDs[h.Job.ConfigID] = true + } + } + + // Create a map to store all configs by their ID + configsMap := make(map[uint]db.TransferConfig) + + // Load all necessary configurations + if len(configIDs) > 0 { + var configsList []db.TransferConfig + configIDsList := make([]uint, 0, len(configIDs)) + + // Extract config IDs from the map + for id := range configIDs { + configIDsList = append(configIDsList, id) + } + + // Load all configurations in one query + if err := h.DB.Where("id IN ?", configIDsList).Find(&configsList).Error; err == nil { + // Create the lookup map + for _, config := range configsList { + configsMap[config.ID] = config + } + } + } + data := components.HistoryData{ History: history, CurrentPage: page, @@ -117,6 +194,7 @@ func (h *Handlers) HandleHistory(c *gin.Context) { SearchTerm: searchTerm, PageSize: pageSize, Total: int(total), + Configs: configsMap, } // If this is an HTMX request, only render the history content component diff --git a/internal/web/handlers/job_handlers.go b/internal/web/handlers/job_handlers.go index 5fe5a2f..bded0e4 100644 --- a/internal/web/handlers/job_handlers.go +++ b/internal/web/handlers/job_handlers.go @@ -3,6 +3,7 @@ package handlers import ( "fmt" "net/http" + "strconv" "github.com/gin-gonic/gin" "github.com/starfleetcptn/gomft/components" @@ -12,12 +13,28 @@ import ( // HandleJobs handles the GET /jobs route func (h *Handlers) HandleJobs(c *gin.Context) { userID := c.GetUint("userID") - + var jobs []db.Job h.DB.Where("created_by = ?", userID).Preload("Config").Find(&jobs) + // Create a map to store config counts for each job + configCount := make(map[uint]int) + + // Count configurations for each job + for _, job := range jobs { + // Get all configurations for this job + configs, err := h.DB.GetConfigsForJob(job.ID) + if err != nil { + c.Error(fmt.Errorf("error loading configurations for job %d: %v", job.ID, err)) + configCount[job.ID] = 0 + } else { + configCount[job.ID] = len(configs) + } + } + data := components.JobsData{ - Jobs: jobs, + Jobs: jobs, + ConfigCount: configCount, } components.Jobs(c, data).Render(c, c.Writer) } @@ -26,40 +43,49 @@ func (h *Handlers) HandleJobs(c *gin.Context) { func (h *Handlers) HandleJobRunDetails(c *gin.Context) { userID := c.GetUint("userID") jobID := c.Param("id") - + // Get job history var jobHistory db.JobHistory if err := h.DB.First(&jobHistory, jobID).Error; err != nil { c.String(http.StatusNotFound, "Job not found") return } - + // Get job var job db.Job if err := h.DB.First(&job, jobHistory.JobID).Error; err != nil { c.String(http.StatusNotFound, "Job not found") return } - + // Verify that the user owns this job if job.CreatedBy != userID { c.String(http.StatusForbidden, "You don't have permission to view this job run") return } - + // Get the config var config db.TransferConfig - if err := h.DB.First(&config, job.ConfigID).Error; err != nil { + + // First try to get the specific config used in this job history record + configID := jobHistory.ConfigID + + // If no ConfigID is set in the history, fall back to the job's primary ConfigID + if configID == 0 { + configID = job.ConfigID + } + + if err := h.DB.First(&config, configID).Error; err != nil { c.String(http.StatusNotFound, "Configuration not found") return } - + data := components.JobRunDetailsData{ JobHistory: jobHistory, Job: job, Config: config, } - + components.JobRunDetails(c.Request.Context(), data).Render(c, c.Writer) } @@ -82,7 +108,7 @@ func (h *Handlers) HandleNewJob(c *gin.Context) { func (h *Handlers) HandleEditJob(c *gin.Context) { id := c.Param("id") userID := c.GetUint("userID") - + var job db.Job if err := h.DB.First(&job, id).Error; err != nil { c.Redirect(http.StatusFound, "/jobs") @@ -113,36 +139,83 @@ func (h *Handlers) HandleEditJob(c *gin.Context) { // HandleCreateJob handles the POST /jobs route func (h *Handlers) HandleCreateJob(c *gin.Context) { + userID := c.GetUint("userID") + + // Parse form data var job db.Job if err := c.ShouldBind(&job); err != nil { c.String(http.StatusBadRequest, "Invalid form data") return } - userID := c.GetUint("userID") - job.CreatedBy = userID - - // Verify that the config exists and belongs to the user - var config db.TransferConfig - if err := h.DB.First(&config, job.ConfigID).Error; err != nil { - c.String(http.StatusBadRequest, "Invalid configuration selected") + // Get multiple config IDs from form + configIDs := c.PostFormArray("config_ids[]") + if len(configIDs) == 0 { + c.String(http.StatusBadRequest, "At least one configuration must be selected") return } - // Check if the config belongs to the user - if config.CreatedBy != userID { - // Check if user is admin - isAdmin, exists := c.Get("isAdmin") - if !exists || isAdmin != true { - c.String(http.StatusForbidden, "You do not have permission to use this configuration") + // Process config IDs + var configIDsList []uint + for _, configIDStr := range configIDs { + configID, err := strconv.ParseUint(configIDStr, 10, 32) + if err != nil { + c.String(http.StatusBadRequest, "Invalid configuration ID format") return } + + // Verify that the config exists and belongs to the user + var config db.TransferConfig + if err := h.DB.First(&config, configID).Error; err != nil { + c.String(http.StatusBadRequest, "Invalid configuration selected") + return + } + + // Check if the config belongs to the user + if config.CreatedBy != userID { + // Check if user is admin + isAdmin, exists := c.Get("isAdmin") + if !exists || isAdmin != true { + c.String(http.StatusForbidden, "You do not have permission to use this configuration") + return + } + } + + configIDsList = append(configIDsList, uint(configID)) + } + + // Set the first config ID for backward compatibility + if len(configIDsList) > 0 { + job.ConfigID = configIDsList[0] + + // Verify that the config exists and belongs to the user (using the first config as primary) + var config db.TransferConfig + if err := h.DB.First(&config, job.ConfigID).Error; err != nil { + c.String(http.StatusBadRequest, "Invalid configuration selected") + return + } + + // Check if the config belongs to the user + if config.CreatedBy != userID { + // Check if user is admin + isAdmin, exists := c.Get("isAdmin") + if !exists || isAdmin != true { + c.String(http.StatusForbidden, "You do not have permission to use this configuration") + return + } + } + + // If job name is empty, use the primary config name + if job.Name == "" { + job.Name = config.Name + } } - // If job name is empty, use the config name - if job.Name == "" { - job.Name = config.Name - } + // Set the config IDs list + job.SetConfigIDsList(configIDsList) + + // Set created by user + job.CreatedBy = userID // Clear the Config field to prevent GORM from creating a new config job.Config = db.TransferConfig{} @@ -166,7 +239,7 @@ func (h *Handlers) HandleCreateJob(c *gin.Context) { func (h *Handlers) HandleUpdateJob(c *gin.Context) { id := c.Param("id") userID := c.GetUint("userID") - + var job db.Job if err := h.DB.First(&job, id).Error; err != nil { c.String(http.StatusNotFound, "Job not found") @@ -186,38 +259,66 @@ func (h *Handlers) HandleUpdateJob(c *gin.Context) { // Get the old job values for comparison oldJob := job - // Bind form data to job + // Parse form data if err := c.ShouldBind(&job); err != nil { c.String(http.StatusBadRequest, "Invalid form data") return } - // Verify that the config exists and belongs to the user - var config db.TransferConfig - if err := h.DB.First(&config, job.ConfigID).Error; err != nil { - c.String(http.StatusBadRequest, "Invalid configuration selected") + // Get multiple config IDs from form + configIDs := c.PostFormArray("config_ids[]") + if len(configIDs) == 0 { + c.String(http.StatusBadRequest, "At least one configuration must be selected") return } - // Check if the config belongs to the user - if config.CreatedBy != userID { - // Check if user is admin - isAdmin, exists := c.Get("isAdmin") - if !exists || isAdmin != true { - c.String(http.StatusForbidden, "You do not have permission to use this configuration") + // Process config IDs + var configIDsList []uint + for _, configIDStr := range configIDs { + configID, err := strconv.ParseUint(configIDStr, 10, 32) + if err != nil { + c.String(http.StatusBadRequest, "Invalid configuration ID format") return } + + // Verify that the config exists + var config db.TransferConfig + if err := h.DB.First(&config, configID).Error; err != nil { + c.String(http.StatusBadRequest, "Invalid configuration selected") + return + } + + // Check if the config belongs to the user + if config.CreatedBy != userID { + // Check if user is admin + isAdmin, exists := c.Get("isAdmin") + if !exists || isAdmin != true { + c.String(http.StatusForbidden, "You do not have permission to use this configuration") + return + } + } + + configIDsList = append(configIDsList, uint(configID)) + } + + // Set the first config ID for backward compatibility + if len(configIDsList) > 0 { + job.ConfigID = configIDsList[0] + + // If job name is empty, use the primary config name + var config db.TransferConfig + if err := h.DB.First(&config, job.ConfigID).Error; err == nil && job.Name == "" { + job.Name = config.Name + } } - // If job name is empty, use the config name - if job.Name == "" { - job.Name = config.Name - } + // Set the config IDs list + job.SetConfigIDsList(configIDsList) // Preserve fields that shouldn't be updated job.CreatedBy = oldJob.CreatedBy job.ID = oldJob.ID - + // Clear the Config field to prevent GORM from updating or creating a new config job.Config = db.TransferConfig{} @@ -239,7 +340,7 @@ func (h *Handlers) HandleUpdateJob(c *gin.Context) { func (h *Handlers) HandleDeleteJob(c *gin.Context) { id := c.Param("id") userID := c.GetUint("userID") - + var job db.Job if err := h.DB.First(&job, id).Error; err != nil { c.JSON(http.StatusNotFound, gin.H{"error": "Job not found"}) @@ -272,7 +373,7 @@ func (h *Handlers) HandleDeleteJob(c *gin.Context) { func (h *Handlers) HandleRunJob(c *gin.Context) { id := c.Param("id") userID := c.GetUint("userID") - + var job db.Job if err := h.DB.First(&job, id).Error; err != nil { c.Header("Content-Type", "text/html") @@ -314,8 +415,8 @@ func (h *Handlers) HandleRunJob(c *gin.Context) { // Set custom header with job name for HTMX to use in the toast notification c.Header("HX-Job-Name", jobName) c.Header("Content-Type", "text/html") - + // Return HTML with JavaScript to trigger the notification successScript := fmt.Sprintf("", jobName) c.String(http.StatusOK, successScript) -} \ No newline at end of file +}