URBM/internal/service/service.go

package service

import (
	"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) 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) execute(ctx context.Context, run model.Run) model.Run {
	ctx = restic.WithRunID(ctx, run.ID)
	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")
	}
	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 {
		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"
	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()
		}
	}()
	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
	}
	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"
	}
	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) 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)
	if err := s.restic.Restore(ctx, mounted.Repository, task); err != nil {
		failedRun := failedError(run, err)
		appendLiveLog(&live, "Restore fehlgeschlagen: "+err.Error())
		failedRun.LiveLog = live.LiveLog
		return failedRun
	}
	run.Status, run.Message, run.JobID = "success", "restore completed", ""
	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" {
		go s.notify(job.Name, result)
	}
	return result
}

func (s *Service) notify(name string, run model.Run) {
	message := notificationMessage(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, "URBM: "+name, message, run.Status); err != nil {
					s.log.Warn("send notification failed", "target", target.ID, "type", target.Type, "error", err)
				}
				break
			}
		}
	}
}

func notificationMessage(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
	}
	lines := []string{"Status: " + status}
	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,
			"Neu gespeichert: "+formatBytes(run.BytesAdded),
			fmt.Sprintf("Neue Dateien: %d", run.FilesNew),
			fmt.Sprintf("Geänderte Dateien: %d", run.FilesChanged),
			"Gesamtgröße dieses Backups: "+formatBytes(run.BytesProcessed),
			fmt.Sprintf("Dateien insgesamt: %d", run.FilesProcessed),
		)
		if run.SnapshotID != "" {
			lines = append(lines, "Snapshot: "+run.SnapshotID)
		}
	}
	if run.Message != "" && run.Message != "backup completed" && run.Message != "rsync completed" {
		lines = append(lines, "Details: "+run.Message)
	}
	return strings.Join(lines, "\n")
}

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()) }