deploy-felhom-compose/controller/internal/storage/migrate.go

package storage

import (
	"bufio"
	"fmt"
	"io"
	"os"
	"os/exec"
	"path/filepath"
	"strings"
	"sync"
	"time"
)

// MigrateRequest holds parameters for migrating app data.
type MigrateRequest struct {
	StackName      string   // e.g., "immich"
	DisplayName    string   // e.g., "Immich"
	CurrentHDDPath string   // e.g., "/mnt/hdd_placeholder"
	TargetPath     string   // e.g., "/mnt/hdd_1"
	HDDMounts      []string // host-side paths to rsync (e.g., ["/mnt/hdd_placeholder/storage/immich"])
}

// MigrateProgress tracks migration state.
type MigrateProgress struct {
	Step           string // "stopping","copying","updating","starting","done","error","rolling_back"
	Message        string
	BytesCopied    int64
	BytesTotal     int64
	Percent        int
	Error          string
	ElapsedSeconds int
}

// StopFunc stops an app's containers. Returns error if stop fails.
type StopFunc func(stackName string) error

// StartFunc starts an app's containers. Returns error if start fails.
type StartFunc func(stackName string) error

// UpdateHDDPathFunc updates the HDD_PATH in app.yaml. Returns error on failure.
type UpdateHDDPathFunc func(stackName, newPath string) error

// MigrateAppData moves app data from current to target storage path.
// stopFn and startFn are called to stop/start the app containers.
// updateFn is called to update the app's HDD_PATH configuration.
func MigrateAppData(
	req MigrateRequest,
	stopFn StopFunc,
	startFn StartFunc,
	updateFn UpdateHDDPathFunc,
	progress chan<- MigrateProgress,
) error {
	start := time.Now()

	send := func(step, msg string, pct int, bytesCopied, bytesTotal int64) {
		progress <- MigrateProgress{
			Step:           step,
			Message:        msg,
			Percent:        pct,
			BytesCopied:    bytesCopied,
			BytesTotal:     bytesTotal,
			ElapsedSeconds: int(time.Since(start).Seconds()),
		}
	}

	fail := func(step, msg string, err error) error {
		errStr := ""
		if err != nil {
			errStr = err.Error()
		}
		progress <- MigrateProgress{
			Step:           "error",
			Message:        msg,
			Error:          errStr,
			ElapsedSeconds: int(time.Since(start).Seconds()),
		}
		return fmt.Errorf("%s: %w", msg, err)
	}

	// --- Step 1: Validate ---
	if req.CurrentHDDPath == "" {
		return fail("validating", "A jelenlegi tárhely nem megadott", fmt.Errorf("empty current HDD path"))
	}
	if req.TargetPath == "" {
		return fail("validating", "A cél tárhely nem megadott", fmt.Errorf("empty target path"))
	}
	if req.CurrentHDDPath == req.TargetPath {
		return fail("validating", "A forrás és a cél tárhely azonos", fmt.Errorf("source equals target"))
	}
	if _, err := os.Stat(req.TargetPath); err != nil {
		return fail("validating", "A cél tárhely nem létezik: "+req.TargetPath, err)
	}
	if len(req.HDDMounts) == 0 {
		return fail("validating", "Nincsenek HDD csatlakozások az alkalmazáshoz", fmt.Errorf("no HDD mounts"))
	}

	// Estimate total size
	var totalBytes int64
	for _, m := range req.HDDMounts {
		if info, err := os.Stat(m); err == nil && info.IsDir() {
			totalBytes += dirSize(m)
		}
	}

	// Check free space on target
	freeBytes := getFreeBytes(req.TargetPath)
	if freeBytes > 0 && totalBytes > 0 && int64(float64(totalBytes)*1.05) > freeBytes {
		return fail("validating", fmt.Sprintf(
			"Nincs elég szabad hely a céltárolón: szükséges ~%s, szabad %s",
			bytesHuman(totalBytes), bytesHuman(freeBytes),
		), fmt.Errorf("insufficient disk space"))
	}

	send("stopping", "Alkalmazás leállítása...", 5, 0, totalBytes)

	// --- Step 2: Stop app ---
	if err := stopFn(req.StackName); err != nil {
		return fail("stopping", "Alkalmazás leállítása sikertelen", err)
	}

	send("stopping", "Alkalmazás leállítva", 10, 0, totalBytes)

	// --- Step 3: rsync ---
	var bytesCopied int64
	for i, srcPath := range req.HDDMounts {
		// Determine destination path: replace CurrentHDDPath prefix with TargetPath.
		// H13: Require trailing separator to prevent /mnt/hdd matching /mnt/hdd_backup/data.
		if srcPath != req.CurrentHDDPath && !strings.HasPrefix(srcPath, req.CurrentHDDPath+"/") {
			continue
		}
		relPath := strings.TrimPrefix(srcPath, req.CurrentHDDPath)
		dstPath := filepath.Join(req.TargetPath, relPath)

		// Ensure destination parent exists
		if err := os.MkdirAll(filepath.Dir(dstPath), 0755); err != nil {
			// Rollback
			send("rolling_back", "Hiba: mappa létrehozása sikertelen, visszagörgetés...", 0, bytesCopied, totalBytes)
			_ = startFn(req.StackName)
			return fail("copying", "Cél mappa létrehozása sikertelen: "+filepath.Dir(dstPath), err)
		}

		mountPct := 10 + (i * 60 / len(req.HDDMounts))

		send("copying", fmt.Sprintf("Adatok másolása (%d/%d): %s...", i+1, len(req.HDDMounts), filepath.Base(srcPath)),
			mountPct, bytesCopied, totalBytes)

		var rsyncErr error
		bytesCopied, rsyncErr = runRsync(srcPath, dstPath, totalBytes, bytesCopied, mountPct, progress, start)
		if rsyncErr != nil {
			// Rollback
			send("rolling_back", "rsync sikertelen, alkalmazás visszaállítása az eredeti tárolóra...", 0, bytesCopied, totalBytes)
			_ = startFn(req.StackName)
			return fail("copying", "Adatmásolás sikertelen", rsyncErr)
		}
	}

	send("updating", "Konfiguráció frissítése...", 75, bytesCopied, totalBytes)

	// --- Step 4: Update app.yaml HDD_PATH ---
	if err := updateFn(req.StackName, req.TargetPath); err != nil {
		send("rolling_back", "Konfiguráció frissítése sikertelen, visszaállítás...", 0, bytesCopied, totalBytes)
		_ = startFn(req.StackName)
		return fail("updating", "HDD_PATH frissítése sikertelen", err)
	}

	send("starting", "Alkalmazás indítása az új tárolóról...", 85, bytesCopied, totalBytes)

	// --- Step 5: Start app ---
	if err := startFn(req.StackName); err != nil {
		// Revert config and restart with old path
		_ = updateFn(req.StackName, req.CurrentHDDPath)
		_ = startFn(req.StackName)
		return fail("starting", "Alkalmazás indítása sikertelen az új tárolóról", err)
	}

	send("done",
		fmt.Sprintf("Áthelyezés kész! Az alkalmazás az új tárolóról fut. (Régi adat: %s, idő: %ds)",
			req.CurrentHDDPath, int(time.Since(start).Seconds())),
		100, bytesCopied, totalBytes)

	return nil
}

