URBM/internal/service/service.go

package service

import (
	"bytes"
	"context"
	"crypto/rand"
	"encoding/hex"
	"errors"
	"fmt"
	"log/slog"
	"os"
	"path/filepath"
	"sort"
	"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
	snapshotMu      sync.Mutex
	snapshotTree    map[string]snapshotTreeCache
	repositoryMu    sync.Mutex
	repositoryLocks map[string]chan struct{}
	persistRuns     chan []model.Run
	persistStop     chan struct{}
	persistDone     chan struct{}
	persistStopOnce sync.Once
	metricsMu       sync.RWMutex
	backupMetrics   []model.BackupMetric
	metricRecords   int
	metricsAppendOK bool
}

const (
	maxBackupMetrics       = 5000
	compactBackupMetricsAt = 5500
)

type SnapshotBrowserNode struct {
	Path        string `json:"path"`
	Name        string `json:"name"`
	Type        string `json:"type"`
	Size        int64  `json:"size,omitempty"`
	HasChildren bool   `json:"hasChildren"`
}

type SnapshotBrowserPage struct {
	Path      string                `json:"path"`
	Items     []SnapshotBrowserNode `json:"items"`
	Total     int                   `json:"total"`
	Cached    bool                  `json:"cached"`
	Generated time.Time             `json:"generated"`
}

type DashboardStatus struct {
	LastSuccessfulBackup *model.BackupMetric `json:"lastSuccessfulBackup,omitempty"`
	LastFailedBackup     *model.Run          `json:"lastFailedBackup,omitempty"`
	OverdueJobs          []OverdueBackupJob  `json:"overdueJobs"`
}

type OverdueBackupJob struct {
	JobID        string     `json:"jobId"`
	Name         string     `json:"name"`
	ScheduledAt  time.Time  `json:"scheduledAt"`
	LastBackupAt *time.Time `json:"lastBackupAt,omitempty"`
}

type snapshotTreeCache struct {
	generated time.Time
	total     int
	children  map[string][]SnapshotBrowserNode
}

type snapshotTreeNode struct {
	path     string
	name     string
	nodeType string
	size     int64
	children map[string]*snapshotTreeNode
}

type snapshotTreeBuilder struct {
	root  *snapshotTreeNode
	nodes map[string]*snapshotTreeNode
	total int
}

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,
		repositoryLocks: map[string]chan struct{}{}, persistRuns: make(chan []model.Run, 1), persistStop: make(chan struct{}), persistDone: make(chan struct{}),
	}
	if metrics, legacy, err := st.LoadBackupMetrics(); err != nil {
		log.Error("load backup metrics", "error", err)
	} else {
		s.backupMetrics = normalizeBackupMetrics(metrics)
		s.metricRecords = len(s.backupMetrics)
		if legacy || len(metrics) != len(s.backupMetrics) {
			if err := st.SaveBackupMetrics(s.backupMetrics); err != nil {
				log.Error("migrate backup metrics to append format", "error", err)
			} else {
				s.metricsAppendOK = true
			}
		} else {
			s.metricsAppendOK = true
		}
	}
	go s.runPersistence()
	s.queue = queue.New(s.execute, s.queuePersistence)
	if runs, err := st.LoadRuns(); err == nil {
		s.queue.RestoreHistory(runs)
	} else {
		log.Error("load run history", "error", err)
	}
	s.queuePersistence(s.queue.Snapshot())
	return s, nil
}

