URBM/internal/service/service.go

package service

import (
	"bytes"
	"context"
	"crypto/rand"
	"encoding/hex"
	"errors"
	"fmt"
	"log/slog"
	"os"
	"path/filepath"
	"strings"
	"sync"
	"syscall"
	"time"

	"git.casaderoll.de/michael/urbm/internal/model"
	"git.casaderoll.de/michael/urbm/internal/notify"
	"git.casaderoll.de/michael/urbm/internal/platform"
	"git.casaderoll.de/michael/urbm/internal/queue"
	"git.casaderoll.de/michael/urbm/internal/restic"
	"git.casaderoll.de/michael/urbm/internal/rsync"
	"git.casaderoll.de/michael/urbm/internal/scheduler"
	"git.casaderoll.de/michael/urbm/internal/store"
)

type SecretStore interface {
	Get(string) (string, error)
	Put(string, string, string) error
	Delete(string) error
}

type Service struct {
	store     *store.Store
	secrets   SecretStore
	restic    *restic.Runner
	rsync     *rsync.Runner
	mounts    *platform.MountManager
	workloads *platform.WorkloadManager
	notifier  *notify.Sender
	log       *slog.Logger
	queue     *queue.Queue
	mu        sync.RWMutex
	config    model.Config
}

func New(st *store.Store, secrets SecretStore, rr *restic.Runner, rs *rsync.Runner, mounts *platform.MountManager, workloads *platform.WorkloadManager, notifier *notify.Sender, log *slog.Logger) (*Service, error) {
	config, err := st.LoadConfig()
	if err != nil {
		return nil, err
	}
	s := &Service{store: st, secrets: secrets, restic: rr, rsync: rs, mounts: mounts, workloads: workloads, notifier: notifier, log: log, config: config}
	s.queue = queue.New(s.execute, func(runs []model.Run) {
		if err := st.SaveRuns(runs); err != nil {
			log.Error("save run history", "error", err)
		}
	})
	if runs, err := st.LoadRuns(); err == nil {
		s.queue.RestoreHistory(runs)
	}
	return s, nil
}

func (s *Service) RunQueue(ctx context.Context) { s.queue.Run(ctx) }
func (s *Service) Stop()                        { s.queue.Stop() }
func (s *Service) Config() model.Config         { s.mu.RLock(); defer s.mu.RUnlock(); return s.config }
func (s *Service) Runs() []model.Run            { return s.queue.Snapshot() }
func (s *Service) Logs() map[string]any {
	return map[string]any{
		"daemon": tailFile("/var/log/urbm.log", 512*1024, 300),
		"runs":   runsWithLogs(s.queue.Snapshot(), 25),
	}
}

func runsWithLogs(runs []model.Run, limit int) []model.Run {
	result := make([]model.Run, 0, limit)
	for index := len(runs) - 1; index >= 0 && len(result) < limit; index-- {
		if len(runs[index].LiveLog) == 0 {
			continue
		}
		result = append(result, runs[index])
	}
	return result
}

func tailFile(path string, maxBytes int64, maxLines int) map[string]any {
	data, err := os.ReadFile(path)
	if err != nil {
		if errors.Is(err, os.ErrNotExist) {
			return map[string]any{"path": path, "lines": []string{}, "error": "Logdatei ist noch nicht vorhanden"}
		}
		return map[string]any{"path": path, "lines": []string{}, "error": err.Error()}
	}
	truncatedBytes := 0
	if int64(len(data)) > maxBytes {
		truncatedBytes = len(data) - int(maxBytes)
		data = data[len(data)-int(maxBytes):]
		if index := bytes.IndexByte(data, '\n'); index >= 0 && index+1 < len(data) {
			data = data[index+1:]
		}
	}
	rawLines := strings.Split(strings.TrimRight(string(data), "\n"), "\n")
	if len(rawLines) == 1 && rawLines[0] == "" {
		rawLines = []string{}
	}
	truncatedLines := 0
	if len(rawLines) > maxLines {
		truncatedLines = len(rawLines) - maxLines
		rawLines = rawLines[len(rawLines)-maxLines:]
	}
	return map[string]any{"path": path, "lines": rawLines, "truncatedBytes": truncatedBytes, "truncatedLines": truncatedLines}
}

func (s *Service) BrowseDirectories(path string) ([]platform.DirectoryEntry, error) {
	if path == "" || path == "/" {
		return platform.BrowseRoots(), nil
	}
	return platform.BrowseDirectories(path)
}

func (s *Service) DiscoverWorkloads(ctx context.Context, kind model.JobType) (platform.WorkloadDiscovery, error) {
	return s.workloads.Discover(ctx, kind)
}

func (s *Service) LastRun(jobID string) time.Time {
	var latest time.Time
	for _, run := range s.queue.Snapshot() {
		if run.JobID == jobID && run.CreatedAt.After(latest) {
			latest = run.CreatedAt
		}
	}
	return latest
}

func (s *Service) SaveConfig(config model.Config) error {
	config = model.NormalizeConfig(config)
	for _, job := range config.Jobs {
		if job.Schedule.Cron != "" {
			if _, err := scheduler.Parse(job.Schedule.Cron); err != nil {
				return fmt.Errorf("validation: job %q schedule: %w", job.ID, err)
			}
		}
		if job.Schedule.Timezone != "" {
			if _, err := time.LoadLocation(job.Schedule.Timezone); err != nil {
				return fmt.Errorf("validation: job %q timezone: %w", job.ID, err)
			}
		}
	}
	for name, spec := range map[string]string{"check": config.Settings.CheckCron, "prune": config.Settings.PruneCron} {
		if spec != "" {
			if _, err := scheduler.Parse(spec); err != nil {
				return fmt.Errorf("validation: %s schedule: %w", name, err)
			}
		}
	}
	if err := s.store.SaveConfig(config); err != nil {
		return err
	}
	s.mu.Lock()
	s.config = config
	s.mu.Unlock()
	s.restic.Binary = config.Settings.ResticPath
	s.rsync.Binary = config.Settings.RsyncPath
	return nil
}

