felhom-controller/controller/internal/stacks/migrate.go

package stacks

// Data-migration engine (TASK B1). Moves an app-data felhom-data NAMESPACE from one storage drive
// to another, in-process over the controller's /mnt:/mnt:rslave RW mount. Crash-safe + resumable via
// a single durable journal (<dataDir>/migration.json); one migration (app OR all) at a time.
//
// Two entry points share ONE pipeline:
//   - MigrateAll(source, target): the whole namespace — every app on the source drive + the
//     conflict-merge walk for non-app/customer content. Used by the decommission flow.
//   - MigrateApp(app, target): one app's data subtree only (appdata + its recovery unit). Handles
//     drive→drive AND SSD→drive (an SSD-resident app gaining HDD_PATH for the first time).
//
// Move primitive = additive `rsync -a --checksum` (NO --delete — never destroys the target). The
// non-app content uses a custom conflict-merge walk (skip-identical / rename-on-differ to a
// `<base>(N)<ext>` sibling; never overwrite). The ONLY destructive step is CLEANUP, which removes the
// SOURCE and runs ONLY after every unit verified AND every app redeployed.

import (
	"bufio"
	"bytes"
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"fmt"
	"io"
	"io/fs"
	"log"
	"os"
	"os/exec"
	"path/filepath"
	"sort"
	"strconv"
	"strings"
	"time"

	"gitea.dooplex.hu/admin/felhom-controller/internal/appbackup"
	"gitea.dooplex.hu/admin/felhom-controller/internal/settings"
	"gitea.dooplex.hu/admin/felhom-controller/internal/system"
)

const (
	migrationJournalFile = "migration.json"
	migrateCopyTimeout   = 6 * time.Hour
	migrateVerifyTimeout = 60 * time.Minute
	nonAppUnit           = "" // map key for the non-app-content unit (MigrateAll only)
)

// MigrationState is the per-unit progress state, persisted in the journal.
type MigrationState string

const (
	UnitPending    MigrationState = "pending"
	UnitCopied     MigrationState = "copied"
	UnitVerified   MigrationState = "verified"
	UnitFlipped    MigrationState = "flipped"
	UnitRedeployed MigrationState = "redeployed"
	UnitCleaned    MigrationState = "cleaned"
)

func stateRank(s MigrationState) int {
	switch s {
	case UnitCopied:
		return 1
	case UnitVerified:
		return 2
	case UnitFlipped:
		return 3
	case UnitRedeployed:
		return 4
	case UnitCleaned:
		return 5
	default:
		return 0 // pending / unknown
	}
}

// MigrationPhase is the pipeline phase, persisted so RecoverMigration resumes from the right step.
type MigrationPhase string

const (
	PhaseStop     MigrationPhase = "stop"
	PhaseCopy     MigrationPhase = "copy"
	PhaseVerify   MigrationPhase = "verify"
	PhaseFlip     MigrationPhase = "flip"
	PhaseRedeploy MigrationPhase = "redeploy"
	PhaseCleanup  MigrationPhase = "cleanup"
	PhaseDone     MigrationPhase = "done"
	PhaseAborted  MigrationPhase = "aborted"
)

// MigUnit is one journal unit: an affected app, or the non-app-content sentinel ("").
type MigUnit struct {
	App   string         `json:"app"`
	State MigrationState `json:"state"`
	Error string         `json:"error,omitempty"`
}

// MigrationJob is BOTH the live status (returned by MigrationStatus, polled by the UI) and the
// durable journal (written to <dataDir>/migration.json atomically at every transition).
type MigrationJob struct {
	ID         string              `json:"id"`
	Scope      string              `json:"scope"` // "app" | "all"
	Phase      MigrationPhase      `json:"phase"`
	Source     string              `json:"source"`    // source drive path (the HDD_PATH value, or the SSD path)
	Target     string              `json:"target"`    // target drive path
	SourceNS   string              `json:"source_ns"` // resolved felhom-data namespace root on source
	TargetNS   string              `json:"target_ns"`
	Apps       []string            `json:"apps"` // affected app names, stable order
	Units      map[string]*MigUnit `json:"units"`
	CurrentApp string              `json:"current_app,omitempty"`
	BytesTotal int64               `json:"bytes_total"`
	BytesDone  int64               `json:"bytes_done"`
	Error      string              `json:"error,omitempty"`
	StartedAt  time.Time           `json:"started_at"`
	UpdatedAt  time.Time           `json:"updated_at"`
	FinishedAt time.Time           `json:"finished_at,omitempty"`

	// DecommissionOnDone: when set, the migration was started by the decommission flow — on successful
	// completion the done-hook soft-marks the source registry path + tells the agent to decommission it.
	// The engine itself does NOT decommission; it only fires the hook (the policy lives in the caller).
	DecommissionOnDone bool `json:"decommission_on_done,omitempty"`
}

func (j *MigrationJob) clone() *MigrationJob {
	cp := *j
	cp.Apps = append([]string(nil), j.Apps...)
	cp.Units = make(map[string]*MigUnit, len(j.Units))
	for k, u := range j.Units {
		uc := *u
		cp.Units[k] = &uc
	}
	return &cp
}