func (s *Service) RunQueue(ctx context.Context) { s.queue.Run(ctx) }
func (s *Service) Stop()                        { s.queue.Stop() }
func (s *Service) Wait(ctx context.Context) error {
	if err := s.queue.Wait(ctx); err != nil {
		return err
	}
	s.persistStopOnce.Do(func() { close(s.persistStop) })
	select {
	case <-s.persistDone:
		return nil
	case <-ctx.Done():
		return ctx.Err()
	}
}
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) BackupMetrics() []model.BackupMetric {
	s.metricsMu.RLock()
	defer s.metricsMu.RUnlock()
	metrics := make([]model.BackupMetric, len(s.backupMetrics))
	copy(metrics, s.backupMetrics)
	return metrics
}
func (s *Service) Logs() map[string]any {
	return map[string]any{
		"daemon": tailFile("/var/log/urbm.log", 1024*1024, 1000),
		"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 (s *Service) queuePersistence(runs []model.Run) {
	s.captureBackupMetrics(runs)
	select {
	case s.persistRuns <- runs:
		return
	default:
	}
	select {
	case <-s.persistRuns:
	default:
	}
	select {
	case s.persistRuns <- runs:
	default:
	}
}

func (s *Service) runPersistence() {
	defer close(s.persistDone)
	for {
		select {
		case runs := <-s.persistRuns:
			s.saveRunState(runs)
		case <-s.persistStop:
			var latest []model.Run
			for {
				select {
				case latest = <-s.persistRuns:
				default:
					if latest != nil {
						s.saveRunState(latest)
					}
					return
				}
			}
		}
	}
}

func (s *Service) saveRunState(runs []model.Run) {
	if err := s.store.SaveRuns(runs); err != nil {
		s.log.Error("save run history", "error", err)
	}
	if err := s.persistRunLogs(runs); err != nil {
		s.log.Error("save persistent run logs", "error", err)
	}
}

func (s *Service) captureBackupMetrics(runs []model.Run) {
	if s.store == nil {
		return
	}
	s.metricsMu.Lock()
	known := make(map[string]struct{}, len(s.backupMetrics))
	for _, metric := range s.backupMetrics {
		known[metric.RunID] = struct{}{}
	}
	added := make([]model.BackupMetric, 0, 1)
	for _, run := range runs {
		if run.TaskType != "backup" || run.FinishedAt == nil || (run.Status != "success" && run.Status != "warning") {
			continue
		}
		if _, exists := known[run.ID]; exists {
			continue
		}
		metric := model.BackupMetric{
			SchemaVersion: model.SchemaVersion, RunID: run.ID, JobID: run.JobID, SnapshotID: run.SnapshotID,
			FinishedAt: *run.FinishedAt, Status: run.Status, BytesProcessed: run.BytesProcessed,
			FilesProcessed: run.FilesProcessed, BytesAdded: run.BytesAdded, FilesNew: run.FilesNew, FilesChanged: run.FilesChanged,
		}
		s.backupMetrics = append(s.backupMetrics, metric)
		added = append(added, metric)
		known[run.ID] = struct{}{}
	}
	if len(added) == 0 {
		s.metricsMu.Unlock()
		return
	}
	s.backupMetrics = normalizeBackupMetrics(s.backupMetrics)
	if len(added) == 1 && s.metricsAppendOK && s.metricRecords < compactBackupMetricsAt {
		if err := s.store.AppendBackupMetric(added[0]); err == nil {
			s.metricRecords++
			s.metricsMu.Unlock()
			return
		} else if s.log != nil {
			s.log.Error("append backup metric", "error", err)
		}
	}
	if err := s.store.SaveBackupMetrics(s.backupMetrics); err != nil {
		if s.log != nil {
			s.log.Error("compact backup metrics", "error", err)
		}
	} else {
		s.metricRecords = len(s.backupMetrics)
		s.metricsAppendOK = true
	}
	s.metricsMu.Unlock()
}

func normalizeBackupMetrics(metrics []model.BackupMetric) []model.BackupMetric {
	known := make(map[string]struct{}, len(metrics))
	normalized := make([]model.BackupMetric, 0, len(metrics))
	for _, metric := range metrics {
		if metric.RunID == "" {
			continue
		}
		if _, exists := known[metric.RunID]; exists {
			continue
		}
		known[metric.RunID] = struct{}{}
		normalized = append(normalized, metric)
	}
	sort.SliceStable(normalized, func(i, j int) bool { return normalized[i].FinishedAt.Before(normalized[j].FinishedAt) })
	if len(normalized) > maxBackupMetrics {
		normalized = append([]model.BackupMetric(nil), normalized[len(normalized)-maxBackupMetrics:]...)
	}
	return normalized
}

func (s *Service) persistRunLogs(runs []model.Run) error {
	dir := strings.TrimSpace(s.Config().Settings.PersistentLogDir)
	if dir == "" {
		return nil
	}
	if err := os.MkdirAll(dir, 0700); err != nil {
		return err
	}
	wanted := make(map[string]struct{})
	for _, run := range runs {
		if run.FinishedAt == nil || len(run.LiveLog) == 0 {
			continue
		}
		name := "urbm-" + safeLogName(run.ID) + ".log"
		wanted[name] = struct{}{}
		path := filepath.Join(dir, name)
		if _, err := os.Stat(path); err == nil {
			continue
		}
		if err := writeTextAtomic(path, strings.Join(run.LiveLog, "\n")+"\n"); err != nil {
			return err
		}
	}
	entries, err := os.ReadDir(dir)
	if err != nil {
		return err
	}
	for _, entry := range entries {
		if entry.IsDir() || !strings.HasPrefix(entry.Name(), "urbm-") || !strings.HasSuffix(entry.Name(), ".log") {
			continue
		}
		if _, ok := wanted[entry.Name()]; !ok {
			if err := os.Remove(filepath.Join(dir, entry.Name())); err != nil && !errors.Is(err, os.ErrNotExist) {
				return err
			}
		}
	}
	return nil
}

func safeLogName(value string) string {
	return strings.Map(func(char rune) rune {
		if char >= 'a' && char <= 'z' || char >= 'A' && char <= 'Z' || char >= '0' && char <= '9' || char == '-' || char == '_' {
			return char
		}
		return '_'
	}, value)
}

func writeTextAtomic(path, value string) error {
	tmp, err := os.CreateTemp(filepath.Dir(path), ".urbm-log-*")
	if err != nil {
		return err
	}
	tmpPath := tmp.Name()
	defer os.Remove(tmpPath)
	if err := tmp.Chmod(0600); err != nil {
		tmp.Close()
		return err
	}
	if _, err := tmp.WriteString(value); err != nil {
		tmp.Close()
		return err
	}
	if err := tmp.Close(); err != nil {
		return err
	}
	return os.Rename(tmpPath, path)
}

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
		}
	}
	s.metricsMu.RLock()
	defer s.metricsMu.RUnlock()
	for _, metric := range s.backupMetrics {
		if metric.JobID == jobID && metric.FinishedAt.After(latest) {
			latest = metric.FinishedAt
		}
	}
	return latest
}