func (s *Service) PutSecret(id, kind, value string) error {
	if id == "" || value == "" {
		return errors.New("validation: id and value are required")
	}
	return s.secrets.Put(id, kind, value)
}

func (s *Service) DeleteSecret(id string) error { return s.secrets.Delete(id) }

func (s *Service) ChangeRepositoryPassword(ctx context.Context, repoID, newPassword string) error {
	if err := s.validatePasswordChange(newPassword); err != nil {
		return err
	}
	repo, ok := s.repository(repoID)
	if !ok {
		return errors.New("validation: unknown repository")
	}
	oldPassword, err := s.secrets.Get(repo.PasswordRef)
	if err != nil {
		return fmt.Errorf("authentication: load current repository password: %w", err)
	}
	if oldPassword == newPassword {
		return errors.New("validation: Das neue Repository-Passwort muss sich vom bisherigen Passwort unterscheiden")
	}
	mounted, err := s.mounts.Prepare(ctx, repo)
	if err != nil {
		return err
	}
	defer s.mounts.Cleanup(context.Background(), mounted)
	if err := s.restic.ChangePassword(ctx, mounted.Repository, oldPassword, newPassword); err != nil {
		return err
	}
	if err := s.secrets.Put(repo.PasswordRef, "restic-password", newPassword); err != nil {
		rollbackCtx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
		defer cancel()
		rollbackErr := s.restic.ChangePassword(rollbackCtx, mounted.Repository, newPassword, oldPassword)
		if rollbackErr != nil {
			return fmt.Errorf("internal: Neues Repository-Passwort wurde aktiviert, konnte aber nicht in URBM gespeichert werden; automatischer Rollback schlug ebenfalls fehl: %v; ursprünglicher Speicherfehler: %w", rollbackErr, err)
		}
		return fmt.Errorf("internal: Neues Passwort konnte nicht in URBM gespeichert werden; Repository wurde erfolgreich auf das bisherige Passwort zurückgesetzt: %w", err)
	}
	if _, err := s.restic.Snapshots(ctx, mounted.Repository); err != nil {
		rollbackCtx, cancel := context.WithTimeout(context.Background(), 5*time.Minute)
		defer cancel()
		rollbackErr := s.restic.ChangePassword(rollbackCtx, mounted.Repository, newPassword, oldPassword)
		secretErr := s.secrets.Put(repo.PasswordRef, "restic-password", oldPassword)
		if rollbackErr != nil || secretErr != nil {
			return fmt.Errorf("internal: Prüfung des neuen Passworts fehlgeschlagen und Rollback war unvollständig (repository: %v, secret: %v): %w", rollbackErr, secretErr, err)
		}
		return fmt.Errorf("authentication: Prüfung des neuen Passworts fehlgeschlagen; Änderung wurde zurückgesetzt: %w", err)
	}
	return nil
}

func (s *Service) AdoptRepositoryPassword(ctx context.Context, repoID, password string) error {
	if err := s.validatePasswordChange(password); err != nil {
		return err
	}
	repo, ok := s.repository(repoID)
	if !ok {
		return errors.New("validation: unknown repository")
	}
	mounted, err := s.mounts.Prepare(ctx, repo)
	if err != nil {
		return err
	}
	defer s.mounts.Cleanup(context.Background(), mounted)
	if err := s.restic.TestPassword(ctx, mounted.Repository, password); err != nil {
		return fmt.Errorf("authentication: Das angegebene Passwort wurde vom Repository nicht akzeptiert: %w", err)
	}
	return s.secrets.Put(repo.PasswordRef, "restic-password", password)
}

func (s *Service) validatePasswordChange(password string) error {
	if len(password) < 8 {
		return errors.New("validation: Das Repository-Passwort muss mindestens 8 Zeichen lang sein")
	}
	for _, run := range s.queue.Snapshot() {
		if run.Status == "queued" || run.Status == "running" || run.Status == "paused" {
			return errors.New("validation: Repository-Passwörter können nur geändert werden, wenn keine Aufgabe wartet, läuft oder pausiert ist")
		}
	}
	return nil
}

func (s *Service) EnqueueJob(jobID string, priority int) (model.Run, error) {
	job, ok := s.job(jobID)
	if !ok {
		return model.Run{}, errors.New("validation: unknown job")
	}
	taskType := "backup"
	if job.Type == model.JobRsync {
		taskType = "rsync"
	}
	run := model.Run{SchemaVersion: model.SchemaVersion, ID: newID(), JobID: job.ID, TaskType: taskType, Priority: priority, CreatedAt: time.Now().UTC()}
	return run, s.queue.Enqueue(run)
}

func (s *Service) EnqueueScheduled(job model.Job) error {
	_, err := s.EnqueueJob(job.ID, 0)
	return err
}

func (s *Service) EnqueueMaintenance(repoID, action string) (model.Run, error) {
	if action != "check" && action != "prune" {
		return model.Run{}, errors.New("validation: invalid maintenance action")
	}
	if _, ok := s.repository(repoID); !ok {
		return model.Run{}, errors.New("validation: unknown repository")
	}
	run := model.Run{SchemaVersion: model.SchemaVersion, ID: newID(), JobID: repoID, TaskType: action, Priority: 5, CreatedAt: time.Now().UTC()}
	return run, s.queue.Enqueue(run)
}

func (s *Service) EnqueueRestore(task model.RestoreTask) (model.Run, error) {
	config := s.Config()
	target, err := platform.ValidateRestoreTarget(task.Target, config.Settings.RestoreRoot, task.InPlace, task.Confirmed)
	if err != nil {
		return model.Run{}, err
	}
	task.Target = target
	if task.ID == "" {
		task.ID = newID()
	}
	task.SchemaVersion = model.SchemaVersion
	encoded := strings.Join(append([]string{task.RepositoryID, task.SnapshotID, task.Target, fmt.Sprint(task.InPlace)}, task.Includes...), "\x00")
	run := model.Run{SchemaVersion: model.SchemaVersion, ID: task.ID, JobID: encoded, TaskType: "restore", Priority: 10, CreatedAt: time.Now().UTC()}
	return run, s.queue.Enqueue(run)
}

