jellybit/internal/worker/reconcile.go

package worker

import (
	"context"
	"errors"
	"fmt"
	"os"
	"strings"

	"git.vakhrushev.me/av/jellybit/internal/layout"
	"git.vakhrushev.me/av/jellybit/internal/logctx"
	"git.vakhrushev.me/av/jellybit/internal/qbt"
	"git.vakhrushev.me/av/jellybit/internal/store"
)

// desyncStates — состояния, которые ведёт сверка с реальностью: уже
// разложенные (done) и восстановимые состояния рассинхрона. deleted сюда не
// входит — оно терминально и сверкой не переоценивается (источник к
// терминальной задаче не вернётся из-за идемпотентности, а цель отбирается
// переходом владения путём; см. state-reconciliation). Активные и
// пользовательски-терминальные (reverted/cancelled/failed/stuck) сверка не
// трогает.
var desyncStates = []store.State{
	store.StateDone,
	store.StateTargetMissing,
	store.StateOrphaned,
}

// deriveState выводит состояние задачи из двумерной матрицы «источник × цель»
// (см. state-reconciliation, D1).
func deriveState(sourcePresent, targetPresent bool) store.State {
	switch {
	case sourcePresent && targetPresent:
		return store.StateDone
	case sourcePresent && !targetPresent:
		return store.StateTargetMissing
	case !sourcePresent && targetPresent:
		return store.StateOrphaned
	default:
		return store.StateDeleted
	}
}

// reconcileDesync сверяет разложенные/desync-задачи с реальностью. Вызывается
// из Poll под w.mu. byHash — карта присутствующих в qBittorrent раздач.
func (w *Worker) reconcileDesync(ctx context.Context, byHash map[string]qbt.Torrent) {
	cands, err := w.store.ListDownloadsByState(ctx, desyncStates...)
	if err != nil {
		w.log.Warn("reconcile list desync failed", "capability", capIngest, "error", err)
		return
	}
	for _, d := range cands {
		w.reconcileOneDesync(ctx, d, byHash)
	}
}

// reconcileOneDesync сверяет одну задачу: вычисляет присутствие источника (с
// дебаунсом) и цели, выводит состояние и переходит при изменении.
func (w *Worker) reconcileOneDesync(ctx context.Context, d store.Download, byHash map[string]qbt.Torrent) {
	if !d.Infohash.Valid {
		return // нечем сопоставить источник
	}
	ctx = w.scoped(ctx, capIngest, d.ID, d.Infohash.String)
	_, sourceSeen := byHash[strings.ToLower(d.Infohash.String)]

	// Дебаунс пропажи источника: считаем удалённым только после порога подряд
	// идущих промахов; любое появление сбрасывает счётчик.
	sourcePresent := w.debounceSource(ctx, d, sourceSeen)

	targetPresent, err := w.targetPresent(ctx, d.ID)
	if err != nil {
		logctx.From(ctx).Warn("reconcile target probe failed", "error", err)
		return // не можем проверить цель — не дёргаем состояние
	}

	want := deriveState(sourcePresent, targetPresent)
	if want == d.State {
		return
	}
	w.transition(ctx, d, want, "reconcile", reconcileReason(sourcePresent, targetPresent))
}

// debounceSource обновляет счётчик промахов источника и возвращает, считать ли
// источник присутствующим с учётом порога. Запись счётчика — только при его
// изменении (без лишних UPDATE на каждом тике).
func (w *Worker) debounceSource(ctx context.Context, d store.Download, sourceSeen bool) bool {
	threshold := max(w.cfg.SourceMissingThreshold, 1)
	if sourceSeen {
		if d.SourceMissCount != 0 {
			if err := w.store.SetSourceMissCount(ctx, d.ID, 0); err != nil {
				logctx.From(ctx).Warn("reconcile reset miss count failed", "error", err)
			}
		}
		return true
	}
	miss := d.SourceMissCount + 1
	if err := w.store.SetSourceMissCount(ctx, d.ID, miss); err != nil {
		logctx.From(ctx).Warn("reconcile bump miss count failed", "error", err)
	}
	// До порога источник трактуется как присутствующий — задача не дёргается.
	return miss < threshold
}