func (s *Service) DashboardStatus(now time.Time) DashboardStatus {
	status := DashboardStatus{OverdueJobs: []OverdueBackupJob{}}
	metrics := s.BackupMetrics()
	lastByJob := make(map[string]time.Time)
	for index := range metrics {
		metric := metrics[index]
		if metric.Status == "success" && (status.LastSuccessfulBackup == nil || metric.FinishedAt.After(status.LastSuccessfulBackup.FinishedAt)) {
			copy := metric
			status.LastSuccessfulBackup = &copy
		}
		if metric.FinishedAt.After(lastByJob[metric.JobID]) {
			lastByJob[metric.JobID] = metric.FinishedAt
		}
	}
	runs := s.queue.Snapshot()
	activeJobs := make(map[string]bool)
	for index := range runs {
		run := runs[index]
		if run.TaskType != "backup" {
			continue
		}
		if run.Status == "queued" || run.Status == "running" || run.Status == "paused" {
			activeJobs[run.JobID] = true
		}
		if run.Status == "failed" && (status.LastFailedBackup == nil || run.CreatedAt.After(status.LastFailedBackup.CreatedAt)) {
			copy := run
			status.LastFailedBackup = &copy
		}
	}
	for _, job := range s.Config().Jobs {
		if job.Type == model.JobRsync || !job.Enabled || job.Schedule.Cron == "" || activeJobs[job.ID] {
			continue
		}
		scheduledAt, err := scheduler.PreviousScheduled(job.Schedule.Cron, job.Schedule.Timezone, now, 370*24*time.Hour)
		if err != nil || scheduledAt.IsZero() || !lastByJob[job.ID].Before(scheduledAt) {
			continue
		}
		overdue := OverdueBackupJob{JobID: job.ID, Name: job.Name, ScheduledAt: scheduledAt}
		if last := lastByJob[job.ID]; !last.IsZero() {
			lastCopy := last
			overdue.LastBackupAt = &lastCopy
		}
		status.OverdueJobs = append(status.OverdueJobs, overdue)
	}
	sort.SliceStable(status.OverdueJobs, func(i, j int) bool {
		return status.OverdueJobs[i].ScheduledAt.Before(status.OverdueJobs[j].ScheduledAt)
	})
	return status
}

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.SetBinary(config.Settings.ResticPath)
	s.rsync.SetBinary(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")
	}
	unlock, err := s.lockRepository(ctx, repo)
	if err != nil {
		return err
	}
	defer unlock()
	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")
	}
	unlock, err := s.lockRepository(ctx, repo)
	if err != nil {
		return err
	}
	defer unlock()
	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) {
	if err := model.ValidateRestoreTask(task); err != nil {
		return model.Run{}, fmt.Errorf("validation: %w", err)
	}
	if _, ok := s.repository(task.RepositoryID); !ok {
		return model.Run{}, errors.New("validation: unknown repository")
	}
	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
	task.ID = newID()
	task.SchemaVersion = model.SchemaVersion
	run := model.Run{SchemaVersion: model.SchemaVersion, ID: task.ID, JobID: task.RepositoryID, TaskType: "restore", Priority: 10, CreatedAt: time.Now().UTC(), Restore: &task}
	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")
	}
	unlock, err := s.lockRepository(ctx, repo)
	if err != nil {
		return nil, err
	}
	defer unlock()
	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}
		unlock, err := s.lockRepository(ctx, repo)
		if err != nil {
			item.Error = err.Error()
			result = append(result, item)
			break
		}
		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
			}
		}
		unlock()
		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) (SnapshotBrowserPage, error) {
	started := time.Now()
	s.log.Info("snapshot file listing started", "repoID", repoID, "snapshot", snapshot, "path", path)
	if err := model.ValidateSnapshotID(snapshot); err != nil {
		return SnapshotBrowserPage{}, fmt.Errorf("validation: %w", err)
	}
	repo, ok := s.repository(repoID)
	if !ok {
		return SnapshotBrowserPage{}, errors.New("validation: unknown repository")
	}

	normalizedPath := normalizeSnapshotPath(path)
	key := snapshotCacheKey(repoID, snapshot)
	if page, ok := s.snapshotBrowserPage(key, normalizedPath); ok {
		s.log.Info("snapshot file listing loaded from cache", "repoID", repoID, "snapshot", snapshot, "path", normalizedPath, "items", len(page.Items), "total", page.Total, "durationMs", time.Since(started).Milliseconds())
		return page, nil
	}
	unlock, err := s.lockRepository(ctx, repo)
	if err != nil {
		return SnapshotBrowserPage{}, err
	}
	defer unlock()
	if page, ok := s.snapshotBrowserPage(key, normalizedPath); ok {
		return page, nil
	}

	mounted, err := s.mounts.Prepare(ctx, repo)
	if err != nil {
		s.log.Warn("snapshot file listing failed", "repoID", repoID, "snapshot", snapshot, "path", normalizedPath, "durationMs", time.Since(started).Milliseconds(), "error", err)
		return SnapshotBrowserPage{}, err
	}
	defer s.mounts.Cleanup(context.Background(), mounted)

	const maxSnapshotEntries = 1_000_000
	builder := newSnapshotTreeBuilder()
	tooLarge := false
	err = s.restic.WalkList(ctx, mounted.Repository, snapshot, "", func(item map[string]any) {
		if builder.total >= maxSnapshotEntries {
			tooLarge = true
			return
		}
		builder.add(item)
	})
	if err != nil {
		s.log.Warn("snapshot file listing failed", "repoID", repoID, "snapshot", snapshot, "path", normalizedPath, "durationMs", time.Since(started).Milliseconds(), "error", err)
		return SnapshotBrowserPage{}, err
	}
	if tooLarge {
		return SnapshotBrowserPage{}, fmt.Errorf("environment: snapshot contains more than %d entries; narrow the backup or use restic directly", maxSnapshotEntries)
	}
	cache := builder.cache()
	s.storeSnapshotTree(key, cache)
	page, _ := s.snapshotBrowserPage(key, normalizedPath)
	s.log.Info("snapshot file listing loaded", "repoID", repoID, "snapshot", snapshot, "path", normalizedPath, "items", len(page.Items), "total", cache.total, "durationMs", time.Since(started).Milliseconds())
	return page, nil
}