func (s *Service) Cancel(runID string) bool { return s.queue.Cancel(runID) }
func (s *Service) ClearRunHistory() int     { return s.queue.ClearHistory() }

func (s *Service) Pause(runID string) error {
	runner := s.restic.Pause
	resume := s.restic.Resume
	if s.runTaskType(runID) == "rsync" {
		runner, resume = s.rsync.Pause, s.rsync.Resume
	}
	if err := runner(runID); err != nil {
		return fmt.Errorf("pause: %w", err)
	}
	if !s.queue.SetPaused(runID, true) {
		_ = resume(runID)
		return errors.New("pause: run is not currently running")
	}
	return nil
}

func (s *Service) Resume(runID string) error {
	runner := s.restic.Resume
	if s.runTaskType(runID) == "rsync" {
		runner = s.rsync.Resume
	}
	if err := runner(runID); err != nil {
		return fmt.Errorf("resume: %w", err)
	}
	if !s.queue.SetPaused(runID, false) {
		return errors.New("resume: run is not currently paused")
	}
	return nil
}

func (s *Service) runTaskType(runID string) string {
	for _, run := range s.queue.Snapshot() {
		if run.ID == runID {
			return run.TaskType
		}
	}
	return ""
}

func (s *Service) Snapshots(ctx context.Context, repoID string) ([]model.Snapshot, error) {
	repo, ok := s.repository(repoID)
	if !ok {
		return nil, errors.New("validation: unknown repository")
	}
	mounted, err := s.mounts.Prepare(ctx, repo)
	if err != nil {
		return nil, err
	}
	defer s.mounts.Cleanup(context.Background(), mounted)
	return s.restic.Snapshots(ctx, mounted.Repository)
}

func (s *Service) RepositoryStats(ctx context.Context) []model.RepositoryStats {
	config := s.Config()
	result := make([]model.RepositoryStats, 0, len(config.Repositories))
	for _, repo := range config.Repositories {
		item := model.RepositoryStats{RepositoryID: repo.ID, RepositoryName: repo.Name}
		mounted, err := s.mounts.Prepare(ctx, repo)
		if err == nil {
			var stored, files restic.Stats
			stored, err = s.restic.Stats(ctx, mounted.Repository, "raw-data")
			if err == nil {
				files, err = s.restic.Stats(ctx, mounted.Repository, "restore-size")
			}
			if err == nil {
				var snapshots []model.Snapshot
				snapshots, err = s.restic.Snapshots(ctx, mounted.Repository)
				item.StoredBytes = stored.TotalSize
				item.FileCount = files.TotalFileCount
				item.SnapshotCount = len(snapshots)
			}
			if cleanupErr := s.mounts.Cleanup(context.Background(), mounted); err == nil && cleanupErr != nil {
				err = cleanupErr
			}
		}
		if err != nil {
			item.Error = err.Error()
		}
		result = append(result, item)
		if ctx.Err() != nil {
			break
		}
	}
	return result
}

func (s *Service) SnapshotFiles(ctx context.Context, repoID, snapshot, path string) ([]map[string]any, error) {
	repo, ok := s.repository(repoID)
	if !ok {
		return nil, errors.New("validation: unknown repository")
	}
	mounted, err := s.mounts.Prepare(ctx, repo)
	if err != nil {
		return nil, err
	}
	defer s.mounts.Cleanup(context.Background(), mounted)
	return s.restic.List(ctx, mounted.Repository, snapshot, path)
}

func (s *Service) TestRepository(ctx context.Context, repoID string, initialize bool) error {
	repo, ok := s.repository(repoID)
	if !ok {
		return errors.New("validation: unknown repository")
	}
	mounted, err := s.mounts.Prepare(ctx, repo)
	if err != nil {
		return err
	}
	defer s.mounts.Cleanup(context.Background(), mounted)
	if initialize {
		return s.restic.Init(ctx, mounted.Repository)
	}
	if mounted.Repository.Type != model.RepositorySFTP {
		if err := validateExistingRepositoryPath(mounted.Repository.Location); err != nil {
			return err
		}
	}
	_, err = s.restic.Snapshots(ctx, mounted.Repository)
	return err
}

func validateExistingRepositoryPath(location string) error {
	configPath := filepath.Join(location, "config")
	if info, err := os.Stat(configPath); err == nil && !info.IsDir() {
		return nil
	}
	entries, readErr := os.ReadDir(location)
	if readErr != nil {
		return fmt.Errorf("repository: Repository-Ordner %q kann nicht gelesen werden: %w", location, readErr)
	}
	candidates := make([]string, 0, 3)
	for _, entry := range entries {
		if !entry.IsDir() {
			continue
		}
		candidate := filepath.Join(location, entry.Name())
		if info, err := os.Stat(filepath.Join(candidate, "config")); err == nil && !info.IsDir() {
			candidates = append(candidates, candidate)
			if len(candidates) == 3 {
				break
			}
		}
	}
	if len(candidates) > 0 {
		return fmt.Errorf("repository: Im gewählten Ordner liegt keine Restic-config. Verwende direkt den Repository-Ordner: %s", strings.Join(candidates, ", "))
	}
	return fmt.Errorf("repository: Im gewählten Ordner %q wurde keine Restic-config gefunden. Ein vorhandenes Repository muss direkt auf den Ordner zeigen, der config, data, index, keys und snapshots enthält", location)
}

func (s *Service) UnlockRepository(ctx context.Context, repoID string) error {
	repo, ok := s.repository(repoID)
	if !ok {
		return errors.New("validation: unknown repository")
	}
	mounted, err := s.mounts.Prepare(ctx, repo)
	if err != nil {
		return err
	}
	defer s.mounts.Cleanup(context.Background(), mounted)
	return s.restic.Unlock(ctx, mounted.Repository)
}

func (s *Service) ForceUnlockRepository(ctx context.Context, repoID string) error {
	repo, ok := s.repository(repoID)
	if !ok {
		return errors.New("validation: unknown repository")
	}
	mounted, err := s.mounts.Prepare(ctx, repo)
	if err != nil {
		return err
	}
	defer s.mounts.Cleanup(context.Background(), mounted)
	return s.restic.ForceUnlock(ctx, mounted.Repository)
}