// migSeams lets tests drive the pipeline without docker/rsync. Nil = use the real implementations.
type migSeams struct {
	copy         func(ctx context.Context, src, dst string, onBytes func(int64)) error
	verify       func(ctx context.Context, src, dst string) error
	stop         func(name string) error
	flipRedeploy func(name, target string) error
}

// SetMigrationDeps wires the registry + the backup-running check (mutual exclusion, Change 3).
// Call once from main after SetEncryptionKey, before RecoverMigration.
func (m *Manager) SetMigrationDeps(sett *settings.Settings, backupRunning func() bool) {
	m.mu.Lock()
	defer m.mu.Unlock()
	m.settings = sett
	m.backupRunning = backupRunning
	m.sysDataPath = m.cfg.Paths.SystemDataPath
}

// IsMigrating reports whether a migration is in progress (used by the backup orchestrator's guard).
func (m *Manager) IsMigrating() bool {
	m.migrateMu.Lock()
	defer m.migrateMu.Unlock()
	return m.migrating
}

func (m *Manager) acquireMigrating() error {
	m.migrateMu.Lock()
	defer m.migrateMu.Unlock()
	if m.migrating {
		return fmt.Errorf("migráció már folyamatban")
	}
	m.migrating = true
	return nil
}

func (m *Manager) releaseMigrating() {
	m.migrateMu.Lock()
	m.migrating = false
	m.migrateMu.Unlock()
}

func (m *Manager) setJob(j *MigrationJob) {
	m.migrateMu.Lock()
	m.migJob = j
	m.migrateMu.Unlock()
}

// MigrationStatus returns a deep copy of the live job (nil when idle).
func (m *Manager) MigrationStatus() *MigrationJob {
	m.migrateMu.Lock()
	defer m.migrateMu.Unlock()
	if m.migJob == nil {
		return nil
	}
	return m.migJob.clone()
}

// SetMigrationDoneHook registers a callback fired (in the migration goroutine) when a migration
// completes successfully. The decommission flow uses it to soft-mark + agent-decommission the source.
func (m *Manager) SetMigrationDoneHook(fn func(*MigrationJob)) { m.migDoneHook = fn }

// MigrateAll moves the whole namespace off sourcePath onto targetPath.
func (m *Manager) MigrateAll(ctx context.Context, sourcePath, targetPath string) (string, error) {
	return m.startMigration("all", sourcePath, "", targetPath, false)
}

// MigrateAllAndDecommission moves the whole namespace then (on success) fires the done-hook so the
// caller decommissions the now-empty source drive.
func (m *Manager) MigrateAllAndDecommission(ctx context.Context, sourcePath, targetPath string) (string, error) {
	return m.startMigration("all", sourcePath, "", targetPath, true)
}

// MigrateApp moves a single app's data subtree onto targetPath.
func (m *Manager) MigrateApp(ctx context.Context, appName, targetPath string) (string, error) {
	return m.startMigration("app", "", appName, targetPath, false)
}

func (m *Manager) startMigration(scope, sourcePath, appName, targetPath string, decommission bool) (string, error) {
	if err := m.acquireMigrating(); err != nil {
		return "", err
	}
	launched := false
	defer func() {
		if !launched {
			m.releaseMigrating()
		}
	}()

	j := &MigrationJob{
		Scope:              scope,
		Target:             filepath.Clean(targetPath),
		Units:              map[string]*MigUnit{},
		StartedAt:          time.Now().UTC(),
		DecommissionOnDone: decommission,
	}
	j.ID = "mig-" + j.StartedAt.Format("20060102-150405")

	// Resolve affected apps + the source namespace.
	var apps []string
	switch scope {
	case "all":
		j.Source = filepath.Clean(sourcePath)
		j.SourceNS = appbackup.NamespaceRoot(j.Source, m.inGuest(j.Source))
		apps = append(apps, m.appsOnDrive(j.Source)...)
		j.Units[nonAppUnit] = &MigUnit{App: nonAppUnit, State: UnitPending}
	case "app":
		cfg := m.LoadAppConfigByName(appName)
		if cfg == nil {
			return "", fmt.Errorf("alkalmazás nem található: %s", appName)
		}
		j.Source, j.SourceNS = m.appSourceNS(cfg)
		apps = []string{appName}
	default:
		return "", fmt.Errorf("ismeretlen migrációs hatókör: %s", scope)
	}

	j.TargetNS = appbackup.NamespaceRoot(j.Target, m.inGuest(j.Target))
	sort.Strings(apps)
	j.Apps = apps
	for _, a := range apps {
		j.Units[a] = &MigUnit{App: a, State: UnitPending}
	}

	// VALIDATE (synchronous — refusals reach the caller and change nothing).
	if err := m.migValidate(j); err != nil {
		return "", err
	}

	j.Phase = PhaseStop
	j.UpdatedAt = time.Now().UTC()
	m.setJob(j)
	if err := m.writeJournal(j); err != nil {
		return "", fmt.Errorf("migrációs napló írása: %w", err)
	}

	launched = true
	go m.runMigration(context.Background(), j)
	return j.ID, nil
}

