zurg/internal/torrent/refresh.go

package torrent

import (
	"fmt"
	"path/filepath"
	"strings"
	"sync"
	"time"

	"github.com/debridmediamanager/zurg/internal/config"
	"github.com/debridmediamanager/zurg/pkg/realdebrid"
	mapset "github.com/deckarep/golang-set/v2"
	cmap "github.com/orcaman/concurrent-map/v2"
)

func (t *TorrentManager) RefreshTorrents() []string {
	instances, _, err := t.Api.GetTorrents(0, false)
	if err != nil {
		t.log.Warnf("Cannot get torrents: %v", err)
		return nil
	}
	infoChan := make(chan *Torrent, len(instances))
	var wg sync.WaitGroup

	allTorrents, _ := t.DirectoryMap.Get(INT_ALL)
	doesNotExist := t.deleteOnceDone.Clone()
	for i := range instances {
		idx := i
		doesNotExist.Remove(instances[idx].ID)
		wg.Add(1)
		_ = t.workerPool.Submit(func() {
			defer wg.Done()
			if !t.fixers.Has(instances[idx].ID) {
				// not a fixer, just regular torrent
				infoChan <- t.getMoreInfo(instances[idx])
				return
			} else if instances[idx].IsDone() {
				// fixer is done, let's check if it fixed the torrent
				infoChan <- t.handleFixers(instances[idx])
				return
			}
			infoChan <- nil
		})
	}

	wg.Wait()
	close(infoChan)
	t.log.Debugf("Fetched info for %d torrents", len(instances))

	// delete expired fixers
	doesNotExist.Each(func(fixerID string) bool {
		t.fixers.Remove(fixerID)
		t.deleteOnceDone.Remove(fixerID)
		return false
	})
	// ensure delete
	infoCache, _ := t.DirectoryMap.Get(INT_INFO_CACHE)
	t.deleteOnceDone.Each(func(fixerID string) bool {
		torrent, exists := infoCache.Get(fixerID)
		if exists && torrent.AnyInProgress() {
			return false
		}
		t.log.Debugf("Ensuring that torrent id=%s is deleted", fixerID)
		t.Delete(t.GetKey(torrent), true)
		t.Api.DeleteTorrent(fixerID)
		infoCache.Remove(fixerID)
		return false
	})

	newlyFetchedKeys := mapset.NewSet[string]()
	noInfoCount := 0
	for info := range infoChan {
		if info == nil {
			noInfoCount++
			continue
		}
		accessKey := t.GetKey(info)
		if !info.AnyInProgress() {
			newlyFetchedKeys.Add(accessKey)
		}
		if torrent, exists := allTorrents.Get(accessKey); !exists {
			allTorrents.Set(accessKey, info)
		} else if !info.DownloadedIDs.Difference(torrent.DownloadedIDs).IsEmpty() {
			mainTorrent := t.mergeToMain(torrent, info)
			allTorrents.Set(accessKey, &mainTorrent)
		}
	}
	t.log.Infof("Compiled into %d torrents, %d were missing info", allTorrents.Count(), noInfoCount)

	var updatedPaths []string
	// torrents yet to be assigned in a directory
	newlyFetchedKeys.Difference(t.allAccessKeys).Each(func(accessKey string) bool {
		// assign to directories
		tor, ok := allTorrents.Get(accessKey)
		if !ok {
			return false
		}
		var directories []string
		t.assignedDirectoryCb(tor, func(directory string) {
			torrents, _ := t.DirectoryMap.Get(directory)
			torrents.Set(accessKey, tor)
			updatedPaths = append(updatedPaths, fmt.Sprintf("%s/%s", directory, accessKey))
			// this is just for the logs
			if directory != config.ALL_TORRENTS {
				directories = append(directories, directory)
			}
		})
		// t.log.Debugf("Added %s to %v", accessKey, directories)
		t.allAccessKeys.Add(accessKey)
		return false
	})
	// removed torrents
	t.allAccessKeys.Difference(newlyFetchedKeys).Each(func(accessKey string) bool {
		t.Delete(accessKey, false)
		return false
	})

	return updatedPaths
}