// targetPresent сообщает, существуют ли разложенные хардлинки задачи. Цель
// считается присутствующей, только если существуют ВСЕ ссылки последнего
// батча; частичная пропажа — это отсутствие цели (библиотека сломана → relink).
func (w *Worker) targetPresent(ctx context.Context, id int64) (bool, error) {
	batch, err := w.store.LatestBatchID(ctx, id)
	if err != nil {
		return false, fmt.Errorf("latest batch: %w", err)
	}
	if batch == "" {
		return false, nil // ничего не разложено — цели нет
	}
	rows, err := w.store.ListFileLinksByBatch(ctx, batch)
	if err != nil {
		return false, fmt.Errorf("list links: %w", err)
	}
	laidOut := 0
	for _, r := range rows {
		// Все статусы, означающие реальный файл на диске: только что слинкован,
		// скопирован (фолбэк) или уже существовал тем же inode (идемпотентный
		// повтор apply). Прочие (collision и т.п.) — не наша разложенная цель.
		if !isLaidOut(r.Status) {
			continue
		}
		laidOut++
		if _, err := os.Lstat(r.DstPath); err != nil {
			if errors.Is(err, os.ErrNotExist) {
				return false, nil
			}
			return false, fmt.Errorf("stat target %q: %w", r.DstPath, err)
		}
	}
	return laidOut > 0, nil
}

// isLaidOut сообщает, означает ли статус file_link реально лежащий на ФС файл
// нашей раскладки (а не коллизию или иной не-разложенный исход).
func isLaidOut(status string) bool {
	switch layout.LinkStatus(status) {
	case layout.StatusLinked, layout.StatusCopied, layout.StatusExists:
		return true
	default:
		return false
	}
}

// reconcileReason — человекочитаемая причина перехода для лога/UI.
func reconcileReason(sourcePresent, targetPresent bool) string {
	switch {
	case sourcePresent && targetPresent:
		return "источник и цель на месте"
	case sourcePresent && !targetPresent:
		return "цель удалена, источник на месте"
	case !sourcePresent && targetPresent:
		return "источник удалён, цель (последняя копия) на месте"
	default:
		return "удалены и источник, и цель"
	}
}

// --- Синхронный preflight перед действием (не доверяем state в БД) ---

// ensureSourcePresent синхронно (без дебаунса) проверяет, что раздача есть в
// qBittorrent прямо сейчас. При отсутствии приводит состояние к реальности и
// возвращает ErrConflict. Недоступность qBittorrent — честный отказ операции.
func (w *Worker) ensureSourcePresent(ctx context.Context, d *store.Download, op string) error {
	if !d.Infohash.Valid {
		return fmt.Errorf("%s: download %d has no infohash", op, d.ID)
	}
	_, ok, err := w.torrentByInfohash(ctx, d.Infohash.String)
	if err != nil {
		return fmt.Errorf("%s: %w", op, err)
	}
	if ok {
		return nil
	}
	// Источник пропал — немедленно приводим состояние к реальности.
	w.reconcileToReality(ctx, *d, false)
	return fmt.Errorf("%s: источник удалён из qBittorrent: %w", op, ErrConflict)
}

// reconcileToReality выводит и проставляет состояние по уже известному факту об
// источнике (sourcePresent) и фактически проверенной цели. Используется
// preflight-проверками: немедленно, без дебаунса.
func (w *Worker) reconcileToReality(ctx context.Context, d store.Download, sourcePresent bool) {
	targetPresent, err := w.targetPresent(ctx, d.ID)
	if err != nil {
		logctx.FromOr(ctx, w.log).Warn("preflight target probe failed", "error", err)
		return
	}
	want := deriveState(sourcePresent, targetPresent)
	if want == d.State {
		return
	}
	w.transition(ctx, d, want, "reconcile", reconcileReason(sourcePresent, targetPresent))
}