nodemaster/models/backup.go

package models

import (
	"bytes"
	"encoding/json"
	"fmt"
	"os"
	"os/exec"
	"regexp"
	"strings"
	"time"

	"github.com/robfig/cron/v3"
)

// validateBackupTarget checks that a backup target is well-formed enough to
// store: a known method, and (if set) a parseable cron schedule.
func validateBackupTarget(t BackupTarget) error {
	switch t.Method {
	case MethodRsync, MethodRestic, MethodExternal:
	default:
		return fmt.Errorf("unknown backup method: %s", t.Method)
	}
	if t.Schedule != "" {
		if _, err := cron.ParseStandard(t.Schedule); err != nil {
			return fmt.Errorf("invalid schedule %q: %w", t.Schedule, err)
		}
	}
	return nil
}

// RunBackup executes backup for all targets against sourcePath and returns one
// execution record per target. scope identifies where progress is published
// ("node" or "svc:<serviceID>").
func RunBackup(scope string, targets []BackupTarget, sourcePath string) []BackupExecution {
	results := make([]BackupExecution, 0, len(targets))
	for _, t := range targets {
		results = append(results, runTarget(scope, t, sourcePath))
	}
	return results
}

func runTarget(scope string, t BackupTarget, sourcePath string) BackupExecution {
	start := time.Now()
	ex := BackupExecution{
		ID:        fmt.Sprintf("bex_%d", start.UnixNano()),
		Timestamp: start.UTC(),
		TargetID:  t.ID,
	}
	key := progressKey(scope, t.ID)
	startedAt := start.UTC()
	setProgress(key, RunState{Running: true, TargetID: t.ID, Method: t.Method, StartedAt: &startedAt})

	if t.Method == MethodExternal {
		ex.Success = true
		ex.Message = "external backup managed externally; no action taken by API"
		setProgress(key, RunState{Running: false, TargetID: t.ID, Method: t.Method, Percent: 100, Message: ex.Message, StartedAt: &startedAt})
		return ex
	}

	var cmd *exec.Cmd
	switch t.Method {
	case MethodRsync:
		args := []string{"-avz", "--info=progress2"}
		for k, v := range t.Params {
			args = append(args, "--"+k+"="+v)
		}
		args = append(args, sourcePath, t.Remote)
		cmd = exec.Command("rsync", args...)
	case MethodRestic:
		password := GetResticPassword()
		if password == "" {
			ex.Success = false
			ex.Message = "no restic repository password configured on this node; set restic_password via PUT /v1/node"
			setProgress(key, RunState{Running: false, TargetID: t.ID, Method: t.Method, Message: ex.Message, StartedAt: &startedAt})
			return ex
		}
		env := append(os.Environ(), "RESTIC_REPOSITORY="+t.Remote, "RESTIC_PASSWORD="+password)
		if err := ensureResticRepo(env); err != nil {
			ex.Success = false
			ex.Message = fmt.Sprintf("restic init failed: %v", err)
			setProgress(key, RunState{Running: false, TargetID: t.ID, Method: t.Method, Message: ex.Message, StartedAt: &startedAt})
			return ex
		}
		args := []string{"backup", sourcePath, "--json"}
		for k, v := range t.Params {
			args = append(args, "--"+k+"="+v)
		}
		cmd = exec.Command("restic", args...)
		cmd.Env = env
	default:
		ex.Success = false
		ex.Message = fmt.Sprintf("unknown backup method: %s", t.Method)
		setProgress(key, RunState{Running: false, TargetID: t.ID, Method: t.Method, Message: ex.Message, StartedAt: &startedAt})
		return ex
	}

	pw := &progressWriter{key: key, targetID: t.ID, method: t.Method, startedAt: startedAt}
	cmd.Stdout = pw
	cmd.Stderr = pw

	err := cmd.Run()
	ex.Duration = int64(time.Since(start).Seconds())
	out := strings.TrimSpace(pw.buf.String())
	if err != nil {
		ex.Success = false
		ex.Message = fmt.Sprintf("%v | %s", err, out)
	} else {
		ex.Success = true
		if pw.lastMessage != "" {
			ex.Message = pw.lastMessage
		} else {
			ex.Message = out
		}
	}

	final := pw.state()
	final.Running = false
	if ex.Success {
		final.Percent = 100
	}
	final.Message = ex.Message
	setProgress(key, final)

	return ex
}

// ensureResticRepo initializes the restic repository named by env's
// RESTIC_REPOSITORY on first use. Running "init" against a repository that
// already exists fails with a message containing "already"; that case is
// treated as success so callers don't need to track init state themselves.
func ensureResticRepo(env []string) error {
	cmd := exec.Command("restic", "init")
	cmd.Env = env
	out, err := cmd.CombinedOutput()
	if err == nil {
		return nil
	}
	if strings.Contains(strings.ToLower(string(out)), "already") {
		return nil
	}
	return fmt.Errorf("%v: %s", err, strings.TrimSpace(string(out)))
}

