package service import ( "context" "crypto/rand" "encoding/hex" "errors" "fmt" "log/slog" "os" "path/filepath" "strings" "sync" "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/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 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, 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, 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 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) EnqueueJob(jobID string, priority int) (model.Run, error) { job, ok := s.job(jobID) if !ok { return model.Run{}, errors.New("validation: unknown job") } run := model.Run{SchemaVersion: model.SchemaVersion, ID: newID(), JobID: job.ID, TaskType: "backup", 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 { if err := s.restic.Pause(runID); err != nil { return fmt.Errorf("pause: %w", err) } if !s.queue.SetPaused(runID, true) { _ = s.restic.Resume(runID) return errors.New("pause: run is not currently running") } return nil } func (s *Service) Resume(runID string) error { if err := s.restic.Resume(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) 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) 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) } _, err = s.restic.Snapshots(ctx, mounted.Repository) return err } 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 == "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) 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) } } 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, run.BytesAdded, run.FilesNew, run.FilesChanged = summary.SnapshotID, summary.DataAdded, summary.FilesNew, summary.FilesChanged run.BytesProcessed, run.FilesProcessed = summary.TotalBytesProcessed, summary.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.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" { 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()) }