jellybit/internal/worker/worker.go

// Package worker — владелец машины состояний. Поллит qBittorrent по
// категории, переводит задачи между состояниями и сериализует команды
// транспортов (cancel/retry), чтобы два транспорта не гонялись за одно
// состояние.
//
// Ф1 ведёт задачу downloading → completed, плюс stuck/failed по таймаутам и
// ошибкам qBittorrent. Ф3 продолжает: completed → recognizing (вызов
// recognize) → review; команды ревью (apply/refine/reject/defer/undo,
// переключение типа, пометка «игнор») раскладывают файлы хардлинками через
// layout. Распознавание зовётся в поллинг-цикле, команды — из транспортов;
// всё под per-download блокировкой w.mu.
package worker

import (
	"context"
	"fmt"
	"log/slog"
	"strings"
	"sync"
	"time"

	"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/recognize"
	"git.vakhrushev.me/av/jellybit/internal/store"
)

// Стадии (capability) — адресуют запись к подсистеме при корреляции по
// download_id. Уровень логов от стадии не зависит.
const (
	capIngest     = "ingest"      // приём/скачивание (поллинг, reconcile, переходы)
	capRecognize  = "recognition" // распознавание фильма/сериала
	capFileLayout = "file-layout" // раскладка хардлинками
	capReview     = "review"      // ручные команды ревью
)

// Store — нужная worker часть хранилища.
type Store interface {
	ListDownloadsByState(ctx context.Context, states ...store.State) ([]store.Download, error)
	ListRecoverable(ctx context.Context, codes ...string) ([]store.Download, error)
	GetDownload(ctx context.Context, id int64) (*store.Download, error)
	SetDownloadState(ctx context.Context, id int64, state store.State, errCode, errMsg string) error
	SetSourceMissCount(ctx context.Context, id int64, n int) error
	SetSourceAddedAt(ctx context.Context, id int64, t time.Time) error

	// Discovery (усыновление раздач по категории/тегу).
	ExistsByInfohash(ctx context.Context, infohash string) (bool, error)
	FindActiveByInfohash(ctx context.Context, infohash string) (*store.Download, error)
	CreateDownload(ctx context.Context, d *store.Download) (int64, error)

	// Ф3: распознавание, ревью, раскладка.
	CreateRecognition(ctx context.Context, r *store.Recognition, reasons []string) (int64, error)
	GetCurrentRecognition(ctx context.Context, downloadID int64) (*store.Recognition, error)
	AddHint(ctx context.Context, downloadID int64, text string) error
	ListHints(ctx context.Context, downloadID int64) ([]string, error)
	SetOverride(ctx context.Context, downloadID int64, field, value string) error
	ListOverrides(ctx context.Context, downloadID int64) (map[string]string, error)
	CreateFileLinks(ctx context.Context, links []store.FileLink) error
	SupersedeForeignLinks(ctx context.Context, downloadID int64, dstPaths []string) error
	LatestBatchID(ctx context.Context, downloadID int64) (string, error)
	ListFileLinksByBatch(ctx context.Context, batchID string) ([]store.FileLink, error)
	DeleteFileLinksByBatch(ctx context.Context, batchID string) error

	// Кандидаты базы метаданных (ручной выбор в review).
	CreateCandidates(ctx context.Context, cands []store.MetadataCandidate) error
	ListCandidatesByRecognition(ctx context.Context, recognitionID int64) ([]store.MetadataCandidate, error)
	GetCandidate(ctx context.Context, id int64) (*store.MetadataCandidate, error)
	SetCandidateChosen(ctx context.Context, recognitionID, candidateID int64) error
}

// QBittorrent — нужная worker часть клиента qBittorrent.
type QBittorrent interface {
	Torrents(ctx context.Context, category string) ([]qbt.Torrent, error)
	Add(ctx context.Context, ar qbt.AddRequest) error
	Files(ctx context.Context, hash string) ([]qbt.File, error)
}

// Recognizer — распознаватель (recognize.Recognizer).
type Recognizer interface {
	Recognize(ctx context.Context, in recognize.Input) (recognize.Result, error)
}

