backup-master/internal/agent/run.go

package agent

import (
	"context"
	"crypto/sha256"
	"encoding/hex"
	"encoding/json"
	"errors"
	"fmt"
	"io"
	"log"
	"os"
	"path/filepath"
	"runtime"
	"strings"
	"time"

	"gitea.hsrv.site/mrleo1nid/backup-master/internal/api"
	"gitea.hsrv.site/mrleo1nid/backup-master/internal/archive"
	"gitea.hsrv.site/mrleo1nid/backup-master/internal/catalog"
	"gitea.hsrv.site/mrleo1nid/backup-master/internal/manifest"
)

type Options struct {
	ConfigPath   string
	AgentVersion string
	// Output — записать части локально в каталог вместо отправки (для отладки).
	Output string
	// ForceHeavy — включить heavy-части независимо от расписания.
	ForceHeavy bool
	// NoUpdate — не обновлять агента, даже если сервер предлагает.
	NoUpdate bool
}

// session — подключение к серверу и его ответ для одного запуска агента.
type session struct {
	cfg    *Config
	client *api.Client
	remote *api.AgentConfig
}

// connect читает конфиг и спрашивает у сервера настройки (каталог, обновление).
func connect(ctx context.Context, opts Options) (*session, error) {
	cfg, err := LoadConfig(opts.ConfigPath)
	if err != nil {
		return nil, err
	}
	client := api.NewClient(cfg.Server, cfg.Token)
	remote, err := client.AgentConfig(ctx, runtime.GOARCH, opts.AgentVersion)
	if err != nil {
		return nil, fmt.Errorf("конфиг с сервера: %w", err)
	}
	return &session{cfg: cfg, client: client, remote: remote}, nil
}

// catalogFor выбирает каталог: итоговый с сервера, а если его нет (старый сервер)
// или он не читается этой версией агента — встроенный + переопределения клиента.
func catalogFor(remote *api.AgentConfig) (*catalog.Config, []string, error) {
	var warns []string
	if strings.TrimSpace(remote.CatalogYAML) != "" {
		cat, err := catalog.Parse([]byte(remote.CatalogYAML))
		if err == nil {
			err = cat.Validate()
		}
		if err == nil {
			return cat, nil, nil
		}
		warns = append(warns, fmt.Sprintf("каталог с сервера не подошёл этой версии агента (%v) — использую встроенный", err))
	}
	cat, err := catalog.Default()
	if err != nil {
		return nil, warns, err
	}
	if strings.TrimSpace(remote.ConfigYAML) != "" {
		over, err := catalog.Parse([]byte(remote.ConfigYAML))
		if err != nil {
			return nil, warns, fmt.Errorf("конфиг клиента с сервера: %w", err)
		}
		cat = catalog.Merge(cat, over)
	}
	return cat, warns, cat.Validate()
}

// prepare собирает план бэкапа по каталогу.
func (s *session) prepare(ctx context.Context, opts Options, spool string, warns []string) (*Plan, error) {
	cat, catWarns, err := catalogFor(s.remote)
	if err != nil {
		return nil, err
	}
	plan, err := Collect(ctx, cat, spool, s.remote.HeavyDue || opts.ForceHeavy)
	if err != nil {
		return nil, err
	}
	plan.Manifest.Domain = s.remote.Domain
	plan.Manifest.AgentVersion = opts.AgentVersion
	plan.Manifest.Warnings = append(append(warns, catWarns...), plan.Manifest.Warnings...)
	return plan, nil
}

func newSpool(base string) (string, error) {
	dir := filepath.Join(base, time.Now().Format("20060102-150405"))
	return dir, os.MkdirAll(dir, 0o700)
}