func (s *Service) execute(ctx context.Context, run model.Run) (result model.Run) {
	ctx = restic.WithRunID(ctx, run.ID)
	maintenanceRepoName := ""
	defer func() {
		if result.TaskType == "prune" && (result.Status == "success" || result.Status == "warning" || result.Status == "failed") {
			go s.notify("Repository-Bereinigung", maintenanceRepoName, result)
		}
	}()
	if run.TaskType == "backup" {
		return s.executeBackup(ctx, run)
	}
	if run.TaskType == "rsync" {
		return s.executeRsync(ctx, run)
	}
	if run.TaskType == "restore" {
		return s.executeRestore(ctx, run)
	}
	repo, ok := s.repository(run.JobID)
	if !ok {
		return failed(run, "validation", "repository no longer exists")
	}
	maintenanceRepoName = repo.Name
	var pruneStatsBefore restic.Stats
	pruneStatsBeforeOK := false
	live := run
	appendLiveLog(&live, strings.ToUpper(run.TaskType)+" wird vorbereitet")
	s.updateActive(live)
	mounted, err := s.mounts.Prepare(ctx, repo)
	if err != nil {
		failedRun := failedError(run, err)
		appendLiveLog(&live, "Vorbereitung fehlgeschlagen: "+err.Error())
		failedRun.LiveLog = live.LiveLog
		return failedRun
	}
	defer s.mounts.Cleanup(context.Background(), mounted)
	appendLiveLog(&live, "Repository bereit; Restic "+run.TaskType+" läuft")
	s.updateActive(live)
	if run.TaskType == "check" {
		err = s.restic.Check(ctx, mounted.Repository)
	} else {
		if stats, statsErr := s.restic.Stats(ctx, mounted.Repository, "raw-data"); statsErr == nil {
			pruneStatsBefore, pruneStatsBeforeOK = stats, true
			appendLiveLog(&live, "Repository-Größe vor Bereinigung: "+formatBytes(stats.TotalSize))
			s.updateActive(live)
		} else {
			appendLiveLog(&live, "Repository-Größe vor Bereinigung konnte nicht ermittelt werden: "+statsErr.Error())
		}
		err = s.restic.Prune(ctx, mounted.Repository)
	}
	if err != nil {
		failedRun := failedError(run, err)
		appendLiveLog(&live, strings.ToUpper(run.TaskType)+" fehlgeschlagen: "+err.Error())
		failedRun.LiveLog = live.LiveLog
		return failedRun
	}
	run.Status, run.Message = "success", run.TaskType+" completed"
	if run.TaskType == "prune" && pruneStatsBeforeOK {
		run.RepositoryBefore = pruneStatsBefore.TotalSize
		if stats, statsErr := s.restic.Stats(ctx, mounted.Repository, "raw-data"); statsErr == nil {
			run.RepositoryAfter = stats.TotalSize
			if run.RepositoryBefore > run.RepositoryAfter {
				run.RepositoryFreed = run.RepositoryBefore - run.RepositoryAfter
			}
			appendLiveLog(&live, fmt.Sprintf("Repository-Größe nach Bereinigung: %s, freigegeben: %s", formatBytes(run.RepositoryAfter), formatBytes(run.RepositoryFreed)))
		} else {
			appendLiveLog(&live, "Repository-Größe nach Bereinigung konnte nicht ermittelt werden: "+statsErr.Error())
		}
	}
	appendLiveLog(&live, strings.ToUpper(run.TaskType)+" abgeschlossen")
	run.LiveLog = live.LiveLog
	return run
}

func (s *Service) executeRsync(ctx context.Context, run model.Run) (result model.Run) {
	job, ok := s.job(run.JobID)
	if !ok {
		return failed(run, "validation", "job no longer exists")
	}
	defer func() { result = s.finishJob(run, job, result) }()
	live := run
	appendLiveLog(&live, fmt.Sprintf("Rsync-Kopie wird vorbereitet: %d Quelle(n) nach %s", len(job.Sources), job.Rsync.Target))
	s.updateActive(live)
	for _, source := range job.Sources {
		if _, err := os.Stat(source.Path); err != nil {
			return failed(run, "source", "source unavailable: "+source.Path)
		}
	}
	if info, err := os.Stat(job.Rsync.Target); err != nil || !info.IsDir() {
		if err == nil {
			err = errors.New("target is not a directory")
		}
		return failed(run, "environment", "rsync target unavailable: "+err.Error())
	}
	if !job.Rsync.DryRun && !job.Rsync.SkipSpaceCheck {
		appendLiveLog(&live, "Rsync-Vorabprüfung ermittelt den tatsächlich zu übertragenden Datenumfang")
		s.updateActive(live)
		estimate, err := s.rsync.Estimate(ctx, run.ID, job)
		if err != nil {
			return failedError(run, err)
		}
		available, err := availableBytes(job.Rsync.Target)
		if err != nil {
			return failed(run, "environment", "read rsync target capacity: "+err.Error())
		}
		appendLiveLog(&live, fmt.Sprintf("Rsync-Vorabprüfung: %s zu übertragen, %s am Ziel frei", formatBytes(estimate.Bytes), formatBytes(available)))
		s.updateActive(live)
		if estimate.Bytes > available {
			return failed(run, "environment", fmt.Sprintf("Rsync-Ziel hat nicht genügend freien Speicher: ungefähr %s für %d Dateien erforderlich, aber nur %s frei unter %s", formatBytes(estimate.Bytes), estimate.Files, formatBytes(available), job.Rsync.Target))
		}
	}
	started := time.Now()
	lastLoggedPercent := -10
	summary, err := s.rsync.Run(ctx, run.ID, job, func(progress rsync.Progress) {
		if progress.Percent > 0 || progress.Bytes > 0 {
			live.ProgressPercent = progress.Percent
			live.ProgressBytes = progress.Bytes
			if progress.Percent > 0 {
				live.ProgressTotal = int64(float64(progress.Bytes) / (progress.Percent / 100))
			}
			if elapsed := time.Since(started).Seconds(); elapsed > 0 {
				live.BytesPerSecond = float64(progress.Bytes) / elapsed
			}
			percent := int(progress.Percent)
			if percent >= lastLoggedPercent+10 {
				appendLiveLog(&live, fmt.Sprintf("Rsync-Fortschritt: %.0f%%, %s übertragen", progress.Percent, formatBytes(progress.Bytes)))
				lastLoggedPercent = percent - percent%10
			}
		}
		if progress.CurrentFile != "" {
			live.CurrentFile = progress.CurrentFile
		}
		s.updateActive(live)
	})
	if err != nil {
		appendLiveLog(&live, "Rsync-Kopie fehlgeschlagen: "+err.Error())
		failedRun := failedError(run, err)
		failedRun.LiveLog = live.LiveLog
		return failedRun
	}
	run.Status, run.Message = "success", "rsync completed"
	run.BytesAdded, run.BytesProcessed = summary.Bytes, summary.Bytes
	run.FilesNew, run.FilesProcessed = summary.Files, summary.Files
	run.ProgressPercent, run.ProgressBytes, run.ProgressTotal = 100, summary.Bytes, summary.Bytes
	run.ProgressFiles, run.ProgressFileTotal = summary.Files, summary.Files
	run.BytesPerSecond, run.CurrentFile = live.BytesPerSecond, live.CurrentFile
	appendLiveLog(&live, fmt.Sprintf("Rsync-Kopie abgeschlossen: %d Dateien, %s", summary.Files, formatBytes(summary.Bytes)))
	run.LiveLog = live.LiveLog
	return run
}