// Layouter — раскладчик хардлинками (layout.Layouter).
type Layouter interface {
	BuildLinks(p layout.Plan) ([]layout.Link, error)
	Apply(ctx context.Context, links []layout.Link) ([]layout.Result, error)
	Undo(ctx context.Context, links []layout.Link) (int, error)
}

// NotifyEvent — повод позвать пользователя.
type NotifyEvent string

const (
	EventReview        NotifyEvent = "review"         // задача ждёт подтверждения
	EventDone          NotifyEvent = "done"           // раскладка завершена
	EventOrphaned      NotifyEvent = "orphaned"       // источник пропал, цель — последняя копия
	EventTargetMissing NotifyEvent = "target_missing" // цель удалена, доступен relink
	EventFailed        NotifyEvent = "failed"         // задача упала/зависла (failed/stuck)
)

// Коды ошибок (error_code) при переходе в failed/stuck. Восстановимые
// (magnet_timeout/stalled) — следствие нашей нетерпеливости: сверка воскрешает
// такие задачи при оживлении источника (см. reconcileRecovery). qbit_error —
// реальная ошибка qBittorrent, восстановлению не подлежит.
const (
	errCodeMagnetTimeout = "magnet_timeout"
	errCodeStalled       = "stalled"
	errCodeQbitError     = "qbit_error"
)

// Notifier — исходящие пинги (Telegram). Вызывается неблокирующе.
type Notifier interface {
	Notify(ctx context.Context, downloadID int64, event NotifyEvent)
}

// Scanner — триггер пересканирования медиатеки Jellyfin. Вызывается
// неблокирующе после успешной раскладки, чтобы новые файлы быстрее появились
// в проигрывателе.
type Scanner interface {
	RefreshLibraries(ctx context.Context) error
}

// Config — параметры воркера.
type Config struct {
	Category      string
	Tag           string // метка для усыновления существующих раздач (discovery)
	SavePath      string
	PathMap       map[string]string // трансляция save_path qBit → хост-путь (обычно пусто)
	PollInterval  time.Duration
	StuckAfter    time.Duration // stalledDL дольше → stuck
	MagnetTimeout time.Duration // metaDL дольше → failed
	// SourceMissingThreshold — порог дебаунса пропажи источника (тиков сверки).
	// <1 трактуется как 1 (помечаем при первой же устойчивой пропаже).
	SourceMissingThreshold int
}

// Live — живая телеметрия одной раздачи из снимка воркера. Курированный срез
// qbt.Torrent: читатели (httpapi) не зависят от пакета qbt, контракт чтения
// узкий. Seeding вычисляется воркером (classify), чтобы трактовка завершённости
// не дублировалась в транспорте.
type Live struct {
	Progress float64 // доля 0..1
	DlSpeed  int64   // скорость загрузки, байт/с
	ETA      int64   // оценка до завершения, с (8640000 ≈ ∞)
	State    string  // сырое состояние qBittorrent
	Seeding  bool    // торрент завершён и раздаётся

	Ratio    float64 // рейтинг отдачи (может быть <0 = ∞/н/д)
	Seeds    int     // подключённые сиды
	Peers    int     // подключённые личи
	Uploaded int64   // отдано всего, байт
	UpSpeed  int64   // скорость отдачи, байт/с
}

// liveFrom собирает Live из торрента qBittorrent (Seeding — через classify).
func liveFrom(t qbt.Torrent) Live {
	return Live{
		Progress: t.Progress,
		DlSpeed:  t.Dlspeed,
		ETA:      t.Eta,
		State:    t.State,
		Seeding:  classify(t.State) == classReady,
		Ratio:    t.Ratio,
		Seeds:    t.NumSeeds,
		Peers:    t.NumLeechs,
		Uploaded: t.Uploaded,
		UpSpeed:  t.Upspeed,
	}
}