// startRefreshJob periodically refreshes the torrents
func (t *TorrentManager) startRefreshJob() {
	t.log.Info("Starting periodic refresh")
	for {
		<-time.After(time.Duration(t.Config.GetRefreshEverySeconds()) * time.Second)

		checksum := t.getCurrentState()
		if t.latestState.equal(checksum) {
			continue
		}
		t.SetNewLatestState(checksum)
		t.log.Infof("Detected changes! Refreshing %d torrents", checksum.TotalCount)

		updatedPaths := t.RefreshTorrents()
		t.log.Info("Finished refreshing torrents")

		t.TriggerHookOnLibraryUpdate(updatedPaths)

		if t.Config.EnableRepair() {
			t.RepairAll()
		} else {
			t.log.Info("Repair is disabled, skipping repair check")
		}
	}
}

// getMoreInfo gets original name, size and files for a torrent
func (t *TorrentManager) getMoreInfo(rdTorrent realdebrid.Torrent) *Torrent {
	infoCache, _ := t.DirectoryMap.Get(INT_INFO_CACHE)
	if torrentFromCache, exists := infoCache.Get(rdTorrent.ID); exists &&
		!torrentFromCache.AnyInProgress() &&
		torrentFromCache.SelectedFiles.Count() == len(rdTorrent.Links) {

		return torrentFromCache
	}

	torrentFromFile := t.readTorrentFromFile(rdTorrent.ID)
	if torrentFromFile != nil &&
		!torrentFromFile.AnyInProgress() &&
		torrentFromFile.SelectedFiles.Count() == len(rdTorrent.Links) {

		hasBrokenFiles := false
		torrentFromFile.SelectedFiles.IterCb(func(filepath string, file *File) {
			if !strings.HasPrefix(file.Link, "http") && file.Link != "unselect" {
				hasBrokenFiles = true
			}
		})

		if !hasBrokenFiles {
			infoCache.Set(rdTorrent.ID, torrentFromFile)
			return torrentFromFile
		} else {
			t.log.Warnf("Torrent %s has broken files, will not save on info cache", rdTorrent.ID)
		}
	}

	info, err := t.Api.GetTorrentInfo(rdTorrent.ID)
	if err != nil {
		t.log.Warnf("Cannot get info for id=%s: %v", rdTorrent.ID, err)
		return nil
	}

	torrent := Torrent{
		Name:         info.Name,
		OriginalName: info.OriginalName,
		Added:        info.Added,
		Hash:         info.Hash,
	}
	// SelectedFiles is a subset of Files with only the selected ones
	// it also has a Link field, which can be empty
	// if it is empty, it means the file is no longer available
	// Files+Links together are the same as SelectedFiles
	var selectedFiles []*File
	// if some Links are empty, we need to repair it
	for _, file := range info.Files {
		if file.Selected == 0 {
			continue
		}
		selectedFiles = append(selectedFiles, &File{
			File:  file,
			Ended: info.Ended,
			Link:  "", // no link yet
		})
	}
	if len(selectedFiles) == len(info.Links) {
		// all links are still intact! good!
		for i, file := range selectedFiles {
			file.Link = info.Links[i]
		}
		torrent.UnassignedLinks = mapset.NewSet[string]()
	} else {
		torrent.UnassignedLinks = mapset.NewSet[string](info.Links...)
	}
	torrent.SelectedFiles = cmap.New[*File]()
	for _, file := range selectedFiles {
		// todo better handling of duplicate filenames
		if torrent.SelectedFiles.Has(filepath.Base(file.Path)) {
			oldName := filepath.Base(file.Path)
			ext := filepath.Ext(oldName)
			filename := strings.TrimSuffix(oldName, ext)
			newName := fmt.Sprintf("%s (%d)%s", filename, file.ID, ext)
			torrent.SelectedFiles.Set(newName, file)
		} else {
			torrent.SelectedFiles.Set(filepath.Base(file.Path), file)
		}
	}
	torrent.DownloadedIDs = mapset.NewSet[string]()
	torrent.InProgressIDs = mapset.NewSet[string]()
	if info.IsDone() {
		torrent.DownloadedIDs.Add(info.ID)
	} else {
		torrent.InProgressIDs.Add(info.ID)
	}
	torrent.BrokenLinks = mapset.NewSet[string]()

	infoCache.Set(rdTorrent.ID, &torrent)
	t.writeTorrentToFile(rdTorrent.ID, &torrent)

	return &torrent
}