func snapshotCacheKey(repoID, snapshot string) string {
	return repoID + "\x00" + snapshot
}

func (s *Service) snapshotBrowserPage(key, path string) (SnapshotBrowserPage, bool) {
	s.snapshotMu.Lock()
	defer s.snapshotMu.Unlock()
	cache, ok := s.snapshotTree[key]
	if !ok || time.Since(cache.generated) > 30*time.Minute {
		return SnapshotBrowserPage{}, false
	}
	items, ok := cache.children[path]
	if !ok {
		items = []SnapshotBrowserNode{}
	}
	return SnapshotBrowserPage{Path: path, Items: items, Total: cache.total, Cached: true, Generated: cache.generated}, true
}

func (s *Service) storeSnapshotTree(key string, cache snapshotTreeCache) {
	s.snapshotMu.Lock()
	defer s.snapshotMu.Unlock()
	if s.snapshotTree == nil {
		s.snapshotTree = map[string]snapshotTreeCache{}
	}
	for cacheKey, value := range s.snapshotTree {
		if time.Since(value.generated) > 30*time.Minute {
			delete(s.snapshotTree, cacheKey)
		}
	}
	for len(s.snapshotTree) >= 3 {
		oldestKey := ""
		var oldest time.Time
		for cacheKey, value := range s.snapshotTree {
			if oldestKey == "" || value.generated.Before(oldest) {
				oldestKey, oldest = cacheKey, value.generated
			}
		}
		delete(s.snapshotTree, oldestKey)
	}
	s.snapshotTree[key] = cache
}