// Worker — поллер и владелец переходов.
type Worker struct {
	store      Store
	qbt        QBittorrent
	recognizer Recognizer
	layouter   Layouter
	cfg        Config
	log        *slog.Logger

	mu       sync.Mutex       // сериализует переходы (поллинг + команды)
	now      func() time.Time // подменяется в тестах
	newID    func() string    // генератор apply_batch_id (подменяется в тестах)
	notifier Notifier         // опц. исходящие пинги
	scanner  Scanner          // опц. пересканирование Jellyfin

	// live — снимок живой телеметрии раздач (ключ — lowercase infohash, по три
	// ключа на торрент, как byHash). Обновляется атомарным свопом карты на
	// каждом тике Poll. Отдельный RWMutex (не w.mu): UI читает телеметрию часто,
	// смешивать частые чтения с замком переходов — лишняя конкуренция. Снимок
	// волатилен, в БД не хранится.
	liveMu sync.RWMutex
	live   map[string]Live

	// failNotified — дебаунс повторных EventFailed по задаче (download_id →
	// время последнего пинга). Мерцающий stalled-торрент колеблется
	// stuck↔downloading; без дебаунса каждый цикл слал бы уведомление. Память
	// процесса: при рестарте дебаунс сбрасывается — допустимо. Доступ под w.mu.
	failNotified map[int64]time.Time
}

// failNotifyDebounce — минимальный интервал между уведомлениями о падении
// одной задачи (см. failNotified).
const failNotifyDebounce = time.Hour

// SetNotifier подключает исходящие пинги (до запуска Run).
func (w *Worker) SetNotifier(n Notifier) { w.notifier = n }

// SetScanner подключает пересканирование Jellyfin (до запуска Run).
func (w *Worker) SetScanner(s Scanner) { w.scanner = s }

// New собирает воркер. recognizer/layouter могут быть nil (Ф1 без Ф3-ступеней
// распознавания и раскладки) — тогда completed-задачи не двигаются дальше.
func New(st Store, qb QBittorrent, rec Recognizer, lay Layouter, cfg Config, log *slog.Logger) *Worker {
	return &Worker{
		store:        st,
		qbt:          qb,
		recognizer:   rec,
		layouter:     lay,
		cfg:          cfg,
		log:          log,
		now:          time.Now,
		newID:        defaultBatchID,
		failNotified: map[int64]time.Time{},
		live:         map[string]Live{},
	}
}

// Live возвращает живую телеметрию раздачи по infohash (любому из v1/v2/hash).
// ok=false, если infohash пуст или раздачи не было в последнем тике поллинга —
// тогда читатель деградирует без живых значений. Чтение под RLock.
func (w *Worker) Live(infohash string) (Live, bool) {
	if infohash == "" {
		return Live{}, false
	}
	w.liveMu.RLock()
	defer w.liveMu.RUnlock()
	l, ok := w.live[strings.ToLower(infohash)]
	return l, ok
}

// setLive атомарно подменяет снимок телеметрии готовой картой.
func (w *Worker) setLive(snap map[string]Live) {
	w.liveMu.Lock()
	w.live = snap
	w.liveMu.Unlock()
}

// defaultBatchID — уникальный идентификатор батча раскладки.
func defaultBatchID() string {
	return fmt.Sprintf("b-%d", time.Now().UnixNano())
}

// scoped кладёт в ctx scoped-логгер загрузки (capability + download_id
// [+ infohash]); стадии и внешние клиенты достают его из ctx и дописывают эти
// ключи на каждую запись сами — без ручного доклеивания download_id.
func (w *Worker) scoped(ctx context.Context, capability string, id int64, infohash string) context.Context {
	log := w.log.With("capability", capability, "download_id", id)
	if infohash != "" {
		log = log.With("infohash", infohash)
	}
	return logctx.With(ctx, log)
}

// Run крутит цикл поллинга до отмены ctx.
func (w *Worker) Run(ctx context.Context) {
	w.log.Info("worker started", "poll_interval", w.cfg.PollInterval, "category", w.cfg.Category)
	t := time.NewTicker(w.cfg.PollInterval)
	defer t.Stop()

	w.pollOnce(ctx)
	for {
		select {
		case <-ctx.Done():
			w.log.Info("worker stopped")
			return
		case <-t.C:
			w.pollOnce(ctx)
		}
	}
}

func (w *Worker) pollOnce(ctx context.Context) {
	if err := w.Poll(ctx); err != nil {
		w.log.Warn("poll failed", "error", err)
	}
	// Ф3: распознаём завершённые загрузки (и перезапускаем по подсказке).
	if w.recognizer != nil {
		w.recognizePending(ctx)
	}
}