func availableBytes(path string) (int64, error) {
	var stats syscall.Statfs_t
	if err := syscall.Statfs(path, &stats); err != nil {
		return 0, err
	}
	return int64(stats.Bavail) * int64(stats.Bsize), nil
}

func (s *Service) executeBackup(ctx context.Context, run model.Run) (result model.Run) {
	job, ok := s.job(run.JobID)
	if !ok {
		return failed(run, "validation", "job no longer exists")
	}
	defer func() { result = s.finishJob(run, job, result) }()
	repo, ok := s.repository(job.RepositoryID)
	if !ok {
		return failed(run, "validation", "repository no longer exists")
	}
	live := run
	appendLiveLog(&live, "Backup wird vorbereitet")
	s.updateActive(live)
	mounted, err := s.mounts.Prepare(ctx, repo)
	if err != nil {
		return failedError(run, err)
	}
	defer func() {
		if err := s.mounts.Cleanup(context.Background(), mounted); err != nil && result.Status == "success" {
			result.Status, result.ErrorCode, result.Message = "warning", "connectivity", result.Message+"; repository unmount failed: "+err.Error()
		}
	}()
	previousSnapshots, previousSnapshotErr := s.jobSnapshots(ctx, mounted.Repository, job.ID)
	previousSnapshot, previousSnapshotOK := latestSnapshot(previousSnapshots)
	if previousSnapshotErr != nil {
		appendLiveLog(&live, "Änderungsliste: vorheriger Snapshot konnte nicht ermittelt werden: "+previousSnapshotErr.Error())
		s.updateActive(live)
	}
	prepared, err := s.workloads.Prepare(ctx, job)
	if err != nil {
		return failedError(run, err)
	}
	defer func() {
		if err := s.workloads.Cleanup(context.Background(), prepared); err != nil {
			if result.Status == "success" {
				result.Status, result.ErrorCode = "warning", "source"
			}
			result.Message += "; workload restart failed: " + err.Error()
		}
	}()
	if job.Type == model.JobFlash && len(prepared.Sources) == 0 {
		prepared.Sources = []string{"/boot"}
	}
	for _, path := range prepared.Sources {
		if _, err := os.Stat(path); err != nil {
			return failed(run, "source", "source unavailable: "+path)
		}
	}
	var imageSummary restic.Summary
	if job.Type == model.JobFlash && job.FlashImage {
		device, err := s.workloads.FlashDevice(ctx)
		if err != nil {
			return failedError(run, err)
		}
		image, err := os.Open(device)
		if err != nil {
			return failed(run, "source", "open USB flash device: "+err.Error())
		}
		appendLiveLog(&live, "USB-Rohabbild wird direkt aus "+device+" zu Restic übertragen")
		s.updateActive(live)
		imageStarted := time.Now()
		imageSummary, err = s.restic.BackupImage(ctx, mounted.Repository, job, image, func(progress restic.Progress) {
			live.ProgressPercent = 0
			live.ProgressBytes, live.ProgressTotal = progress.BytesDone, 0
			live.ProgressFiles, live.ProgressFileTotal = 0, 0
			live.SecondsRemaining = 0
			if elapsed := time.Since(imageStarted).Seconds(); elapsed > 0 {
				live.BytesPerSecond = float64(progress.BytesDone) / elapsed
			}
			live.CurrentFile = "urbm-usb-flash.img"
			s.updateActive(live)
		})
		closeErr := image.Close()
		if err != nil {
			appendLiveLog(&live, "USB-Rohabbild fehlgeschlagen: "+err.Error())
			failedRun := failedError(run, err)
			failedRun.LiveLog = live.LiveLog
			return failedRun
		}
		if closeErr != nil {
			return failed(run, "source", "close USB flash device: "+closeErr.Error())
		}
		appendLiveLog(&live, "USB-Rohabbild als urbm-usb-flash.img gesichert")
		s.updateActive(live)
	}
	appendLiveLog(&live, fmt.Sprintf("Restic-Backup gestartet: %d Quelle(n)", len(prepared.Sources)))
	s.updateActive(live)
	lastLoggedPercent := -5
	started := time.Now()
	summary, err := s.restic.Backup(ctx, mounted.Repository, job, prepared.Sources, func(progress restic.Progress) {
		live.ProgressPercent = progress.PercentDone * 100
		live.ProgressBytes, live.ProgressTotal = progress.BytesDone, progress.TotalBytes
		live.ProgressFiles, live.ProgressFileTotal = progress.FilesDone, progress.TotalFiles
		live.SecondsRemaining = progress.SecondsRemaining
		if elapsed := time.Since(started).Seconds(); elapsed > 0 {
			live.BytesPerSecond = float64(progress.BytesDone) / elapsed
		}
		if len(progress.CurrentFiles) > 0 {
			live.CurrentFile = progress.CurrentFiles[len(progress.CurrentFiles)-1]
		}
		percent := int(live.ProgressPercent)
		if percent >= lastLoggedPercent+5 {
			appendLiveLog(&live, fmt.Sprintf("Fortschritt: %.1f%%, %d/%d Dateien, %s/%s", live.ProgressPercent, live.ProgressFiles, live.ProgressFileTotal, formatBytes(live.ProgressBytes), formatBytes(live.ProgressTotal)))
			lastLoggedPercent = percent - percent%5
		}
		s.updateActive(live)
	})
	if err != nil {
		appendLiveLog(&live, "Backup fehlgeschlagen: "+err.Error())
		failedRun := failedError(run, err)
		failedRun.LiveLog = live.LiveLog
		return failedRun
	}
	if previousSnapshotOK && summary.SnapshotID != "" {
		s.appendBackupDiffLog(ctx, &live, mounted.Repository, previousSnapshot, summary.SnapshotID)
		s.updateActive(live)
	}
	appendLiveLog(&live, "Backup-Daten übertragen; Aufbewahrung wird angewendet")
	s.updateActive(live)
	if err := s.restic.Forget(ctx, mounted.Repository, job); err != nil {
		run.Status, run.Message, run.ErrorCode = "warning", "backup succeeded but retention failed: "+err.Error(), "repository"
	} else {
		run.Status, run.Message = "success", "backup completed"
		if summary.SnapshotID != "" && previousSnapshotErr == nil {
			run.RetentionBefore = len(previousSnapshots)
			if remainingSnapshots, remainingErr := s.jobSnapshots(ctx, mounted.Repository, job.ID); remainingErr == nil {
				run.RetentionAfter = len(remainingSnapshots)
				expectedAfter := run.RetentionBefore + 1
				if expectedAfter > run.RetentionAfter {
					run.RetentionRemoved = expectedAfter - run.RetentionAfter
				}
				appendLiveLog(&live, fmt.Sprintf("Aufbewahrung angewendet: %d Snapshot(s) entfernt, %d bleiben", run.RetentionRemoved, run.RetentionAfter))
				s.updateActive(live)
			} else {
				appendLiveLog(&live, "Aufbewahrung angewendet; verbleibende Snapshots konnten nicht ermittelt werden: "+remainingErr.Error())
			}
		}
	}
	run.SnapshotID = summary.SnapshotID
	run.BytesAdded = summary.DataAdded + imageSummary.DataAdded
	run.FilesNew = summary.FilesNew + imageSummary.FilesNew
	run.FilesChanged = summary.FilesChanged + imageSummary.FilesChanged
	run.BytesProcessed = summary.TotalBytesProcessed + imageSummary.TotalBytesProcessed
	run.FilesProcessed = summary.TotalFilesProcessed + imageSummary.TotalFilesProcessed
	run.ProgressPercent, run.ProgressBytes, run.ProgressTotal = 100, live.ProgressBytes, live.ProgressTotal
	run.ProgressFiles, run.ProgressFileTotal = live.ProgressFiles, live.ProgressFileTotal
	run.BytesPerSecond, run.SecondsRemaining, run.CurrentFile = live.BytesPerSecond, 0, live.CurrentFile
	appendLiveLog(&live, "Backup abgeschlossen")
	run.LiveLog = live.LiveLog
	return run
}

