Add migrating to ffmpeg 6

app/ffmigrate/main.go

@@ -4,6 +4,9 @@ import (
 	"fmt"
 	"os"
 	"regexp"
+	"strconv"
+	"strings"
+	"time"
 
 	cfgstore "github.com/datarhei/core/v16/config/store"
 	cfgvars "github.com/datarhei/core/v16/config/vars"
@@ -11,6 +14,7 @@ import (
 	"github.com/datarhei/core/v16/io/file"
 	"github.com/datarhei/core/v16/io/fs"
 	"github.com/datarhei/core/v16/log"
+	"github.com/datarhei/core/v16/restream/app"
 	"github.com/datarhei/core/v16/restream/store"
 
 	"github.com/Masterminds/semver/v3"
@@ -18,10 +22,7 @@ import (
 )
 
 func main() {
-	logger := log.New("Migration").WithOutput(log.NewConsoleWriter(os.Stderr, log.Linfo, true)).WithFields(log.Fields{
-		"from": "ffmpeg4",
-		"to":   "ffmpeg5",
-	})
+	logger := log.New("Migration").WithOutput(log.NewConsoleWriter(os.Stderr, log.Linfo, true))
 
 	configfile := cfgstore.Location(os.Getenv("CORE_CONFIGFILE"))
 
@@ -34,6 +35,7 @@ func main() {
 	}
 
 	if err := doMigration(logger, configstore); err != nil {
+		logger.Error().WithError(err).Log("Migration failed")
 		os.Exit(1)
 	}
 }
@@ -82,17 +84,6 @@ func doMigration(logger log.Logger, configstore cfgstore.Store) error {
 		return fmt.Errorf("parsing FFmpeg version failed: %w", err)
 	}
 
-	// The current FFmpeg version is 4. Nothing to do.
-	if version.Major() == 4 {
-		return nil
-	}
-
-	if version.Major() != 5 {
-		err := fmt.Errorf("unknown FFmpeg version found: %d", version.Major())
-		logger.Error().WithError(err).Log("Unsupported FFmpeg version found")
-		return fmt.Errorf("unsupported FFmpeg version found: %w", err)
-	}
-
 	// Check if there's a DB file
 	dbFilepath := cfg.DB.Dir + "/db.json"
 
@@ -102,29 +93,20 @@ func doMigration(logger log.Logger, configstore cfgstore.Store) error {
 		return nil
 	}
 
-	// Check if we already have a backup
-	backupFilepath := cfg.DB.Dir + "/db_ff4.json"
-
-	if _, err = os.Stat(backupFilepath); err == nil {
-		// Yes, we have a backup. The migration already happened
-		logger.Info().WithField("backup", backupFilepath).Log("Migration already done")
-		return nil
-	}
-
-	// Create a backup
-	if err := file.Copy(dbFilepath, backupFilepath); err != nil {
-		logger.Error().WithError(err).Log("Creating backup file failed")
-		return fmt.Errorf("creating backup file failed: %w", err)
-	}
-
-	logger.Info().WithField("backup", backupFilepath).Log("Backup created")
-
 	// Load the existing DB
+	diskfs, err := fs.NewDiskFilesystem(fs.DiskConfig{})
+	if err != nil {
+		logger.Error().WithError(err).Log("Accessing disk filesystem failed")
+		return fmt.Errorf("accessing disk filesystem failed: %w", err)
+	}
+
 	datastore, err := store.NewJSON(store.JSONConfig{
-		Filepath: cfg.DB.Dir + "/db.json",
+		Filesystem: diskfs,
+		Filepath:   cfg.DB.Dir + "/db.json",
 	})
 	if err != nil {
-		return err
+		logger.Error().WithField("db", dbFilepath).WithError(err).Log("Creating JSON store failed")
+		return fmt.Errorf("creating JSON store failed: %w", err)
 	}
 
 	data, err := datastore.Load()
@@ -133,17 +115,102 @@ func doMigration(logger log.Logger, configstore cfgstore.Store) error {
 		return fmt.Errorf("loading database failed: %w", err)
 	}
 
+	// Migrate processes
 	logger.Info().Log("Migrating processes ...")
 
-	// Migrate the processes to version 5
-	// Only this happens:
-	// - for RTSP inputs, replace -stimeout with -timeout
+	migrated := false
 
-	reRTSP := regexp.MustCompile(`^rtsps?://`)
 	for id, p := range data.Process {
-		logger.Info().WithField("processid", p.ID).Log("")
+		ok, err := migrateProcessConfig(logger.WithField("processid", p.ID), p.Config, version.String())
+		if err != nil {
+			logger.Info().WithField("processid", p.ID).WithError(err).Log("Migrating process failed")
+			return fmt.Errorf("migrating process failed: %w", err)
+		}
 
-		for index, input := range p.Config.Input {
+		data.Process[id] = p
+
+		if ok {
+			migrated = true
+		}
+	}
+
+	logger.Info().Log("Migrating processes done")
+
+	if migrated {
+		// Create backup if something has been changed.
+		backupFilepath := cfg.DB.Dir + "/db." + strconv.FormatInt(time.Now().UnixMilli(), 10) + ".json"
+
+		if _, err = os.Stat(backupFilepath); err == nil {
+			// Yes, we have a backup. The migration already happened
+			logger.Info().WithField("backup", backupFilepath).Log("Migration already done")
+			return nil
+		}
+
+		// Create a backup
+		if err := file.Copy(dbFilepath, backupFilepath); err != nil {
+			logger.Error().WithError(err).Log("Creating backup file failed")
+			return fmt.Errorf("creating backup file failed: %w", err)
+		}
+
+		logger.Info().WithField("backup", backupFilepath).Log("Backup created")
+
+		// Store the modified DB
+		if err := datastore.Store(data); err != nil {
+			logger.Error().WithError(err).Log("Storing database failed")
+			return fmt.Errorf("storing database failed: %w", err)
+		}
+	}
+
+	logger.Info().Log("Completed")
+
+	return nil
+}
+
+func migrateProcessConfig(logger log.Logger, config *app.Config, version string) (bool, error) {
+	migrated := false
+
+	vtarget, err := semver.NewVersion(version)
+	if err != nil {
+		logger.Error().WithError(err).Log("Parsing target FFmpeg version failed")
+		return false, fmt.Errorf("parsing target FFmpeg version failed: %w", err)
+	}
+
+	targetmajor := vtarget.Major()
+	currentmajor := uint64(4)
+
+	if len(config.FFVersion) != 0 {
+		vcurrent, err := semver.NewVersion(strings.TrimPrefix(config.FFVersion, "^"))
+		if err != nil {
+			logger.Error().WithError(err).Log("Parsing current FFmpeg version failed")
+			return false, fmt.Errorf("parsing current FFmpeg version failed: %w", err)
+		}
+
+		currentmajor = vcurrent.Major()
+	}
+
+	if currentmajor < 4 {
+		err := fmt.Errorf("unknown FFmpeg version found: %d", currentmajor)
+		logger.Error().WithError(err).Log("Unsupported FFmpeg version")
+		return false, fmt.Errorf("unsupported FFmpeg version: %w", err)
+	}
+
+	if targetmajor > 6 {
+		err := fmt.Errorf("unknown FFmpeg version found: %d", targetmajor)
+		logger.Error().WithError(err).Log("Unsupported FFmpeg version")
+		return false, fmt.Errorf("unsupported FFmpeg version: %w", err)
+	}
+
+	if currentmajor != targetmajor {
+		migrated = true
+	}
+
+	if currentmajor == 4 && targetmajor > 4 {
+		// Migration from version 4 to version 5
+		// Only this happens:
+		// - for RTSP inputs, replace -stimeout with -timeout
+		reRTSP := regexp.MustCompile(`^rtsps?://`)
+
+		for index, input := range config.Input {
 			if !reRTSP.MatchString(input.Address) {
 				continue
 			}
@@ -156,21 +223,27 @@ func doMigration(logger log.Logger, configstore cfgstore.Store) error {
 				input.Options[i] = "-timeout"
 			}
 
-			p.Config.Input[index] = input
+			config.Input[index] = input
 		}
-		p.Config.FFVersion = version.String()
-		data.Process[id] = p
+
+		currentmajor = 5
 	}
 
-	logger.Info().Log("Migrating processes done")
+	if currentmajor == 5 && targetmajor > 5 {
+		// Migration from version 5 to version 6
+		// Nothing happens
 
-	// Store the modified DB
-	if err := datastore.Store(data); err != nil {
-		logger.Error().WithError(err).Log("Storing database failed")
-		return fmt.Errorf("storing database failed: %w", err)
+		currentmajor = 6
 	}
 
-	logger.Info().Log("Completed")
+	if migrated {
+		logger.Info().WithFields(log.Fields{
+			"from": config.FFVersion,
+			"to":   "^" + version,
+		}).Log("Migrated")
+	}
 
-	return nil
+	config.FFVersion = "^" + version
+
+	return migrated, nil
 }