func buildSnapshotTreeCache(items []map[string]any) snapshotTreeCache {
	builder := newSnapshotTreeBuilder()
	for _, item := range items {
		builder.add(item)
	}
	return builder.cache()
}

func newSnapshotTreeBuilder() *snapshotTreeBuilder {
	root := &snapshotTreeNode{path: "/", name: "/", nodeType: "dir", children: map[string]*snapshotTreeNode{}}
	return &snapshotTreeBuilder{root: root, nodes: map[string]*snapshotTreeNode{"/": root}}
}

func (b *snapshotTreeBuilder) ensure(path string, nodeType string, size int64) *snapshotTreeNode {
	path = normalizeSnapshotPath(path)
	if node, ok := b.nodes[path]; ok {
		if nodeType != "" && nodeType != "dir" {
			node.nodeType, node.size = nodeType, size
		}
		return node
	}
	parentPath := filepath.Dir(path)
	if parentPath == "." {
		parentPath = "/"
	}
	parent := b.ensure(parentPath, "dir", 0)
	node := &snapshotTreeNode{path: path, name: filepath.Base(path), nodeType: nodeType, size: size, children: map[string]*snapshotTreeNode{}}
	b.nodes[path] = node
	parent.children[path] = node
	return node
}

func (b *snapshotTreeBuilder) add(item map[string]any) {
	path := snapshotItemPath(item)
	if path == "/" {
		return
	}
	nodeType := "file"
	if value, _ := item["type"].(string); value == "dir" {
		nodeType = "dir"
	}
	b.ensure(path, nodeType, snapshotItemSize(item))
	b.total++
}