// inGuest reports whether drivePath is a user drive (its in-guest mount IS the felhom-data namespace
// root) vs the system/SSD path (which holds a felhom-data SUBDIR). Compares cleaned paths so the
// decision is stable regardless of slash style.
func (m *Manager) inGuest(drivePath string) bool {
	return filepath.Clean(drivePath) != filepath.Clean(m.sysDataPath)
}

// appSourceNS resolves an app's current source drive path + felhom-data namespace root from its
// config. An app with no HDD_PATH lives on the system/SSD path (the SSD→drive case).
func (m *Manager) appSourceNS(cfg *AppConfig) (src, ns string) {
	src = cfg.Env["HDD_PATH"]
	if src == "" {
		src = m.sysDataPath
	}
	src = filepath.Clean(src)
	return src, appbackup.NamespaceRoot(src, m.inGuest(src))
}

// appsOnDrive returns the names of deployed apps whose HDD_PATH equals sourcePath.
func (m *Manager) appsOnDrive(sourcePath string) []string {
	var out []string
	for _, st := range m.GetStacks() {
		if !st.Deployed {
			continue
		}
		cfg := m.LoadAppConfigByName(st.Name)
		if cfg != nil && cfg.Env["HDD_PATH"] == sourcePath {
			out = append(out, st.Name)
		}
	}
	return out
}

// migValidate runs the pre-flight checks. Any failure aborts before any side-effect.
func (m *Manager) migValidate(j *MigrationJob) error {
	if m.backupRunning != nil && m.backupRunning() {
		return fmt.Errorf("biztonsági mentés folyamatban, próbáld újra")
	}
	if m.testSeams == nil { // real rsync is only required when not using injected copy/verify seams
		if _, err := exec.LookPath("rsync"); err != nil {
			return fmt.Errorf("az rsync nem érhető el a rendszeren")
		}
	}
	if j.Target == j.Source {
		return fmt.Errorf("a cél és a forrás tároló megegyezik")
	}
	if m.settings == nil || !m.settings.IsStoragePathSchedulable(j.Target) {
		return fmt.Errorf("a céltároló nem elérhető vagy nem választható")
	}
	// App-dir collision: refuse if the same app dir already exists at the target.
	var collide []string
	for _, app := range j.Apps {
		if pathExists(appbackup.AppDataDir(j.TargetNS, app)) {
			collide = append(collide, app)
		}
	}
	if len(collide) > 0 {
		return fmt.Errorf("ütközés a céltárolón — már létezik ezeknek az alkalmazásoknak az adata: %s", strings.Join(collide, ", "))
	}
	// Free-space check (best-effort; GetDiskUsage is nil on non-linux).
	need := m.migSourceSize(j)
	j.BytesTotal = need
	if du := system.GetDiskUsage(j.Target); du != nil {
		avail := int64(du.AvailGB * 1e9)
		if need > avail {
			return fmt.Errorf("nincs elég hely a céltárolón (kb. %d GB szükséges, %.1f GB szabad)", need/1_000_000_000, du.AvailGB)
		}
	}
	return nil
}

// migSourceSize estimates the bytes to move (conservative; dedup may write less).
func (m *Manager) migSourceSize(j *MigrationJob) int64 {
	if j.Scope == "all" {
		return dirBytes(j.SourceNS)
	}
	var total int64
	for _, app := range j.Apps {
		total += dirBytes(appbackup.AppDataDir(j.SourceNS, app))
		total += dirBytes(appbackup.RecoveryUnitPath(j.SourceNS, app))
	}
	return total
}

// runMigration drives the phase machine. Used both for a fresh start and for resume; each phase is
// idempotent and skips already-completed units, so re-entry never re-copies or re-removes.
func (m *Manager) runMigration(ctx context.Context, j *MigrationJob) {
	defer m.releaseMigrating()
	m.logger.Printf("[INFO] [migrate] %s running: %s → %s scope=%s apps=%d phase=%s",
		j.ID, j.Source, j.Target, j.Scope, len(j.Apps), j.Phase)
	for {
		m.logger.Printf("[DEBUG] [migrate] %s phase %s", j.ID, j.Phase)
		var err error
		switch j.Phase {
		case PhaseStop:
			err = m.migStop(j)
			if err == nil {
				j.Phase = PhaseCopy
			}
		case PhaseCopy:
			err = m.migCopy(ctx, j)
			if err == nil {
				j.Phase = PhaseVerify
			}
		case PhaseVerify:
			err = m.migVerify(ctx, j)
			if err == nil {
				j.Phase = PhaseFlip
			}
		case PhaseFlip, PhaseRedeploy:
			err = m.migFlipRedeploy(j)
			if err == nil {
				j.Phase = PhaseCleanup
			}
		case PhaseCleanup:
			err = m.migCleanup(j)
			if err == nil {
				j.Phase = PhaseDone
				j.FinishedAt = time.Now().UTC()
			}
		case PhaseDone, PhaseAborted:
			return
		default:
			err = fmt.Errorf("ismeretlen migrációs fázis: %q", j.Phase)
		}
		if err != nil {
			m.migAbort(j, err)
			return
		}
		j.CurrentApp = ""
		if perr := m.persistJob(j); perr != nil {
			m.logger.Printf("[ERROR] [migrate] journal write failed: %v", perr)
		}
		if j.Phase == PhaseDone {
			m.logger.Printf("[INFO] [migrate] %s complete: %s → %s (%d app(s), %s)",
				j.ID, j.Source, j.Target, len(j.Apps), j.FinishedAt.Sub(j.StartedAt).Round(time.Second))
			if m.migDoneHook != nil {
				m.migDoneHook(j.clone()) // decommission policy (soft-mark + agent) lives in the hook
			}
			return
		}
	}
}