// Poll сверяет активные задачи с состоянием qBittorrent и двигает их.
// Листаем все торренты (а не только свою категорию), чтобы reconcile нашёл и
// усыновлённые по тегу раздачи, а discovery — увидел новые.
func (w *Worker) Poll(ctx context.Context) error {
	torrents, err := w.qbt.Torrents(ctx, "")
	if err != nil {
		return fmt.Errorf("poll: list torrents: %w", err)
	}
	byHash := make(map[string]qbt.Torrent, len(torrents)*2)
	live := make(map[string]Live, len(torrents)*2)
	for _, t := range torrents {
		l := liveFrom(t)
		for _, h := range []string{t.Hash, t.InfohashV1, t.InfohashV2} {
			if h != "" {
				key := strings.ToLower(h)
				byHash[key] = t
				live[key] = l
			}
		}
	}
	// Снимок зависит только от torrents — свопаем сразу, до store-операций
	// ниже (их ранний return по ошибке не должен лишать UI свежей телеметрии).
	w.setLive(live)

	w.mu.Lock()
	defer w.mu.Unlock()

	// Усыновляем новые раздачи с нашей категорией/тегом до reconcile.
	w.discover(ctx, torrents)

	active, err := w.store.ListDownloadsByState(ctx, store.StateDownloading)
	if err != nil {
		return fmt.Errorf("poll: list active: %w", err)
	}
	for _, d := range active {
		if !d.Infohash.Valid {
			continue // нечем сопоставить (в Ф1 не случается: magnet всегда с infohash)
		}
		t, ok := byHash[strings.ToLower(d.Infohash.String)]
		if !ok {
			w.log.Warn("active download not found in qbittorrent",
				"capability", capIngest, "download_id", d.ID, "infohash", d.Infohash.String)
			continue
		}
		w.captureSourceAddedAt(ctx, d, t)
		w.reconcile(ctx, d, t)
	}

	// Сверка разложенных задач с реальностью (источник в qBit + хардлинки на ФС)
	// — отдельно от активных, по двумерной матрице (см. state-reconciliation).
	w.reconcileDesync(ctx, byHash)

	// Восстановление задач, упавших по нашей нетерпеливости (magnet_timeout/
	// stalled), если их источник в qBittorrent ожил и продвинулся.
	w.reconcileRecovery(ctx, byHash)
	return nil
}

// reconcile двигает одну задачу по состоянию её торрента. Вызывается под
// w.mu.
func (w *Worker) reconcile(ctx context.Context, d store.Download, t qbt.Torrent) {
	ctx = w.scoped(ctx, capIngest, d.ID, d.Infohash.String)
	switch classify(t.State) {
	case classReady:
		w.transition(ctx, d, store.StateCompleted, "", "")
	case classErrored:
		w.transition(ctx, d, store.StateFailed, errCodeQbitError, "qBittorrent state: "+t.State)
	case classDownloading:
		w.checkTimeouts(ctx, d, t)
	case classBusy:
		// moving/checking — ждём, файлы ещё не на финальном месте.
	}
}

// checkTimeouts помечает зависшие задачи. Возраст считаем от факта в
// qBittorrent (added_on), а не от created_at: базис переживает retry и
// усыновление раздачи (см. design download-failure-recovery). magnet_timeout —
// редкий страховочный предохранитель (дефолт 24h); настоящие провалы ловит
// classErrored, а ожившие задачи воскрешает reconcileRecovery.
func (w *Worker) checkTimeouts(ctx context.Context, d store.Download, t qbt.Torrent) {
	age := w.torrentAge(d, t)

	switch {
	case isMeta(t.State) && w.cfg.MagnetTimeout > 0 && age > w.cfg.MagnetTimeout:
		w.transition(ctx, d, store.StateFailed, errCodeMagnetTimeout,
			fmt.Sprintf("no metadata after %s", age.Truncate(time.Second)))
	case isStalledDL(t.State) && w.cfg.StuckAfter > 0 && age > w.cfg.StuckAfter:
		w.transition(ctx, d, store.StateStuck, errCodeStalled,
			fmt.Sprintf("stalled for %s", age.Truncate(time.Second)))
	}
}