func (s *Service) jobSnapshots(ctx context.Context, repo model.Repository, jobID string) ([]model.Snapshot, error) {
	snapshots, err := s.restic.Snapshots(ctx, repo)
	if err != nil {
		return nil, err
	}
	tag := "job:" + jobID
	filtered := make([]model.Snapshot, 0, len(snapshots))
	for _, snapshot := range snapshots {
		if hasTag(snapshot.Tags, tag) && !hasTag(snapshot.Tags, "usb-image") {
			filtered = append(filtered, snapshot)
		}
	}
	return filtered, nil
}

func latestSnapshot(snapshots []model.Snapshot) (model.Snapshot, bool) {
	var latest model.Snapshot
	found := false
	for _, snapshot := range snapshots {
		if !found || snapshot.Time.After(latest.Time) {
			latest = snapshot
			found = true
		}
	}
	return latest, found
}

func hasTag(tags []string, wanted string) bool {
	for _, tag := range tags {
		if tag == wanted {
			return true
		}
	}
	return false
}

func (s *Service) appendBackupDiffLog(ctx context.Context, run *model.Run, repo model.Repository, previous model.Snapshot, currentSnapshotID string) {
	const limit = 50
	diff, err := s.restic.Diff(ctx, repo, previous.ID, currentSnapshotID, limit)
	if err != nil {
		appendLiveLog(run, "Änderungsliste konnte nicht erstellt werden: "+err.Error())
		return
	}
	if diff.Total == 0 {
		appendLiveLog(run, "Änderungsliste gegenüber "+shortID(previous.ID)+": keine geänderten Pfade erkannt")
		return
	}
	if diff.Total > len(diff.Entries) {
		appendLiveLog(run, fmt.Sprintf("Änderungsliste gegenüber %s: letzte %d von %d geänderten Pfaden", shortID(previous.ID), len(diff.Entries), diff.Total))
	} else {
		appendLiveLog(run, fmt.Sprintf("Änderungsliste gegenüber %s: %d geänderte Pfade", shortID(previous.ID), diff.Total))
	}
	for _, entry := range diff.Entries {
		appendLiveLog(run, fmt.Sprintf("%s %s", diffKindLabel(entry.Kind), entry.Path))
	}
}

func diffKindLabel(kind string) string {
	switch kind {
	case "new":
		return "NEU"
	case "removed":
		return "GELÖSCHT"
	case "modified":
		return "GEÄNDERT"
	default:
		return strings.ToUpper(kind)
	}
}