func (b *snapshotTreeBuilder) cache() snapshotTreeCache {
	children := make(map[string][]SnapshotBrowserNode, len(b.nodes))
	for _, node := range b.nodes {
		if len(node.children) == 0 {
			continue
		}
		list := make([]SnapshotBrowserNode, 0, len(node.children))
		for _, child := range node.children {
			list = append(list, SnapshotBrowserNode{Path: child.path, Name: child.name, Type: child.nodeType, Size: child.size, HasChildren: len(child.children) > 0})
		}
		sort.Slice(list, func(i, j int) bool {
			if list[i].Type != list[j].Type {
				return list[i].Type == "dir"
			}
			return strings.ToLower(list[i].Name) < strings.ToLower(list[j].Name)
		})
		children[node.path] = list
	}
	return snapshotTreeCache{generated: time.Now().UTC(), total: b.total, children: children}
}

func snapshotItemPath(item map[string]any) string {
	for _, key := range []string{"path", "name"} {
		if value, ok := item[key].(string); ok && value != "" {
			return normalizeSnapshotPath(value)
		}
	}
	return "/"
}

func snapshotItemSize(item map[string]any) int64 {
	switch value := item["size"].(type) {
	case float64:
		return int64(value)
	case int64:
		return value
	case int:
		return int64(value)
	default:
		return 0
	}
}

func normalizeSnapshotPath(path string) string {
	path = filepath.Clean("/" + strings.TrimSpace(path))
	if path == "." {
		return "/"
	}
	return 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")
	}
	unlock, err := s.lockRepository(ctx, repo)
	if err != nil {
		return err
	}
	defer unlock()
	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")
	}
	unlock, err := s.lockRepository(ctx, repo)
	if err != nil {
		return err
	}
	defer unlock()
	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")
	}
	if s.repositoryBusy(repoID) {
		return errors.New("validation: repository has a queued, running, or paused task")
	}
	unlock, err := s.lockRepository(ctx, repo)
	if err != nil {
		return err
	}
	defer unlock()
	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")
	}
	unlock, err := s.lockRepository(ctx, repo)
	if err != nil {
		return failedError(run, err)
	}
	defer unlock()
	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) }()
	if err := platform.ValidateRsyncPaths(job); err != nil {
		return failedError(run, err)
	}
	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")
	}
	unlock, err := s.lockRepository(ctx, repo)
	if err != nil {
		return failedError(run, err)
	}
	defer unlock()
	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)
	if run.Restore == nil {
		return failed(run, "internal", "invalid restore payload")
	}
	task := *run.Restore
	if err := model.ValidateRestoreTask(task); err != nil {
		return failed(run, "validation", err.Error())
	}
	config := s.Config()
	target, err := platform.ValidateRestoreTarget(task.Target, config.Settings.RestoreRoot, task.InPlace, task.Confirmed)
	if err != nil {
		return failedError(run, err)
	}
	task.Target = target
	repo, ok := s.repository(task.RepositoryID)
	if !ok {
		return failed(run, "validation", "repository no longer exists")
	}
	unlock, err := s.lockRepository(ctx, repo)
	if err != nil {
		return failedError(run, err)
	}
	defer unlock()
	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 (s *Service) lockRepository(ctx context.Context, repo model.Repository) (func(), error) {
	key := string(repo.Type) + ":" + filepath.Clean(repo.Location)
	if repo.Mount != nil && repo.Mount.Managed {
		key = "mount:" + filepath.Clean(repo.Mount.MountPoint)
	}
	s.repositoryMu.Lock()
	lock := s.repositoryLocks[key]
	if lock == nil {
		lock = make(chan struct{}, 1)
		s.repositoryLocks[key] = lock
	}
	s.repositoryMu.Unlock()
	select {
	case lock <- struct{}{}:
		return func() { <-lock }, nil
	case <-ctx.Done():
		return nil, ctx.Err()
	}
}

func (s *Service) repositoryBusy(repoID string) bool {
	for _, run := range s.queue.Snapshot() {
		if run.Status != "queued" && run.Status != "running" && run.Status != "paused" {
			continue
		}
		if run.TaskType == "restore" && run.Restore != nil && run.Restore.RepositoryID == repoID {
			return true
		}
		if run.TaskType == "check" || run.TaskType == "prune" {
			if run.JobID == repoID {
				return true
			}
			continue
		}
		if job, ok := s.job(run.JobID); ok && job.RepositoryID == repoID {
			return true
		}
	}
	return 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()) }