func (m *Manager) migAbort(j *MigrationJob, cause error) {
	j.Phase = PhaseAborted
	j.Error = cause.Error()
	j.FinishedAt = time.Now().UTC()
	_ = m.persistJob(j)
	m.logger.Printf("[ERROR] [migrate] %s ABORTED at copy/verify/flip — source intact: %v", j.ID, cause)
}

// migStop stops every affected app (idempotent — compose down on a stopped stack is a no-op).
func (m *Manager) migStop(j *MigrationJob) error {
	for _, app := range j.Apps {
		if err := m.doStop(app); err != nil {
			return fmt.Errorf("alkalmazás leállítása sikertelen (%s): %w", app, err)
		}
	}
	return nil
}

// migCopy copies each app subtree (rsync, additive) and — for MigrateAll — the non-app content via
// the conflict-merge walk. Skips units already past `copied` so resume never re-copies.
func (m *Manager) migCopy(ctx context.Context, j *MigrationJob) error {
	for _, app := range j.Apps {
		u := j.Units[app]
		if stateRank(u.State) >= stateRank(UnitCopied) {
			continue
		}
		j.CurrentApp = app
		_ = m.persistJob(j)
		// appdata subtree (collision-free post-validate)
		if err := m.copySubtree(ctx, j, appbackup.AppDataDir(j.SourceNS, app), appbackup.AppDataDir(j.TargetNS, app)); err != nil {
			u.Error = err.Error()
			return fmt.Errorf("másolás sikertelen (%s appdata): %w", app, err)
		}
		// the app's recovery unit (db-dumps + volume-dumps + compose + manifest)
		if err := m.copySubtree(ctx, j, appbackup.RecoveryUnitPath(j.SourceNS, app), appbackup.RecoveryUnitPath(j.TargetNS, app)); err != nil {
			u.Error = err.Error()
			return fmt.Errorf("másolás sikertelen (%s mentés): %w", app, err)
		}
		u.State = UnitCopied
		_ = m.persistJob(j)
	}
	if j.Scope == "all" {
		u := j.Units[nonAppUnit]
		if stateRank(u.State) < stateRank(UnitCopied) {
			j.CurrentApp = ""
			if err := walkMerge(m.logger, j.SourceNS, j.TargetNS, m.appDataSkipSet(j), false, func(b int64) { j.BytesDone += b }); err != nil {
				u.Error = err.Error()
				return fmt.Errorf("ügyfél-adatok összefésülése sikertelen: %w", err)
			}
			u.State = UnitCopied
			_ = m.persistJob(j)
		}
	}
	return nil
}

// copySubtree rsyncs src→dst if src exists; a missing source is a no-op (nothing to move).
func (m *Manager) copySubtree(ctx context.Context, j *MigrationJob, src, dst string) error {
	if !pathExists(src) {
		return nil
	}
	return m.doCopy(ctx, src, dst, func(b int64) { j.BytesDone = b })
}

// migVerify confirms every source byte landed at the target before any source mutation.
func (m *Manager) migVerify(ctx context.Context, j *MigrationJob) error {
	for _, app := range j.Apps {
		u := j.Units[app]
		if stateRank(u.State) >= stateRank(UnitVerified) {
			continue
		}
		if err := m.verifySubtree(ctx, appbackup.AppDataDir(j.SourceNS, app), appbackup.AppDataDir(j.TargetNS, app)); err != nil {
			u.Error = err.Error()
			return fmt.Errorf("ellenőrzés sikertelen (%s appdata): %w", app, err)
		}
		if err := m.verifySubtree(ctx, appbackup.RecoveryUnitPath(j.SourceNS, app), appbackup.RecoveryUnitPath(j.TargetNS, app)); err != nil {
			u.Error = err.Error()
			return fmt.Errorf("ellenőrzés sikertelen (%s mentés): %w", app, err)
		}
		u.State = UnitVerified
		_ = m.persistJob(j)
	}
	if j.Scope == "all" {
		u := j.Units[nonAppUnit]
		if stateRank(u.State) < stateRank(UnitVerified) {
			// assert-only merge walk: every source file has a content-identical counterpart at target.
			if err := walkMerge(m.logger, j.SourceNS, j.TargetNS, m.appDataSkipSet(j), true, nil); err != nil {
				u.Error = err.Error()
				return fmt.Errorf("ügyfél-adatok ellenőrzése sikertelen: %w", err)
			}
			u.State = UnitVerified
			_ = m.persistJob(j)
		}
	}
	return nil
}

