Update repair; fix minor bugs with namings

pkg/repair/repair.go

@@ -10,7 +10,7 @@ import (
 	"github.com/sirrobot01/debrid-blackhole/internal/logger"
 	"github.com/sirrobot01/debrid-blackhole/internal/request"
 	"github.com/sirrobot01/debrid-blackhole/pkg/arr"
-	"github.com/sirrobot01/debrid-blackhole/pkg/debrid/types"
+	"github.com/sirrobot01/debrid-blackhole/pkg/debrid/debrid"
 	"golang.org/x/sync/errgroup"
 	"net"
 	"net/http"
@@ -29,7 +29,7 @@ import (
 type Repair struct {
 	Jobs        map[string]*Job
 	arrs        *arr.Storage
-	deb         types.Client
+	deb         *debrid.Engine
 	duration    time.Duration
 	runOnStart  bool
 	ZurgURL     string
@@ -39,7 +39,7 @@ type Repair struct {
 	filename    string
 }
 
-func New(arrs *arr.Storage) *Repair {
+func New(arrs *arr.Storage, engine *debrid.Engine) *Repair {
 	cfg := config.GetConfig()
 	duration, err := parseSchedule(cfg.Repair.Interval)
 	if err != nil {
@@ -53,6 +53,7 @@ func New(arrs *arr.Storage) *Repair {
 		ZurgURL:     cfg.Repair.ZurgURL,
 		autoProcess: cfg.Repair.AutoProcess,
 		filename:    filepath.Join(cfg.Path, "repair.json"),
+		deb:         engine,
 	}
 	if r.ZurgURL != "" {
 		r.IsZurg = true
@@ -66,10 +67,11 @@ func New(arrs *arr.Storage) *Repair {
 type JobStatus string
 
 const (
-	JobStarted   JobStatus = "started"
-	JobPending   JobStatus = "pending"
-	JobFailed    JobStatus = "failed"
-	JobCompleted JobStatus = "completed"
+	JobStarted    JobStatus = "started"
+	JobPending    JobStatus = "pending"
+	JobFailed     JobStatus = "failed"
+	JobCompleted  JobStatus = "completed"
+	JobProcessing JobStatus = "processing"
 )
 
 type Job struct {
@@ -185,12 +187,21 @@ func (r *Repair) AddJob(arrsNames []string, mediaIDs []string, autoProcess, recu
 	r.reset(job)
 	r.Jobs[key] = job
 	go r.saveToFile()
-	err := r.repair(job)
-	go r.saveToFile()
-	return err
+	go func() {
+		if err := r.repair(job); err != nil {
+			r.logger.Error().Err(err).Msg("Error running repair")
+			r.logger.Error().Err(err).Msg("Error running repair")
+			job.FailedAt = time.Now()
+			job.Error = err.Error()
+			job.Status = JobFailed
+			job.CompletedAt = time.Now()
+		}
+	}()
+	return nil
 }
 
 func (r *Repair) repair(job *Job) error {
+	defer r.saveToFile()
 	if err := r.preRunChecks(); err != nil {
 		return err
 	}
@@ -331,6 +342,161 @@ func (r *Repair) Start(ctx context.Context) error {
 	}
 }
 
+func (r *Repair) getUniquePaths(media arr.Content) map[string]string {
+	// Use zurg setup to check file availability with zurg
+	// This reduces bandwidth usage significantly
+
+	uniqueParents := make(map[string]string)
+	files := media.Files
+	for _, file := range files {
+		target := getSymlinkTarget(file.Path)
+		if target != "" {
+			file.IsSymlink = true
+			dir, f := filepath.Split(target)
+			parent := filepath.Base(filepath.Clean(dir))
+			// Set target path folder/file.mkv
+			file.TargetPath = f
+			uniqueParents[parent] = target
+		}
+	}
+	return uniqueParents
+}
+
+func (r *Repair) clean(job *Job) error {
+	// Create a new error group
+	g, ctx := errgroup.WithContext(context.Background())
+
+	uniqueItems := make(map[string]string)
+	mu := sync.Mutex{}
+
+	// Limit concurrent goroutines
+	g.SetLimit(runtime.NumCPU() * 4)
+
+	for _, a := range job.Arrs {
+		a := a // Capture range variable
+		g.Go(func() error {
+			// Check if context was canceled
+			select {
+			case <-ctx.Done():
+				return ctx.Err()
+			default:
+			}
+
+			items, err := r.cleanArr(job, a, "")
+			if err != nil {
+				r.logger.Error().Err(err).Msgf("Error cleaning %s", a)
+				return err
+			}
+
+			// Safely append the found items to the shared slice
+			if len(items) > 0 {
+				mu.Lock()
+				for k, v := range items {
+					uniqueItems[k] = v
+				}
+				mu.Unlock()
+			}
+
+			return nil
+		})
+	}
+
+	if err := g.Wait(); err != nil {
+		return err
+	}
+
+	if len(uniqueItems) == 0 {
+		job.CompletedAt = time.Now()
+		job.Status = JobCompleted
+
+		go func() {
+			if err := request.SendDiscordMessage("repair_clean_complete", "success", job.discordContext()); err != nil {
+				r.logger.Error().Msgf("Error sending discord message: %v", err)
+			}
+		}()
+
+		return nil
+	}
+
+	cache := r.deb.Caches["realdebrid"]
+	if cache == nil {
+		return fmt.Errorf("cache not found")
+	}
+	torrents := cache.GetTorrents()
+
+	dangling := make([]string, 0)
+	for _, t := range torrents {
+		if _, ok := uniqueItems[t.Name]; !ok {
+			dangling = append(dangling, t.Id)
+		}
+	}
+
+	r.logger.Info().Msgf("Found %d delapitated items", len(dangling))
+
+	if len(dangling) == 0 {
+		job.CompletedAt = time.Now()
+		job.Status = JobCompleted
+		return nil
+	}
+
+	client := r.deb.Clients["realdebrid"]
+	if client == nil {
+		return fmt.Errorf("client not found")
+	}
+	for _, id := range dangling {
+		client.DeleteTorrent(id)
+	}
+
+	return nil
+}
+
+func (r *Repair) cleanArr(j *Job, _arr string, tmdbId string) (map[string]string, error) {
+	uniqueItems := make(map[string]string)
+	a := r.arrs.Get(_arr)
+
+	r.logger.Info().Msgf("Starting repair for %s", a.Name)
+	media, err := a.GetMedia(tmdbId)
+	if err != nil {
+		r.logger.Info().Msgf("Failed to get %s media: %v", a.Name, err)
+		return uniqueItems, err
+	}
+
+	// Create a new error group
+	g, ctx := errgroup.WithContext(context.Background())
+
+	mu := sync.Mutex{}
+
+	// Limit concurrent goroutines
+	g.SetLimit(runtime.NumCPU() * 4)
+
+	for _, m := range media {
+		m := m // Create a new variable scoped to the loop iteration
+		g.Go(func() error {
+			// Check if context was canceled
+			select {
+			case <-ctx.Done():
+				return ctx.Err()
+			default:
+			}
+
+			u := r.getUniquePaths(m)
+			for k, v := range u {
+				mu.Lock()
+				uniqueItems[k] = v
+				mu.Unlock()
+			}
+			return nil
+		})
+	}
+
+	if err := g.Wait(); err != nil {
+		return uniqueItems, err
+	}
+
+	r.logger.Info().Msgf("Repair completed for %s. %d unique items", a.Name, len(uniqueItems))
+	return uniqueItems, nil
+}
+
 func (r *Repair) repairArr(j *Job, _arr string, tmdbId string) ([]arr.ContentFile, error) {
 	brokenItems := make([]arr.ContentFile, 0)
 	a := r.arrs.Get(_arr)
@@ -575,6 +741,7 @@ func (r *Repair) ProcessJob(id string) error {
 	if job == nil {
 		return fmt.Errorf("job %s not found", id)
 	}
+	// All validation checks remain the same
 	if job.Status != JobPending {
 		return fmt.Errorf("job %s not pending", id)
 	}
@@ -598,6 +765,7 @@ func (r *Repair) ProcessJob(id string) error {
 
 	// Create a new error group
 	g := new(errgroup.Group)
+	g.SetLimit(runtime.NumCPU() * 4)
 
 	for arrName, items := range brokenItems {
 		items := items
@@ -612,7 +780,6 @@ func (r *Repair) ProcessJob(id string) error {
 			if err := a.DeleteFiles(items); err != nil {
 				r.logger.Error().Err(err).Msgf("Failed to delete broken items for %s", arrName)
 				return nil
-
 			}
 			// Search for missing items
 			if err := a.SearchMissing(items); err != nil {
@@ -620,20 +787,29 @@ func (r *Repair) ProcessJob(id string) error {
 				return nil
 			}
 			return nil
-
 		})
 	}
 
-	if err := g.Wait(); err != nil {
-		job.FailedAt = time.Now()
-		job.Error = err.Error()
-		job.CompletedAt = time.Now()
-		job.Status = JobFailed
-		return err
-	}
+	// Update job status to in-progress
+	job.Status = JobProcessing
+	r.saveToFile()
 
-	job.CompletedAt = time.Now()
-	job.Status = JobCompleted
+	// Launch a goroutine to wait for completion and update the job
+	go func() {
+		if err := g.Wait(); err != nil {
+			job.FailedAt = time.Now()
+			job.Error = err.Error()
+			job.CompletedAt = time.Now()
+			job.Status = JobFailed
+			r.logger.Error().Err(err).Msgf("Job %s failed", id)
+		} else {
+			job.CompletedAt = time.Now()
+			job.Status = JobCompleted
+			r.logger.Info().Msgf("Job %s completed successfully", id)
+		}
+
+		r.saveToFile()
+	}()
 
 	return nil
 }