healthlog/cmd/healthlog/reindex.go

package main

import (
	"context"
	"errors"
	"flag"
	"fmt"
	"io"
	"log/slog"
	"os"
	"os/signal"
	"path/filepath"
	"syscall"
	"time"

	"git.vakhrushev.me/av/healthlog/internal/archive"
	"git.vakhrushev.me/av/healthlog/internal/config"
	"git.vakhrushev.me/av/healthlog/internal/fold"
	"git.vakhrushev.me/av/healthlog/internal/ident"
	"git.vakhrushev.me/av/healthlog/internal/logging"
	"git.vakhrushev.me/av/healthlog/internal/replay"
	"git.vakhrushev.me/av/healthlog/internal/store"
)

// rebuildSuffix — как зовётся собранная витрина рядом с рабочей базой.
// Соседом, а не во временном каталоге: подмена обязана быть переименованием
// внутри одной файловой системы.
const rebuildSuffix = ".rebuild"

// partialSuffix — под каким именем витрина собирается, пока не готова.
//
// Полусобранная база выглядит как обычная, и файл с именем результата человек
// подменит по напечатанной процедуре не глядя. Поэтому имя результата
// появляется последним шагом успеха, а не первым шагом работы.
const partialSuffix = ".partial"

// progressInterval — как часто печатается прогресс. Прогон на полном архиве
// идёт минутами и молчит; зависший при этом неотличим от идущего.
const progressInterval = 5 * time.Second

// errNothingReplayed — журнал пуст или не свернулось ничего.
var errNothingReplayed = errors.New("проигрывать нечего")

func runReindex(args []string) error {
	fs := flag.NewFlagSet("reindex", flag.ContinueOnError)
	cfgPath := fs.String("config", config.DefaultPath, "путь к config.toml")
	out := fs.String("out", "", "куда собрать витрину (по умолчанию — рабочая база с суффиксом "+rebuildSuffix+")")
	force := fs.Bool("force", false, "перезаписать существующий файл назначения")
	if err := fs.Parse(args); err != nil {
		if errors.Is(err, flag.ErrHelp) {
			// Справка — не отказ: иначе `reindex -h` печатает usage и выходит
			// со словом «fatal» и кодом 1.
			return nil
		}
		return fmt.Errorf("parse flags: %w", err)
	}

	cfg, err := config.Load(*cfgPath)
	if err != nil {
		return err
	}
	// Лог — в stderr: stdout занят отчётом человеку, и лог в том же потоке
	// сделал бы отчёт неразбираемым.
	log := logging.NewErr(cfg.Log.Level, cfg.Log.Format)

	target, err := resolveTarget(cfg.Storage.DBPath, *out, *force)
	if err != nil {
		return err
	}

	// Отмена приходит из сигнала: команду прерывает человек, и без этого вся
	// логика отмены недостижима — процесс умирал бы мимо неё.
	ctx, stop := signal.NotifyContext(context.Background(), syscall.SIGINT, syscall.SIGTERM)
	defer stop()

	// Прогресс — в поток ошибок: stdout занят отчётом, который человек
	// перенаправляет и читает глазами.
	rep, err := rebuild(ctx, cfg, target, log, os.Stderr)
	if err != nil {
		return err
	}

	writeReport(os.Stdout, rep)
	if rep.replay.Canceled {
		return errors.New("пересборка отменена")
	}
	if rep.replay.Bodies == 0 || rep.replay.Folded == 0 {
		// Пустая витрина совпадает по отпечатку с пустой витриной, то есть
		// пустой прогон выглядит идеальной сходимостью. Успехом он быть не
		// может: человек, выполнивший напечатанную процедуру, заменил бы
		// накопленное пустым.
		return errNothingReplayed
	}
	return nil
}

// target — куда собираем и как называется промежуточный файл.
type target struct {
	final   string
	partial string
}