// runRsync runs rsync from srcPath to dstPath and reports progress.
func runRsync(srcPath, dstPath string, totalBytes, prevCopied int64, basePct int, progress chan<- MigrateProgress, start time.Time) (int64, error) {
	// Ensure src ends with / for rsync to sync contents (not the directory itself)
	if !strings.HasSuffix(srcPath, "/") {
		srcPath += "/"
	}

	cmd := exec.Command(
		"rsync", "-a", "--info=progress2", "--human-readable",
		srcPath, dstPath,
	)

	stdout, err := cmd.StdoutPipe()
	if err != nil {
		return prevCopied, err
	}
	stderr, err := cmd.StderrPipe()
	if err != nil {
		return prevCopied, err
	}

	if err := cmd.Start(); err != nil {
		return prevCopied, fmt.Errorf("rsync start failed: %w", err)
	}

	var bytesCopied int64 = prevCopied
	var mu sync.Mutex

	// Parse stdout progress
	go func() {
		scanner := bufio.NewScanner(stdout)
		for scanner.Scan() {
			line := scanner.Text()
			if b, pct, ok := parseRsyncProgress(line); ok {
				mu.Lock()
				bytesCopied = prevCopied + b
				// Scale pct into our range
				scaledPct := basePct + pct*40/100
				if scaledPct > 99 {
					scaledPct = 99
				}
				mu.Unlock()
				progress <- MigrateProgress{
					Step:           "copying",
					Message:        fmt.Sprintf("Adatok másolása... %s / %s", bytesHuman(b), bytesHuman(totalBytes)),
					Percent:        scaledPct,
					BytesCopied:    bytesCopied,
					BytesTotal:     totalBytes,
					ElapsedSeconds: int(time.Since(start).Seconds()),
				}
			}
		}
	}()

	var stderrBuf strings.Builder
	io.Copy(&stderrBuf, stderr)

	if err := cmd.Wait(); err != nil {
		// H11: Read bytesCopied under lock to avoid data race with the progress goroutine.
		mu.Lock()
		copied := bytesCopied
		mu.Unlock()
		return copied, fmt.Errorf("rsync failed: %w — %s", err, stderrBuf.String())
	}

	mu.Lock()
	finalCopied := bytesCopied
	mu.Unlock()
	return finalCopied, nil
}

// dirSize returns the total bytes in a directory (best-effort).
func dirSize(path string) int64 {
	var total int64
	filepath.Walk(path, func(_ string, info os.FileInfo, err error) error {
		if err != nil || info.IsDir() {
			return nil
		}
		total += info.Size()
		return nil
	})
	return total
}

// getFreeBytes returns available bytes on the filesystem at path.
func getFreeBytes(path string) int64 {
	// Use df to get available bytes — works cross-platform within Linux container
	out, err := exec.Command("df", "-B1", "--output=avail", path).Output()
	if err != nil {
		return 0
	}
	lines := strings.Split(strings.TrimSpace(string(out)), "\n")
	if len(lines) < 2 {
		return 0
	}
	var avail int64
	fmt.Sscanf(strings.TrimSpace(lines[1]), "%d", &avail)
	return avail
}

// bytesHuman converts a byte count to human-readable string.
func bytesHuman(b int64) string {
	const (
		KB = 1024
		MB = KB * 1024
		GB = MB * 1024
	)
	switch {
	case b >= GB:
		return fmt.Sprintf("%.1f GB", float64(b)/float64(GB))
	case b >= MB:
		return fmt.Sprintf("%.0f MB", float64(b)/float64(MB))
	case b >= KB:
		return fmt.Sprintf("%.0f KB", float64(b)/float64(KB))
	default:
		return fmt.Sprintf("%d B", b)
	}
}