fix: some fixes after linter corrections

docs: update readme latest changes
feat: PGS conversion shows progress in worker cli

Author: iaymerich

model/model.go

@@ -234,6 +234,7 @@ type JobRequest struct {
 	SourcePath     string `json:"sourcePath"`
 	ForceCompleted bool   `json:"forceCompleted"`
 	ForceFailed    bool   `json:"forceFailed"`
+	ForceCanceled  bool   `json:"forceCanceled"`
 	ForceAssigned  bool   `json:"forceAssigned"`
 	SourceSize     int64
 	TargetPath     string

readme.md

@@ -26,19 +26,18 @@ You can have as many workers you want, locally or remote in multiple machines.
 
 Replace it by your own values, remember to create manually the DB scheme and PG user.
 ```yaml
-database:
-  host: 192.168.1.55
-  user: db_user
-  password: db_pass
-  scheme: encode
-
+server:
+    database:
+      host: 192.168.1.55
+      user: db_user
+      password: db_pass
+      scheme: encode    
+    scheduler:
+      sourcePath: /mnt/media
+      deleteOnComplete: false # if true, the original file will be deleted after job is completed
+      minFileSize: 100 # minimum file size to be considered for encoding
 web:
   token: my_secret_token # replace it by your own secret
-
-scheduler:
-  sourcePath: /mnt/media
-  deleteOnComplete: false # if true, the original file will be deleted after job is completed
-  minFileSize: 100 # minimum file size to be considered for encoding
 ```
 
 ### Run

server/repository/repository.go

@@ -513,7 +513,7 @@ func (s *SQLRepository) AddNewTaskEvent(ctx context.Context, event *model.TaskEv
 }
 
 func (s *SQLRepository) addNewTaskEvent(ctx context.Context, tx SQLDBOperations, event *model.TaskEvent) error {
-	rows, err := tx.QueryContext(ctx, "select max(job_event_id) from job_events where job_id=$1", event.Id.String())
+	rows, err := tx.QueryContext(ctx, "select COALESCE(max(job_event_id),-1) from job_events where job_id=$1", event.Id.String())
 	if err != nil {
 		return err
 	}

server/scheduler/scheduler.go

@@ -211,7 +211,7 @@ func (r *RuntimeScheduler) scheduleRoutine(ctx context.Context) {
 			r.pathChecksumMap[checksumPath.path] = checksumPath.checksum
 		case <-time.After(r.config.ScheduleTime):
 			if err := r.jobMaintenance(ctx); err != nil {
-				log.Errorf("Errorf on job maintenance %s", err)
+				log.Errorf("Error on job maintenance %s", err)
 			}
 
 		}
@@ -280,65 +280,63 @@ type ScheduleJobRequestResult struct {
 }
 
 func (r *RuntimeScheduler) scheduleJobRequest(ctx context.Context, jobRequest *model.JobRequest) (job *model.Job, err error) {
+	l := log.WithFields(log.Fields{
+		"source_path": jobRequest.SourcePath,
+	})
 	err = r.repo.WithTransaction(ctx, func(ctx context.Context, tx repository.Repository) error {
 		job, err = tx.GetJobByPath(ctx, jobRequest.SourcePath)
 		if err != nil {
 			return err
 		}
 
-		l := log.WithFields(log.Fields{
-			"source_path": jobRequest.SourcePath,
-		})
-
 		var eventsToAdd []*model.TaskEvent
 		if job == nil {
-			newUUID, _ := uuid.NewUUID()
-			job = &model.Job{
-				SourcePath: jobRequest.SourcePath,
-				SourceSize: jobRequest.SourceSize,
-				TargetPath: jobRequest.TargetPath,
-				Id:         newUUID,
-			}
-			l.WithField("job_id", job.Id.String()).Info("Creating new job")
-			err = tx.AddJob(ctx, job)
+			job, err = r.newJob(ctx, tx, jobRequest)
 			if err != nil {
 				return err
 			}
-			startEvent := job.AddEvent(model.NotificationEvent, model.JobNotification, model.QueuedNotificationStatus)
-			eventsToAdd = append(eventsToAdd, startEvent)
+			eventsToAdd = job.Events
 		} else {
 			// If job exist we check if we can retry the job
-			lastEvent := job.Events.GetLatestPerNotificationType(model.JobNotification)
-			status := job.Events.GetStatus()
-			if jobRequest.ForceAssigned && (status == model.AssignedNotificationStatus || status == model.StartedNotificationStatus) {
-				cancelEvent := job.AddEvent(model.NotificationEvent, model.JobNotification, model.CanceledNotificationStatus)
-				eventsToAdd = append(eventsToAdd, cancelEvent)
-
-			}
-			if (jobRequest.ForceCompleted && status == model.CompletedNotificationStatus) ||
-				(jobRequest.ForceFailed && (status == model.FailedNotificationStatus || status == model.CanceledNotificationStatus)) ||
-				(jobRequest.ForceAssigned && (status == model.StartedNotificationStatus || status == model.AssignedNotificationStatus)) {
-				requeueEvent := job.AddEvent(model.NotificationEvent, model.JobNotification, model.QueuedNotificationStatus)
-				eventsToAdd = append(eventsToAdd, requeueEvent)
-			} else if !(jobRequest.ForceAssigned && status == model.QueuedNotificationStatus) {
-				return fmt.Errorf("%s (%s) job is in %s state by %s, can not be rescheduled", job.Id.String(), jobRequest.SourcePath, lastEvent.Status, lastEvent.WorkerName)
+			eventsToAdd, err = r.updateTerminatedJobByRequest(job, jobRequest)
+			if err != nil {
+				return err
 			}
 		}
-		if len(eventsToAdd) > 0 {
-			for _, taskEvent := range eventsToAdd {
-				err = tx.AddNewTaskEvent(ctx, taskEvent)
-				if err != nil {
-					return err
-				}
-				l.WithField("job_id", job.Id.String()).Infof("job is now %s", taskEvent.Status)
+
+		for _, taskEvent := range eventsToAdd {
+			err = tx.AddNewTaskEvent(ctx, taskEvent)
+			if err != nil {
+				return err
 			}
+			l.WithField("job_id", job.Id.String()).Infof("job is now %s", taskEvent.Status)
 		}
 
 		return nil
 	})
 	return job, err
 }
 
+func (r *RuntimeScheduler) newJob(ctx context.Context, tx repository.Repository, jobRequest *model.JobRequest) (*model.Job, error) {
+	l := log.WithFields(log.Fields{
+		"source_path": jobRequest.SourcePath,
+	})
+	newUUID, _ := uuid.NewUUID()
+	job := &model.Job{
+		SourcePath: jobRequest.SourcePath,
+		SourceSize: jobRequest.SourceSize,
+		TargetPath: jobRequest.TargetPath,
+		Id:         newUUID,
+	}
+	l.WithField("job_id", job.Id.String()).Info("Creating new job")
+	err := tx.AddJob(ctx, job)
+	if err != nil {
+		return nil, err
+	}
+	job.AddEvent(model.NotificationEvent, model.JobNotification, model.QueuedNotificationStatus)
+	return job, nil
+}
+
 func (r *RuntimeScheduler) ScheduleJobRequests(ctx context.Context, jobRequest *model.JobRequest) (result *ScheduleJobRequestResult, returnError error) {
 	result = &ScheduleJobRequestResult{}
 	searchJobRequestChan := make(chan *JobRequestResult, 10)
@@ -546,6 +544,27 @@ func (r *RuntimeScheduler) assignedJobMaintenance(ctx context.Context) error {
 	return nil
 }
 
+func (r *RuntimeScheduler) updateTerminatedJobByRequest(job *model.Job, jobRequest *model.JobRequest) ([]*model.TaskEvent, error) {
+	var eventsToAdd []*model.TaskEvent
+	lastEvent := job.Events.GetLatestPerNotificationType(model.JobNotification)
+	status := lastEvent.Status
+
+	switch {
+	case jobRequest.ForceAssigned && (status == model.AssignedNotificationStatus || status == model.StartedNotificationStatus):
+		eventsToAdd = append(eventsToAdd, job.AddEvent(model.NotificationEvent, model.JobNotification, model.CanceledNotificationStatus))
+		eventsToAdd = append(eventsToAdd, job.AddEvent(model.NotificationEvent, model.JobNotification, model.QueuedNotificationStatus))
+
+	case jobRequest.ForceCompleted && status == model.CompletedNotificationStatus,
+		jobRequest.ForceFailed && status == model.FailedNotificationStatus,
+		jobRequest.ForceCanceled && status == model.CanceledNotificationStatus:
+		requeueEvent := job.AddEvent(model.NotificationEvent, model.JobNotification, model.QueuedNotificationStatus)
+		eventsToAdd = append(eventsToAdd, requeueEvent)
+	default:
+		return nil, fmt.Errorf("%s (%s) job is in %s state by %s, can not be rescheduled", job.Id.String(), jobRequest.SourcePath, lastEvent.Status, lastEvent.WorkerName)
+	}
+	return eventsToAdd, nil
+}
+
 func simpleRegex(pattern string, string string) bool {
 	m, err := regexp.MatchString(strings.ToLower(pattern), strings.ToLower(string))
 	if err != nil {

server/web/web.go

@@ -15,7 +15,6 @@ import (
 	"strings"
 	"sync"
 	"sync/atomic"
-	"syscall"
 	"time"
 	"transcoder/model"
 	"transcoder/server/scheduler"
@@ -52,7 +51,7 @@ func (s *Server) requestJob(writer http.ResponseWriter, request *http.Request) {
 	writer.WriteHeader(200)
 	_, err = writer.Write(b)
 	if err != nil {
-		log.Errorf("Errorf writing response %v", err)
+		log.Errorf("Error writing response %v", err)
 	}
 }
 
@@ -101,7 +100,7 @@ func (s *Server) addJobs(writer http.ResponseWriter, request *http.Request) {
 	writer.WriteHeader(200)
 	_, err = writer.Write(b)
 	if err != nil {
-		log.Errorf("Errorf writing response %v", err)
+		log.Errorf("Error writing response %v", err)
 	}
 }
 
@@ -164,24 +163,18 @@ loop:
 		case <-request.Context().Done():
 			return
 		default:
-			readedBytes, err := reader.Read(b)
-			if err != nil {
-				if err == io.EOF {
-					break loop
-				}
-				log.Errorf("Errorreading from reader: %v", err)
-				http.Error(writer, "Error during upload", http.StatusInternalServerError)
-				return
-			}
-
+			readedBytes, readErr := reader.Read(b)
 			readed += uint64(readedBytes)
-			_, err = uploadStream.Write(b[0:readedBytes])
-			if err != nil {
-				if err == io.EOF {
-					break loop
-				}
-				log.Errorf("Error writing: %v", err)
-				http.Error(writer, "Error during upload", http.StatusInternalServerError)
+			_, writeErr := uploadStream.Write(b[0:readedBytes])
+			if writeErr != nil {
+				break loop
+			}
+			if errors.Is(readErr, io.EOF) {
+				break loop
+			}
+			if readErr != nil {
+				log.Errorf("Error reading from download stream: %v", readErr)
+				return
 			}
 		}
 	}
@@ -238,28 +231,16 @@ loop:
 		case <-request.Context().Done():
 			return
 		default:
-			readedBytes, err := downloadStream.Read(b)
-			if err != nil {
-				if err == io.EOF {
-					break loop
-				}
-				log.Errorf("Error reading from download stream: %v", err)
-				// Send an appropriate HTTP error code or terminate gracefully.
-				http.Error(writer, "Error during download", http.StatusInternalServerError)
-				return
+			readedBytes, readErr := downloadStream.Read(b)
+			_, writeErr := writer.Write(b[0:readedBytes])
+			if writeErr != nil {
+				break loop
 			}
-			_, err = writer.Write(b[0:readedBytes])
-			if err != nil {
-				if err == io.EOF {
-					break loop
-				}
-				log.Errorf("Error writing to response: %v", err)
-				// Gracefully terminate if the client cannot receive data.
-				if errors.Is(err, syscall.EPIPE) { // Broken pipe error
-					log.Warn("Client disconnected during download")
-				} else {
-					http.Error(writer, "Error during download", http.StatusInternalServerError)
-				}
+			if errors.Is(readErr, io.EOF) {
+				break loop
+			}
+			if readErr != nil {
+				log.Errorf("Error reading from download stream: %v", readErr)
 				return
 			}
 
@@ -283,7 +264,7 @@ func (s *Server) checksum(writer http.ResponseWriter, request *http.Request) {
 	writer.WriteHeader(200)
 	_, err = writer.Write([]byte(checksum))
 	if err != nil {
-		log.Errorf("Errorf writing response %v", err)
+		log.Errorf("Error writing response %v", err)
 	}
 }
 
@@ -436,7 +417,7 @@ func writeUnauthorized(w http.ResponseWriter) {
 	w.WriteHeader(401)
 	_, err := w.Write([]byte("Unauthorised.\n"))
 	if err != nil {
-		log.Errorf("Errorf writing response %v", err)
+		log.Errorf("Error writing response %v", err)
 	}
 }
 

worker/serverclient/server_client.go

@@ -50,7 +50,7 @@ func (s *ServerClient) PublishEvent(event model.TaskEvent) error {
 	}
 	req, err := s.request("POST", "/api/v1/event", bytes.NewBuffer(b))
 	if err != nil {
-		fmt.Printf("Errorf creating request: %v\n", err)
+		fmt.Printf("Error creating request: %v\n", err)
 		return err
 	}
 
@@ -118,7 +118,7 @@ func (s *ServerClient) GetURL(uri string) string {
 func (s *ServerClient) request(method string, uri string, body io.Reader) (*http.Request, error) {
 	req, err := http.NewRequest(method, s.GetURL(uri), body)
 	if err != nil {
-		fmt.Printf("Errorf creating request: %v\n", err)
+		fmt.Printf("Error creating request: %v\n", err)
 		return nil, err
 	}
 

worker/task/coordinator.go

@@ -59,7 +59,7 @@ func (q *ServerCoordinator) heartbeatRoutine(ctx context.Context) {
 			return
 		case <-time.After(time.Second * 30):
 			if err := q.serverClient.PublishPing(); err != nil {
-				q.printer.Errorf("Errorf Publishing Ping Event: %v", err)
+				q.printer.Errorf("Error Publishing Ping Event: %v", err)
 			}
 		}
 	}
@@ -74,7 +74,7 @@ func (q *ServerCoordinator) requestTaskRoutine(ctx context.Context) {
 			if q.worker.AcceptJobs() {
 				release, requireUpdate, err := q.updater.CheckForUpdate()
 				if err != nil {
-					q.printer.Errorf("Errorf Checking For Update: %v", err)
+					q.printer.Errorf("Error Checking For Update: %v", err)
 					continue
 				}
 				if requireUpdate {
@@ -85,13 +85,13 @@ func (q *ServerCoordinator) requestTaskRoutine(ctx context.Context) {
 				taskJob, err := q.serverClient.RequestJob(q.worker.GetName())
 				if err != nil {
 					if !errors.Is(err, serverclient.NoJobAvailable) {
-						q.printer.Errorf("Errorf Requesting Job: %v", err)
+						q.printer.Errorf("Error Requesting Job: %v", err)
 					}
 					continue
 				}
 
 				if err := q.worker.Execute(taskJob); err != nil {
-					q.printer.Errorf("Errorf Preparing Job Execution: %v", err)
+					q.printer.Errorf("Error Preparing Job Execution: %v", err)
 				}
 			}
 		}