// resolveTarget выбирает файл назначения и проверяет, что писать в него можно.
func resolveTarget(dbPath, out string, force bool) (target, error) {
	final := out
	if final == "" {
		final = dbPath + rebuildSuffix
	}

	// Тождество определяется файлом, а не строкой пути: `..`, симлинк или
	// другой префикс монтирования дают ту же цель при другой строке, а ошибка
	// здесь означает проигрывание журнала прямо в живую рабочую базу.
	same, err := sameFile(final, dbPath)
	if err != nil {
		return target{}, err
	}
	if same {
		return target{}, fmt.Errorf("файл назначения %q — это рабочая база", final)
	}

	if _, err := os.Stat(final); err == nil && !force {
		return target{}, fmt.Errorf("файл назначения %q уже существует (--force перезапишет)", final)
	} else if err != nil && !errors.Is(err, os.ErrNotExist) {
		return target{}, fmt.Errorf("stat %q: %w", final, err)
	}

	// Имя промежуточного файла уникально: фиксированное затирало бы чужой файл
	// с тем же именем ДО всякой проверки, то есть мимо правила «без --force не
	// перезаписываем», и обломок прошлого прогона блокировал бы следующий.
	return target{final: final, partial: final + "." + ident.NewID() + partialSuffix}, nil
}

// sameFile отвечает, ведут ли два пути к одному файлу.
//
// Когда файла назначения ещё нет, сравниваются каталог-родитель и имя: сам файл
// сравнить не с чем, а совпадение каталога и имени — это и есть тождество
// будущего файла.
func sameFile(a, b string) (bool, error) {
	// Совпадение очищенных путей — тождество независимо от того, существуют ли
	// файлы. Без этой проверки `--out <db_path>` при отсутствующей рабочей базе
	// устанавливал бы витрину прямо на её место, минуя всё правило «подмену
	// делает человек при остановленном сервисе».
	if filepath.Clean(a) == filepath.Clean(b) {
		return true, nil
	}

	fa, errA := os.Stat(a)
	fb, errB := os.Stat(b)
	switch {
	case errA == nil && errB == nil:
		return os.SameFile(fa, fb), nil
	case errB != nil:
		// Рабочей базы нет: сравнивать не с чем, а совпадение строк уже
		// исключено выше.
		return false, nil
	}

	da, err := os.Stat(filepath.Dir(a))
	if err != nil {
		return false, fmt.Errorf("stat %q: %w", filepath.Dir(a), err)
	}
	db, err := os.Stat(filepath.Dir(b))
	if err != nil {
		return false, fmt.Errorf("stat %q: %w", filepath.Dir(b), err)
	}
	return os.SameFile(da, db) && filepath.Base(a) == filepath.Base(b), nil
}

// report — всё, что печатается человеку.
type report struct {
	replay replay.Report

	target        string
	dbPath        string
	sourcePrint   string
	sourceBuckets int64
	// sourceWorkouts и sourceRecords — то же «было» для остальных единиц
	// хранения витрины. Отпечаток отвечает «да/нет» за витрину целиком, поэтому
	// единица, которой нет в счётчиках, делает расхождение безадресным.
	sourceWorkouts int64
	sourceRecords  int64
	sourceBefore   int64
	sourceAfter    int64
	sourceMissing  bool
}