func (m *Manager) verifySubtree(ctx context.Context, src, dst string) error {
	if !pathExists(src) {
		return nil
	}
	return m.doVerify(ctx, src, dst)
}

// migFlipRedeploy rewrites each app's HDD_PATH to the target and redeploys it (RedeployFromEnv as one
// idempotent unit). Journals `flipped` before the call and `redeployed` after, so resume re-calls.
func (m *Manager) migFlipRedeploy(j *MigrationJob) error {
	for _, app := range j.Apps {
		u := j.Units[app]
		if stateRank(u.State) >= stateRank(UnitRedeployed) {
			continue
		}
		j.CurrentApp = app
		u.State = UnitFlipped
		_ = m.persistJob(j)
		if err := m.doFlipRedeploy(app, j.Target); err != nil {
			u.Error = err.Error()
			return fmt.Errorf("újratelepítés sikertelen (%s): %w", app, err)
		}
		u.State = UnitRedeployed
		_ = m.persistJob(j)
	}
	return nil
}

// migCleanup removes the SOURCE content. THE irreversible step: gated on every unit verified AND
// every app redeployed. Idempotent (RemoveAll of an absent path is a no-op).
func (m *Manager) migCleanup(j *MigrationJob) error {
	if err := m.migCleanupAllowed(j); err != nil {
		return err
	}
	for _, app := range j.Apps {
		u := j.Units[app]
		if u.State == UnitCleaned {
			continue
		}
		if err := os.RemoveAll(appbackup.AppDataDir(j.SourceNS, app)); err != nil {
			return fmt.Errorf("forrás törlése sikertelen (%s appdata): %w", app, err)
		}
		if err := os.RemoveAll(appbackup.RecoveryUnitPath(j.SourceNS, app)); err != nil {
			return fmt.Errorf("forrás törlése sikertelen (%s mentés): %w", app, err)
		}
		u.State = UnitCleaned
		_ = m.persistJob(j)
	}
	if j.Scope == "all" {
		u := j.Units[nonAppUnit]
		if u.State != UnitCleaned {
			// Remove every remaining child of the source namespace (the non-app/customer content).
			entries, err := os.ReadDir(j.SourceNS)
			if err != nil && !os.IsNotExist(err) {
				return fmt.Errorf("forrás névtér olvasása sikertelen: %w", err)
			}
			for _, e := range entries {
				if err := os.RemoveAll(filepath.Join(j.SourceNS, e.Name())); err != nil {
					return fmt.Errorf("forrás törlése sikertelen (%s): %w", e.Name(), err)
				}
			}
			u.State = UnitCleaned
			_ = m.persistJob(j)
		}
	}
	return nil
}

// migCleanupAllowed is the load-bearing gate: cleanup runs ONLY when all units are verified AND all
// apps are redeployed.
func (m *Manager) migCleanupAllowed(j *MigrationJob) error {
	for key, u := range j.Units {
		if stateRank(u.State) < stateRank(UnitVerified) {
			return fmt.Errorf("cleanup gate: a(z) %q egység nincs ellenőrizve (állapot=%s)", key, u.State)
		}
	}
	for _, app := range j.Apps {
		if stateRank(j.Units[app].State) < stateRank(UnitRedeployed) {
			return fmt.Errorf("cleanup gate: a(z) %q alkalmazás nincs újratelepítve (állapot=%s)", app, j.Units[app].State)
		}
	}
	return nil
}

// appDataSkipSet returns the source appdata dirs (rsync'd separately) to skip in the merge walk.
func (m *Manager) appDataSkipSet(j *MigrationJob) map[string]bool {
	skip := map[string]bool{}
	for _, app := range j.Apps {
		skip[filepath.Clean(appbackup.AppDataDir(j.SourceNS, app))] = true
	}
	return skip
}

// RecoverMigration resumes a crashed migration on startup (no-op if none or terminal).
func (m *Manager) RecoverMigration(ctx context.Context) {
	j, err := m.loadJournal()
	if err != nil {
		m.logger.Printf("[WARN] [migrate] could not read migration journal: %v", err)
		return
	}
	if j == nil || j.Phase == PhaseDone || j.Phase == PhaseAborted {
		return
	}
	if err := m.acquireMigrating(); err != nil {
		return
	}
	m.logger.Printf("[WARN] [migrate] resuming interrupted migration %s from phase=%s", j.ID, j.Phase)
	m.setJob(j)
	go m.runMigration(context.Background(), j)
}

// --- seam dispatch (real impls; tests override via m.testSeams) ---

func (m *Manager) doStop(name string) error {
	if m.testSeams != nil && m.testSeams.stop != nil {
		return m.testSeams.stop(name)
	}
	return m.StopStack(name)
}