func (s *Service) updateActive(run model.Run) {
	s.queue.UpdateActive(run.ID, func(active *model.Run) {
		active.ProgressPercent, active.ProgressBytes, active.ProgressTotal = run.ProgressPercent, run.ProgressBytes, run.ProgressTotal
		active.ProgressFiles, active.ProgressFileTotal = run.ProgressFiles, run.ProgressFileTotal
		active.BytesPerSecond, active.SecondsRemaining = run.BytesPerSecond, run.SecondsRemaining
		active.CurrentFile = run.CurrentFile
		active.LiveLog = append([]string(nil), run.LiveLog...)
	})
}

func appendLiveLog(run *model.Run, message string) {
	line := time.Now().Format("15:04:05") + "  " + strings.ReplaceAll(strings.TrimSpace(message), "\n", " ")
	run.LiveLog = append(run.LiveLog, line)
	if len(run.LiveLog) > 100 {
		run.LiveLog = run.LiveLog[len(run.LiveLog)-100:]
	}
}

func (s *Service) executeRestore(ctx context.Context, run model.Run) model.Run {
	live := run
	appendLiveLog(&live, "Restore wird vorbereitet")
	s.updateActive(live)
	parts := strings.Split(run.JobID, "\x00")
	if len(parts) < 4 {
		return failed(run, "internal", "invalid restore payload")
	}
	repo, ok := s.repository(parts[0])
	if !ok {
		return failed(run, "validation", "repository no longer exists")
	}
	task := model.RestoreTask{SchemaVersion: model.SchemaVersion, ID: run.ID, RepositoryID: parts[0], SnapshotID: parts[1], Target: parts[2], InPlace: parts[3] == "true", Confirmed: true, Includes: parts[4:]}
	if err := os.MkdirAll(task.Target, 0750); err != nil {
		failedRun := failedError(run, err)
		appendLiveLog(&live, "Restore-Ziel konnte nicht vorbereitet werden: "+err.Error())
		failedRun.LiveLog = live.LiveLog
		return failedRun
	}
	mounted, err := s.mounts.Prepare(ctx, repo)
	if err != nil {
		failedRun := failedError(run, err)
		appendLiveLog(&live, "Repository konnte nicht vorbereitet werden: "+err.Error())
		failedRun.LiveLog = live.LiveLog
		return failedRun
	}
	defer s.mounts.Cleanup(context.Background(), mounted)
	appendLiveLog(&live, "Restic stellt den Snapshot wieder her")
	s.updateActive(live)
	lastLoggedPercent := -10
	started := time.Now()
	if err := s.restic.Restore(ctx, mounted.Repository, task, func(progress restic.Progress) {
		live.ProgressPercent = progress.PercentDone * 100
		live.ProgressBytes, live.ProgressTotal = progress.BytesDone, progress.TotalBytes
		live.ProgressFiles, live.ProgressFileTotal = progress.FilesDone, progress.TotalFiles
		live.SecondsRemaining = progress.SecondsRemaining
		if elapsed := time.Since(started).Seconds(); elapsed > 0 {
			live.BytesPerSecond = float64(progress.BytesDone) / elapsed
		}
		if len(progress.CurrentFiles) > 0 {
			live.CurrentFile = progress.CurrentFiles[len(progress.CurrentFiles)-1]
		}
		percent := int(live.ProgressPercent)
		if percent >= lastLoggedPercent+10 {
			appendLiveLog(&live, fmt.Sprintf("Restore-Fortschritt: %.1f%%, %d/%d Dateien, %s/%s", live.ProgressPercent, live.ProgressFiles, live.ProgressFileTotal, formatBytes(live.ProgressBytes), formatBytes(live.ProgressTotal)))
			lastLoggedPercent = percent - percent%10
		}
		s.updateActive(live)
	}); err != nil {
		failedRun := failedError(run, err)
		appendLiveLog(&live, "Restore fehlgeschlagen: "+err.Error())
		failedRun.LiveLog = live.LiveLog
		return failedRun
	}
	run.Status, run.Message = "success", "restore completed"
	run.ProgressPercent, run.ProgressBytes, run.ProgressTotal = 100, live.ProgressBytes, live.ProgressTotal
	run.ProgressFiles, run.ProgressFileTotal = live.ProgressFiles, live.ProgressFileTotal
	run.BytesPerSecond, run.SecondsRemaining, run.CurrentFile = live.BytesPerSecond, 0, live.CurrentFile
	appendLiveLog(&live, "Restore abgeschlossen")
	run.LiveLog = live.LiveLog
	return run
}

func (s *Service) finishJob(run model.Run, job model.Job, result model.Run) model.Run {
	if result.Status == "success" || result.Status == "warning" || result.Status == "failed" {
		repoName := ""
		if job.RepositoryID != "" {
			if repo, ok := s.repository(job.RepositoryID); ok {
				repoName = repo.Name
			}
		}
		go s.notify(job.Name, repoName, result)
	}
	return result
}

func (s *Service) notify(name, repository string, run model.Run) {
	title := notificationTitle(name, repository, run)
	message := notificationMessage(name, repository, run, time.Now().UTC())
	for _, target := range s.Config().Notifications {
		for _, event := range target.Events {
			if event == run.Status {
				if err := s.notifier.Send(context.Background(), target, title, message, run.Status); err != nil {
					s.log.Warn("send notification failed", "target", target.ID, "type", target.Type, "error", err)
				}
				break
			}
		}
	}
}

func notificationTitle(name, repository string, run model.Run) string {
	icon := map[string]string{"success": "✅", "warning": "⚠️", "failed": "❌"}[run.Status]
	if icon == "" {
		icon = "ℹ️"
	}
	task := map[string]string{"backup": "Backup", "rsync": "Rsync", "restore": "Restore", "check": "Prüfung", "prune": "Bereinigung"}[run.TaskType]
	if task == "" {
		task = run.TaskType
		if task != "" {
			task = strings.ToUpper(task[:1]) + task[1:]
		}
	}
	if repository != "" {
		return fmt.Sprintf("%s URBM %s: %s", icon, task, repository)
	}
	return fmt.Sprintf("%s URBM %s: %s", icon, task, name)
}

