Fix uploader (#12)

.golangci.yaml

@@ -61,7 +61,7 @@ linters-settings:
   dupl:
     threshold: 150
   funlen:
-    lines: 160
+    lines: 180
     statements: 100
   goconst:
     min-len: 2

internal/pkg/api/v1/transcoding/handler.go

@@ -60,7 +60,11 @@ func (t *trascodeHandler) ServeHTTP(w http.ResponseWriter, r *http.Request) {
 		return
 	}
 
-	t.scheduler.Schedule(&task)
+	if err := t.scheduler.Schedule(&task); err != nil {
+		w.WriteHeader(http.StatusTooManyRequests)
+		return
+	}
+
 	w.WriteHeader(http.StatusOK)
 }
 

internal/pkg/mediaconvert/command.go

@@ -32,7 +32,7 @@ import (
 // Options represents configuration for the ffmpeg command
 type Options struct {
 	Input         string
-	OuputDir      string
+	OutputDir     string
 	ScalingLevels []string
 	Level         string
 	Threads       int
@@ -56,7 +56,7 @@ func newFfmpegCommand(ctx context.Context, in io.Reader, args []string) (*exec.C
 	return result, nil
 }
 
-func buildCommonComamnd(opts *Options) []string {
+func buildCommonCommand(opts *Options) []string {
 	return []string{
 		"-threads", fmt.Sprint(opts.Threads),
 		"-i", opts.Input,
@@ -65,24 +65,24 @@ func buildCommonComamnd(opts *Options) []string {
 
 // BuildAudioCommand returns flags for getting the audio from the input
 func BuildAudioCommand(opts *Options) []string {
-	var commonPart = buildCommonComamnd(opts)
+	var commonPart = buildCommonCommand(opts)
 
 	return append(commonPart,
 		"-vn", "-acodec",
-		"copy", filepath.Join(opts.OuputDir, opts.UploadID),
+		"copy", filepath.Join(opts.OutputDir, opts.UploadID),
 	)
 }
 
 // BuildRawVideoCommand returns an extremely lightweight ffmpeg command for converting raw video without extra cost.
 func BuildRawVideoCommand(opts *Options) []string {
-	return append(buildCommonComamnd(opts),
+	return append(buildCommonCommand(opts),
 		"-c:a", "copy", // Copy audio stream
 		"-c:v", "copy", // Copy video stream
 		"-hls_time", "5",
 		"-hls_flags", "split_by_time",
 		"-hls_list_size", "0",
-		"-hls_segment_filename", filepath.Join(opts.OuputDir, opts.UploadID, fmt.Sprintf("%s_%s_%s.ts", opts.UploadID, "%03d", opts.Level)),
-		filepath.Join(opts.OuputDir, opts.UploadID, fmt.Sprintf("%s_%s_master.m3u8", opts.UploadID, opts.Level)))
+		"-hls_segment_filename", filepath.Join(opts.OutputDir, opts.UploadID, fmt.Sprintf("%s_%s_%s.ts", opts.UploadID, "%03d", opts.Level)),
+		filepath.Join(opts.OutputDir, opts.UploadID, fmt.Sprintf("%s_%s_master.m3u8", opts.UploadID, opts.Level)))
 }
 
 // BuildThumbnailCommand creates a command that creates a thumbnail for the input video
@@ -90,13 +90,13 @@ func BuildThumbnailCommand(opts *Options) []string {
 	return append([]string{},
 		"-i", opts.Input,
 		"-vframes", "1",
-		filepath.Join(opts.OuputDir, opts.UploadID, opts.UploadID+".jpg"),
+		filepath.Join(opts.OutputDir, opts.UploadID, opts.UploadID+".jpg"),
 	)
 }
 
 // BuildScalingVideoCommand returns flags for ffmpeg for video scaling
 func BuildScalingVideoCommand(opts *Options) []string {
-	var result = buildCommonComamnd(opts)
+	var result = buildCommonCommand(opts)
 
 	for _, level := range opts.ScalingLevels {
 		result = append(result,
@@ -109,8 +109,8 @@ func BuildScalingVideoCommand(opts *Options) []string {
 			"-hls_time", "5",
 			"-hls_flags", "split_by_time",
 			"-hls_list_size", "0",
-			"-hls_segment_filename", filepath.Join(opts.OuputDir, opts.UploadID, fmt.Sprintf("%s_%s_%s.ts", opts.UploadID, "%03d", level)),
-			filepath.Join(opts.OuputDir, opts.UploadID, fmt.Sprintf("%s_%s_master.m3u8", opts.UploadID, level)))
+			"-hls_segment_filename", filepath.Join(opts.OutputDir, opts.UploadID, fmt.Sprintf("%s_%s_%s.ts", opts.UploadID, "%03d", level)),
+			filepath.Join(opts.OutputDir, opts.UploadID, fmt.Sprintf("%s_%s_master.m3u8", opts.UploadID, level)))
 	}
 
 	return result

internal/pkg/mediaconvert/command_test.go

@@ -24,7 +24,7 @@ import (
 
 func Test_BuildVideoCommand_Scaling(t *testing.T) {
 	var scaleCommand = mediaconvert.BuildScalingVideoCommand(&mediaconvert.Options{
-		OuputDir:      "test",
+		OutputDir:     "test",
 		Input:         "pipe:0",
 		UploadID:      "1",
 		Threads:       4,
@@ -38,11 +38,11 @@ func Test_BuildVideoCommand_Scaling(t *testing.T) {
 
 func Test_BuildVideoCommand_Raw(t *testing.T) {
 	var rawCommand = mediaconvert.BuildRawVideoCommand(&mediaconvert.Options{
-		OuputDir: "test",
-		Input:    "pipe:0",
-		UploadID: "1",
-		Threads:  4,
-		Level:    resconv.Level("651:490"),
+		OutputDir: "test",
+		Input:     "pipe:0",
+		UploadID:  "1",
+		Threads:   4,
+		Level:     resconv.Level("651:490"),
 	})
 
 	const expected = `"-threads 4 -i pipe:0 -c:a copy -c:v copy -hls_time 5 -hls_flags split_by_time -hls_list_size 0 -hls_segment_filename test/1/1_%03d_480p.ts test/1/1_480p_master.m3u8`

internal/pkg/mediaconvert/coordinator.go

@@ -97,7 +97,7 @@ func (s *StreamCoordinator) NewUpload(ctx context.Context, info handler.FileInfo
 
 	var commandOptions = Options{
 		Input:         "pipe:0",
-		OuputDir:      s.conf.OutputDir,
+		OutputDir:     s.conf.OutputDir,
 		Threads:       s.conf.MaxThreadCount,
 		UploadID:      info.ID,
 		Level:         level,

internal/pkg/mediaconvert/scheduler.go

@@ -62,15 +62,15 @@ type Scheduler struct {
 }
 
 // Schedule schedules a task to transcode
-func (p *Scheduler) Schedule(t *Task) {
+func (p *Scheduler) Schedule(t *Task) error {
 	t.ID = uuid.NewString()
 	t.Status = "planned"
 
 	select {
 	case p.taskCh <- t:
-		p.logger.Sugar().Debugf("task %v is scheduled", t)
+		return nil
 	default:
-		p.logger.Error("task channel is full")
+		return fmt.Errorf("task queue is full")
 	}
 }
 
@@ -127,6 +127,12 @@ func (p *Scheduler) processTask(ctx context.Context, task *Task) {
 		return
 	}
 
+	defer func() {
+		if err = os.RemoveAll(destinationFolder); err != nil {
+			logger.Error("failed to cleanup temporary folder", zap.Error(err))
+		}
+	}()
+
 	logger.Debug("phase 3: get the remote file")
 
 	remoteStorage, err := storage.NewStorageByURL(ctx, p.cfg.Endpoint(), p.cfg.EndpointURL.Scheme, tokenString, task.Workspace)
@@ -176,7 +182,7 @@ func (p *Scheduler) processTask(ctx context.Context, task *Task) {
 	var level = resconv.Level(res)
 	var opts = Options{
 		Input:         sourceFilePath,
-		OuputDir:      p.cfg.OutputDir,
+		OutputDir:     p.cfg.OutputDir,
 		Level:         level,
 		ScalingLevels: append(resconv.SubLevels(res), level),
 		UploadID:      task.ID,

internal/pkg/mediaconvert/stream.go

@@ -115,7 +115,7 @@ func (s *StreamCoordinator) AsConcatableUpload(upload handler.Upload) handler.Co
 func (w *Stream) start(ctx context.Context, options *Options) error {
 	defer w.logger.Debug("start done")
 	w.reader = w.writer.Transpile()
-	if err := manifest.GenerateHLSPlaylist(append(options.ScalingLevels, options.Level), options.OuputDir, options.UploadID); err != nil {
+	if err := manifest.GenerateHLSPlaylist(append(options.ScalingLevels, options.Level), options.OutputDir, options.UploadID); err != nil {
 		return err
 	}
 	w.commandGroup.Add(1)

internal/pkg/uploader/uploader.go

@@ -91,20 +91,10 @@ func New(ctx context.Context, s storage.Storage, opts Options) Uploader {
 
 	res.uploadCtx, res.uploadCancel = context.WithCancel(context.Background())
 
-	_ = os.MkdirAll(opts.Dir, os.ModePerm)
-	res.workerWaitGroup.Add(1)
-
-	go func() {
-		defer res.workerWaitGroup.Done()
-		initFiles, _ := os.ReadDir(opts.Dir)
-		for _, f := range initFiles {
-			var filePath = filepath.Join(opts.Dir, f.Name())
-			if filePath == opts.SourceFile {
-				continue
-			}
-			res.filesCh <- filePath
-		}
-	}()
+	err := os.MkdirAll(opts.Dir, os.ModePerm)
+	if err != nil {
+		res.logger.Error("can not create upload directory", zap.Error(err), zap.String("dir", opts.Dir))
+	}
 
 	return res
 }
@@ -117,6 +107,34 @@ func (u *uploaderImpl) Cancel() {
 	u.stop(true)
 }
 
+func (u *uploaderImpl) scanInitialFiles() {
+	u.workerWaitGroup.Add(1)
+
+	go func() {
+		defer u.workerWaitGroup.Done()
+
+		initFiles, err := os.ReadDir(u.options.Dir)
+		if err != nil {
+			u.logger.Error("failed to read initial files", zap.Error(err), zap.String("dir", u.options.Dir))
+			return
+		}
+
+		for _, f := range initFiles {
+			if f.IsDir() {
+				continue
+			}
+
+			var filePath = filepath.Join(u.options.Dir, f.Name())
+			if filePath == u.options.SourceFile {
+				continue
+			}
+			u.filesCh <- filePath
+		}
+
+		u.logger.Info("initial file scan complete", zap.String("dir", u.options.Dir), zap.Int("count", len(initFiles)))
+	}()
+}
+
 func (u *uploaderImpl) stop(rollback bool) {
 	close(u.watcherStopCh)
 	<-u.watcherDoneCh
@@ -147,8 +165,14 @@ func (u *uploaderImpl) stop(rollback bool) {
 }
 
 func (u *uploaderImpl) Start() {
+	watcherReady := make(chan struct{})
+
 	u.startWorkers()
-	u.startWatch()
+	go u.startWatch(watcherReady)
+
+	<-watcherReady
+
+	u.scanInitialFiles()
 }
 
 func (u *uploaderImpl) startWorkers() {
@@ -230,13 +254,11 @@ func (u *uploaderImpl) uploadAndDelete(f string) {
 		if err != nil {
 			logger.Error("attempt failed", zap.Error(err))
 		} else {
-			if !u.shouldDeleteOnStop(f) {
-				_ = os.Remove(f)
-				logger.Debug("removed file locally")
-			}
+			// Mark the file as uploaded
 			u.sentFiles.Store(f, struct{}{})
 			logger.Debug("file uploaded")
 
+			// Update the file's parent if SourceFile is set
 			if u.options.SourceFile != "" {
 				err = u.storage.SetParent(ctx, f, u.options.SourceFile)
 				if err != nil {
@@ -244,6 +266,15 @@ func (u *uploaderImpl) uploadAndDelete(f string) {
 				}
 			}
 
+			// Delete the file locally if it should be deleted
+			if !u.shouldDeleteOnStop(f) {
+				if err := os.Remove(f); err != nil {
+					logger.Error("failed to remove file locally", zap.Error(err), zap.String("file", f))
+				} else {
+					logger.Debug("removed file locally")
+				}
+			}
+
 			break
 		}
 
@@ -251,28 +282,30 @@ func (u *uploaderImpl) uploadAndDelete(f string) {
 	}
 }
 
-func (u *uploaderImpl) startWatch() {
+func (u *uploaderImpl) startWatch(ready chan<- struct{}) {
+	defer close(u.watcherDoneCh)
+
 	var logger = u.logger.With(zap.String("func", "startWatch"))
 	var watcher, err = inotify.NewWatcher()
 
 	if err != nil {
 		logger.Error("can not start file watcher", zap.Error(err))
-		return
-	}
-
-	if err := watcher.AddWatch(u.options.Dir, inotifyCloseWrite); err != nil {
-		logger.Error("can not start watching for close write", zap.Error(err))
-		return
-	}
-	if err := watcher.AddWatch(u.options.Dir, inotifyMovedTo); err != nil {
-		logger.Error("can not start watching for moved to", zap.Error(err))
+		close(ready)
 		return
 	}
 	defer func() {
-		_ = watcher.Close()
-		close(u.watcherDoneCh)
+		if err := watcher.Close(); err != nil {
+			logger.Error("can not close watcher", zap.Error(err))
+		}
 	}()
 
+	if err := watcher.AddWatch(u.options.Dir, inotifyCloseWrite|inotifyMovedTo); err != nil {
+		logger.Error("can not start watching", zap.Error(err))
+		close(ready)
+		return
+	}
+
+	close(ready)
 	logger.Debug("watching for file updates")
 	defer logger.Debug("done")