// captureSourceAddedAt однократно сохраняет время добавления торрента в
// qBittorrent (added_on) у задачи — базис сортировки списка. Пишем только при
// первом наблюдении (в БД source_added_at ещё пуст, SQL-гард в store); значение
// неизменно, поэтому повторные тики его не трогают. Учётная операция: её сбой не
// двигает задачу, лишь логируем WARN. Вызывается под w.mu.
func (w *Worker) captureSourceAddedAt(ctx context.Context, d store.Download, t qbt.Torrent) {
	if d.SourceAddedAt.Valid || t.AddedOn <= 0 {
		return
	}
	if err := w.store.SetSourceAddedAt(ctx, d.ID, time.Unix(t.AddedOn, 0)); err != nil {
		w.log.Warn("capture source_added_at failed",
			"capability", capIngest, "download_id", d.ID, "error", err)
	}
}

// torrentAge — возраст торрента: от added_on в qBittorrent (надёжный базис,
// переживает retry/усыновление), с фолбэком на created_at задачи, если qBit не
// отдал added_on.
func (w *Worker) torrentAge(d store.Download, t qbt.Torrent) time.Duration {
	if t.AddedOn > 0 {
		return w.now().Sub(time.Unix(t.AddedOn, 0).UTC())
	}
	created, err := d.CreatedTime()
	if err != nil {
		// Ни added_on от qBit, ни разбираемого created_at — возраст неизвестен,
		// таймауты не сработают; фиксируем диагностикой.
		w.log.Warn("cannot determine torrent age",
			"capability", capIngest, "download_id", d.ID,
			"created_at", d.CreatedAt, "error", err)
		return 0
	}
	return w.now().Sub(created)
}

// transition пишет новое состояние и логирует переход.
func (w *Worker) transition(ctx context.Context, d store.Download, state store.State, code, msg string) {
	// FromOr, а не From: если вызывающий не завёл scoped-логгер, падаем на
	// w.log (настроенный), а не на slog.Default().
	log := logctx.FromOr(ctx, w.log)
	if err := w.store.SetDownloadState(ctx, d.ID, state, code, msg); err != nil {
		log.Error("state transition failed", "from", d.State, "to", state, "error", err)
		return
	}
	log.Info("state transition", "from", d.State, "to", state, "code", code)

	// Пинги — неблокирующе и в отдельном контексте: вызов уходит в сеть, а
	// мы под w.mu (Notify читает состояние уже после освобождения замка).
	if w.notifier != nil {
		switch state {
		case store.StateReview:
			go w.notifier.Notify(context.Background(), d.ID, EventReview)
		case store.StateDone:
			go w.notifier.Notify(context.Background(), d.ID, EventDone)
		case store.StateOrphaned:
			go w.notifier.Notify(context.Background(), d.ID, EventOrphaned)
		case store.StateTargetMissing:
			go w.notifier.Notify(context.Background(), d.ID, EventTargetMissing)
		case store.StateFailed, store.StateStuck:
			if w.shouldNotifyFail(d.ID) {
				go w.notifier.Notify(context.Background(), d.ID, EventFailed)
			}
		}
	}

	// Раскладка завершена — просим Jellyfin пересканировать библиотеку, чтобы
	// новые файлы быстрее появились в проигрывателе. Тоже неблокирующе и вне
	// w.mu; недоступность Jellyfin не влияет на состояние задачи.
	if w.scanner != nil && state == store.StateDone {
		// Скан Jellyfin — неблокирующе и вне w.mu, в фоновом ctx со scoped-логгером
		// (download_id для корреляции ext.*-записи клиента). Недоступность Jellyfin
		// на задачу не влияет; ошибку вызова логирует сам клиент (ext.*), здесь гасим.
		gctx := w.scoped(context.Background(), capFileLayout, d.ID, d.Infohash.String)
		go func() { _ = w.scanner.RefreshLibraries(gctx) }()
	}
}

