jellybit/internal/worker/discover.go

package worker

import (
	"context"
	"slices"
	"strings"
	"time"

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

// discover усыновляет новые раздачи: для каждого торрента с нашей категорией
// ИЛИ тегом, чьих хешей ещё нет в БД, заводит задачу downloading. Дальше
// её ведёт обычный reconcile. Вызывается под w.mu.
//
// Корректность при гонке с Ingest (другая горутина): и adopt, и приём идут
// через store.CreateDownloadIfNoActive — атомарный check-then-insert в одной
// write-транзакции; при столкновении второй участник получает существующую
// активную задачу и просто пропускает.
func (w *Worker) discover(ctx context.Context, torrents []qbt.Torrent) {
	for _, t := range torrents {
		if w.tracked(t) {
			w.adopt(ctx, t)
		}
	}
}

// tracked — относится ли торрент к jellybit (категория или тег из конфига).
func (w *Worker) tracked(t qbt.Torrent) bool {
	if w.cfg.Category != "" && t.Category == w.cfg.Category {
		return true
	}
	return hasTag(t.Tags, w.cfg.Tag)
}

// adopt заводит задачу под торрент, если его ещё не видели.
func (w *Worker) adopt(ctx context.Context, t qbt.Torrent) {
	hashes := torrentHashes(t)
	if len(hashes) == 0 {
		return // нечем идентифицировать (напр. ещё metaDL без хэша)
	}
	exists, err := w.store.ExistsByInfohash(ctx, hashes...)
	if err != nil {
		w.log.Warn("discover exists check failed", "capability", capIngest, "infohash", hashes[0], "error", err)
		return
	}
	if exists {
		return // уже усыновлён ранее (или обработан) — не трогаем
	}

	d := &store.Download{
		SourceType:  store.SourceMagnet,
		SourceRef:   magnetURN(hashes[0]),
		DisplayName: t.Name, // усыновление: приёма/rename нет, берём имя торрента из qBittorrent
		State:       store.StateDownloading,
	}
	existing, err := w.store.CreateDownloadIfNoActive(ctx, d, hashes)
	if err != nil {
		w.log.Error("discover adopt failed", "capability", capIngest, "infohash", hashes[0], "error", err)
		return
	}
	if existing != nil {
		return // гонка с Ingest/другим тиком: задача уже заведена — всё ок
	}
	// Базис сортировки — время добавления в источник; у усыновлённого оно уже
	// известно (created_at задачи было бы моментом усыновления, не добавления).
	if t.AddedOn > 0 {
		if err := w.store.SetSourceAddedAt(ctx, d.ID, time.Unix(t.AddedOn, 0)); err != nil {
			w.log.Warn("adopt set source_added_at failed",
				"capability", capIngest, "download_id", d.ID, "error", err)
		}
	}
	w.log.Info("discover adopted torrent",
		"capability", capIngest, "download_id", d.ID, "infohash", hashes[0], "name", t.Name,
		"category", t.Category, "tags", t.Tags)
}

// hasTag сообщает, есть ли tag среди списка тегов qBit (через запятую).
func hasTag(tags, tag string) bool {
	if tag == "" {
		return false
	}
	for _, x := range strings.Split(tags, ",") {
		if strings.TrimSpace(x) == tag {
			return true
		}
	}
	return false
}

// torrentHashes — все непустые хеши торрента (нижний регистр, без дублей,
// v1-приоритетный порядок: v1 раньше v2). Единственный сборщик хешей
// торрента для записи в БД: t.Hash берётся только когда qBittorrent не
// отдал infohash_v1/v2 (старые версии API); у v2-only раздачи t.Hash — это
// УСЕЧЁННЫЙ до 40 hex v2-хеш, хранить его нельзя (по длине он неотличим от
// v1 и порождает битые btih-magnet при retry).
func torrentHashes(t qbt.Torrent) []string {
	cands := []string{t.InfohashV1, t.InfohashV2}
	if t.InfohashV1 == "" && t.InfohashV2 == "" {
		cands = append(cands, t.Hash)
	}
	var out []string
	for _, h := range cands {
		h = store.NormalizeHash(h)
		if h != "" && !slices.Contains(out, h) {
			out = append(out, h)
		}
	}
	return out
}

// magnetURN — синтетический источник усыновлённой раздачи по её хешу:
// btih для v1, btmh (multihash sha256, префикс 1220) для v2. Хеш обязан
// быть полноразмерным (torrentHashes усечённые не отдаёт).
func magnetURN(h string) string {
	if store.HashKind(h) == store.HashV2 {
		return "magnet:?xt=urn:btmh:1220" + h
	}
	return "magnet:?xt=urn:btih:" + h
}