// PrintPlan выводит, что попадёт в бэкап. Дампы БД при этом реально выполняются (во временный каталог).
func PrintPlan(ctx context.Context, opts Options, w io.Writer) error {
	sess, err := connect(ctx, opts)
	if err != nil {
		return err
	}
	spool, err := newSpool(sess.cfg.SpoolDir)
	if err != nil {
		return err
	}
	defer os.RemoveAll(spool)
	plan, err := sess.prepare(ctx, opts, spool, nil)
	if err != nil {
		return err
	}
	m := &plan.Manifest
	fmt.Fprintf(w, "Агент %s, сервер %s\n", opts.AgentVersion, sess.remote.ServerVersion)
	if u := sess.remote.Update; u != nil {
		fmt.Fprintf(w, "Доступно обновление агента до %s — установится при следующем бэкапе\n", u.Version)
	}
	fmt.Fprintf(w, "Домен: %s, hostname: %s, %s (%s)\n", m.Domain, m.Hostname, m.OS.PrettyName, m.OS.Arch)
	if len(m.Packages) > 0 {
		fmt.Fprintf(w, "Пакеты:")
		for k, v := range m.Packages {
			fmt.Fprintf(w, " %s=%s", k, v)
		}
		fmt.Fprintln(w)
	}
	for _, c := range m.Categories {
		line := fmt.Sprintf("\n[%s] %s", c.Name, c.Type)
		if c.Auto {
			line += " (найден автоматически)"
		}
		if c.Version != "" {
			line += ", версия " + c.Version
		}
		if c.Service != nil {
			line += fmt.Sprintf(", сервис %s enabled=%v active=%v", c.Service.Unit, c.Service.Enabled, c.Service.Active)
		}
		fmt.Fprintln(w, line)
		if c.Compose != nil {
			fmt.Fprintf(w, "  проект %s, запущен=%v\n", c.Compose.Project, c.Compose.Running)
			for svc, img := range c.Compose.Images {
				ref := img.Digest
				if img.Built {
					ref = "собирается из исходников"
				} else if ref == "" {
					ref = img.Ref + " (digest не определён)"
				}
				fmt.Fprintf(w, "  образ %s: %s\n", svc, ref)
			}
		}
		for _, d := range c.Dumps {
			fmt.Fprintf(w, "  дамп %s %s\n", d.Kind, d.Name)
		}
		for _, v := range c.Volumes {
			fmt.Fprintf(w, "  том %s\n", v.Name)
		}
		for _, h := range c.Heavy {
			note := "в этот раз"
			if c.HeavySkipped {
				note = "по расписанию, не в этот раз"
			}
			fmt.Fprintf(w, "  heavy %s (%s)\n", h, note)
		}
	}
	fmt.Fprintln(w, "\nСодержимое частей:")
	for _, part := range m.Parts {
		var total int64
		var files int
		for _, it := range plan.Parts[part] {
			size, n := treeSize(it.Src, it.Skip)
			total += size
			files += n
		}
		fmt.Fprintf(w, "  %s: %d файлов, %s до сжатия\n", part, files, human(total))
	}
	if len(m.Warnings) > 0 {
		fmt.Fprintln(w, "\nПредупреждения:")
		for _, wr := range m.Warnings {
			fmt.Fprintln(w, "  ! "+wr)
		}
	}
	return nil
}

// Backup выполняет полный цикл: сбор, отправка частей, подтверждение.
func Backup(ctx context.Context, opts Options) (err error) {
	unlock, err := lock("/run/backup-agent.lock")
	if err != nil {
		return err
	}
	defer unlock()

	sess, err := connect(ctx, opts)
	if err != nil {
		return err
	}
	// Самообновление — до сбора: при успехе процесс перезапускается новой версией
	// (exec не возвращается), при ошибке бэкап идёт текущей версией с предупреждением.
	var warns []string
	if u := sess.remote.Update; u != nil && !opts.NoUpdate && os.Getenv(updatedEnv) == "" {
		log.Printf("обновление агента %s → %s", opts.AgentVersion, u.Version)
		if err := selfUpdate(ctx, sess.client, u, opts.AgentVersion); err != nil {
			log.Printf("! обновление не удалось: %v", err)
			warns = append(warns, fmt.Sprintf("обновление агента до %s не удалось: %v", u.Version, err))
		}
	}
	if from := os.Getenv(updatedEnv); from != "" {
		log.Printf("агент обновлён: %s → %s", from, opts.AgentVersion)
	}
	client := sess.client
	spool, err := newSpool(sess.cfg.SpoolDir)
	if err != nil {
		return err
	}
	defer os.RemoveAll(spool)

	start := time.Now()
	plan, err := sess.prepare(ctx, opts, spool, warns)
	if err != nil {
		return err
	}
	m := &plan.Manifest
	log.Printf("собрано: %d категорий, части %v, предупреждений %d", len(m.Categories), m.Parts, len(m.Warnings))
	for _, w := range m.Warnings {
		log.Printf("! %s", w)
	}
	if len(m.Categories) == 0 {
		// пустой «успешный» бэкап хуже ошибки: сервер решит, что всё в порядке
		return errors.New("ни одна категория не найдена на сервере — проверьте каталог (backupctl plan)")
	}

	if opts.Output != "" {
		return writeLocal(plan, opts.Output)
	}

	id, err := client.CreateBackup(ctx, m)
	if err != nil {
		return fmt.Errorf("создание бэкапа на сервере: %w", err)
	}
	defer func() {
		if err != nil {
			fctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
			defer cancel()
			if ferr := client.Fail(fctx, id, err.Error()); ferr != nil {
				log.Printf("не удалось сообщить серверу об ошибке: %v", ferr)
			}
		}
	}()

	var done api.CompleteRequest
	for _, part := range m.Parts {
		info, warns, err := uploadWithRetry(ctx, client, id, plan, part)
		if err != nil {
			return fmt.Errorf("часть %s: %w", part, err)
		}
		done.Parts = append(done.Parts, *info)
		done.Warnings = append(done.Warnings, warns...)
		log.Printf("часть %s отправлена: %s", part, human(info.Size))
	}
	for _, w := range done.Warnings {
		log.Printf("! %s", w)
	}
	if err := client.Complete(ctx, id, done); err != nil {
		return fmt.Errorf("подтверждение: %w", err)
	}
	log.Printf("бэкап #%d готов за %s", id, time.Since(start).Round(time.Second))
	return nil
}