// shouldNotifyFail дебаунсит повторные уведомления о падении одной задачи
// (мерцающий stalled-торрент: stuck↔downloading), чтобы не спамить. Вызывается
// под w.mu. НЕ сбрасываем запись при восстановлении — иначе дебаунс не гасил бы
// флаппинг.
func (w *Worker) shouldNotifyFail(id int64) bool {
	now := w.now()
	if last, ok := w.failNotified[id]; ok && now.Sub(last) < failNotifyDebounce {
		return false
	}
	w.failNotified[id] = now
	// Лёгкая чистка устаревших записей, чтобы карта не росла без предела.
	for k, t := range w.failNotified {
		if now.Sub(t) >= failNotifyDebounce {
			delete(w.failNotified, k)
		}
	}
	return true
}

// Cancel отклоняет задачу. Торрент в qBittorrent не трогаем — он продолжает
// раздачу (источник неприкосновенен).
func (w *Worker) Cancel(ctx context.Context, id int64) error {
	w.mu.Lock()
	defer w.mu.Unlock()

	d, err := w.store.GetDownload(ctx, id)
	if err != nil {
		return fmt.Errorf("cancel: %w", err)
	}
	if d.State.IsTerminal() {
		return fmt.Errorf("cancel: download %d is already terminal (%s)", id, d.State)
	}
	if err := w.store.SetDownloadState(ctx, id, store.StateCancelled, "", ""); err != nil {
		return fmt.Errorf("cancel: %w", err)
	}
	logctx.From(w.scoped(ctx, capReview, id, d.Infohash.String)).Info("download cancelled", "from", d.State)
	return nil
}

// Retry повторяет застрявшую/упавшую задачу: заново отдаёт источник в
// qBittorrent и возвращает в downloading.
func (w *Worker) Retry(ctx context.Context, id int64) error {
	w.mu.Lock()
	defer w.mu.Unlock()

	d, err := w.store.GetDownload(ctx, id)
	if err != nil {
		return fmt.Errorf("retry: %w", err)
	}
	if d.State != store.StateFailed && d.State != store.StateStuck {
		return fmt.Errorf("retry: download %d is %s, only failed/stuck are retriable", id, d.State)
	}
	// Если раздача уже жива в qBittorrent — перецепляемся к ней, повторный Add
	// не нужен (и вреден: вслепую дублировал бы торрент). Add — только когда
	// источника в qBittorrent нет. Базис таймаута берётся от added_on, поэтому
	// возврат в downloading не роняет задачу снова на ближайшем тике.
	alive := false
	if d.Infohash.Valid {
		_, alive, err = w.torrentByInfohash(ctx, d.Infohash.String)
		if err != nil {
			return fmt.Errorf("retry: %w", err)
		}
	}
	if !alive && d.SourceType == store.SourceMagnet {
		if err := w.qbt.Add(ctx, qbt.AddRequest{
			URLs:     []string{d.SourceRef},
			Category: w.cfg.Category,
			SavePath: w.cfg.SavePath,
		}); err != nil {
			return fmt.Errorf("retry: add to qbittorrent: %w", err)
		}
	}
	if err := w.store.SetDownloadState(ctx, id, store.StateDownloading, "", ""); err != nil {
		return fmt.Errorf("retry: %w", err)
	}
	logctx.From(w.scoped(ctx, capReview, id, d.Infohash.String)).Info("download retried", "from", d.State)
	return nil
}

// class — класс состояния торрента qBittorrent.
type class int

const (
	classDownloading class = iota // ещё качается
	classReady                    // готов к раскладке
	classErrored                  // ошибка
	classBusy                     // moving/checking — переходный момент, ждём
)

// classify относит состояние qBittorrent к классу (см. architecture.md,
// «Завершение в qBittorrent»). Учитываем и v5-имена (stopped* вместо
// paused*).
func classify(state string) class {
	switch state {
	case "uploading", "stalledUP", "pausedUP", "stoppedUP", "queuedUP", "forcedUP":
		return classReady
	case "error", "missingFiles":
		return classErrored
	case "moving", "checkingUP", "checkingResumeData", "allocating":
		return classBusy
	default:
		// downloading, stalledDL, metaDL, forcedMetaDL, queuedDL, checkingDL,
		// forcedDL, pausedDL, stoppedDL, unknown — считаем «ещё качается».
		return classDownloading
	}
}

func isMeta(state string) bool {
	return state == "metaDL" || state == "forcedMetaDL"
}

func isStalledDL(state string) bool {
	return state == "stalledDL"
}