// rebuild собирает витрину в промежуточный файл и переименовывает его в файл
// назначения последним шагом успеха.
func rebuild(ctx context.Context, cfg *config.Config, t target, log *slog.Logger, progress io.Writer) (report, error) {
	rep := report{target: t.final, dbPath: cfg.Storage.DBPath}

	// Отмена — не отказ пересборки, а требование прекратить работу, и застать
	// она может на любом шаге, включая снятие отпечатка рабочей витрины.
	stopped := func(err error) bool {
		return errors.Is(err, context.Canceled) || errors.Is(err, context.DeadlineExceeded)
	}
	if ctx.Err() != nil {
		rep.replay.Canceled = true
		return rep, nil
	}

	arch, err := archive.Existing(cfg.Storage.ArchiveDir)
	if err != nil {
		return rep, err
	}

	// Рабочей базы может не быть вовсе — журнал тогда состоит из одних
	// подобранных тел. Это законный вход: восстановление после её потери. Но
	// заголовки доставок при этом не воскресают, они жили только в ней.
	var src *store.Store
	if _, err := os.Stat(cfg.Storage.DBPath); errors.Is(err, os.ErrNotExist) {
		rep.sourceMissing = true
	} else if err != nil {
		return rep, fmt.Errorf("stat %q: %w", cfg.Storage.DBPath, err)
	} else {
		src, err = store.OpenForRead(cfg.Storage.DBPath)
		if err != nil {
			return rep, err
		}
		defer func() { _ = src.Close() }()

		// Отпечаток рабочей витрины снимается ДО проигрывания, иначе под живым
		// приёмом он всегда движется, и оракул отвечает «разошлись» независимо
		// от того, разошёлся ли разбор.
		if rep.sourcePrint, err = src.Fingerprint(ctx); err != nil {
			return canceledOr(rep, err, stopped)
		}
		if rep.sourceBefore, err = src.CountDeliveries(ctx); err != nil {
			return canceledOr(rep, err, stopped)
		}
		if rep.sourceBuckets, err = src.CountBuckets(ctx); err != nil {
			return canceledOr(rep, err, stopped)
		}
		if rep.sourceWorkouts, err = src.CountWorkouts(ctx); err != nil {
			return canceledOr(rep, err, stopped)
		}
		if rep.sourceRecords, err = src.CountRecords(ctx); err != nil {
			return canceledOr(rep, err, stopped)
		}
	}

	removeDB(t.partial)
	dst, err := store.Open(t.partial)
	if err != nil {
		return rep, err
	}

	rep.replay, err = replay.Run(ctx, replay.Options{
		Archive: arch,
		Source:  src,
		Target:  dst,
		// `mode=replay` в логе не украшение: за один прогон через слияние
		// проходит вся история, и её WARN о перезаписях иначе неотличимы от
		// аномалий живого приёма в общем логе.
		Fold:     fold.New(arch, dst, int64(cfg.Ingest.MaxBodyMB)<<20, log.With("mode", "replay")),
		Progress: progressEvery(progress, progressInterval, time.Now),
		Log:      log,
	})
	if cerr := dst.Close(); err == nil {
		err = cerr
	}
	if err != nil {
		removeDB(t.partial)
		return rep, err
	}

	if src != nil && !rep.replay.Canceled {
		if rep.sourceAfter, err = src.CountDeliveries(ctx); err != nil {
			removeDB(t.partial)
			return canceledOr(rep, err, stopped)
		}
	}

	ok := !rep.replay.Canceled && rep.replay.Bodies > 0 && rep.replay.Folded > 0
	if !ok {
		removeDB(t.partial)
		return rep, nil
	}
	if err := os.Rename(t.partial, t.final); err != nil {
		removeDB(t.partial)
		return rep, fmt.Errorf("переименование в %q: %w", t.final, err)
	}
	return rep, nil
}

// progressEvery печатает прогресс не чаще интервала.
//
// Живёт в команде, а не в пакете проигрывания: «куда и как часто печатать» —
// забота адресата вывода. Часы параметром, чтобы функция была проверяема, не
// завися от настоящего времени.
func progressEvery(w io.Writer, every time.Duration, now func() time.Time) func(done, total int) {
	last := now()
	return func(done, total int) {
		if done < total && now().Sub(last) < every {
			return
		}
		last = now()
		_, _ = fmt.Fprintf(w, "проиграно %d из %d\n", done, total)
	}
}

// canceledOr отличает отмену от настоящего отказа: первая не является ошибкой
// команды, вторая является.
func canceledOr(rep report, err error, stopped func(error) bool) (report, error) {
	if stopped(err) {
		rep.replay.Canceled = true
		return rep, nil
	}
	return rep, err
}

// removeDB убирает файл базы вместе со спутниками журнала SQLite: оставленный
// `-wal` подцепится к следующему файлу с тем же именем.
func removeDB(path string) {
	for _, s := range []string{"", "-wal", "-shm"} {
		_ = os.Remove(path + s)
	}
}