func (m *Manager) doCopy(ctx context.Context, src, dst string, onBytes func(int64)) error {
	if m.testSeams != nil && m.testSeams.copy != nil {
		return m.testSeams.copy(ctx, src, dst, onBytes)
	}
	return rsyncCopy(ctx, src, dst, onBytes)
}

func (m *Manager) doVerify(ctx context.Context, src, dst string) error {
	if m.testSeams != nil && m.testSeams.verify != nil {
		return m.testSeams.verify(ctx, src, dst)
	}
	return rsyncVerify(ctx, src, dst)
}

func (m *Manager) doFlipRedeploy(name, target string) error {
	if m.testSeams != nil && m.testSeams.flipRedeploy != nil {
		return m.testSeams.flipRedeploy(name, target)
	}
	cfg := m.LoadAppConfigByName(name)
	if cfg == nil {
		return fmt.Errorf("app config not found")
	}
	if err := m.RedeployFromEnv(name, flipEnv(cfg.Env, target)); err != nil {
		return err
	}
	if !m.waitHealthy(name) {
		return fmt.Errorf("az alkalmazás nem indult el az új tárhelyen")
	}
	return nil
}

// flipEnv returns a copy of env with HDD_PATH set to target.
func flipEnv(env map[string]string, target string) map[string]string {
	out := make(map[string]string, len(env)+1)
	for k, v := range env {
		out[k] = v
	}
	out["HDD_PATH"] = target
	return out
}

// waitHealthy polls until the stack is up (running/unhealthy) or times out.
func (m *Manager) waitHealthy(name string) bool {
	deadline := time.Now().Add(90 * time.Second)
	for {
		_ = m.RefreshStatus()
		if st, ok := m.GetStack(name); ok {
			if st.State == StateRunning || st.State == StateUnhealthy {
				return true
			}
		}
		if time.Now().After(deadline) {
			if st, ok := m.GetStack(name); ok {
				return st.State == StateRunning || st.State == StateUnhealthy || st.State == StateStarting
			}
			return false
		}
		time.Sleep(3 * time.Second)
	}
}

// --- journal persistence ---

func (m *Manager) journalPath() string {
	return filepath.Join(m.cfg.Paths.DataDir, migrationJournalFile)
}

// persistJob stamps UpdatedAt and writes the journal atomically.
func (m *Manager) persistJob(j *MigrationJob) error {
	j.UpdatedAt = time.Now().UTC()
	return m.writeJournal(j)
}

func (m *Manager) writeJournal(j *MigrationJob) error {
	data, err := json.MarshalIndent(j, "", "  ")
	if err != nil {
		return err
	}
	if err := os.MkdirAll(filepath.Dir(m.journalPath()), 0o755); err != nil {
		return err
	}
	tmp := m.journalPath() + ".tmp"
	if err := os.WriteFile(tmp, data, 0o600); err != nil {
		os.Remove(tmp)
		return err
	}
	return os.Rename(tmp, m.journalPath())
}

func (m *Manager) loadJournal() (*MigrationJob, error) {
	data, err := os.ReadFile(m.journalPath())
	if err != nil {
		if os.IsNotExist(err) {
			return nil, nil
		}
		return nil, err
	}
	var j MigrationJob
	if err := json.Unmarshal(data, &j); err != nil {
		return nil, err
	}
	return &j, nil
}

// --- rsync + filesystem primitives (additive — NO --delete) ---

// rsyncCopy mirrors src→dst additively with `rsync -a --checksum` (NEVER --delete). onBytes receives
// the running transferred-bytes total parsed from --info=progress2 (best-effort).
func rsyncCopy(ctx context.Context, src, dst string, onBytes func(int64)) error {
	if err := os.MkdirAll(dst, 0o755); err != nil {
		return fmt.Errorf("mkdir %s: %w", dst, err)
	}
	cctx, cancel := context.WithTimeout(ctx, migrateCopyTimeout)
	defer cancel()
	cmd := exec.CommandContext(cctx, "rsync", "-a", "--checksum", "--info=progress2",
		strings.TrimRight(src, "/")+"/", strings.TrimRight(dst, "/")+"/")
	stdout, err := cmd.StdoutPipe()
	if err != nil {
		return err
	}
	var stderr bytes.Buffer
	cmd.Stderr = &stderr
	if err := cmd.Start(); err != nil {
		return err
	}
	scanner := bufio.NewScanner(stdout)
	for scanner.Scan() {
		if b, ok := parseProgress2Bytes(scanner.Text()); ok && onBytes != nil {
			onBytes(b)
		}
	}
	if err := cmd.Wait(); err != nil {
		return fmt.Errorf("rsync: %v: %s", err, strings.TrimSpace(stderr.String()))
	}
	return nil
}

