Serialize transcode jobs

handlers.go

@@ -12,7 +12,6 @@ import (
 	"strings"
 	"time"
 
-	"github.com/google/uuid"
 	"github.com/labstack/echo/v4"
 	"golang.org/x/crypto/bcrypt"
 )
@@ -104,7 +103,10 @@ type Meta struct {
 }
 
 func getMeta(url string) (Meta, error) {
-	cmd := exec.Command("yt-dlp", "--simulate", "--print", "%(title)s.%(ext)s", url)
+	ytdlp := "yt-dlp"
+	args := []string{"--simulate", "--print", "%(title)s.%(ext)s", url}
+	fmt.Println(ytdlp, strings.Join(args, " "))
+	cmd := exec.Command(ytdlp, args...)
 
 	var stdout bytes.Buffer
 	cmd.Stdout = &stdout
@@ -288,92 +290,71 @@ func getVideoMeta(path string) (VideoMeta, error) {
 	}, nil
 }
 
-func videoToAudio(audioID uint, bitrate uint, videoFilepath string) {
-	audioFilename := uuid.Must(uuid.NewV7()).String()
-	audioFilename = fmt.Sprintf("%s.mp3", audioFilename)
-	audioFilepath := filepath.Join(getDataDir(), audioFilename)
-	audioDir := filepath.Dir(audioFilepath)
-	fmt.Println("Create", audioDir)
-	err := os.MkdirAll(audioDir, 0700)
-	if err != nil {
-		fmt.Println("Error: couldn't create", audioDir)
-		db.Model(&Audio{}).Where("id = ?", audioID).Update("status", "failed")
-		return
-	}
-	ffmpeg := "ffmpeg"
-	ffmpegArgs := []string{"-i", videoFilepath, "-vn", "-acodec",
-		"mp3", "-b:a",
-		fmt.Sprintf("%dk", bitrate),
-		audioFilepath}
-	fmt.Println(ffmpeg, strings.Join(ffmpegArgs, " "))
-	cmd := exec.Command(ffmpeg, ffmpegArgs...)
-	err = cmd.Run()
-	if err != nil {
-		fmt.Println("Error: convert to audio file", videoFilepath, "->", audioFilepath)
+func processOriginal(originalID uint, videoFilename string, origMeta Meta) {
+
+	videoFilepath := filepath.Join(getDataDir(), videoFilename)
+	_, err := os.Stat(videoFilepath)
+	if os.IsNotExist(err) {
+		fmt.Println("Skipping non-existant file for processOriginal")
 		return
 	}
 
-	fileSize, err := getSize(audioFilepath)
+	// create video entry for original
+	video := Video{
+		OriginalID: originalID,
+		Filename:   videoFilename,
+		Source:     "original",
+		Type:       origMeta.ext,
+	}
+	fmt.Println("create Video", video)
+	if err := db.Create(&video).Error; err != nil {
+		fmt.Println(err)
+	}
+
+	videoMeta, err := getVideoMeta(videoFilepath)
+	if err != nil {
+		fmt.Println(err)
+	} else {
+		fmt.Println(videoMeta)
+		db.Model(&Video{}).Where("id = ?", video.ID).Update("fps", videoMeta.fps)
+		db.Model(&Video{}).Where("id = ?", video.ID).Update("width", videoMeta.width)
+		db.Model(&Video{}).Where("id = ?", video.ID).Update("height", videoMeta.height)
+	}
+
+	videoSize, err := getSize(videoFilepath)
 	if err == nil {
-		db.Model(&Audio{}).Where("id = ?", audioID).Update("size", humanSize(fileSize))
+		db.Model(&Video{}).Where("id = ?", video.ID).Update("size", humanSize(videoSize))
 	}
 
-	db.Model(&Audio{}).Where("id = ?", audioID).Update("filename", audioFilename)
-	db.Model(&Audio{}).Where("id = ?", audioID).Update("status", "completed")
-}
-
-func videoToVideo(videoID uint, height uint, videoFilepath string) {
-	dstFilename := uuid.Must(uuid.NewV7()).String()
-	dstFilename = fmt.Sprintf("%s.mp4", dstFilename)
-	dstFilepath := filepath.Join(getDataDir(), dstFilename)
-	dstDir := filepath.Dir(dstFilepath)
-	fmt.Println("Create", dstDir)
-	err := os.MkdirAll(dstDir, 0700)
-	if err != nil {
-		fmt.Println("Error: couldn't create", dstDir)
-		db.Model(&Video{}).Where("id = ?", videoID).Update("status", "failed")
-		return
-	}
-	var audioBitrate uint = 160
-	if height <= 144 {
-		audioBitrate = 64
-	} else if height <= 240 {
-		audioBitrate = 96
-	} else if height < 720 {
-		audioBitrate = 128
+	// create audio transcodes
+	for _, bitrate := range []uint{64, 96, 128, 160, 192} {
+		t := Transcode{
+			SrcID:      video.ID,
+			OriginalID: originalID,
+			SrcKind:    "video",
+			DstKind:    "audio",
+			Rate:       bitrate,
+			TimeSubmit: time.Now(),
+			Status:     "pending",
+		}
+		db.Create(&t)
 	}
 
-	ffmpeg := "ffmpeg"
-	ffmpegArgs := []string{"-i", videoFilepath,
-		"-vf", fmt.Sprintf("scale=-2:%d", height), "-c:v", "libx264",
-		"-crf", "23", "-preset", "veryfast", "-c:a", "aac", "-b:a", fmt.Sprintf("%dk", audioBitrate),
-		dstFilepath}
-	fmt.Println(ffmpeg, strings.Join(ffmpegArgs, " "))
-	cmd := exec.Command(ffmpeg, ffmpegArgs...)
-	var stdout bytes.Buffer
-	var stderr bytes.Buffer
-	cmd.Stdout, cmd.Stderr = &stdout, &stderr
-	err = cmd.Run()
-	if err != nil {
-		fmt.Println("Error: convert to video file", videoFilepath, "->", dstFilepath, stdout.String(), stderr.String())
-		return
+	// create video transcodes
+	for _, targetHeight := range []uint{144, 240, 360, 480, 720, 1080} {
+		if targetHeight <= videoMeta.height {
+			t := Transcode{
+				SrcID:      video.ID,
+				OriginalID: originalID,
+				SrcKind:    "video",
+				DstKind:    "video",
+				Height:     targetHeight,
+				TimeSubmit: time.Now(),
+				Status:     "pending",
+			}
+			db.Create(&t)
+		}
 	}
-	db.Model(&Video{}).Where("id = ?", videoID).Update("filename", dstFilename)
-
-	fileSize, err := getSize(dstFilepath)
-	if err == nil {
-		db.Model(&Video{}).Where("id = ?", videoID).Update("size", humanSize(fileSize))
-	}
-
-	meta, err := getVideoMeta(dstFilepath)
-	fmt.Println("meta for", dstFilepath, meta)
-	if err == nil {
-		db.Model(&Video{}).Where("id = ?", videoID).Update("width", meta.width)
-		db.Model(&Video{}).Where("id = ?", videoID).Update("height", meta.height)
-		db.Model(&Video{}).Where("id = ?", videoID).Update("fps", meta.fps)
-	}
-
-	db.Model(&Video{}).Where("id = ?", videoID).Update("status", "completed")
 }
 
 func startDownload(originalID uint, videoURL string) {
@@ -403,59 +384,7 @@ func startDownload(originalID uint, videoURL string) {
 	}
 	db.Model(&Original{}).Where("id = ?", originalID).Update("status", "completed")
 
-	// create video entry for original
-	video := Video{
-		OriginalID: originalID,
-		Filename:   videoFilename,
-		Source:     "original",
-		Type:       origMeta.ext,
-		Status:     "completed",
-	}
-	if err := db.Create(&video).Error; err != nil {
-		fmt.Println(err)
-	}
-
-	videoMeta, err := getVideoMeta(videoFilepath)
-	if err != nil {
-		fmt.Println(err)
-	} else {
-		fmt.Println(videoMeta)
-		db.Model(&Video{}).Where("id = ?", video.ID).Update("fps", videoMeta.fps)
-		db.Model(&Video{}).Where("id = ?", video.ID).Update("width", videoMeta.width)
-		db.Model(&Video{}).Where("id = ?", video.ID).Update("height", videoMeta.height)
-	}
-
-	videoSize, err := getSize(videoFilepath)
-	if err == nil {
-		db.Model(&Video{}).Where("id = ?", video.ID).Update("size", humanSize(videoSize))
-	}
-
-	// create audio transcodes
-	for _, bitrate := range []uint{64, 96, 128, 160, 192} {
-		audio := Audio{
-			OriginalID: originalID,
-			Rate:       fmt.Sprintf("%dk", bitrate),
-			Type:       "mp3",
-			Status:     "pending",
-		}
-		db.Create(&audio)
-		go videoToAudio(audio.ID, bitrate, videoFilepath)
-	}
-
-	// create video transcodes
-	for _, targetHeight := range []uint{144, 240, 360, 480, 720, 1080} {
-		if targetHeight <= videoMeta.height {
-			newVideo := Video{
-				OriginalID: originalID,
-				Type:       "mp4",
-				Status:     "pending",
-				Source:     "transcode",
-			}
-			db.Create(&newVideo)
-			videoToVideo(newVideo.ID, targetHeight, videoFilepath)
-		}
-	}
-
+	processOriginal(originalID, videoFilename, origMeta)
 }
 
 func videosHandler(c echo.Context) error {
@@ -525,8 +454,6 @@ func videoCancelHandler(c echo.Context) error {
 	}
 
 	// Cancel the download (this is a simplified version, you might need to implement a more robust cancellation mechanism)
-	video.Status = "cancelled"
-	db.Save(&video)
 
 	return c.Redirect(http.StatusSeeOther, "/videos")
 }
@@ -552,7 +479,10 @@ func videoDeleteHandler(c echo.Context) error {
 		return c.Redirect(http.StatusSeeOther, "/videos")
 	}
 
-	// TODO: delete all files
+	fmt.Println("Delete Transcode entries for Original", id)
+	db.Delete(&Transcode{}, "original_id = ?", id)
+
+	// delete videos
 	var videos []Video
 	db.Where("original_id = ?", id).Find(&videos)
 	for _, video := range videos {
@@ -563,8 +493,9 @@ func videoDeleteHandler(c echo.Context) error {
 			fmt.Println("error removing", path, err)
 		}
 	}
-	db.Delete(videos)
+	db.Delete(&Video{}, "original_id = ?", id)
 
+	// delete audios
 	var audios []Audio
 	db.Where("original_id = ?", id).Find(&audios)
 	for _, audio := range audios {
@@ -575,9 +506,9 @@ func videoDeleteHandler(c echo.Context) error {
 			fmt.Println("error removing", path, err)
 		}
 	}
-	db.Delete(audios)
+	db.Delete(&Video{}, "original_id = ?", id)
 
-	// Delete from database
+	// Delete original
 	db.Delete(&orig)
 
 	return c.Redirect(http.StatusSeeOther, "/videos")

main.go

@@ -51,7 +51,7 @@ func main() {
 	}
 
 	// Migrate the schema
-	db.AutoMigrate(&Original{}, &Video{}, &Audio{}, &User{}, &TempURL{})
+	db.AutoMigrate(&Original{}, &Video{}, &Audio{}, &User{}, &TempURL{}, &Transcode{})
 	go PeriodicCleanup()
 
 	// create a user
@@ -108,6 +108,9 @@ func main() {
 		Secure:   false, // needed for session to work over http
 	}
 
+	// start the transcode worker
+	go transcodeWorker()
+
 	// Start server
 	e.Logger.Fatal(e.Start(":8080"))
 }

middleware.go

@@ -22,7 +22,6 @@ func authMiddleware(next echo.HandlerFunc) echo.HandlerFunc {
 			// return c.String(http.StatusForbidden, "not logged in")
 			return c.Redirect(http.StatusSeeOther, "/login")
 		}
-		fmt.Println("set user_id", userID, "in context")
 		c.Set("user_id", userID)
 		return next(c)
 	}

models.go

@@ -22,7 +22,9 @@ type Original struct {
 
 type Video struct {
 	gorm.Model
-	OriginalID uint // Original.ID
+	OriginalID uint   // Original.ID
+	Source     string // "original", "transcode"
+	Filename   string
 	Width      uint
 	Height     uint
 	FPS        float64
@@ -30,9 +32,25 @@ type Video struct {
 	Size       string
 	Type       string
 	Codec      string
-	Filename   string
-	Status     string // "pending", "completed"
-	Source     string // "original", "ffmpeg"
+}
+
+type Transcode struct {
+	gorm.Model
+	Status     string // "pending", "running", "failed"
+	SrcID      uint   // Video.ID of the source file
+	OriginalID uint   // Original.ID
+	SrcKind    string // "video", "audio"
+	DstKind    string // "video", "audio"
+	TimeSubmit time.Time
+	TimeStart  time.Time
+
+	// video fields
+	Height uint // target height
+	Width  uint // target width
+	FPS    uint // target FPS
+
+	// audio & video fields
+	Rate uint
 }
 
 type Audio struct {
@@ -138,7 +156,8 @@ func CreateTempURL(filePath string) (TempURL, error) {
 }
 
 func cleanupExpiredURLs() {
-	result := db.Where("expires_at < ?", time.Now()).Delete(&TempURL{})
+	fmt.Println("cleanupExpiredURLs...")
+	result := db.Unscoped().Where("expires_at < ?", time.Now()).Delete(&TempURL{})
 	if result.Error != nil {
 		fmt.Printf("Error cleaning up expired URLs: %v\n", result.Error)
 	} else {
@@ -153,9 +172,10 @@ func vacuumDatabase() {
 }
 
 func PeriodicCleanup() {
-	ticker := time.NewTicker(12 * time.Hour)
+	cleanupExpiredURLs()
+	vacuumDatabase()
+	ticker := time.NewTicker(1 * time.Hour)
 	for range ticker.C {
-		fmt.Println("PeriodicCleanup...")
 		cleanupExpiredURLs()
 		vacuumDatabase()
 	}

templates/video.html

@@ -46,10 +46,9 @@
     <h1>{{.original.Title}}</h1>
 
     {{range .videos}}
-    {{if eq .Status "completed"}}
     <h2>{{.Source}} {{.Width}} x {{.Height}} @ {{.FPS}}</h2>
     <div class="video-container">
-        <video controls playsinline preload="metadata">
+        <video controls playsinline preload="none">
             <source src="/temp/{{.Token}}" type="video/mp4">
             Your browser does not support the video tag.
         </video>
@@ -57,16 +56,12 @@
     <div class="video-download">
         <a href="/data/{{.Filename}}" download>Download ({{.Size}})</a>
     </div>
-    {{else}}
-    <h2>Video {{.Source}} {{.Status}}</h2>
-    {{end}}
     {{end}}
 
     {{range .audios}}
-    {{if eq .Status "completed"}}
     <h2>{{.Rate}}</h2>
     <div class="audio-container">
-        <audio controls playsinline preload="metadata">
+        <audio controls playsinline preload="none">
             <source src="/temp/{{.Token}}">
             Your browser does not support the audio tag.
         </audio>
@@ -74,9 +69,6 @@
     <div class="audio-download">
         <a href="/data/{{.Filename}}" download>Download ({{.Size}})</a>
     </div>
-    {{else}}
-    <h2>Audio {{.Status}}</h2>
-    {{end}}
     {{end}}
 
 </body>

workers.go

@@ -0,0 +1,192 @@
+package main
+
+import (
+	"bytes"
+	"fmt"
+	"os"
+	"os/exec"
+	"path/filepath"
+	"strings"
+	"time"
+
+	"github.com/google/uuid"
+	"gorm.io/gorm"
+)
+
+func ensureDirFor(path string) error {
+	dir := filepath.Dir(path)
+	fmt.Println("Create", dir)
+	return os.MkdirAll(dir, 0700)
+}
+
+func videoToVideo(transID uint, height uint, srcFilepath string) {
+
+	// determine destination path
+	dstFilename := uuid.Must(uuid.NewV7()).String()
+	dstFilename = fmt.Sprintf("%s.mp4", dstFilename)
+	dstFilepath := filepath.Join(getDataDir(), dstFilename)
+
+	err := ensureDirFor(dstFilepath)
+	if err != nil {
+		fmt.Println("Error: couldn't create dir for ", dstFilepath, err)
+		db.Model(&Transcode{}).Where("id = ?", transID).Update("status", "failed")
+		return
+	}
+
+	// FIXME: ignoring any requested audio bitrate
+
+	// determine audio bitrate
+	var audioBitrate uint = 160
+	if height <= 144 {
+		audioBitrate = 64
+	} else if height <= 240 {
+		audioBitrate = 96
+	} else if height < 720 {
+		audioBitrate = 128
+	}
+
+	// start ffmpeg
+	ffmpeg := "ffmpeg"
+	ffmpegArgs := []string{"-i", srcFilepath,
+		"-vf", fmt.Sprintf("scale=-2:%d", height), "-c:v", "libx264",
+		"-crf", "23", "-preset", "veryfast", "-c:a", "aac", "-b:a", fmt.Sprintf("%dk", audioBitrate),
+		dstFilepath}
+	fmt.Println(ffmpeg, strings.Join(ffmpegArgs, " "))
+	cmd := exec.Command(ffmpeg, ffmpegArgs...)
+	var stdout bytes.Buffer
+	var stderr bytes.Buffer
+	cmd.Stdout, cmd.Stderr = &stdout, &stderr
+	db.Model(&Transcode{}).Where("id = ?", transID).Update("status", "running")
+	err = cmd.Run()
+	if err != nil {
+		fmt.Println("Error: convert to video file", srcFilepath, "->", dstFilepath, stdout.String(), stderr.String())
+		return
+	}
+
+	// look up original
+	var trans Transcode
+	db.First(&trans, transID)
+	var orig Original
+	db.First(&orig, "id = ?", trans.OriginalID)
+
+	// create video record
+	video := Video{OriginalID: orig.ID, Source: "transcode", Filename: dstFilename}
+
+	fileSize, err := getSize(dstFilepath)
+	if err == nil {
+		video.Size = humanSize(fileSize)
+	}
+
+	meta, err := getVideoMeta(dstFilepath)
+	fmt.Println("meta for", dstFilepath, meta)
+	if err == nil {
+		video.Width = meta.width
+		video.Height = meta.height
+		video.FPS = meta.fps
+	}
+
+	db.Create(&video)
+
+	// complete transcode
+	db.Delete(&trans)
+}
+
+func videoToAudio(transID uint, bitrate uint, videoFilepath string) {
+
+	// determine destination path
+	audioFilename := uuid.Must(uuid.NewV7()).String()
+	audioFilename = fmt.Sprintf("%s.mp3", audioFilename)
+	audioFilepath := filepath.Join(getDataDir(), audioFilename)
+
+	// ensure destination directory
+	err := ensureDirFor(audioFilepath)
+	if err != nil {
+		fmt.Println("Error: couldn't create dir for ", audioFilepath, err)
+		db.Model(&Transcode{}).Where("id = ?", transID).Update("status", "failed")
+		return
+	}
+
+	ffmpeg := "ffmpeg"
+	ffmpegArgs := []string{"-i", videoFilepath, "-vn", "-acodec",
+		"mp3", "-b:a",
+		fmt.Sprintf("%dk", bitrate),
+		audioFilepath}
+	fmt.Println(ffmpeg, strings.Join(ffmpegArgs, " "))
+	cmd := exec.Command(ffmpeg, ffmpegArgs...)
+	db.Model(&Transcode{}).Where("id = ?", transID).Update("status", "running")
+	err = cmd.Run()
+	if err != nil {
+		fmt.Println("Error: convert to audio file", videoFilepath, "->", audioFilepath)
+		db.Model(&Transcode{}).Where("id = ?", transID).Update("status", "failed")
+		return
+	}
+
+	// look up original
+	var trans Transcode
+	db.First(&trans, "id = ?", transID)
+	var orig Original
+	db.First(&orig, "id = ?", trans.OriginalID)
+
+	// create audio record
+	audio := Audio{OriginalID: orig.ID, Filename: audioFilename}
+
+	fileSize, err := getSize(audioFilepath)
+	if err == nil {
+		audio.Size = humanSize(fileSize)
+	}
+
+	db.Create(&audio)
+
+	// complete transcode
+	db.Delete(&trans)
+}
+
+func transcodePending() {
+	fmt.Println("transcodePending...")
+
+	// any running jobs here got stuck or dead in the midde, so reset them
+	db.Model(&Transcode{}).Where("status = ?", "running").Update("status", "pending")
+
+	// loop until no more pending jobs
+	for {
+		var trans Transcode
+		err := db.First(&trans, "status = ?", "pending").Error
+		if err == gorm.ErrRecordNotFound {
+			fmt.Println("no pending transcode jobs")
+			break // no more pending jobs
+		}
+
+		if trans.SrcKind == "video" {
+
+			var srcVideo Video
+			err = db.First(&srcVideo, "id = ?", trans.SrcID).Error
+			if err != nil {
+				fmt.Println("no such source video for video Transcode", trans)
+				db.Delete(&trans)
+				continue
+			}
+			srcFilepath := filepath.Join(getDataDir(), srcVideo.Filename)
+
+			if trans.DstKind == "video" {
+				videoToVideo(trans.ID, trans.Height, srcFilepath)
+			} else if trans.DstKind == "audio" {
+				videoToAudio(trans.ID, trans.Rate, srcFilepath)
+			} else {
+				fmt.Println("unexpected src/dst kinds for Transcode", trans)
+				db.Delete(&trans)
+			}
+		} else {
+			fmt.Println("unexpected src kind for Transcode", trans)
+			db.Delete(&trans)
+		}
+	}
+
+}
+
+func transcodeWorker() {
+	transcodePending()
+	ticker := time.NewTicker(10 * time.Second)
+	for range ticker.C {
+		transcodePending()
+	}
+}