fix: enable conversion for mp4 files only

Signed-off-by: Alexander Onnikov <Alexander.Onnikov@xored.com>

internal/pkg/mediaconvert/scheduler.go

@@ -121,20 +121,37 @@ func (p *Scheduler) processTask(ctx context.Context, task *Task) {
 	var destinationFolder = filepath.Join(p.cfg.OutputDir, task.ID)
 	var _, filename = filepath.Split(task.Source)
 	var sourceFilePath = filepath.Join(destinationFolder, filename)
-	_ = os.MkdirAll(destinationFolder, os.ModePerm)
+	err = os.MkdirAll(destinationFolder, os.ModePerm)
+	if err != nil {
+		logger.Error("can not create temporary folder", zap.Error(err))
+		return
+	}
 
 	logger.Debug("phase 3: get the remote file")
 
 	remoteStorage, err := storage.NewStorageByURL(ctx, p.cfg.Endpoint(), p.cfg.EndpointURL.Scheme, tokenString, task.Workspace)
 
 	if err != nil {
-		logger.Error("can not create storage by url", zap.Error(err))
+		logger.Error("can not create storage by url", zap.Error(err), zap.String("url", p.cfg.EndpointURL.String()))
+		_ = os.RemoveAll(destinationFolder)
+		return
+	}
+
+	stat, err := remoteStorage.StatFile(ctx, task.Source)
+	if err != nil {
+		logger.Error("can not stat a file", zap.Error(err), zap.String("filepath", task.Source))
+		_ = os.RemoveAll(destinationFolder)
+		return
+	}
+
+	if !IsSupportedMediaType(stat.Type) {
+		logger.Info("unsupported media type", zap.String("type", stat.Type))
 		_ = os.RemoveAll(destinationFolder)
 		return
 	}
 
 	if err = remoteStorage.GetFile(ctx, task.Source, sourceFilePath); err != nil {
-		logger.Error("can not download a file", zap.Error(err))
+		logger.Error("can not download a file", zap.Error(err), zap.String("filepath", task.Source))
 		_ = os.RemoveAll(destinationFolder)
 		// TODO: reschedule
 		return
@@ -177,7 +194,12 @@ func (p *Scheduler) processTask(ctx context.Context, task *Task) {
 		SourceFile:  sourceFilePath,
 	})
 
-	_ = manifest.GenerateHLSPlaylist(opts.ScalingLevels, p.cfg.OutputDir, opts.UploadID)
+	err = manifest.GenerateHLSPlaylist(opts.ScalingLevels, p.cfg.OutputDir, opts.UploadID)
+	if err != nil {
+		logger.Error("can not generate hls playlist", zap.String("out", p.cfg.OutputDir), zap.String("uploadID", opts.UploadID))
+		_ = os.RemoveAll(destinationFolder)
+		return
+	}
 
 	go uploader.Start()
 
@@ -240,3 +262,15 @@ func (p *Scheduler) processTask(ctx context.Context, task *Task) {
 		}
 	}
 }
+
+func IsSupportedMediaType(mediaType string) bool {
+	// Explicitly disable conversion for video/mp2t and video/x-mpegurl
+	switch mediaType {
+	case "video/mp2t", "video/x-mpegurl":
+		return false
+	case "video/mp4":
+		return true
+	default:
+		return false
+	}
+}

internal/pkg/storage/datalake.go

@@ -271,5 +271,37 @@ func (d *DatalakeStorage) GetFile(ctx context.Context, filename, destination str
 	return nil
 }
 
+func (d *DatalakeStorage) StatFile(ctx context.Context, filename string) (*BlobInfo, error) {
+	var logger = d.logger.With(zap.String("head", d.workspace), zap.String("fileName", filename))
+	logger.Debug("start")
+
+	var objectKey = getObjectKey(filename)
+
+	req := fasthttp.AcquireRequest()
+	defer fasthttp.ReleaseRequest(req)
+	req.SetRequestURI(d.baseURL + "/blob/" + d.workspace + "/" + objectKey)
+	req.Header.SetMethod(fasthttp.MethodHead)
+
+	resp := fasthttp.AcquireResponse()
+	defer fasthttp.ReleaseResponse(resp)
+
+	if err := d.client.Do(req, resp); err != nil {
+		return nil, err
+	}
+
+	// Check the response status code
+	if resp.StatusCode() != fasthttp.StatusOK {
+		var err = fmt.Errorf("unexpected status code: %d", resp.StatusCode())
+		logger.Debug("bad status code", zap.Error(err))
+		return nil, err
+	}
+
+	var info BlobInfo
+	info.Size = int64(resp.Header.ContentLength())
+	info.Type = string(resp.Header.ContentType())
+	info.ETag = string(resp.Header.Peek("ETag"))
+	return &info, nil
+}
+
 var _ Storage = (*DatalakeStorage)(nil)
 var _ MetaProvider = (*DatalakeStorage)(nil)

internal/pkg/storage/s3.go

@@ -164,3 +164,24 @@ func (u *S3Storage) GetFile(ctx context.Context, filename, dest string) error {
 
 	return nil
 }
+
+func (u *S3Storage) StatFile(ctx context.Context, filename string) (*BlobInfo, error) {
+	var logger = u.logger.With(zap.String("head", u.bucketName), zap.String("fileName", filename))
+
+	var head, err = u.client.HeadObject(ctx, &s3.HeadObjectInput{
+		Bucket: &u.bucketName,
+		Key:    &filename,
+	})
+
+	if err != nil {
+		logger.Error("failed to head object", zap.Error(err))
+		return nil, err
+	}
+
+	var info BlobInfo
+	info.Size = *head.ContentLength
+	info.Type = *head.ContentType
+	info.ETag = *head.ETag
+
+	return &info, nil
+}

internal/pkg/storage/storage.go

@@ -30,11 +30,18 @@ type MetaProvider interface {
 	PatchMeta(ctx context.Context, filename string, value *Metadata) error
 }
 
+type BlobInfo struct {
+	Size int64
+	Type string
+	ETag string
+}
+
 // Storage represents file-based storage
 type Storage interface {
 	PutFile(ctx context.Context, fileName string) error
 	DeleteFile(ctx context.Context, fileName string) error
 	GetFile(ctx context.Context, fileName, destination string) error
+	StatFile(ctx context.Context, fileName string) (*BlobInfo, error)
 }
 
 // NewStorageByURL creates a new storage based on the type from the url scheme, for example "datalake://my-datalake-endpoint"