// rsyncVerify dry-runs `rsync -ani --checksum` and fails if ANY content transfer/create is still
// pending (itemize lines beginning with <, >, or c). Attr-only `.`-prefixed lines are ignored.
func rsyncVerify(ctx context.Context, src, dst string) error {
	cctx, cancel := context.WithTimeout(ctx, migrateVerifyTimeout)
	defer cancel()
	cmd := exec.CommandContext(cctx, "rsync", "-a", "-n", "-i", "--checksum",
		strings.TrimRight(src, "/")+"/", strings.TrimRight(dst, "/")+"/")
	out, err := cmd.CombinedOutput()
	if err != nil {
		return fmt.Errorf("verify rsync: %v: %s", err, strings.TrimSpace(string(out)))
	}
	var pending []string
	for _, ln := range strings.Split(string(out), "\n") {
		ln = strings.TrimRight(ln, "\r")
		if isPendingTransfer(ln) {
			pending = append(pending, strings.TrimSpace(ln))
		}
	}
	if len(pending) > 0 {
		return fmt.Errorf("%d függőben lévő átvitel maradt, pl. %q", len(pending), pending[0])
	}
	return nil
}

// isPendingTransfer reports whether an rsync -i itemize line denotes a content transfer/create
// (file or dir). Attr-only changes ('.' prefix) and messages ('*') are NOT transfers.
func isPendingTransfer(line string) bool {
	if len(line) < 2 {
		return false
	}
	switch line[0] {
	case '>', '<', 'c':
		return true
	default:
		return false
	}
}

// parseProgress2Bytes extracts the running transferred-bytes total from an --info=progress2 line
// (first field is a comma-grouped byte count, e.g. "1,234,567  45% ...").
func parseProgress2Bytes(line string) (int64, bool) {
	fields := strings.Fields(strings.TrimSpace(line))
	if len(fields) == 0 {
		return 0, false
	}
	digits := strings.ReplaceAll(fields[0], ",", "")
	n, err := strconv.ParseInt(digits, 10, 64)
	if err != nil {
		return 0, false
	}
	return n, true
}

// walkMerge merges source ns into target ns with the conflict rules (Change: non-app content only).
// assertOnly=true performs no writes — it verifies every source regular file has a content-identical
// counterpart at target (the file itself or a `(N)` sibling), returning an error if any does not.
// skip names absolute SOURCE dirs (app appdata dirs) handled by rsync, which are pruned.
func walkMerge(lg *log.Logger, srcNS, dstNS string, skip map[string]bool, assertOnly bool, onBytes func(int64)) error {
	if !pathExists(srcNS) {
		return nil
	}
	return filepath.WalkDir(srcNS, func(path string, d fs.DirEntry, err error) error {
		if err != nil {
			return err
		}
		if filepath.Clean(path) == filepath.Clean(srcNS) {
			return nil
		}
		rel, rerr := filepath.Rel(srcNS, path)
		if rerr != nil {
			return rerr
		}
		dst := filepath.Join(dstNS, rel)

		if d.IsDir() {
			if skip[filepath.Clean(path)] {
				return filepath.SkipDir
			}
			if assertOnly {
				return nil
			}
			if err := os.MkdirAll(dst, 0o755); err != nil {
				return err
			}
			// M3: userdata dirs ALWAYS get the 2775-setgid/gid-1000 convention RE-ASSERTED (not merely
			// source-preserved), so a PRE-EXISTING stale 755 target dir is corrected regardless of the
			// source mode. Other dirs keep #8's source-mode preservation.
			if isUserdataDir(rel) {
				return appbackup.EnsureUserdataDir(dst)
			}
			return preserveDirOwnership(dst, d)
		}

		// Symlink: recreate-if-absent (copy mode); ignored in assert mode.
		if d.Type()&fs.ModeSymlink != 0 {
			if assertOnly {
				return nil
			}
			return mergeSymlink(path, dst)
		}
		if !d.Type().IsRegular() {
			if lg != nil {
				lg.Printf("[WARN] [migrate] skipping non-regular file %s", path)
			}
			return nil
		}

		srcSum, serr := fileSum(path)
		if serr != nil {
			return serr
		}
		matched, merr := matchesExistingTarget(dst, srcSum)
		if merr != nil {
			return merr
		}
		if matched {
			return nil // dedup / already present
		}
		if assertOnly {
			return fmt.Errorf("a forrásfájlnak nincs azonos másolata a célon: %s", rel)
		}
		out := dst
		if pathExists(dst) {
			out = lowestFreeSibling(dst) // never overwrite an existing target file
		}
		n, cerr := copyFile(path, out)
		if cerr != nil {
			return cerr
		}
		if onBytes != nil {
			onBytes(n)
		}
		return nil
	})
}

// matchesExistingTarget reports whether srcSum equals the checksum of dst OR any `(N)` sibling.
func matchesExistingTarget(dst, srcSum string) (bool, error) {
	if pathExists(dst) {
		s, err := fileSum(dst)
		if err != nil {
			return false, err
		}
		if s == srcSum {
			return true, nil
		}
	}
	for n := 1; ; n++ {
		sib := siblingName(dst, n)
		if !pathExists(sib) {
			break
		}
		s, err := fileSum(sib)
		if err != nil {
			return false, err
		}
		if s == srcSum {
			return true, nil
		}
	}
	return false, nil
}

