core/app/ffmigrate/main.go

package main

import (
	"fmt"
	"os"
	"regexp"
	"strconv"
	"strings"
	"time"

	cfgstore "github.com/datarhei/core/v16/config/store"
	cfgvars "github.com/datarhei/core/v16/config/vars"
	"github.com/datarhei/core/v16/ffmpeg"
	"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"
	_ "github.com/joho/godotenv/autoload"
)

func main() {
	logger := log.New("Migration").WithOutput(log.NewConsoleWriter(os.Stderr, log.Linfo, true))

	configfile := cfgstore.Location(os.Getenv("CORE_CONFIGFILE"))

	diskfs, _ := fs.NewDiskFilesystem(fs.DiskConfig{})

	configstore, err := cfgstore.NewJSON(diskfs, configfile, nil)
	if err != nil {
		logger.Error().WithError(err).Log("Loading configuration failed")
		os.Exit(1)
	}

	if err := doMigration(logger, diskfs, configstore); err != nil {
		logger.Error().WithError(err).Log("Migration failed")
		os.Exit(1)
	}
}

func doMigration(logger log.Logger, fs fs.Filesystem, configstore cfgstore.Store) error {
	if logger == nil {
		logger = log.New("")
	}

	cfg := configstore.Get()

	// Merging the persisted config with the environment variables
	cfg.Merge()

	cfg.Validate(false)
	if cfg.HasErrors() {
		logger.Error().Log("The configuration contains errors")
		messages := []string{}
		cfg.Messages(func(level string, v cfgvars.Variable, message string) {
			if level == "error" {
				logger.Error().WithFields(log.Fields{
					"variable":    v.Name,
					"value":       v.Value,
					"env":         v.EnvName,
					"description": v.Description,
				}).Log(message)

				messages = append(messages, v.Name+": "+message)
			}
		})

		return fmt.Errorf("the configuration contains errors: %v", messages)
	}

	ff, err := ffmpeg.New(ffmpeg.Config{
		Binary: cfg.FFmpeg.Binary,
	})
	if err != nil {
		logger.Error().WithError(err).Log("Loading FFmpeg binary failed")
		return fmt.Errorf("loading FFmpeg binary failed: %w", err)
	}

	version, err := semver.NewVersion(ff.Skills().FFmpeg.Version)
	if err != nil {
		logger.Error().WithError(err).Log("Parsing FFmpeg version failed")
		return fmt.Errorf("parsing FFmpeg version failed: %w", err)
	}

	// Check if there's a DB file
	dbFilepath := cfg.DB.Dir + "/db.json"

	if _, err = os.Stat(dbFilepath); err != nil {
		// There's no DB to backup
		logger.Info().WithField("db", dbFilepath).Log("Database not found. Migration not required")
		return nil
	}

	// Load the existing DB
	datastore, err := store.NewJSON(store.JSONConfig{
		Filesystem: fs,
		Filepath:   cfg.DB.Dir + "/db.json",
	})
	if err != nil {
		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()
	if err != nil {
		logger.Error().WithError(err).Log("Loading database failed")
		return fmt.Errorf("loading database failed: %w", err)
	}

	// Migrate processes
	logger.Info().Log("Migrating processes ...")

	migrated := false

	for id, p := range data.Process {
		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)
		}

		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
			}

			for i, o := range input.Options {
				if o != "-stimeout" {
					continue
				}

				input.Options[i] = "-timeout"
			}

			config.Input[index] = input
		}

		currentmajor = 5
	}

	if currentmajor == 5 && targetmajor > 5 {
		// Migration from version 5 to version 6
		// Nothing happens

		currentmajor = 6
	}

	if migrated {
		logger.Info().WithFields(log.Fields{
			"from": config.FFVersion,
			"to":   "^" + version,
		}).Log("Migrated")
	}

	config.FFVersion = "^" + version

	return migrated, nil
}