func (t *TorrentManager) mergeToMain(existing, toMerge *Torrent) Torrent {
	mainTorrent := Torrent{
		Name:          existing.Name,
		OriginalName:  existing.OriginalName,
		Rename:        existing.Rename,
		Hash:          existing.Hash,
		DownloadedIDs: mapset.NewSet[string](),
		InProgressIDs: mapset.NewSet[string](),
		// UnassignedLinks: mapset.NewSet[string](),
		UnassignedLinks: existing.UnassignedLinks.Union(toMerge.UnassignedLinks),
		BrokenLinks:     existing.BrokenLinks.Union(toMerge.BrokenLinks),
	}

	// unrepairable reason
	if existing.UnrepairableReason != "" && toMerge.UnrepairableReason != "" && existing.UnrepairableReason != toMerge.UnrepairableReason {
		mainTorrent.UnrepairableReason = fmt.Sprintf("%s, %s", existing.UnrepairableReason, toMerge.UnrepairableReason)
	} else if existing.UnrepairableReason != "" {
		mainTorrent.UnrepairableReason = existing.UnrepairableReason
	} else if toMerge.UnrepairableReason != "" {
		mainTorrent.UnrepairableReason = toMerge.UnrepairableReason
	}

	// this function triggers only when we have a new DownloadedID
	toMerge.DownloadedIDs.Difference(existing.DownloadedIDs).Each(func(id string) bool {
		mainTorrent.DownloadedIDs.Add(id)
		mainTorrent.InProgressIDs.Remove(id)
		return false
	})

	// the link can have the following values
	// 1. https://*** - the file is available
	// 2. repair - the link is there but we need to repair it
	// 3. unselect - the file is deleted
	// 4. empty - the file is not available
	mainTorrent.SelectedFiles = existing.SelectedFiles
	toMerge.SelectedFiles.IterCb(func(filepath string, fileToMerge *File) {
		// see if it already exists in the main torrent
		if originalFile, ok := mainTorrent.SelectedFiles.Get(filepath); !ok || fileToMerge.Link == "unselect" {
			// if it doesn't exist in the main torrent, add it
			mainTorrent.SelectedFiles.Set(filepath, fileToMerge)
		} else if originalFile.Link != "unselect" {
			// if it exists, compare the Added property and the link
			if existing.Added < toMerge.Added {
				//  && strings.HasPrefix(fileToMerge.Link, "http")
				// if torrentToMerge is more recent and its file has a link, update the main torrent's file
				mainTorrent.SelectedFiles.Set(filepath, fileToMerge)
			}
			// else do nothing, the main torrent's file is more recent or has a valid link
		}
	})

	// broken links
	if mainTorrent.BrokenLinks.Cardinality() > 0 {
		mainTorrent.SelectedFiles.IterCb(func(_ string, file *File) {
			mainTorrent.BrokenLinks.Each(func(brokenLink string) bool {
				if file.Link == brokenLink {
					file.Link = ""
				}
				return file.Link == brokenLink
			})
		})
	}

	if existing.Added < toMerge.Added {
		mainTorrent.Added = toMerge.Added
	} else {
		mainTorrent.Added = existing.Added
	}

	return mainTorrent
}