- Revamp decypharr arch \n
- Add callback_ur, download_folder to addContent API \n - Fix few bugs \n - More declarative UI keywords - Speed up repairs - Few other improvements/bug fixes
This commit is contained in:
Mukhtar Akere
858
pkg/debrid/store/cache.go
Normal file
858
pkg/debrid/store/cache.go
Normal file
@@ -0,0 +1,858 @@
|
||||
package store
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"cmp"
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"github.com/sirrobot01/decypharr/pkg/debrid/types"
|
||||
"os"
|
||||
"path"
|
||||
"path/filepath"
|
||||
"regexp"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
"encoding/json"
|
||||
"github.com/go-co-op/gocron/v2"
|
||||
"github.com/rs/zerolog"
|
||||
"github.com/sirrobot01/decypharr/internal/config"
|
||||
"github.com/sirrobot01/decypharr/internal/logger"
|
||||
"github.com/sirrobot01/decypharr/internal/utils"
|
||||
_ "time/tzdata"
|
||||
)
|
||||
|
||||
type WebDavFolderNaming string
|
||||
|
||||
const (
|
||||
WebDavUseFileName WebDavFolderNaming = "filename"
|
||||
WebDavUseOriginalName WebDavFolderNaming = "original"
|
||||
WebDavUseFileNameNoExt WebDavFolderNaming = "filename_no_ext"
|
||||
WebDavUseOriginalNameNoExt WebDavFolderNaming = "original_no_ext"
|
||||
WebDavUseID WebDavFolderNaming = "id"
|
||||
WebdavUseHash WebDavFolderNaming = "infohash"
|
||||
)
|
||||
|
||||
type CachedTorrent struct {
|
||||
*types.Torrent
|
||||
AddedOn time.Time `json:"added_on"`
|
||||
IsComplete bool `json:"is_complete"`
|
||||
Bad bool `json:"bad"`
|
||||
}
|
||||
|
||||
func (c CachedTorrent) copy() CachedTorrent {
|
||||
return CachedTorrent{
|
||||
Torrent: c.Torrent,
|
||||
AddedOn: c.AddedOn,
|
||||
IsComplete: c.IsComplete,
|
||||
Bad: c.Bad,
|
||||
}
|
||||
}
|
||||
|
||||
type RepairType string
|
||||
|
||||
const (
|
||||
RepairTypeReinsert RepairType = "reinsert"
|
||||
RepairTypeDelete RepairType = "delete"
|
||||
)
|
||||
|
||||
type RepairRequest struct {
|
||||
Type RepairType
|
||||
TorrentID string
|
||||
Priority int
|
||||
FileName string
|
||||
}
|
||||
|
||||
type Cache struct {
|
||||
dir string
|
||||
client types.Client
|
||||
logger zerolog.Logger
|
||||
|
||||
torrents *torrentCache
|
||||
downloadLinks *downloadLinkCache
|
||||
invalidDownloadLinks sync.Map
|
||||
folderNaming WebDavFolderNaming
|
||||
|
||||
listingDebouncer *utils.Debouncer[bool]
|
||||
// monitors
|
||||
repairRequest sync.Map
|
||||
failedToReinsert sync.Map
|
||||
downloadLinkRequests sync.Map
|
||||
|
||||
// repair
|
||||
repairChan chan RepairRequest
|
||||
|
||||
// readiness
|
||||
ready chan struct{}
|
||||
|
||||
// config
|
||||
workers int
|
||||
torrentRefreshInterval string
|
||||
downloadLinksRefreshInterval string
|
||||
autoExpiresLinksAfterDuration time.Duration
|
||||
|
||||
// refresh mutex
|
||||
downloadLinksRefreshMu sync.RWMutex // for refreshing download links
|
||||
torrentsRefreshMu sync.RWMutex // for refreshing torrents
|
||||
|
||||
scheduler gocron.Scheduler
|
||||
cetScheduler gocron.Scheduler
|
||||
|
||||
saveSemaphore chan struct{}
|
||||
|
||||
config config.Debrid
|
||||
customFolders []string
|
||||
}
|
||||
|
||||
func NewDebridCache(dc config.Debrid, client types.Client) *Cache {
|
||||
cfg := config.Get()
|
||||
cetSc, err := gocron.NewScheduler(gocron.WithLocation(time.UTC))
|
||||
if err != nil {
|
||||
// If we can't create a CET scheduler, fallback to local time
|
||||
cetSc, _ = gocron.NewScheduler(gocron.WithLocation(time.Local))
|
||||
}
|
||||
scheduler, err := gocron.NewScheduler(gocron.WithLocation(time.Local))
|
||||
if err != nil {
|
||||
// If we can't create a local scheduler, fallback to CET
|
||||
scheduler = cetSc
|
||||
}
|
||||
|
||||
autoExpiresLinksAfter, err := time.ParseDuration(dc.AutoExpireLinksAfter)
|
||||
if autoExpiresLinksAfter == 0 || err != nil {
|
||||
autoExpiresLinksAfter = 48 * time.Hour
|
||||
}
|
||||
var customFolders []string
|
||||
dirFilters := map[string][]directoryFilter{}
|
||||
for name, value := range dc.Directories {
|
||||
for filterType, v := range value.Filters {
|
||||
df := directoryFilter{filterType: filterType, value: v}
|
||||
switch filterType {
|
||||
case filterByRegex, filterByNotRegex:
|
||||
df.regex = regexp.MustCompile(v)
|
||||
case filterBySizeGT, filterBySizeLT:
|
||||
df.sizeThreshold, _ = config.ParseSize(v)
|
||||
case filterBLastAdded:
|
||||
df.ageThreshold, _ = time.ParseDuration(v)
|
||||
}
|
||||
dirFilters[name] = append(dirFilters[name], df)
|
||||
}
|
||||
customFolders = append(customFolders, name)
|
||||
|
||||
}
|
||||
_log := logger.New(fmt.Sprintf("%s-webdav", client.GetName()))
|
||||
c := &Cache{
|
||||
dir: filepath.Join(cfg.Path, "cache", dc.Name), // path to save cache files
|
||||
|
||||
torrents: newTorrentCache(dirFilters),
|
||||
client: client,
|
||||
logger: _log,
|
||||
workers: dc.Workers,
|
||||
downloadLinks: newDownloadLinkCache(),
|
||||
torrentRefreshInterval: dc.TorrentsRefreshInterval,
|
||||
downloadLinksRefreshInterval: dc.DownloadLinksRefreshInterval,
|
||||
folderNaming: WebDavFolderNaming(dc.FolderNaming),
|
||||
autoExpiresLinksAfterDuration: autoExpiresLinksAfter,
|
||||
saveSemaphore: make(chan struct{}, 50),
|
||||
cetScheduler: cetSc,
|
||||
scheduler: scheduler,
|
||||
|
||||
config: dc,
|
||||
customFolders: customFolders,
|
||||
|
||||
ready: make(chan struct{}),
|
||||
}
|
||||
|
||||
c.listingDebouncer = utils.NewDebouncer[bool](100*time.Millisecond, func(refreshRclone bool) {
|
||||
c.RefreshListings(refreshRclone)
|
||||
})
|
||||
return c
|
||||
}
|
||||
|
||||
func (c *Cache) IsReady() chan struct{} {
|
||||
return c.ready
|
||||
}
|
||||
|
||||
func (c *Cache) StreamWithRclone() bool {
|
||||
return c.config.ServeFromRclone
|
||||
}
|
||||
|
||||
// Reset clears all internal state so the Cache can be reused without leaks.
|
||||
// Call this after stopping the old Cache (so no goroutines are holding references),
|
||||
// and before you discard the instance on a restart.
|
||||
func (c *Cache) Reset() {
|
||||
|
||||
if err := c.scheduler.StopJobs(); err != nil {
|
||||
c.logger.Error().Err(err).Msg("Failed to stop scheduler jobs")
|
||||
}
|
||||
|
||||
if err := c.scheduler.Shutdown(); err != nil {
|
||||
c.logger.Error().Err(err).Msg("Failed to stop scheduler")
|
||||
}
|
||||
|
||||
// Stop the listing debouncer
|
||||
c.listingDebouncer.Stop()
|
||||
|
||||
// Close the repair channel
|
||||
close(c.repairChan)
|
||||
|
||||
// 1. Reset torrent storage
|
||||
c.torrents.reset()
|
||||
|
||||
// 2. Reset download-link cache
|
||||
c.downloadLinks.reset()
|
||||
|
||||
// 3. Clear any sync.Maps
|
||||
c.invalidDownloadLinks = sync.Map{}
|
||||
c.repairRequest = sync.Map{}
|
||||
c.failedToReinsert = sync.Map{}
|
||||
c.downloadLinkRequests = sync.Map{}
|
||||
|
||||
// 5. Rebuild the listing debouncer
|
||||
c.listingDebouncer = utils.NewDebouncer[bool](
|
||||
100*time.Millisecond,
|
||||
func(refreshRclone bool) {
|
||||
c.RefreshListings(refreshRclone)
|
||||
},
|
||||
)
|
||||
|
||||
// 6. Reset repair channel so the next Start() can spin it up
|
||||
c.repairChan = make(chan RepairRequest, 100)
|
||||
}
|
||||
|
||||
func (c *Cache) Start(ctx context.Context) error {
|
||||
if err := os.MkdirAll(c.dir, 0755); err != nil {
|
||||
return fmt.Errorf("failed to create cache directory: %w", err)
|
||||
}
|
||||
|
||||
if err := c.Sync(ctx); err != nil {
|
||||
return fmt.Errorf("failed to sync cache: %w", err)
|
||||
}
|
||||
|
||||
// initial download links
|
||||
go c.refreshDownloadLinks(ctx)
|
||||
|
||||
if err := c.StartSchedule(ctx); err != nil {
|
||||
c.logger.Error().Err(err).Msg("Failed to start cache worker")
|
||||
}
|
||||
|
||||
c.repairChan = make(chan RepairRequest, 100)
|
||||
go c.repairWorker(ctx)
|
||||
|
||||
// Fire the ready channel
|
||||
close(c.ready)
|
||||
cfg := config.Get()
|
||||
name := c.client.GetName()
|
||||
addr := cfg.BindAddress + ":" + cfg.Port + cfg.URLBase + "webdav/" + name + "/"
|
||||
c.logger.Info().Msgf("%s WebDav server running at %s", name, addr)
|
||||
|
||||
<-ctx.Done()
|
||||
c.logger.Info().Msgf("Stopping %s WebDav server", name)
|
||||
c.Reset()
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *Cache) load(ctx context.Context) (map[string]CachedTorrent, error) {
|
||||
mu := sync.Mutex{}
|
||||
|
||||
if err := os.MkdirAll(c.dir, 0755); err != nil {
|
||||
return nil, fmt.Errorf("failed to create cache directory: %w", err)
|
||||
}
|
||||
|
||||
files, err := os.ReadDir(c.dir)
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("failed to read cache directory: %w", err)
|
||||
}
|
||||
|
||||
// Get only json files
|
||||
var jsonFiles []os.DirEntry
|
||||
for _, file := range files {
|
||||
if !file.IsDir() && filepath.Ext(file.Name()) == ".json" {
|
||||
jsonFiles = append(jsonFiles, file)
|
||||
}
|
||||
}
|
||||
|
||||
if len(jsonFiles) == 0 {
|
||||
return nil, nil
|
||||
}
|
||||
|
||||
// Create channels with appropriate buffering
|
||||
workChan := make(chan os.DirEntry, min(c.workers, len(jsonFiles)))
|
||||
|
||||
// Create a wait group for workers
|
||||
var wg sync.WaitGroup
|
||||
|
||||
torrents := make(map[string]CachedTorrent, len(jsonFiles))
|
||||
|
||||
// Start workers
|
||||
for i := 0; i < c.workers; i++ {
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
|
||||
for {
|
||||
file, ok := <-workChan
|
||||
if !ok {
|
||||
return // Channel closed, exit goroutine
|
||||
}
|
||||
|
||||
fileName := file.Name()
|
||||
filePath := filepath.Join(c.dir, fileName)
|
||||
data, err := os.ReadFile(filePath)
|
||||
if err != nil {
|
||||
c.logger.Error().Err(err).Msgf("Failed to read file: %s", filePath)
|
||||
continue
|
||||
}
|
||||
|
||||
var ct CachedTorrent
|
||||
if err := json.Unmarshal(data, &ct); err != nil {
|
||||
c.logger.Error().Err(err).Msgf("Failed to unmarshal file: %s", filePath)
|
||||
continue
|
||||
}
|
||||
|
||||
isComplete := true
|
||||
if len(ct.GetFiles()) != 0 {
|
||||
// Check if all files are valid, if not, delete the file.json and remove from cache.
|
||||
fs := make(map[string]types.File, len(ct.GetFiles()))
|
||||
for _, f := range ct.GetFiles() {
|
||||
if f.Link == "" {
|
||||
isComplete = false
|
||||
break
|
||||
}
|
||||
f.TorrentId = ct.Id
|
||||
fs[f.Name] = f
|
||||
}
|
||||
|
||||
if isComplete {
|
||||
|
||||
if addedOn, err := time.Parse(time.RFC3339, ct.Added); err == nil {
|
||||
ct.AddedOn = addedOn
|
||||
}
|
||||
ct.IsComplete = true
|
||||
ct.Files = fs
|
||||
ct.Name = path.Clean(ct.Name)
|
||||
mu.Lock()
|
||||
torrents[ct.Id] = ct
|
||||
mu.Unlock()
|
||||
}
|
||||
}
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
// Feed work to workers
|
||||
for _, file := range jsonFiles {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
break // Context cancelled
|
||||
default:
|
||||
workChan <- file
|
||||
}
|
||||
}
|
||||
|
||||
// Signal workers that no more work is coming
|
||||
close(workChan)
|
||||
|
||||
// Wait for all workers to complete
|
||||
wg.Wait()
|
||||
|
||||
return torrents, nil
|
||||
}
|
||||
|
||||
func (c *Cache) Sync(ctx context.Context) error {
|
||||
cachedTorrents, err := c.load(ctx)
|
||||
if err != nil {
|
||||
c.logger.Error().Err(err).Msg("Failed to load cache")
|
||||
}
|
||||
|
||||
torrents, err := c.client.GetTorrents()
|
||||
if err != nil {
|
||||
return fmt.Errorf("failed to sync torrents: %v", err)
|
||||
}
|
||||
|
||||
totalTorrents := len(torrents)
|
||||
|
||||
c.logger.Info().Msgf("%d torrents found from %s", totalTorrents, c.client.GetName())
|
||||
|
||||
newTorrents := make([]*types.Torrent, 0)
|
||||
idStore := make(map[string]struct{}, totalTorrents)
|
||||
for _, t := range torrents {
|
||||
idStore[t.Id] = struct{}{}
|
||||
if _, ok := cachedTorrents[t.Id]; !ok {
|
||||
newTorrents = append(newTorrents, t)
|
||||
}
|
||||
}
|
||||
|
||||
// Check for deleted torrents
|
||||
deletedTorrents := make([]string, 0)
|
||||
for _, t := range cachedTorrents {
|
||||
if _, ok := idStore[t.Id]; !ok {
|
||||
deletedTorrents = append(deletedTorrents, t.Id)
|
||||
}
|
||||
}
|
||||
|
||||
if len(deletedTorrents) > 0 {
|
||||
c.logger.Info().Msgf("Found %d deleted torrents", len(deletedTorrents))
|
||||
for _, id := range deletedTorrents {
|
||||
if _, ok := cachedTorrents[id]; ok {
|
||||
c.deleteTorrent(id, false) // delete from cache
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Write these torrents to the cache
|
||||
c.setTorrents(cachedTorrents, func() {
|
||||
c.listingDebouncer.Call(false)
|
||||
}) // Initial calls
|
||||
c.logger.Info().Msgf("Loaded %d torrents from cache", len(cachedTorrents))
|
||||
|
||||
if len(newTorrents) > 0 {
|
||||
c.logger.Info().Msgf("Found %d new torrents", len(newTorrents))
|
||||
if err := c.sync(ctx, newTorrents); err != nil {
|
||||
return fmt.Errorf("failed to sync torrents: %v", err)
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *Cache) sync(ctx context.Context, torrents []*types.Torrent) error {
|
||||
|
||||
// Create channels with appropriate buffering
|
||||
workChan := make(chan *types.Torrent, min(c.workers, len(torrents)))
|
||||
|
||||
// Use an atomic counter for progress tracking
|
||||
var processed int64
|
||||
var errorCount int64
|
||||
|
||||
// Create a wait group for workers
|
||||
var wg sync.WaitGroup
|
||||
|
||||
// Start workers
|
||||
for i := 0; i < c.workers; i++ {
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
for {
|
||||
select {
|
||||
case t, ok := <-workChan:
|
||||
if !ok {
|
||||
return // Channel closed, exit goroutine
|
||||
}
|
||||
|
||||
if err := c.ProcessTorrent(t); err != nil {
|
||||
c.logger.Error().Err(err).Str("torrent", t.Name).Msg("sync error")
|
||||
atomic.AddInt64(&errorCount, 1)
|
||||
}
|
||||
|
||||
count := atomic.AddInt64(&processed, 1)
|
||||
if count%1000 == 0 {
|
||||
c.logger.Info().Msgf("Progress: %d/%d torrents processed", count, len(torrents))
|
||||
}
|
||||
|
||||
case <-ctx.Done():
|
||||
return // Context cancelled, exit goroutine
|
||||
}
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
// Feed work to workers
|
||||
for _, t := range torrents {
|
||||
select {
|
||||
case workChan <- t:
|
||||
// Work sent successfully
|
||||
case <-ctx.Done():
|
||||
break // Context cancelled
|
||||
}
|
||||
}
|
||||
|
||||
// Signal workers that no more work is coming
|
||||
close(workChan)
|
||||
|
||||
// Wait for all workers to complete
|
||||
wg.Wait()
|
||||
|
||||
c.listingDebouncer.Call(false) // final refresh
|
||||
c.logger.Info().Msgf("Sync complete: %d torrents processed, %d errors", len(torrents), errorCount)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *Cache) GetTorrentFolder(torrent *types.Torrent) string {
|
||||
switch c.folderNaming {
|
||||
case WebDavUseFileName:
|
||||
return path.Clean(torrent.Filename)
|
||||
case WebDavUseOriginalName:
|
||||
return path.Clean(torrent.OriginalFilename)
|
||||
case WebDavUseFileNameNoExt:
|
||||
return path.Clean(utils.RemoveExtension(torrent.Filename))
|
||||
case WebDavUseOriginalNameNoExt:
|
||||
return path.Clean(utils.RemoveExtension(torrent.OriginalFilename))
|
||||
case WebDavUseID:
|
||||
return torrent.Id
|
||||
case WebdavUseHash:
|
||||
return strings.ToLower(torrent.InfoHash)
|
||||
default:
|
||||
return path.Clean(torrent.Filename)
|
||||
}
|
||||
}
|
||||
|
||||
func (c *Cache) setTorrent(t CachedTorrent, callback func(torrent CachedTorrent)) {
|
||||
torrentName := c.GetTorrentFolder(t.Torrent)
|
||||
updatedTorrent := t.copy()
|
||||
if o, ok := c.torrents.getByName(torrentName); ok && o.Id != t.Id {
|
||||
// If another torrent with the same name exists, merge the files, if the same file exists,
|
||||
// keep the one with the most recent added date
|
||||
|
||||
// Save the most recent torrent
|
||||
mergedFiles := mergeFiles(o, updatedTorrent) // Useful for merging files across multiple torrents, while keeping the most recent
|
||||
updatedTorrent.Files = mergedFiles
|
||||
}
|
||||
c.torrents.set(torrentName, t, updatedTorrent)
|
||||
c.SaveTorrent(t)
|
||||
if callback != nil {
|
||||
callback(updatedTorrent)
|
||||
}
|
||||
}
|
||||
|
||||
func (c *Cache) setTorrents(torrents map[string]CachedTorrent, callback func()) {
|
||||
for _, t := range torrents {
|
||||
torrentName := c.GetTorrentFolder(t.Torrent)
|
||||
updatedTorrent := t.copy()
|
||||
if o, ok := c.torrents.getByName(torrentName); ok && o.Id != t.Id {
|
||||
// Save the most recent torrent
|
||||
mergedFiles := mergeFiles(o, updatedTorrent)
|
||||
updatedTorrent.Files = mergedFiles
|
||||
}
|
||||
c.torrents.set(torrentName, t, updatedTorrent)
|
||||
}
|
||||
c.SaveTorrents()
|
||||
if callback != nil {
|
||||
callback()
|
||||
}
|
||||
}
|
||||
|
||||
// GetListing returns a sorted list of torrents(READ-ONLY)
|
||||
func (c *Cache) GetListing(folder string) []os.FileInfo {
|
||||
switch folder {
|
||||
case "__all__", "torrents":
|
||||
return c.torrents.getListing()
|
||||
default:
|
||||
return c.torrents.getFolderListing(folder)
|
||||
}
|
||||
}
|
||||
|
||||
func (c *Cache) GetCustomFolders() []string {
|
||||
return c.customFolders
|
||||
}
|
||||
|
||||
func (c *Cache) Close() error {
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *Cache) GetTorrents() map[string]CachedTorrent {
|
||||
return c.torrents.getAll()
|
||||
}
|
||||
|
||||
func (c *Cache) GetTorrentByName(name string) *CachedTorrent {
|
||||
if torrent, ok := c.torrents.getByName(name); ok {
|
||||
return &torrent
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *Cache) GetTorrent(torrentId string) *CachedTorrent {
|
||||
if torrent, ok := c.torrents.getByID(torrentId); ok {
|
||||
return &torrent
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *Cache) SaveTorrents() {
|
||||
torrents := c.torrents.getAll()
|
||||
for _, torrent := range torrents {
|
||||
c.SaveTorrent(torrent)
|
||||
}
|
||||
}
|
||||
|
||||
func (c *Cache) SaveTorrent(ct CachedTorrent) {
|
||||
marshaled, err := json.MarshalIndent(ct, "", " ")
|
||||
if err != nil {
|
||||
c.logger.Error().Err(err).Msgf("Failed to marshal torrent: %s", ct.Id)
|
||||
return
|
||||
}
|
||||
|
||||
// Store just the essential info needed for the file operation
|
||||
saveInfo := struct {
|
||||
id string
|
||||
jsonData []byte
|
||||
}{
|
||||
id: ct.Torrent.Id,
|
||||
jsonData: marshaled,
|
||||
}
|
||||
|
||||
// Try to acquire semaphore without blocking
|
||||
select {
|
||||
case c.saveSemaphore <- struct{}{}:
|
||||
go func() {
|
||||
defer func() { <-c.saveSemaphore }()
|
||||
c.saveTorrent(saveInfo.id, saveInfo.jsonData)
|
||||
}()
|
||||
default:
|
||||
c.saveTorrent(saveInfo.id, saveInfo.jsonData)
|
||||
}
|
||||
}
|
||||
|
||||
func (c *Cache) saveTorrent(id string, data []byte) {
|
||||
|
||||
fileName := id + ".json"
|
||||
filePath := filepath.Join(c.dir, fileName)
|
||||
|
||||
// Use a unique temporary filename for concurrent safety
|
||||
tmpFile := filePath + ".tmp." + strconv.FormatInt(time.Now().UnixNano(), 10)
|
||||
|
||||
f, err := os.Create(tmpFile)
|
||||
if err != nil {
|
||||
c.logger.Error().Err(err).Msgf("Failed to create file: %s", tmpFile)
|
||||
return
|
||||
}
|
||||
|
||||
// Track if we've closed the file
|
||||
fileClosed := false
|
||||
defer func() {
|
||||
// Only close if not already closed
|
||||
if !fileClosed {
|
||||
_ = f.Close()
|
||||
}
|
||||
// Clean up the temp file if it still exists and rename failed
|
||||
_ = os.Remove(tmpFile)
|
||||
}()
|
||||
|
||||
w := bufio.NewWriter(f)
|
||||
if _, err := w.Write(data); err != nil {
|
||||
c.logger.Error().Err(err).Msgf("Failed to write data: %s", tmpFile)
|
||||
return
|
||||
}
|
||||
|
||||
if err := w.Flush(); err != nil {
|
||||
c.logger.Error().Err(err).Msgf("Failed to flush data: %s", tmpFile)
|
||||
return
|
||||
}
|
||||
|
||||
// Close the file before renaming
|
||||
_ = f.Close()
|
||||
fileClosed = true
|
||||
|
||||
if err := os.Rename(tmpFile, filePath); err != nil {
|
||||
c.logger.Error().Err(err).Msgf("Failed to rename file: %s", tmpFile)
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
func (c *Cache) ProcessTorrent(t *types.Torrent) error {
|
||||
|
||||
isComplete := func(files map[string]types.File) bool {
|
||||
_complete := len(files) > 0
|
||||
for _, file := range files {
|
||||
if file.Link == "" {
|
||||
_complete = false
|
||||
break
|
||||
}
|
||||
}
|
||||
return _complete
|
||||
}
|
||||
|
||||
if !isComplete(t.Files) {
|
||||
if err := c.client.UpdateTorrent(t); err != nil {
|
||||
return fmt.Errorf("failed to update torrent: %w", err)
|
||||
}
|
||||
}
|
||||
|
||||
if !isComplete(t.Files) {
|
||||
c.logger.Debug().Msgf("Torrent %s is still not complete. Triggering a reinsert(disabled)", t.Id)
|
||||
} else {
|
||||
addedOn, err := time.Parse(time.RFC3339, t.Added)
|
||||
if err != nil {
|
||||
addedOn = time.Now()
|
||||
}
|
||||
ct := CachedTorrent{
|
||||
Torrent: t,
|
||||
IsComplete: len(t.Files) > 0,
|
||||
AddedOn: addedOn,
|
||||
}
|
||||
c.setTorrent(ct, func(tor CachedTorrent) {
|
||||
c.listingDebouncer.Call(false)
|
||||
})
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *Cache) Add(t *types.Torrent) error {
|
||||
if len(t.Files) == 0 {
|
||||
if err := c.client.UpdateTorrent(t); err != nil {
|
||||
return fmt.Errorf("failed to update torrent: %w", err)
|
||||
}
|
||||
}
|
||||
addedOn, err := time.Parse(time.RFC3339, t.Added)
|
||||
if err != nil {
|
||||
addedOn = time.Now()
|
||||
}
|
||||
ct := CachedTorrent{
|
||||
Torrent: t,
|
||||
IsComplete: len(t.Files) > 0,
|
||||
AddedOn: addedOn,
|
||||
}
|
||||
c.setTorrent(ct, func(tor CachedTorrent) {
|
||||
c.RefreshListings(true)
|
||||
})
|
||||
go c.GenerateDownloadLinks(ct)
|
||||
return nil
|
||||
|
||||
}
|
||||
|
||||
func (c *Cache) GetClient() types.Client {
|
||||
return c.client
|
||||
}
|
||||
|
||||
func (c *Cache) DeleteTorrent(id string) error {
|
||||
c.torrentsRefreshMu.Lock()
|
||||
defer c.torrentsRefreshMu.Unlock()
|
||||
|
||||
if c.deleteTorrent(id, true) {
|
||||
go c.RefreshListings(true)
|
||||
return nil
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *Cache) validateAndDeleteTorrents(torrents []string) {
|
||||
wg := sync.WaitGroup{}
|
||||
for _, torrent := range torrents {
|
||||
wg.Add(1)
|
||||
go func(t string) {
|
||||
defer wg.Done()
|
||||
// Check if torrent is truly deleted
|
||||
if _, err := c.client.GetTorrent(t); err != nil {
|
||||
c.deleteTorrent(t, false) // Since it's removed from debrid already
|
||||
}
|
||||
}(torrent)
|
||||
}
|
||||
wg.Wait()
|
||||
c.listingDebouncer.Call(true)
|
||||
}
|
||||
|
||||
// deleteTorrent deletes the torrent from the cache and debrid service
|
||||
// It also handles torrents with the same name but different IDs
|
||||
func (c *Cache) deleteTorrent(id string, removeFromDebrid bool) bool {
|
||||
|
||||
if torrent, ok := c.torrents.getByID(id); ok {
|
||||
c.torrents.removeId(id) // Delete id from cache
|
||||
defer func() {
|
||||
c.removeFromDB(id)
|
||||
if removeFromDebrid {
|
||||
_ = c.client.DeleteTorrent(id) // Skip error handling, we don't care if it fails
|
||||
}
|
||||
}() // defer delete from debrid
|
||||
|
||||
torrentName := torrent.Name
|
||||
|
||||
if t, ok := c.torrents.getByName(torrentName); ok {
|
||||
|
||||
newFiles := map[string]types.File{}
|
||||
newId := ""
|
||||
for _, file := range t.GetFiles() {
|
||||
if file.TorrentId != "" && file.TorrentId != id {
|
||||
if newId == "" && file.TorrentId != "" {
|
||||
newId = file.TorrentId
|
||||
}
|
||||
newFiles[file.Name] = file
|
||||
}
|
||||
}
|
||||
if len(newFiles) == 0 {
|
||||
// Delete the torrent since no files are left
|
||||
c.torrents.remove(torrentName)
|
||||
} else {
|
||||
t.Files = newFiles
|
||||
newId = cmp.Or(newId, t.Id)
|
||||
t.Id = newId
|
||||
c.setTorrent(t, nil) // This gets called after calling deleteTorrent
|
||||
}
|
||||
}
|
||||
return true
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
func (c *Cache) DeleteTorrents(ids []string) {
|
||||
c.logger.Info().Msgf("Deleting %d torrents", len(ids))
|
||||
for _, id := range ids {
|
||||
_ = c.deleteTorrent(id, true)
|
||||
}
|
||||
c.listingDebouncer.Call(true)
|
||||
}
|
||||
|
||||
func (c *Cache) removeFromDB(torrentId string) {
|
||||
// Moves the torrent file to the trash
|
||||
filePath := filepath.Join(c.dir, torrentId+".json")
|
||||
|
||||
// Check if the file exists
|
||||
if _, err := os.Stat(filePath); errors.Is(err, os.ErrNotExist) {
|
||||
return
|
||||
}
|
||||
|
||||
// Move the file to the trash
|
||||
trashPath := filepath.Join(c.dir, "trash", torrentId+".json")
|
||||
if err := os.MkdirAll(filepath.Dir(trashPath), 0755); err != nil {
|
||||
return
|
||||
}
|
||||
if err := os.Rename(filePath, trashPath); err != nil {
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
func (c *Cache) OnRemove(torrentId string) {
|
||||
c.logger.Debug().Msgf("OnRemove triggered for %s", torrentId)
|
||||
err := c.DeleteTorrent(torrentId)
|
||||
if err != nil {
|
||||
c.logger.Error().Err(err).Msgf("Failed to delete torrent: %s", torrentId)
|
||||
return
|
||||
}
|
||||
}
|
||||
|
||||
// RemoveFile removes a file from the torrent cache
|
||||
// TODO sends a re-insert that removes the file from debrid
|
||||
func (c *Cache) RemoveFile(torrentId string, filename string) error {
|
||||
c.logger.Debug().Str("torrent_id", torrentId).Msgf("Removing file %s", filename)
|
||||
torrent, ok := c.torrents.getByID(torrentId)
|
||||
if !ok {
|
||||
return fmt.Errorf("torrent %s not found", torrentId)
|
||||
}
|
||||
file, ok := torrent.GetFile(filename)
|
||||
if !ok {
|
||||
return fmt.Errorf("file %s not found in torrent %s", filename, torrentId)
|
||||
}
|
||||
file.Deleted = true
|
||||
torrent.Files[filename] = file
|
||||
|
||||
// If the torrent has no files left, delete it
|
||||
if len(torrent.GetFiles()) == 0 {
|
||||
c.logger.Debug().Msgf("Torrent %s has no files left, deleting it", torrentId)
|
||||
if err := c.DeleteTorrent(torrentId); err != nil {
|
||||
return fmt.Errorf("failed to delete torrent %s: %w", torrentId, err)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
c.setTorrent(torrent, func(torrent CachedTorrent) {
|
||||
c.listingDebouncer.Call(true)
|
||||
}) // Update the torrent in the cache
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *Cache) GetLogger() zerolog.Logger {
|
||||
return c.logger
|
||||
}
|
||||
257
pkg/debrid/store/download_link.go
Normal file
257
pkg/debrid/store/download_link.go
Normal file
@@ -0,0 +1,257 @@
|
||||
package store
|
||||
|
||||
import (
|
||||
"errors"
|
||||
"fmt"
|
||||
"github.com/sirrobot01/decypharr/pkg/debrid/types"
|
||||
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
"github.com/sirrobot01/decypharr/internal/request"
|
||||
)
|
||||
|
||||
type linkCache struct {
|
||||
Id string
|
||||
link string
|
||||
accountId string
|
||||
expiresAt time.Time
|
||||
}
|
||||
|
||||
type downloadLinkCache struct {
|
||||
data map[string]linkCache
|
||||
mu sync.Mutex
|
||||
}
|
||||
|
||||
func newDownloadLinkCache() *downloadLinkCache {
|
||||
return &downloadLinkCache{
|
||||
data: make(map[string]linkCache),
|
||||
}
|
||||
}
|
||||
|
||||
func (c *downloadLinkCache) reset() {
|
||||
c.mu.Lock()
|
||||
c.data = make(map[string]linkCache)
|
||||
c.mu.Unlock()
|
||||
}
|
||||
|
||||
func (c *downloadLinkCache) Load(key string) (linkCache, bool) {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
dl, ok := c.data[key]
|
||||
return dl, ok
|
||||
}
|
||||
func (c *downloadLinkCache) Store(key string, value linkCache) {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
c.data[key] = value
|
||||
}
|
||||
func (c *downloadLinkCache) Delete(key string) {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
delete(c.data, key)
|
||||
}
|
||||
|
||||
func (c *downloadLinkCache) Len() int {
|
||||
c.mu.Lock()
|
||||
defer c.mu.Unlock()
|
||||
return len(c.data)
|
||||
}
|
||||
|
||||
type downloadLinkRequest struct {
|
||||
result string
|
||||
err error
|
||||
done chan struct{}
|
||||
}
|
||||
|
||||
func newDownloadLinkRequest() *downloadLinkRequest {
|
||||
return &downloadLinkRequest{
|
||||
done: make(chan struct{}),
|
||||
}
|
||||
}
|
||||
|
||||
func (r *downloadLinkRequest) Complete(result string, err error) {
|
||||
r.result = result
|
||||
r.err = err
|
||||
close(r.done)
|
||||
}
|
||||
|
||||
func (r *downloadLinkRequest) Wait() (string, error) {
|
||||
<-r.done
|
||||
return r.result, r.err
|
||||
}
|
||||
|
||||
func (c *Cache) GetDownloadLink(torrentName, filename, fileLink string) (string, error) {
|
||||
// Check link cache
|
||||
if dl := c.checkDownloadLink(fileLink); dl != "" {
|
||||
return dl, nil
|
||||
}
|
||||
|
||||
if req, inFlight := c.downloadLinkRequests.Load(fileLink); inFlight {
|
||||
// Wait for the other request to complete and use its result
|
||||
result := req.(*downloadLinkRequest)
|
||||
return result.Wait()
|
||||
}
|
||||
|
||||
// Create a new request object
|
||||
req := newDownloadLinkRequest()
|
||||
c.downloadLinkRequests.Store(fileLink, req)
|
||||
|
||||
downloadLink, err := c.fetchDownloadLink(torrentName, filename, fileLink)
|
||||
|
||||
// Complete the request and remove it from the map
|
||||
req.Complete(downloadLink, err)
|
||||
c.downloadLinkRequests.Delete(fileLink)
|
||||
|
||||
return downloadLink, err
|
||||
}
|
||||
|
||||
func (c *Cache) fetchDownloadLink(torrentName, filename, fileLink string) (string, error) {
|
||||
ct := c.GetTorrentByName(torrentName)
|
||||
if ct == nil {
|
||||
return "", fmt.Errorf("torrent not found")
|
||||
}
|
||||
file, ok := ct.GetFile(filename)
|
||||
if !ok {
|
||||
return "", fmt.Errorf("file %s not found in torrent %s", filename, torrentName)
|
||||
}
|
||||
|
||||
if file.Link == "" {
|
||||
// file link is empty, refresh the torrent to get restricted links
|
||||
ct = c.refreshTorrent(file.TorrentId) // Refresh the torrent from the debrid
|
||||
if ct == nil {
|
||||
return "", fmt.Errorf("failed to refresh torrent")
|
||||
} else {
|
||||
file, ok = ct.GetFile(filename)
|
||||
if !ok {
|
||||
return "", fmt.Errorf("file %s not found in refreshed torrent %s", filename, torrentName)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// If file.Link is still empty, return
|
||||
if file.Link == "" {
|
||||
// Try to reinsert the torrent?
|
||||
newCt, err := c.reInsertTorrent(ct)
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("failed to reinsert torrent. %w", err)
|
||||
}
|
||||
ct = newCt
|
||||
file, ok = ct.GetFile(filename)
|
||||
if !ok {
|
||||
return "", fmt.Errorf("file %s not found in reinserted torrent %s", filename, torrentName)
|
||||
}
|
||||
}
|
||||
|
||||
c.logger.Trace().Msgf("Getting download link for %s(%s)", filename, file.Link)
|
||||
downloadLink, err := c.client.GetDownloadLink(ct.Torrent, &file)
|
||||
if err != nil {
|
||||
if errors.Is(err, request.HosterUnavailableError) {
|
||||
newCt, err := c.reInsertTorrent(ct)
|
||||
if err != nil {
|
||||
return "", fmt.Errorf("failed to reinsert torrent: %w", err)
|
||||
}
|
||||
ct = newCt
|
||||
file, ok = ct.GetFile(filename)
|
||||
if !ok {
|
||||
return "", fmt.Errorf("file %s not found in reinserted torrent %s", filename, torrentName)
|
||||
}
|
||||
// Retry getting the download link
|
||||
downloadLink, err = c.client.GetDownloadLink(ct.Torrent, &file)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
if downloadLink == nil {
|
||||
return "", fmt.Errorf("download link is empty for")
|
||||
}
|
||||
c.updateDownloadLink(downloadLink)
|
||||
return "", nil
|
||||
} else if errors.Is(err, request.TrafficExceededError) {
|
||||
// This is likely a fair usage limit error
|
||||
return "", err
|
||||
} else {
|
||||
return "", fmt.Errorf("failed to get download link: %w", err)
|
||||
}
|
||||
}
|
||||
if downloadLink == nil {
|
||||
return "", fmt.Errorf("download link is empty")
|
||||
}
|
||||
c.updateDownloadLink(downloadLink)
|
||||
return downloadLink.DownloadLink, nil
|
||||
}
|
||||
|
||||
func (c *Cache) GenerateDownloadLinks(t CachedTorrent) {
|
||||
if err := c.client.GenerateDownloadLinks(t.Torrent); err != nil {
|
||||
c.logger.Error().Err(err).Str("torrent", t.Name).Msg("Failed to generate download links")
|
||||
return
|
||||
}
|
||||
for _, file := range t.GetFiles() {
|
||||
if file.DownloadLink != nil {
|
||||
c.updateDownloadLink(file.DownloadLink)
|
||||
}
|
||||
}
|
||||
c.setTorrent(t, nil)
|
||||
}
|
||||
|
||||
func (c *Cache) updateDownloadLink(dl *types.DownloadLink) {
|
||||
c.downloadLinks.Store(dl.Link, linkCache{
|
||||
Id: dl.Id,
|
||||
link: dl.DownloadLink,
|
||||
expiresAt: time.Now().Add(c.autoExpiresLinksAfterDuration),
|
||||
accountId: dl.AccountId,
|
||||
})
|
||||
}
|
||||
|
||||
func (c *Cache) checkDownloadLink(link string) string {
|
||||
if dl, ok := c.downloadLinks.Load(link); ok {
|
||||
if dl.expiresAt.After(time.Now()) && !c.IsDownloadLinkInvalid(dl.link) {
|
||||
return dl.link
|
||||
}
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
func (c *Cache) MarkDownloadLinkAsInvalid(link, downloadLink, reason string) {
|
||||
c.invalidDownloadLinks.Store(downloadLink, reason)
|
||||
// Remove the download api key from active
|
||||
if reason == "bandwidth_exceeded" {
|
||||
if dl, ok := c.downloadLinks.Load(link); ok {
|
||||
if dl.accountId != "" && dl.link == downloadLink {
|
||||
c.client.DisableAccount(dl.accountId)
|
||||
}
|
||||
}
|
||||
}
|
||||
c.removeDownloadLink(link)
|
||||
}
|
||||
|
||||
func (c *Cache) removeDownloadLink(link string) {
|
||||
if dl, ok := c.downloadLinks.Load(link); ok {
|
||||
// Delete dl from cache
|
||||
c.downloadLinks.Delete(link)
|
||||
// Delete dl from debrid
|
||||
if dl.Id != "" {
|
||||
_ = c.client.DeleteDownloadLink(dl.Id)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (c *Cache) IsDownloadLinkInvalid(downloadLink string) bool {
|
||||
if reason, ok := c.invalidDownloadLinks.Load(downloadLink); ok {
|
||||
c.logger.Debug().Msgf("Download link %s is invalid: %s", downloadLink, reason)
|
||||
return true
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
func (c *Cache) GetDownloadByteRange(torrentName, filename string) (*[2]int64, error) {
|
||||
ct := c.GetTorrentByName(torrentName)
|
||||
if ct == nil {
|
||||
return nil, fmt.Errorf("torrent not found")
|
||||
}
|
||||
file := ct.Files[filename]
|
||||
return file.ByteRange, nil
|
||||
}
|
||||
|
||||
func (c *Cache) GetTotalActiveDownloadLinks() int {
|
||||
return c.downloadLinks.Len()
|
||||
}
|
||||
42
pkg/debrid/store/misc.go
Normal file
42
pkg/debrid/store/misc.go
Normal file
@@ -0,0 +1,42 @@
|
||||
package store
|
||||
|
||||
import (
|
||||
"github.com/sirrobot01/decypharr/pkg/debrid/types"
|
||||
"sort"
|
||||
)
|
||||
|
||||
// MergeFiles merges the files from multiple torrents into a single map.
|
||||
// It uses the file name as the key and the file object as the value.
|
||||
// This is useful for deduplicating files across multiple torrents.
|
||||
// The order of the torrents is determined by the AddedOn time, with the earliest added torrent first.
|
||||
// If a file with the same name exists in multiple torrents, the last one will be used.
|
||||
func mergeFiles(torrents ...CachedTorrent) map[string]types.File {
|
||||
merged := make(map[string]types.File)
|
||||
|
||||
// order torrents by added time
|
||||
sort.Slice(torrents, func(i, j int) bool {
|
||||
return torrents[i].AddedOn.Before(torrents[j].AddedOn)
|
||||
})
|
||||
|
||||
for _, torrent := range torrents {
|
||||
for _, file := range torrent.GetFiles() {
|
||||
merged[file.Name] = file
|
||||
}
|
||||
}
|
||||
return merged
|
||||
}
|
||||
|
||||
func (c *Cache) GetIngests() ([]types.IngestData, error) {
|
||||
torrents := c.GetTorrents()
|
||||
debridName := c.client.GetName()
|
||||
var ingests []types.IngestData
|
||||
for _, torrent := range torrents {
|
||||
ingests = append(ingests, types.IngestData{
|
||||
Debrid: debridName,
|
||||
Name: torrent.Filename,
|
||||
Hash: torrent.InfoHash,
|
||||
Size: torrent.Bytes,
|
||||
})
|
||||
}
|
||||
return ingests, nil
|
||||
}
|
||||
267
pkg/debrid/store/refresh.go
Normal file
267
pkg/debrid/store/refresh.go
Normal file
@@ -0,0 +1,267 @@
|
||||
package store
|
||||
|
||||
import (
|
||||
"context"
|
||||
"fmt"
|
||||
"github.com/sirrobot01/decypharr/pkg/debrid/types"
|
||||
"io"
|
||||
"net/http"
|
||||
"os"
|
||||
"strings"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
type fileInfo struct {
|
||||
id string
|
||||
name string
|
||||
size int64
|
||||
mode os.FileMode
|
||||
modTime time.Time
|
||||
isDir bool
|
||||
}
|
||||
|
||||
func (fi *fileInfo) Name() string { return fi.name }
|
||||
func (fi *fileInfo) Size() int64 { return fi.size }
|
||||
func (fi *fileInfo) Mode() os.FileMode { return fi.mode }
|
||||
func (fi *fileInfo) ModTime() time.Time { return fi.modTime }
|
||||
func (fi *fileInfo) IsDir() bool { return fi.isDir }
|
||||
func (fi *fileInfo) ID() string { return fi.id }
|
||||
func (fi *fileInfo) Sys() interface{} { return nil }
|
||||
|
||||
func (c *Cache) RefreshListings(refreshRclone bool) {
|
||||
// Copy the torrents to a string|time map
|
||||
c.torrents.refreshListing() // refresh torrent listings
|
||||
|
||||
if refreshRclone {
|
||||
if err := c.refreshRclone(); err != nil {
|
||||
c.logger.Error().Err(err).Msg("Failed to refresh rclone") // silent error
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (c *Cache) refreshTorrents(ctx context.Context) {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
default:
|
||||
}
|
||||
|
||||
if !c.torrentsRefreshMu.TryLock() {
|
||||
return
|
||||
}
|
||||
defer c.torrentsRefreshMu.Unlock()
|
||||
|
||||
// Get all torrents from the debrid service
|
||||
debTorrents, err := c.client.GetTorrents()
|
||||
if err != nil {
|
||||
c.logger.Error().Err(err).Msg("Failed to get torrents")
|
||||
return
|
||||
}
|
||||
|
||||
if len(debTorrents) == 0 {
|
||||
// Maybe an error occurred
|
||||
return
|
||||
}
|
||||
|
||||
currentTorrentIds := make(map[string]struct{}, len(debTorrents))
|
||||
for _, t := range debTorrents {
|
||||
currentTorrentIds[t.Id] = struct{}{}
|
||||
}
|
||||
|
||||
// Let's implement deleting torrents removed from debrid
|
||||
deletedTorrents := make([]string, 0)
|
||||
cachedTorrents := c.torrents.getIdMaps()
|
||||
for id := range cachedTorrents {
|
||||
if _, exists := currentTorrentIds[id]; !exists {
|
||||
deletedTorrents = append(deletedTorrents, id)
|
||||
}
|
||||
}
|
||||
|
||||
if len(deletedTorrents) > 0 {
|
||||
go c.validateAndDeleteTorrents(deletedTorrents)
|
||||
}
|
||||
|
||||
newTorrents := make([]*types.Torrent, 0)
|
||||
for _, t := range debTorrents {
|
||||
if _, exists := cachedTorrents[t.Id]; !exists {
|
||||
newTorrents = append(newTorrents, t)
|
||||
}
|
||||
}
|
||||
|
||||
if len(newTorrents) == 0 {
|
||||
return
|
||||
}
|
||||
|
||||
c.logger.Trace().Msgf("Found %d new torrents", len(newTorrents))
|
||||
|
||||
workChan := make(chan *types.Torrent, min(100, len(newTorrents)))
|
||||
errChan := make(chan error, len(newTorrents))
|
||||
var wg sync.WaitGroup
|
||||
counter := 0
|
||||
|
||||
for i := 0; i < c.workers; i++ {
|
||||
wg.Add(1)
|
||||
go func() {
|
||||
defer wg.Done()
|
||||
for t := range workChan {
|
||||
if err := c.ProcessTorrent(t); err != nil {
|
||||
c.logger.Error().Err(err).Msgf("Failed to process new torrent %s", t.Id)
|
||||
errChan <- err
|
||||
}
|
||||
counter++
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
for _, t := range newTorrents {
|
||||
workChan <- t
|
||||
}
|
||||
close(workChan)
|
||||
wg.Wait()
|
||||
|
||||
c.listingDebouncer.Call(false)
|
||||
|
||||
c.logger.Debug().Msgf("Processed %d new torrents", counter)
|
||||
}
|
||||
|
||||
func (c *Cache) refreshRclone() error {
|
||||
cfg := c.config
|
||||
|
||||
if cfg.RcUrl == "" {
|
||||
return nil
|
||||
}
|
||||
|
||||
if cfg.RcUrl == "" {
|
||||
return nil
|
||||
}
|
||||
|
||||
client := &http.Client{
|
||||
Timeout: 10 * time.Second,
|
||||
Transport: &http.Transport{
|
||||
MaxIdleConns: 10,
|
||||
IdleConnTimeout: 30 * time.Second,
|
||||
DisableCompression: false,
|
||||
MaxIdleConnsPerHost: 5,
|
||||
},
|
||||
}
|
||||
// Create form data
|
||||
data := ""
|
||||
dirs := strings.FieldsFunc(cfg.RcRefreshDirs, func(r rune) bool {
|
||||
return r == ',' || r == '&'
|
||||
})
|
||||
if len(dirs) == 0 {
|
||||
data = "dir=__all__"
|
||||
} else {
|
||||
for index, dir := range dirs {
|
||||
if dir != "" {
|
||||
if index == 0 {
|
||||
data += "dir=" + dir
|
||||
} else {
|
||||
data += "&dir" + fmt.Sprint(index+1) + "=" + dir
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
sendRequest := func(endpoint string) error {
|
||||
req, err := http.NewRequest("POST", fmt.Sprintf("%s/%s", cfg.RcUrl, endpoint), strings.NewReader(data))
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
req.Header.Set("Content-Type", "application/x-www-form-urlencoded")
|
||||
|
||||
if cfg.RcUser != "" && cfg.RcPass != "" {
|
||||
req.SetBasicAuth(cfg.RcUser, cfg.RcPass)
|
||||
}
|
||||
resp, err := client.Do(req)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
defer resp.Body.Close()
|
||||
|
||||
if resp.StatusCode != 200 {
|
||||
body, _ := io.ReadAll(io.LimitReader(resp.Body, 1024))
|
||||
return fmt.Errorf("failed to perform %s: %s - %s", endpoint, resp.Status, string(body))
|
||||
}
|
||||
|
||||
_, _ = io.Copy(io.Discard, resp.Body)
|
||||
return nil
|
||||
}
|
||||
|
||||
if err := sendRequest("vfs/forget"); err != nil {
|
||||
return err
|
||||
}
|
||||
if err := sendRequest("vfs/refresh"); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
func (c *Cache) refreshTorrent(torrentId string) *CachedTorrent {
|
||||
|
||||
if torrentId == "" {
|
||||
c.logger.Error().Msg("Torrent ID is empty")
|
||||
return nil
|
||||
}
|
||||
|
||||
torrent, err := c.client.GetTorrent(torrentId)
|
||||
if err != nil {
|
||||
c.logger.Error().Err(err).Msgf("Failed to get torrent %s", torrentId)
|
||||
return nil
|
||||
}
|
||||
addedOn, err := time.Parse(time.RFC3339, torrent.Added)
|
||||
if err != nil {
|
||||
addedOn = time.Now()
|
||||
}
|
||||
ct := CachedTorrent{
|
||||
Torrent: torrent,
|
||||
AddedOn: addedOn,
|
||||
IsComplete: len(torrent.Files) > 0,
|
||||
}
|
||||
c.setTorrent(ct, func(torrent CachedTorrent) {
|
||||
go c.listingDebouncer.Call(true)
|
||||
})
|
||||
|
||||
return &ct
|
||||
}
|
||||
|
||||
func (c *Cache) refreshDownloadLinks(ctx context.Context) {
|
||||
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
default:
|
||||
}
|
||||
|
||||
if !c.downloadLinksRefreshMu.TryLock() {
|
||||
return
|
||||
}
|
||||
defer c.downloadLinksRefreshMu.Unlock()
|
||||
|
||||
downloadLinks, err := c.client.GetDownloads()
|
||||
|
||||
if err != nil {
|
||||
c.logger.Error().Err(err).Msg("Failed to get download links")
|
||||
return
|
||||
}
|
||||
for k, v := range downloadLinks {
|
||||
// if link is generated in the last 24 hours, add it to cache
|
||||
timeSince := time.Since(v.Generated)
|
||||
if timeSince < c.autoExpiresLinksAfterDuration {
|
||||
c.downloadLinks.Store(k, linkCache{
|
||||
Id: v.Id,
|
||||
accountId: v.AccountId,
|
||||
link: v.DownloadLink,
|
||||
expiresAt: v.Generated.Add(c.autoExpiresLinksAfterDuration - timeSince),
|
||||
})
|
||||
} else {
|
||||
c.downloadLinks.Delete(k)
|
||||
}
|
||||
}
|
||||
|
||||
c.logger.Trace().Msgf("Refreshed %d download links", len(downloadLinks))
|
||||
|
||||
}
|
||||
257
pkg/debrid/store/repair.go
Normal file
257
pkg/debrid/store/repair.go
Normal file
@@ -0,0 +1,257 @@
|
||||
package store
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"github.com/sirrobot01/decypharr/internal/request"
|
||||
"github.com/sirrobot01/decypharr/internal/utils"
|
||||
"github.com/sirrobot01/decypharr/pkg/debrid/types"
|
||||
"sync"
|
||||
"time"
|
||||
)
|
||||
|
||||
type reInsertRequest struct {
|
||||
result *CachedTorrent
|
||||
err error
|
||||
done chan struct{}
|
||||
}
|
||||
|
||||
func newReInsertRequest() *reInsertRequest {
|
||||
return &reInsertRequest{
|
||||
done: make(chan struct{}),
|
||||
}
|
||||
}
|
||||
|
||||
func (r *reInsertRequest) Complete(result *CachedTorrent, err error) {
|
||||
r.result = result
|
||||
r.err = err
|
||||
close(r.done)
|
||||
}
|
||||
|
||||
func (r *reInsertRequest) Wait() (*CachedTorrent, error) {
|
||||
<-r.done
|
||||
return r.result, r.err
|
||||
}
|
||||
|
||||
func (c *Cache) markAsFailedToReinsert(torrentId string) {
|
||||
c.failedToReinsert.Store(torrentId, struct{}{})
|
||||
|
||||
// Remove the torrent from the directory if it has failed to reinsert, max retries are hardcoded to 5
|
||||
if torrent, ok := c.torrents.getByID(torrentId); ok {
|
||||
torrent.Bad = true
|
||||
c.setTorrent(torrent, func(t CachedTorrent) {
|
||||
c.RefreshListings(false)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func (c *Cache) markAsSuccessfullyReinserted(torrentId string) {
|
||||
if _, ok := c.failedToReinsert.Load(torrentId); !ok {
|
||||
return
|
||||
}
|
||||
c.failedToReinsert.Delete(torrentId)
|
||||
if torrent, ok := c.torrents.getByID(torrentId); ok {
|
||||
torrent.Bad = false
|
||||
c.setTorrent(torrent, func(torrent CachedTorrent) {
|
||||
c.RefreshListings(false)
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
func (c *Cache) GetBrokenFiles(t *CachedTorrent, filenames []string) []string {
|
||||
files := make(map[string]types.File)
|
||||
brokenFiles := make([]string, 0)
|
||||
if len(filenames) > 0 {
|
||||
for name, f := range t.Files {
|
||||
if utils.Contains(filenames, name) {
|
||||
files[name] = f
|
||||
}
|
||||
}
|
||||
} else {
|
||||
files = t.Files
|
||||
}
|
||||
for _, f := range files {
|
||||
// Check if file is missing
|
||||
if f.Link == "" {
|
||||
// refresh torrent and then break
|
||||
if newT := c.refreshTorrent(f.TorrentId); newT != nil {
|
||||
t = newT
|
||||
} else {
|
||||
c.logger.Error().Str("torrentId", t.Torrent.Id).Msg("Failed to refresh torrent")
|
||||
return filenames // Return original filenames if refresh fails(torrent is somehow botched)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if t.Torrent == nil {
|
||||
c.logger.Error().Str("torrentId", t.Torrent.Id).Msg("Failed to refresh torrent")
|
||||
return filenames // Return original filenames if refresh fails(torrent is somehow botched)
|
||||
}
|
||||
|
||||
files = t.Files
|
||||
|
||||
for _, f := range files {
|
||||
// Check if file link is still missing
|
||||
if f.Link == "" {
|
||||
brokenFiles = append(brokenFiles, f.Name)
|
||||
} else {
|
||||
// Check if file.Link not in the downloadLink Cache
|
||||
if err := c.client.CheckLink(f.Link); err != nil {
|
||||
if errors.Is(err, request.HosterUnavailableError) {
|
||||
brokenFiles = append(brokenFiles, f.Name)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Try to reinsert the torrent if it's broken
|
||||
if len(brokenFiles) > 0 && t.Torrent != nil {
|
||||
// Check if the torrent is already in progress
|
||||
if _, err := c.reInsertTorrent(t); err != nil {
|
||||
c.logger.Error().Err(err).Str("torrentId", t.Torrent.Id).Msg("Failed to reinsert torrent")
|
||||
return brokenFiles // Return broken files if reinsert fails
|
||||
}
|
||||
return nil // Return nil if the torrent was successfully reinserted
|
||||
}
|
||||
|
||||
return brokenFiles
|
||||
}
|
||||
|
||||
func (c *Cache) repairWorker(ctx context.Context) {
|
||||
// This watches a channel for torrents to repair and can be cancelled via context
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
|
||||
case req, ok := <-c.repairChan:
|
||||
// Channel was closed
|
||||
if !ok {
|
||||
c.logger.Debug().Msg("Repair channel closed, shutting down worker")
|
||||
return
|
||||
}
|
||||
|
||||
torrentId := req.TorrentID
|
||||
c.logger.Debug().Str("torrentId", req.TorrentID).Msg("Received repair request")
|
||||
|
||||
// Get the torrent from the cache
|
||||
cachedTorrent := c.GetTorrent(torrentId)
|
||||
if cachedTorrent == nil {
|
||||
c.logger.Warn().Str("torrentId", torrentId).Msg("Torrent not found in cache")
|
||||
continue
|
||||
}
|
||||
|
||||
switch req.Type {
|
||||
case RepairTypeReinsert:
|
||||
c.logger.Debug().Str("torrentId", torrentId).Msg("Reinserting torrent")
|
||||
if _, err := c.reInsertTorrent(cachedTorrent); err != nil {
|
||||
c.logger.Error().Err(err).Str("torrentId", cachedTorrent.Id).Msg("Failed to reinsert torrent")
|
||||
continue
|
||||
}
|
||||
case RepairTypeDelete:
|
||||
c.logger.Debug().Str("torrentId", torrentId).Msg("Deleting torrent")
|
||||
if err := c.DeleteTorrent(torrentId); err != nil {
|
||||
c.logger.Error().Err(err).Str("torrentId", torrentId).Msg("Failed to delete torrent")
|
||||
continue
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (c *Cache) reInsertTorrent(ct *CachedTorrent) (*CachedTorrent, error) {
|
||||
// Check if Magnet is not empty, if empty, reconstruct the magnet
|
||||
torrent := ct.Torrent
|
||||
oldID := torrent.Id // Store the old ID
|
||||
if _, ok := c.failedToReinsert.Load(oldID); ok {
|
||||
return ct, fmt.Errorf("can't retry re-insert for %s", torrent.Id)
|
||||
}
|
||||
if reqI, inFlight := c.repairRequest.Load(oldID); inFlight {
|
||||
req := reqI.(*reInsertRequest)
|
||||
c.logger.Debug().Msgf("Waiting for existing reinsert request to complete for torrent %s", oldID)
|
||||
return req.Wait()
|
||||
}
|
||||
req := newReInsertRequest()
|
||||
c.repairRequest.Store(oldID, req)
|
||||
|
||||
// Make sure we clean up even if there's a panic
|
||||
defer func() {
|
||||
c.repairRequest.Delete(oldID)
|
||||
}()
|
||||
|
||||
// Submit the magnet to the debrid service
|
||||
newTorrent := &types.Torrent{
|
||||
Name: torrent.Name,
|
||||
Magnet: utils.ConstructMagnet(torrent.InfoHash, torrent.Name),
|
||||
InfoHash: torrent.InfoHash,
|
||||
Size: torrent.Size,
|
||||
Files: make(map[string]types.File),
|
||||
Arr: torrent.Arr,
|
||||
}
|
||||
var err error
|
||||
newTorrent, err = c.client.SubmitMagnet(newTorrent)
|
||||
if err != nil {
|
||||
c.markAsFailedToReinsert(oldID)
|
||||
// Remove the old torrent from the cache and debrid service
|
||||
return ct, fmt.Errorf("failed to submit magnet: %w", err)
|
||||
}
|
||||
|
||||
// Check if the torrent was submitted
|
||||
if newTorrent == nil || newTorrent.Id == "" {
|
||||
c.markAsFailedToReinsert(oldID)
|
||||
return ct, fmt.Errorf("failed to submit magnet: empty torrent")
|
||||
}
|
||||
newTorrent.DownloadUncached = false // Set to false, avoid re-downloading
|
||||
newTorrent, err = c.client.CheckStatus(newTorrent, true)
|
||||
if err != nil {
|
||||
if newTorrent != nil && newTorrent.Id != "" {
|
||||
// Delete the torrent if it was not downloaded
|
||||
_ = c.client.DeleteTorrent(newTorrent.Id)
|
||||
}
|
||||
c.markAsFailedToReinsert(oldID)
|
||||
return ct, err
|
||||
}
|
||||
|
||||
// Update the torrent in the cache
|
||||
addedOn, err := time.Parse(time.RFC3339, newTorrent.Added)
|
||||
if err != nil {
|
||||
addedOn = time.Now()
|
||||
}
|
||||
for _, f := range newTorrent.GetFiles() {
|
||||
if f.Link == "" {
|
||||
c.markAsFailedToReinsert(oldID)
|
||||
return ct, fmt.Errorf("failed to reinsert torrent: empty link")
|
||||
}
|
||||
}
|
||||
// Set torrent to newTorrent
|
||||
newCt := CachedTorrent{
|
||||
Torrent: newTorrent,
|
||||
AddedOn: addedOn,
|
||||
IsComplete: len(newTorrent.Files) > 0,
|
||||
}
|
||||
c.setTorrent(newCt, func(torrent CachedTorrent) {
|
||||
c.RefreshListings(true)
|
||||
})
|
||||
|
||||
ct = &newCt // Update ct to point to the new torrent
|
||||
|
||||
// We can safely delete the old torrent here
|
||||
if oldID != "" {
|
||||
if err := c.DeleteTorrent(oldID); err != nil {
|
||||
return ct, fmt.Errorf("failed to delete old torrent: %w", err)
|
||||
}
|
||||
}
|
||||
|
||||
req.Complete(ct, err)
|
||||
c.markAsSuccessfullyReinserted(oldID)
|
||||
|
||||
c.logger.Debug().Str("torrentId", torrent.Id).Msg("Torrent successfully reinserted")
|
||||
|
||||
return ct, nil
|
||||
}
|
||||
|
||||
func (c *Cache) resetInvalidLinks() {
|
||||
c.invalidDownloadLinks = sync.Map{}
|
||||
c.client.ResetActiveDownloadKeys() // Reset the active download keys
|
||||
}
|
||||
298
pkg/debrid/store/torrent.go
Normal file
298
pkg/debrid/store/torrent.go
Normal file
@@ -0,0 +1,298 @@
|
||||
package store
|
||||
|
||||
import (
|
||||
"fmt"
|
||||
"os"
|
||||
"regexp"
|
||||
"sort"
|
||||
"strings"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
)
|
||||
|
||||
const (
|
||||
filterByInclude string = "include"
|
||||
filterByExclude string = "exclude"
|
||||
|
||||
filterByStartsWith string = "starts_with"
|
||||
filterByEndsWith string = "ends_with"
|
||||
filterByNotStartsWith string = "not_starts_with"
|
||||
filterByNotEndsWith string = "not_ends_with"
|
||||
|
||||
filterByRegex string = "regex"
|
||||
filterByNotRegex string = "not_regex"
|
||||
|
||||
filterByExactMatch string = "exact_match"
|
||||
filterByNotExactMatch string = "not_exact_match"
|
||||
|
||||
filterBySizeGT string = "size_gt"
|
||||
filterBySizeLT string = "size_lt"
|
||||
|
||||
filterBLastAdded string = "last_added"
|
||||
)
|
||||
|
||||
type directoryFilter struct {
|
||||
filterType string
|
||||
value string
|
||||
regex *regexp.Regexp // only for regex/not_regex
|
||||
sizeThreshold int64 // only for size_gt/size_lt
|
||||
ageThreshold time.Duration // only for last_added
|
||||
}
|
||||
|
||||
type torrentCache struct {
|
||||
mu sync.Mutex
|
||||
byID map[string]CachedTorrent
|
||||
byName map[string]CachedTorrent
|
||||
listing atomic.Value
|
||||
folderListing map[string][]os.FileInfo
|
||||
folderListingMu sync.RWMutex
|
||||
directoriesFilters map[string][]directoryFilter
|
||||
sortNeeded atomic.Bool
|
||||
}
|
||||
|
||||
type sortableFile struct {
|
||||
id string
|
||||
name string
|
||||
modTime time.Time
|
||||
size int64
|
||||
bad bool
|
||||
}
|
||||
|
||||
func newTorrentCache(dirFilters map[string][]directoryFilter) *torrentCache {
|
||||
|
||||
tc := &torrentCache{
|
||||
byID: make(map[string]CachedTorrent),
|
||||
byName: make(map[string]CachedTorrent),
|
||||
folderListing: make(map[string][]os.FileInfo),
|
||||
directoriesFilters: dirFilters,
|
||||
}
|
||||
|
||||
tc.sortNeeded.Store(false)
|
||||
tc.listing.Store(make([]os.FileInfo, 0))
|
||||
return tc
|
||||
}
|
||||
|
||||
func (tc *torrentCache) reset() {
|
||||
tc.mu.Lock()
|
||||
tc.byID = make(map[string]CachedTorrent)
|
||||
tc.byName = make(map[string]CachedTorrent)
|
||||
tc.mu.Unlock()
|
||||
|
||||
// reset the sorted listing
|
||||
tc.sortNeeded.Store(false)
|
||||
tc.listing.Store(make([]os.FileInfo, 0))
|
||||
|
||||
// reset any per-folder views
|
||||
tc.folderListingMu.Lock()
|
||||
tc.folderListing = make(map[string][]os.FileInfo)
|
||||
tc.folderListingMu.Unlock()
|
||||
}
|
||||
|
||||
func (tc *torrentCache) getByID(id string) (CachedTorrent, bool) {
|
||||
tc.mu.Lock()
|
||||
defer tc.mu.Unlock()
|
||||
torrent, exists := tc.byID[id]
|
||||
return torrent, exists
|
||||
}
|
||||
|
||||
func (tc *torrentCache) getByName(name string) (CachedTorrent, bool) {
|
||||
tc.mu.Lock()
|
||||
defer tc.mu.Unlock()
|
||||
torrent, exists := tc.byName[name]
|
||||
return torrent, exists
|
||||
}
|
||||
|
||||
func (tc *torrentCache) set(name string, torrent, newTorrent CachedTorrent) {
|
||||
tc.mu.Lock()
|
||||
// Set the id first
|
||||
tc.byID[newTorrent.Id] = torrent // This is the unadulterated torrent
|
||||
tc.byName[name] = newTorrent // This is likely the modified torrent
|
||||
tc.mu.Unlock()
|
||||
tc.sortNeeded.Store(true)
|
||||
}
|
||||
|
||||
func (tc *torrentCache) getListing() []os.FileInfo {
|
||||
// Fast path: if we have a sorted list and no changes since last sort
|
||||
if !tc.sortNeeded.Load() {
|
||||
return tc.listing.Load().([]os.FileInfo)
|
||||
}
|
||||
|
||||
// Slow path: need to sort
|
||||
tc.refreshListing()
|
||||
return tc.listing.Load().([]os.FileInfo)
|
||||
}
|
||||
|
||||
func (tc *torrentCache) getFolderListing(folderName string) []os.FileInfo {
|
||||
tc.folderListingMu.RLock()
|
||||
defer tc.folderListingMu.RUnlock()
|
||||
if folderName == "" {
|
||||
return tc.getListing()
|
||||
}
|
||||
if folder, ok := tc.folderListing[folderName]; ok {
|
||||
return folder
|
||||
}
|
||||
// If folder not found, return empty slice
|
||||
return []os.FileInfo{}
|
||||
}
|
||||
|
||||
func (tc *torrentCache) refreshListing() {
|
||||
|
||||
tc.mu.Lock()
|
||||
all := make([]sortableFile, 0, len(tc.byName))
|
||||
for name, t := range tc.byName {
|
||||
all = append(all, sortableFile{t.Id, name, t.AddedOn, t.Bytes, t.Bad})
|
||||
}
|
||||
tc.sortNeeded.Store(false)
|
||||
tc.mu.Unlock()
|
||||
|
||||
sort.Slice(all, func(i, j int) bool {
|
||||
if all[i].name != all[j].name {
|
||||
return all[i].name < all[j].name
|
||||
}
|
||||
return all[i].modTime.Before(all[j].modTime)
|
||||
})
|
||||
|
||||
wg := sync.WaitGroup{}
|
||||
|
||||
wg.Add(1) // for all listing
|
||||
go func() {
|
||||
listing := make([]os.FileInfo, len(all))
|
||||
for i, sf := range all {
|
||||
listing[i] = &fileInfo{sf.id, sf.name, sf.size, 0755 | os.ModeDir, sf.modTime, true}
|
||||
}
|
||||
tc.listing.Store(listing)
|
||||
}()
|
||||
wg.Done()
|
||||
|
||||
wg.Add(1)
|
||||
// For __bad__
|
||||
go func() {
|
||||
listing := make([]os.FileInfo, 0)
|
||||
for _, sf := range all {
|
||||
if sf.bad {
|
||||
listing = append(listing, &fileInfo{
|
||||
id: sf.id,
|
||||
name: fmt.Sprintf("%s || %s", sf.name, sf.id),
|
||||
size: sf.size,
|
||||
mode: 0755 | os.ModeDir,
|
||||
modTime: sf.modTime,
|
||||
isDir: true,
|
||||
})
|
||||
}
|
||||
}
|
||||
tc.folderListingMu.Lock()
|
||||
if len(listing) > 0 {
|
||||
tc.folderListing["__bad__"] = listing
|
||||
} else {
|
||||
delete(tc.folderListing, "__bad__")
|
||||
}
|
||||
tc.folderListingMu.Unlock()
|
||||
}()
|
||||
wg.Done()
|
||||
|
||||
now := time.Now()
|
||||
wg.Add(len(tc.directoriesFilters)) // for each directory filter
|
||||
for dir, filters := range tc.directoriesFilters {
|
||||
go func(dir string, filters []directoryFilter) {
|
||||
defer wg.Done()
|
||||
var matched []os.FileInfo
|
||||
for _, sf := range all {
|
||||
if tc.torrentMatchDirectory(filters, sf, now) {
|
||||
matched = append(matched, &fileInfo{
|
||||
id: sf.id,
|
||||
name: sf.name, size: sf.size,
|
||||
mode: 0755 | os.ModeDir, modTime: sf.modTime, isDir: true,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
tc.folderListingMu.Lock()
|
||||
if len(matched) > 0 {
|
||||
tc.folderListing[dir] = matched
|
||||
} else {
|
||||
delete(tc.folderListing, dir)
|
||||
}
|
||||
tc.folderListingMu.Unlock()
|
||||
}(dir, filters)
|
||||
}
|
||||
|
||||
wg.Wait()
|
||||
}
|
||||
|
||||
func (tc *torrentCache) torrentMatchDirectory(filters []directoryFilter, file sortableFile, now time.Time) bool {
|
||||
|
||||
torrentName := strings.ToLower(file.name)
|
||||
for _, filter := range filters {
|
||||
matched := false
|
||||
|
||||
switch filter.filterType {
|
||||
case filterByInclude:
|
||||
matched = strings.Contains(torrentName, filter.value)
|
||||
case filterByStartsWith:
|
||||
matched = strings.HasPrefix(torrentName, filter.value)
|
||||
case filterByEndsWith:
|
||||
matched = strings.HasSuffix(torrentName, filter.value)
|
||||
case filterByExactMatch:
|
||||
matched = torrentName == filter.value
|
||||
case filterByExclude:
|
||||
matched = !strings.Contains(torrentName, filter.value)
|
||||
case filterByNotStartsWith:
|
||||
matched = !strings.HasPrefix(torrentName, filter.value)
|
||||
case filterByNotEndsWith:
|
||||
matched = !strings.HasSuffix(torrentName, filter.value)
|
||||
case filterByRegex:
|
||||
matched = filter.regex.MatchString(torrentName)
|
||||
case filterByNotRegex:
|
||||
matched = !filter.regex.MatchString(torrentName)
|
||||
case filterByNotExactMatch:
|
||||
matched = torrentName != filter.value
|
||||
case filterBySizeGT:
|
||||
matched = file.size > filter.sizeThreshold
|
||||
case filterBySizeLT:
|
||||
matched = file.size < filter.sizeThreshold
|
||||
case filterBLastAdded:
|
||||
matched = file.modTime.After(now.Add(-filter.ageThreshold))
|
||||
}
|
||||
if !matched {
|
||||
return false // All filters must match
|
||||
}
|
||||
}
|
||||
|
||||
// If we get here, all filters matched
|
||||
return true
|
||||
}
|
||||
|
||||
func (tc *torrentCache) getAll() map[string]CachedTorrent {
|
||||
tc.mu.Lock()
|
||||
defer tc.mu.Unlock()
|
||||
result := make(map[string]CachedTorrent)
|
||||
for name, torrent := range tc.byID {
|
||||
result[name] = torrent
|
||||
}
|
||||
return result
|
||||
}
|
||||
|
||||
func (tc *torrentCache) getIdMaps() map[string]struct{} {
|
||||
tc.mu.Lock()
|
||||
defer tc.mu.Unlock()
|
||||
res := make(map[string]struct{}, len(tc.byID))
|
||||
for id := range tc.byID {
|
||||
res[id] = struct{}{}
|
||||
}
|
||||
return res
|
||||
}
|
||||
|
||||
func (tc *torrentCache) removeId(id string) {
|
||||
tc.mu.Lock()
|
||||
defer tc.mu.Unlock()
|
||||
delete(tc.byID, id)
|
||||
tc.sortNeeded.Store(true)
|
||||
}
|
||||
|
||||
func (tc *torrentCache) remove(name string) {
|
||||
tc.mu.Lock()
|
||||
defer tc.mu.Unlock()
|
||||
delete(tc.byName, name)
|
||||
tc.sortNeeded.Store(true)
|
||||
}
|
||||
60
pkg/debrid/store/worker.go
Normal file
60
pkg/debrid/store/worker.go
Normal file
@@ -0,0 +1,60 @@
|
||||
package store
|
||||
|
||||
import (
|
||||
"context"
|
||||
"github.com/go-co-op/gocron/v2"
|
||||
"github.com/sirrobot01/decypharr/internal/utils"
|
||||
)
|
||||
|
||||
func (c *Cache) StartSchedule(ctx context.Context) error {
|
||||
// For now, we just want to refresh the listing and download links
|
||||
|
||||
// Schedule download link refresh job
|
||||
if jd, err := utils.ConvertToJobDef(c.downloadLinksRefreshInterval); err != nil {
|
||||
c.logger.Error().Err(err).Msg("Failed to convert download link refresh interval to job definition")
|
||||
} else {
|
||||
// Schedule the job
|
||||
if _, err := c.scheduler.NewJob(jd, gocron.NewTask(func() {
|
||||
c.refreshDownloadLinks(ctx)
|
||||
}), gocron.WithContext(ctx)); err != nil {
|
||||
c.logger.Error().Err(err).Msg("Failed to create download link refresh job")
|
||||
} else {
|
||||
c.logger.Debug().Msgf("Download link refresh job scheduled for every %s", c.downloadLinksRefreshInterval)
|
||||
}
|
||||
}
|
||||
|
||||
// Schedule torrent refresh job
|
||||
if jd, err := utils.ConvertToJobDef(c.torrentRefreshInterval); err != nil {
|
||||
c.logger.Error().Err(err).Msg("Failed to convert torrent refresh interval to job definition")
|
||||
} else {
|
||||
// Schedule the job
|
||||
if _, err := c.scheduler.NewJob(jd, gocron.NewTask(func() {
|
||||
c.refreshTorrents(ctx)
|
||||
}), gocron.WithContext(ctx)); err != nil {
|
||||
c.logger.Error().Err(err).Msg("Failed to create torrent refresh job")
|
||||
} else {
|
||||
c.logger.Debug().Msgf("Torrent refresh job scheduled for every %s", c.torrentRefreshInterval)
|
||||
}
|
||||
}
|
||||
|
||||
// Schedule the reset invalid links job
|
||||
// This job will run every at 00:00 CET
|
||||
// and reset the invalid links in the cache
|
||||
if jd, err := utils.ConvertToJobDef("00:00"); err != nil {
|
||||
c.logger.Error().Err(err).Msg("Failed to convert link reset interval to job definition")
|
||||
} else {
|
||||
// Schedule the job
|
||||
if _, err := c.cetScheduler.NewJob(jd, gocron.NewTask(func() {
|
||||
c.resetInvalidLinks()
|
||||
}), gocron.WithContext(ctx)); err != nil {
|
||||
c.logger.Error().Err(err).Msg("Failed to create link reset job")
|
||||
} else {
|
||||
c.logger.Debug().Msgf("Link reset job scheduled for every midnight, CET")
|
||||
}
|
||||
}
|
||||
|
||||
// Start the scheduler
|
||||
c.scheduler.Start()
|
||||
c.cetScheduler.Start()
|
||||
return nil
|
||||
}
|
||||
1
pkg/debrid/store/xml.go
Normal file
1
pkg/debrid/store/xml.go
Normal file
@@ -0,0 +1 @@
|
||||
package store
|
||||
Reference in New Issue
Block a user