func notificationMessage(name, repository string, run model.Run, now time.Time) string {
	status := map[string]string{"success": "Erfolgreich", "warning": "Mit Warnung", "failed": "Fehlgeschlagen"}[run.Status]
	if status == "" {
		status = run.Status
	}
	summaryTitle := map[string]string{"backup": "📊 Backup-Zusammenfassung", "rsync": "📊 Rsync-Zusammenfassung", "restore": "📊 Restore-Zusammenfassung", "prune": "🧹 Prune-Zusammenfassung", "check": "🔎 Prüfungs-Zusammenfassung"}[run.TaskType]
	if summaryTitle == "" {
		summaryTitle = "📊 URBM-Zusammenfassung"
	}
	lines := []string{summaryTitle, ""}
	if repository != "" {
		lines = append(lines, "📦 Repository: "+repository)
	}
	if name != "" {
		lines = append(lines, "🗂️ Job: "+name)
	}
	if run.TaskType == "backup" && run.SnapshotID != "" {
		lines = append(lines, "📸 Snapshot: "+shortID(run.SnapshotID))
	}
	if run.StartedAt != nil {
		finished := now
		if run.FinishedAt != nil {
			finished = *run.FinishedAt
		}
		if duration := finished.Sub(*run.StartedAt); duration >= 0 {
			lines = append(lines, "", "⏱️ Dauer: "+formatDuration(duration))
		}
	}
	if (run.TaskType == "backup" || run.TaskType == "rsync") && (run.Status == "success" || run.Status == "warning") {
		lines = append(lines,
			fmt.Sprintf("📂 Dateien: %d verarbeitet", run.FilesProcessed),
			"💾 Verarbeitet: "+formatBytes(run.BytesProcessed),
		)
		if run.TaskType == "backup" {
			lines = append(lines,
				fmt.Sprintf("➕ Neu: %d Dateien", run.FilesNew),
				fmt.Sprintf("✏️ Geändert: %d Dateien", run.FilesChanged),
				"📤 Neu gespeichert: "+formatBytes(run.BytesAdded),
			)
		} else {
			lines = append(lines, "📤 Kopiert: "+formatBytes(run.BytesAdded))
		}
	}
	if run.TaskType == "backup" && run.RetentionAfter > 0 {
		lines = append(lines,
			"",
			fmt.Sprintf("🧹 Aufbewahrung: %d Snapshot(s) entfernt", run.RetentionRemoved),
			fmt.Sprintf("📸 Verbleibende Snapshots: %d", run.RetentionAfter),
		)
	}
	if run.TaskType == "prune" && run.RepositoryBefore > 0 && run.RepositoryAfter > 0 {
		lines = append(lines,
			"",
			"📦 Vorher: "+formatBytes(run.RepositoryBefore),
			"📦 Nachher: "+formatBytes(run.RepositoryAfter),
			"🧹 Freigegeben: "+formatBytes(run.RepositoryFreed),
		)
	}
	lines = append(lines, "", "🔐 Ziel: "+notificationTargetLabel(run.TaskType), statusIcon(run.Status)+" Status: "+status)
	if run.Message != "" && run.Message != run.TaskType+" completed" {
		lines = append(lines, "⚠️ Details: "+run.Message)
	}
	return strings.Join(lines, "\n")
}

func statusIcon(status string) string {
	switch status {
	case "success":
		return "✅"
	case "warning":
		return "⚠️"
	case "failed":
		return "❌"
	default:
		return "ℹ️"
	}
}

func notificationTargetLabel(taskType string) string {
	switch taskType {
	case "backup":
		return "Restic Repository"
	case "rsync":
		return "Rsync Ziel"
	case "restore":
		return "Restore Ziel"
	case "check", "prune":
		return "Restic Repository"
	default:
		return "URBM"
	}
}

func shortID(value string) string {
	if len(value) <= 12 {
		return value
	}
	return value[:12]
}

func formatBytes(value int64) string {
	const unit = int64(1024)
	if value < unit {
		return fmt.Sprintf("%d B", value)
	}
	div, exp := unit, 0
	for n := value / unit; n >= unit && exp < 4; n /= unit {
		div *= unit
		exp++
	}
	return fmt.Sprintf("%.1f %ciB", float64(value)/float64(div), "KMGTPE"[exp])
}

func formatDuration(duration time.Duration) string {
	duration = duration.Round(time.Second)
	if duration < time.Minute {
		return fmt.Sprintf("%d Sekunden", int(duration.Seconds()))
	}
	hours := int(duration / time.Hour)
	minutes := int(duration/time.Minute) % 60
	seconds := int(duration/time.Second) % 60
	if hours > 0 {
		return fmt.Sprintf("%d Std. %d Min. %d Sek.", hours, minutes, seconds)
	}
	return fmt.Sprintf("%d Min. %d Sek.", minutes, seconds)
}

func (s *Service) job(id string) (model.Job, bool) {
	for _, item := range s.Config().Jobs {
		if item.ID == id {
			return item, true
		}
	}
	return model.Job{}, false
}
func (s *Service) repository(id string) (model.Repository, bool) {
	for _, item := range s.Config().Repositories {
		if item.ID == id {
			return item, true
		}
	}
	return model.Repository{}, false
}

func failed(run model.Run, code, message string) model.Run {
	run.Status, run.ErrorCode, run.Message = "failed", code, message
	return run
}
func failedError(run model.Run, err error) model.Run {
	code := "internal"
	message := err.Error()
	for _, candidate := range []string{"validation", "environment", "connectivity", "authentication", "repository", "source", "restic", "cancelled"} {
		if strings.HasPrefix(message, candidate+":") {
			code = candidate
			break
		}
	}
	if errors.Is(err, context.Canceled) {
		code = "cancelled"
		run.Status = "cancelled"
	} else {
		run.Status = "failed"
	}
	run.ErrorCode, run.Message = code, message
	return run
}
func newID() string {
	b := make([]byte, 12)
	if _, err := rand.Read(b); err != nil {
		return fmt.Sprintf("%d", time.Now().UnixNano())
	}
	return hex.EncodeToString(b)
}

func DefaultRestoreTarget(root string) string { return filepath.Join(root, newID()) }