// lowestFreeSibling returns the first `<base>(N)<ext>` name (N≥1) that does not exist.
func lowestFreeSibling(dst string) string {
	for n := 1; ; n++ {
		sib := siblingName(dst, n)
		if !pathExists(sib) {
			return sib
		}
	}
}

// siblingName builds `<stem>(N)<ext>` using the LAST extension (foo.tar.gz → stem "foo.tar", ext ".gz").
func siblingName(dst string, n int) string {
	dir := filepath.Dir(dst)
	base := filepath.Base(dst)
	ext := filepath.Ext(base)
	stem := strings.TrimSuffix(base, ext)
	return filepath.Join(dir, fmt.Sprintf("%s(%d)%s", stem, n, ext))
}

func mergeSymlink(src, dst string) error {
	target, err := os.Readlink(src)
	if err != nil {
		return err
	}
	if existing, lerr := os.Readlink(dst); lerr == nil {
		if existing == target {
			return nil // identical symlink already present
		}
		// differing symlink → rename-sibling
		dst = lowestFreeSibling(dst)
	} else if pathExists(dst) {
		dst = lowestFreeSibling(dst)
	}
	if err := os.MkdirAll(filepath.Dir(dst), 0o755); err != nil {
		return err
	}
	return os.Symlink(target, dst)
}

// copyFile streams src→dst via a temp file + chmod + fsync + atomic rename. Returns bytes written.
func copyFile(src, dst string) (int64, error) {
	in, err := os.Open(src)
	if err != nil {
		return 0, err
	}
	defer in.Close()
	fi, err := in.Stat()
	if err != nil {
		return 0, err
	}
	if err := os.MkdirAll(filepath.Dir(dst), 0o755); err != nil {
		return 0, err
	}
	tmp := dst + ".felhom-mig.tmp"
	out, err := os.OpenFile(tmp, os.O_CREATE|os.O_WRONLY|os.O_TRUNC, fi.Mode().Perm())
	if err != nil {
		return 0, err
	}
	n, cerr := io.Copy(out, in)
	if cerr != nil {
		out.Close()
		os.Remove(tmp)
		return 0, cerr
	}
	if err := out.Sync(); err != nil {
		out.Close()
		os.Remove(tmp)
		return 0, err
	}
	if err := out.Close(); err != nil {
		os.Remove(tmp)
		return 0, err
	}
	// #8 (v0.66.0): preserve the SOURCE file's FULL mode (incl. setgid/setuid/sticky — not .Perm(),
	// which masks them off) + group, so the userdata convention survives a whole-drive migration.
	if err := os.Chmod(tmp, fi.Mode()); err != nil {
		os.Remove(tmp)
		return 0, err
	}
	if gid, ok := appbackup.StatGID(fi); ok {
		_ = os.Chown(tmp, -1, gid) // best-effort; needs root for an arbitrary group (the controller is)
	}
	if err := os.Rename(tmp, dst); err != nil {
		os.Remove(tmp)
		return 0, err
	}
	return n, nil
}

// preserveDirOwnership re-stamps a freshly-created target dir with the SOURCE dir's full mode (incl.
// setgid) and group — part of the #8 fix so the userdata convention survives a migration.
// isUserdataDir reports whether a namespace-relative path is the userdata tree (the customer-facing
// shared-content area) — `userdata` itself or anything under it — which must carry the 2775-setgid
// convention. Normalised to forward slashes so it matches on any host.
func isUserdataDir(rel string) bool {
	r := filepath.ToSlash(rel)
	return r == "userdata" || strings.HasPrefix(r, "userdata/")
}

func preserveDirOwnership(dst string, d fs.DirEntry) error {
	info, err := d.Info()
	if err != nil {
		return err
	}
	if err := os.Chmod(dst, info.Mode()); err != nil {
		return err
	}
	if gid, ok := appbackup.StatGID(info); ok {
		_ = os.Chown(dst, -1, gid) // best-effort; root sets an arbitrary group (the controller is root)
	}
	return nil
}

// fileSum returns the hex sha256 of a file (streaming).
func fileSum(path string) (string, error) {
	f, err := os.Open(path)
	if err != nil {
		return "", err
	}
	defer f.Close()
	h := sha256.New()
	if _, err := io.Copy(h, f); err != nil {
		return "", err
	}
	return hex.EncodeToString(h.Sum(nil)), nil
}

func pathExists(p string) bool {
	_, err := os.Stat(p)
	return err == nil
}

// dirBytes returns the total size of a directory via `du -sb` (0 if absent/error).
func dirBytes(dir string) int64 {
	if !pathExists(dir) {
		return 0
	}
	ctx, cancel := context.WithTimeout(context.Background(), 60*time.Second)
	defer cancel()
	out, err := exec.CommandContext(ctx, "du", "-sb", dir).Output()
	if err != nil {
		return 0
	}
	fields := strings.Fields(string(out))
	if len(fields) == 0 {
		return 0
	}
	n, err := strconv.ParseInt(fields[0], 10, 64)
	if err != nil {
		return 0
	}
	return n
}