URBM/internal/service/service.go

package service

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

	"github.com/backupper-unraid/backupper/internal/model"
	"github.com/backupper-unraid/backupper/internal/notify"
	"github.com/backupper-unraid/backupper/internal/platform"
	"github.com/backupper-unraid/backupper/internal/queue"
	"github.com/backupper-unraid/backupper/internal/restic"
	"github.com/backupper-unraid/backupper/internal/scheduler"
	"github.com/backupper-unraid/backupper/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 {
	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) 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 {
	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")
	}
	mounted, err := s.mounts.Prepare(ctx, repo)
	if err != nil {
		return failedError(run, err)
	}
	defer s.mounts.Cleanup(context.Background(), mounted)
	if run.TaskType == "check" {
		err = s.restic.Check(ctx, mounted.Repository)
	} else {
		err = s.restic.Prune(ctx, mounted.Repository)
	}
	if err != nil {
		return failedError(run, err)
	}
	run.Status, run.Message = "success", run.TaskType+" completed"
	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")
	}
	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)
		}
	}
	summary, err := s.restic.Backup(ctx, mounted.Repository, job, prepared.Sources)
	if err != nil {
		return failedError(run, err)
	}
	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 = summary.SnapshotID, summary.DataAdded, summary.FilesNew
	run.BytesProcessed, run.FilesProcessed = summary.TotalBytesProcessed, summary.TotalFilesProcessed
	return run
}

func (s *Service) executeRestore(ctx context.Context, run model.Run) model.Run {
	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 {
		return failedError(run, err)
	}
	mounted, err := s.mounts.Prepare(ctx, repo)
	if err != nil {
		return failedError(run, err)
	}
	defer s.mounts.Cleanup(context.Background(), mounted)
	if err := s.restic.Restore(ctx, mounted.Repository, task); err != nil {
		return failedError(run, err)
	}
	run.Status, run.Message, run.JobID = "success", "restore completed", ""
	return run
}

func (s *Service) finishJob(run model.Run, job model.Job, result model.Run) model.Run {
	event := result.Status
	for _, wanted := range job.NotifyOn {
		if wanted == event {
			go s.notify(job.Name, result)
			break
		}
	}
	return result
}

func (s *Service) notify(name string, run model.Run) {
	for _, target := range s.Config().Notifications {
		for _, event := range target.Events {
			if event == run.Status {
				_ = s.notifier.Send(context.Background(), target, "Backupper: "+name, run.Message, run.Status)
				break
			}
		}
	}
}

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