feat(runner): VOD upload handling

api/courses.go

@@ -8,9 +8,8 @@
 	"fmt"
 	"io"
 	"net/http"
-	"net/http/httputil"
-	"net/url"
 	"os"
+	"path/filepath"
 	"regexp"
 	"sort"
 	"strconv"
@@ -28,10 +27,15 @@
 	"github.com/TUM-Dev/gocast/model"
 	"github.com/TUM-Dev/gocast/tools"
 	"github.com/TUM-Dev/gocast/tools/tum"
+
+	"github.com/tum-dev/gocast/runner/protobuf"
 )
 
-func configGinCourseRouter(router *gin.Engine, daoWrapper dao.DaoWrapper) {
-	routes := coursesRoutes{daoWrapper}
+func configGinCourseRouter(router *gin.Engine, daoWrapper dao.DaoWrapper, manager runnerManager) {
+	routes := coursesRoutes{
+		DaoWrapper: daoWrapper,
+		manager:    manager,
+	}
 
 	router.POST("/api/course/activate/:token", routes.activateCourseByToken)
 	router.GET("/api/lecture-halls-by-id", routes.lectureHallsByID)
@@ -105,6 +109,11 @@
 
 type coursesRoutes struct {
 	dao.DaoWrapper
+	manager runnerManager
+}
+
+type runnerManager interface {
+	SendVODJob(ctx context.Context, streamID uint, version protobuf.StreamVersion, recordingDir string) error
 }
 
 const (
@@ -466,44 +475,87 @@
 		return
 	}
 
-	key := uuid.NewV4().String()
-	err = r.UploadKeyDao.CreateUploadKey(key, stream.ID, req.VideoType)
+	// Get the uploaded file
+	file, header, err := c.Request.FormFile("file")
+	if err != nil {
+		_ = c.Error(tools.RequestError{
+			Status:        http.StatusBadRequest,
+			CustomMessage: "can not read uploaded file",
+			Err:           err,
+		})
+		return
+	}
+	defer file.Close()
+
+	// Create directory structure in Ceph: mass/streamID/videoType/
+	streamDir := filepath.Join(tools.Cfg.Paths.Mass, fmt.Sprintf("%d", stream.ID), string(req.VideoType))
+	err = os.MkdirAll(streamDir, os.ModePerm)
+	if err != nil {
+		_ = c.Error(tools.RequestError{
+			Status:        http.StatusInternalServerError,
+			CustomMessage: "can not create storage directory",
+			Err:           err,
+		})
+		return
+	}
+
+	// Save file to Ceph
+	destPath := filepath.Join(streamDir, header.Filename)
+	destFile, err := os.Create(destPath)
 	if err != nil {
 		_ = c.Error(tools.RequestError{
 			Status:        http.StatusInternalServerError,
-			CustomMessage: "can not create upload key",
+			CustomMessage: "can not create destination file",
 			Err:           err,
 		})
 		return
 	}
-	workers := r.WorkerDao.GetAliveWorkers()
-	if len(workers) == 0 {
+	defer destFile.Close()
+
+	_, err = io.Copy(destFile, file)
+	if err != nil {
 		_ = c.Error(tools.RequestError{
 			Status:        http.StatusInternalServerError,
-			CustomMessage: "no workers available",
+			CustomMessage: "can not save file",
 			Err:           err,
 		})
 		return
 	}
-	w := workers[getWorkerWithLeastWorkload(workers)]
-	u, err := url.Parse("http://" + w.Host + ":" + WorkerHTTPPort + "/upload?" + c.Request.URL.Query().Encode() + "&key=" + key)
+
+	logger.Info("File saved to Ceph", "path", destPath, "streamID", stream.ID, "videoType", req.VideoType)
+
+	// Send job to runner to process the VOD
+	err = r.sendVODJobToRunner(c.Request.Context(), stream.ID, req.VideoType, streamDir)
 	if err != nil {
+		logger.Error("Failed to send VOD job to runner", "err", err, "streamID", stream.ID)
 		_ = c.Error(tools.RequestError{
 			Status:        http.StatusInternalServerError,
-			CustomMessage: fmt.Sprintf("parse proxy url: %v", err),
+			CustomMessage: "file uploaded but failed to start processing",
 			Err:           err,
 		})
 		return
 	}
-	p := httputil.NewSingleHostReverseProxy(u)
-	p.Director = func(req *http.Request) {
-		req.URL.Scheme = u.Scheme
-		req.URL.Host = u.Host
-		req.Host = u.Host
-		req.URL.Path = u.Path
-		req.URL.RawQuery = u.RawQuery
+
+	c.JSON(http.StatusOK, gin.H{"message": "file uploaded successfully, processing started"})
+}
+
+// sendVODJobToRunner sends a VOD processing job to an available runner
+func (r coursesRoutes) sendVODJobToRunner(ctx context.Context, streamID uint, videoType model.VideoType, recordingDir string) error {
+	// Convert VideoType to StreamVersion
+	var version protobuf.StreamVersion
+	switch videoType {
+	case model.VideoTypeCombined:
+		version = protobuf.StreamVersion_STREAM_VERSION_COMBINED
+	case model.VideoTypePresentation:
+		version = protobuf.StreamVersion_STREAM_VERSION_PRESENTATION
+	case model.VideoTypeCamera:
+		version = protobuf.StreamVersion_STREAM_VERSION_CAMERA
+	default:
+		return fmt.Errorf("unsupported video type: %s", videoType)
 	}
-	p.ServeHTTP(c.Writer, c.Request)
+
+	// Send job to runner via manager
+	return r.manager.SendVODJob(ctx, streamID, version, recordingDir)
 }
 
 // updateSourceSettings updates the CameraPresets of a course

api/router.go

@@ -38,7 +38,7 @@ func ConfigGinRouter(
 
 	configGinStreamRestRouter(router, daoWrapper)
 	configGinUsersRouter(router, daoWrapper)
-	configGinCourseRouter(router, daoWrapper)
+	configGinCourseRouter(router, daoWrapper, manager)
 	configGinDownloadRouter(router, daoWrapper)
 	configGinDownloadICSRouter(router, daoWrapper)
 	configGinLectureHallApiRouter(router, daoWrapper, camService, tools.Cfg.Paths.Static)

dao/upload_key.go

@@ -8,6 +8,7 @@ import (
 
 //go:generate go tool mockgen -source=upload_key.go -destination ../mock_dao/upload_key.go
 
+// deprecated: delete with worker
 type UploadKeyDao interface {
 	GetUploadKey(key string) (model.UploadKey, error)
 	CreateUploadKey(key string, stream uint, videoType model.VideoType) error

model/upload-key.go

@@ -14,6 +14,7 @@ func (v VideoType) Valid() bool {
 	return v == VideoTypeCombined || v == VideoTypePresentation || v == VideoTypeCamera
 }
 
+// deprecated: delete with worker
 // UploadKey represents a key that is created when a user uploads a file,
 // sent to the worker with the upload request and back to TUM-Live to authenticate the request.
 type UploadKey struct {

pkg/runner_manager/manager.go

@@ -237,6 +237,27 @@ func (m *Manager) getClient(ctx context.Context) (protobuf.RunnerServiceClient,
 	return protobuf.NewRunnerServiceClient(conn), nil
 }
 
+// SendVODJob sends a VOD processing job to an available runner
+func (m *Manager) SendVODJob(ctx context.Context, streamID uint, version protobuf.StreamVersion, recordingDir string) error {
+	client, err := m.getClient(ctx)
+	if err != nil {
+		return fmt.Errorf("get runner client: %w", err)
+	}
+
+	streamIDUint64 := uint64(streamID)
+	_, err = client.HandleVOD(ctx, &protobuf.HandleVODRequest{
+		StreamId: &streamIDUint64,
+		Version:  &version,
+		Filepath: &recordingDir,
+	})
+	if err != nil {
+		return fmt.Errorf("send HandleVOD request: %w", err)
+	}
+
+	m.logger.Info("VOD job sent to runner", "streamID", streamID, "version", version)
+	return nil
+}
+
 func (m *Manager) streamStarted(ctx context.Context, req *protobuf.StreamStartNotification) error {
 	// This is usually called in bursts, which introduces a chance for race conditions,
 	// where a stream is fetched and overwrites the url that the other requests added.

runner/handlers.go

@@ -45,3 +45,23 @@ func (r *Runner) RequestStreamEnd(_ context.Context, req *protobuf.StreamEndRequ
 	}
 	return nil, status.Errorf(codes.NotFound, "stream not found")
 }
+
+func (r *Runner) HandleVOD(_ context.Context, req *protobuf.HandleVODRequest) (*protobuf.HandleVODResponse, error) {
+	data := map[string]any{
+		"streamID":      req.GetStreamId(),
+		"streamVersion": req.GetVersion().String(),
+		"recordingDir":  req.GetFilepath(),
+	}
+	r.log.Info("HandleVOD data constructed", "data", data)
+	a := []actions.Action{
+		actions.CheckCodec,
+		actions.MkVOD,
+		actions.CheckVoD,
+		actions.MkThumb,
+	}
+
+	jID := r.RunAction(a, data, r.log.With("stream_id", req.GetStreamId(), "stream_version", req.GetVersion(), "input", req.GetFilepath()))
+	r.log.Info("job added", "ID", jID)
+
+	return &protobuf.HandleVODResponse{JobId: ptr.Take(jID)}, nil
+}

runner/pkg/actions/checkcodec.go

@@ -0,0 +1,58 @@
+package actions
+
+import (
+	"context"
+	"fmt"
+	"log/slog"
+	"path"
+
+	"github.com/tum-dev/gocast/runner/pkg/ffmpeg"
+	"github.com/tum-dev/gocast/runner/pkg/metrics"
+	"github.com/tum-dev/gocast/runner/protobuf"
+)
+
+// CheckCodec probes the recording file and checks if it needs re-encoding.
+// Sets "needsReencode" to true if the video is not h264 or exceeds 3Mbit/s bitrate,
+// or if audio is not AAC.
+func CheckCodec(ctx context.Context, logger *slog.Logger, _ chan *protobuf.Notification, d map[string]any, _ *metrics.Broker) error {
+	recordingDir, ok := d["recordingDir"].(string)
+	if !ok {
+		return AbortingError(fmt.Errorf("no recordingDir in context"))
+	}
+	recording := path.Join(recordingDir, "playlist.m3u8")
+
+	probe, err := ffmpeg.Probe(ctx, recording)
+	if err != nil {
+		return AbortingError(fmt.Errorf("ffprobe failed: %w", err))
+	}
+
+	needsReencode := false
+	for _, stream := range probe.Streams() {
+		if stream.CodecType == "video" {
+			if stream.CodecName != "h264" {
+				needsReencode = true
+				logger.Info("video codec requires re-encoding", "codec", stream.CodecName)
+				break
+			}
+
+			if stream.BitRate > 3000000 { // 3 Mbit/s in bits/s
+				needsReencode = true
+				logger.Info("video bitrate exceeds 3Mbit/s", "bitrate", stream.BitRate)
+				break
+			}
+		}
+
+		if stream.CodecType == "audio" {
+			if stream.CodecName != "aac" {
+				needsReencode = true
+				logger.Info("audio codec requires re-encoding", "codec", stream.CodecName)
+				break
+			}
+		}
+	}
+
+	d["needsReencode"] = needsReencode
+	logger.Info("codec check completed", "needsReencode", needsReencode)
+
+	return nil
+}

runner/pkg/actions/mkthumb.go

@@ -47,13 +47,12 @@ func MkThumb(_ context.Context, logger *slog.Logger, notify chan *protobuf.Notif
 	return nil
 }
 
-// createVideoThumbnail creates a thumbnail from the given video file
-
 const (
 	thumbnailWidth         = 720 // Width of the generated thumbnail in pixels
 	jpegCompressionQuality = 90  // JPEG compression quality (0-100)
 )
 
+// createVideoThumbnail creates a thumbnail from the given video file
 func createVideoThumbnail(source string) ([]byte, error) {
 	g, err := thumbgen.New(source, thumbnailWidth, 1, "", thumbgen.WithJpegCompression(jpegCompressionQuality))
 	if err != nil {

runner/pkg/actions/mkvod.go

@@ -31,9 +31,21 @@ func MkVOD(ctx context.Context, logger *slog.Logger, notify chan *protobuf.Notif
 	if !ok {
 		return AbortingError(fmt.Errorf("no stream version in context"))
 	}
-	recordingDir, ok := d["recordingDir"].(string)
-	if !ok {
-		return AbortingError(fmt.Errorf("no recordingDir in context"))
+	var recording string
+	if rec, ok := d["recording"]; ok {
+		if recStr, ok := rec.(string); ok {
+			recording = recStr
+		} else {
+			return AbortingError(fmt.Errorf("recording value is not a string"))
+		}
+	} else if dir, ok := d["recordingDir"]; ok {
+		if dirStr, ok := dir.(string); ok {
+			recording = path.Join(dirStr, "playlist.m3u8")
+		} else {
+			return AbortingError(fmt.Errorf("recordingDir value is not a string"))
+		}
+	} else {
+		return AbortingError(fmt.Errorf("no recording or recordingDir in context"))
 	}
 
 	metrics.ConvertingProgresses.With(metrics.With().Stream(streamID).L()).Inc()
@@ -45,7 +57,28 @@ func MkVOD(ctx context.Context, logger *slog.Logger, notify chan *protobuf.Notif
 		return AbortingError(fmt.Errorf("create VOD directory: %w", err))
 	}
 
-	err = convertStream(ctx, logger, streamID, path.Join(recordingDir, "playlist.m3u8"), vodDir, "playlist.m3u8")
+	// Check if re-encoding is needed
+	var reencode bool
+	if needsReencode, ok := d["needsReencode"]; ok {
+		if reencodeVal, ok := needsReencode.(bool); ok {
+			reencode = reencodeVal
+		} else {
+			return AbortingError(fmt.Errorf("needsReencode is not a bool"))
+		}
+	}
+
+	var videoCodec, audioCodec string
+	if reencode {
+		logger.Info("re-encoding required, transcoding video")
+		videoCodec = "libx264"
+		audioCodec = "aac"
+	} else {
+		logger.Info("no re-encoding needed, using copy codec")
+		videoCodec = "copy"
+		audioCodec = "copy"
+	}
+
+	err = convertStream(ctx, logger, streamID, recording, vodDir, "playlist.m3u8", videoCodec, audioCodec)
 	if err != nil {
 		return AbortingError(fmt.Errorf("convert stream: %w", err))
 	}
@@ -67,9 +100,18 @@ func MkVOD(ctx context.Context, logger *slog.Logger, notify chan *protobuf.Notif
 	return nil
 }
 
-func convertStream(ctx context.Context, logger *slog.Logger, streamID uint64, streamPath, vodDir string, playlistName string) error {
+func convertStream(ctx context.Context, logger *slog.Logger, streamID uint64, streamPath, vodDir string, playlistName string, videoCodec string, audioCodec string) error {
 	input := "-i " + streamPath
-	options := "-c copy -f hls -hls_time 20 -hls_playlist_type vod -hls_flags append_list -hls_segment_filename " + path.Join(vodDir, "%05d.ts") + " " + path.Join(vodDir, playlistName)
+
+	// Build codec options based on parameters
+	codecOpts := fmt.Sprintf("-c:v %s -c:a %s", videoCodec, audioCodec)
+
+	// Add bitrate limit if re-encoding video
+	if videoCodec != "copy" {
+		codecOpts += " -b:v 3M"
+	}
+
+	options := codecOpts + " -f hls -hls_time 20 -hls_playlist_type vod -hls_flags append_list -hls_segment_filename " + path.Join(vodDir, "%05d.ts") + " " + path.Join(vodDir, playlistName)
 
 	args := strings.Split(input, " ")
 	args = append(args, strings.Split(options, " ")...)