func uploadWithRetry(ctx context.Context, client *api.Client, id int64, plan *Plan, part string) (*api.PartInfo, []string, error) {
	var lastErr error
	for attempt := 1; attempt <= 3; attempt++ {
		if attempt > 1 {
			wait := time.Duration(attempt*attempt) * 20 * time.Second
			log.Printf("часть %s: попытка %d через %s (%v)", part, attempt, wait, lastErr)
			select {
			case <-ctx.Done():
				return nil, nil, ctx.Err()
			case <-time.After(wait):
			}
		}
		info, warns, err := uploadPart(ctx, client, id, plan, part)
		if err == nil {
			return info, warns, nil
		}
		lastErr = err
		var apiErr *api.Error
		if errors.As(err, &apiErr) && apiErr.Status >= 400 && apiErr.Status < 500 && apiErr.Status != 408 && apiErr.Status != 429 {
			break // повтор не поможет
		}
	}
	return nil, nil, lastErr
}

// uploadPart архивирует часть прямо в тело PUT-запроса, параллельно считая sha256.
func uploadPart(ctx context.Context, client *api.Client, id int64, plan *Plan, part string) (*api.PartInfo, []string, error) {
	pr, pw := io.Pipe()
	h := sha256.New()
	cw := &countWriter{w: io.MultiWriter(pw, h)}
	type result struct {
		warns []string
		err   error
	}
	done := make(chan result, 1)
	go func() {
		warns, err := writePart(cw, plan, part)
		pw.CloseWithError(err)
		done <- result{warns, err}
	}()
	info, err := client.UploadPart(ctx, id, part, pr)
	pr.CloseWithError(errors.New("загрузка прервана")) // освобождаем писателя, если сервер ответил раньше
	res := <-done
	if err != nil {
		if res.err != nil && !errors.Is(res.err, io.ErrClosedPipe) && !strings.Contains(res.err.Error(), "загрузка прервана") {
			return nil, nil, fmt.Errorf("архивация: %w", res.err)
		}
		return nil, nil, err
	}
	if res.err != nil {
		return nil, nil, fmt.Errorf("архивация: %w", res.err)
	}
	warns := res.warns
	sum := hex.EncodeToString(h.Sum(nil))
	if info.SHA256 != sum || info.Size != cw.n {
		return nil, nil, fmt.Errorf("сервер получил %d байт (%s), отправлено %d (%s)", info.Size, info.SHA256, cw.n, sum)
	}
	return info, warns, nil
}

func writePart(w io.Writer, plan *Plan, part string) ([]string, error) {
	aw, err := archive.NewWriter(w)
	if err != nil {
		return nil, err
	}
	if part == manifest.MainPart {
		data, err := json.MarshalIndent(plan.Manifest, "", "  ")
		if err != nil {
			return nil, err
		}
		if err := aw.AddBytes(manifest.ManifestFile, data, 0o600); err != nil {
			return nil, err
		}
	}
	for _, it := range plan.Parts[part] {
		if err := aw.AddTree(it.Src, it.Name, it.Skip); err != nil {
			return aw.Warnings, err
		}
	}
	return aw.Warnings, aw.Close()
}

func writeLocal(plan *Plan, dir string) error {
	if err := os.MkdirAll(dir, 0o700); err != nil {
		return err
	}
	for _, part := range plan.Manifest.Parts {
		p := filepath.Join(dir, part+".tar.zst")
		f, err := os.OpenFile(p, os.O_CREATE|os.O_TRUNC|os.O_WRONLY, 0o600)
		if err != nil {
			return err
		}
		warns, err := writePart(f, plan, part)
		if cerr := f.Close(); err == nil {
			err = cerr
		}
		if err != nil {
			return err
		}
		for _, w := range warns {
			log.Printf("! %s", w)
		}
		log.Printf("записано %s", p)
	}
	return nil
}

type countWriter struct {
	w io.Writer
	n int64
}

func (c *countWriter) Write(p []byte) (int, error) {
	n, err := c.w.Write(p)
	c.n += int64(n)
	return n, err
}

func human(n int64) string {
	const unit = 1024
	if n < unit {
		return fmt.Sprintf("%d Б", n)
	}
	div, exp := int64(unit), 0
	for m := n / unit; m >= unit; m /= unit {
		div *= unit
		exp++
	}
	return fmt.Sprintf("%.1f %cБ", float64(n)/float64(div), []rune("КМГТП")[exp])
}