// progressWriter is assigned as both Stdout and Stderr of a backup command. It
// buffers the full output (for the final BackupExecution.Message) while also
// scanning completed lines for progress info to publish via setProgress.
// Best-effort only: a line that doesn't parse just leaves the last known
// percentage/ETA in place rather than failing the backup.
type progressWriter struct {
	buf         bytes.Buffer
	pending     []byte
	key         string
	targetID    string
	method      BackupMethod
	startedAt   time.Time
	percent     float64
	eta         string
	lastMessage string
}

func (w *progressWriter) Write(p []byte) (int, error) {
	w.buf.Write(p)
	w.pending = append(w.pending, p...)
	for {
		idx := bytes.IndexAny(w.pending, "\r\n")
		if idx < 0 {
			break
		}
		line := bytes.TrimSpace(w.pending[:idx])
		w.pending = w.pending[idx+1:]
		if len(line) > 0 {
			w.handleLine(line)
		}
	}
	return len(p), nil
}

func (w *progressWriter) handleLine(line []byte) {
	switch w.method {
	case MethodRsync:
		w.handleRsyncLine(line)
	case MethodRestic:
		w.handleResticLine(line)
	}
	setProgress(w.key, w.state())
}

var rsyncPercentRe = regexp.MustCompile(`(\d+)%`)

func (w *progressWriter) handleRsyncLine(line []byte) {
	w.lastMessage = string(line)
	m := rsyncPercentRe.FindSubmatch(line)
	if m == nil {
		return
	}
	var pct float64
	if _, err := fmt.Sscanf(string(m[1]), "%f", &pct); err == nil {
		w.percent = pct
	}
}

// resticStatusLine models the subset of restic's `backup --json` line kinds
// (message_type "status", "summary", "error") that are useful for progress
// reporting; unrecognized/unparseable lines are ignored.
type resticStatusLine struct {
	MessageType      string  `json:"message_type"`
	PercentDone      float64 `json:"percent_done"`
	SecondsRemaining float64 `json:"seconds_remaining"`
	SnapshotID       string  `json:"snapshot_id"`
	Error            *struct {
		Message string `json:"message"`
	} `json:"error,omitempty"`
}

func (w *progressWriter) handleResticLine(line []byte) {
	var sl resticStatusLine
	if err := json.Unmarshal(line, &sl); err != nil {
		return
	}
	switch sl.MessageType {
	case "status":
		w.percent = sl.PercentDone * 100
		if sl.SecondsRemaining > 0 {
			w.eta = fmt.Sprintf("%.0fs", sl.SecondsRemaining)
		}
	case "summary":
		w.percent = 100
		if sl.SnapshotID != "" {
			w.lastMessage = fmt.Sprintf("snapshot %s created", sl.SnapshotID)
		}
	case "error":
		if sl.Error != nil && sl.Error.Message != "" {
			w.lastMessage = sl.Error.Message
		}
	}
}

func (w *progressWriter) state() RunState {
	startedAt := w.startedAt
	return RunState{
		Running:   true,
		TargetID:  w.targetID,
		Method:    w.method,
		Percent:   w.percent,
		ETA:       w.eta,
		Message:   w.lastMessage,
		StartedAt: &startedAt,
	}
}

// StartService runs the compose start command for the given service.
func StartService(svc Service) error {
	cmd := composeCmd(svc, "start")
	if out, err := cmd.CombinedOutput(); err != nil {
		return fmt.Errorf("%v: %s", err, strings.TrimSpace(string(out)))
	}
	return nil
}

// StopService runs the compose stop command for the given service.
func StopService(svc Service) error {
	cmd := composeCmd(svc, "stop")
	if out, err := cmd.CombinedOutput(); err != nil {
		return fmt.Errorf("%v: %s", err, strings.TrimSpace(string(out)))
	}
	return nil
}

func composeCmd(svc Service, action string) *exec.Cmd {
	switch svc.ComposeType {
	case "docker-compose":
		return exec.Command("docker-compose", "-f", svc.ComposeFile, action)
	case "podman-compose":
		return exec.Command("podman-compose", "-f", svc.ComposeFile, action)
	default:
		return exec.Command("docker", "compose", "-f", svc.ComposeFile, action)
	}
}

// filterTargets returns the subset of targets whose ID is in ids. A nil or
// empty ids means "all targets".
func filterTargets(targets []BackupTarget, ids []string) []BackupTarget {
	if len(ids) == 0 {
		return targets
	}
	want := make(map[string]bool, len(ids))
	for _, id := range ids {
		want[id] = true
	}
	result := make([]BackupTarget, 0, len(ids))
	for _, t := range targets {
		if want[t.ID] {
			result = append(result, t)
		}
	}
	return result
}