ref: Move the yt manager to the worker

Simplify the worker/manager relationship. Now the worker is responsible
for the managing the yt-dlp process as well.

Also introduce chan to report back logs. That is an attempt to decouple
things.

1 files changed, 41 insertions, 6 deletions

diff --git a/worker/worker.go b/worker/worker.go
index 2444e89..525a736 100644
--- a/worker/worker.go
+++ b/worker/worker.go
@@ -1,10 +1,13 @@
 package worker
 
 import (
+	"bufio"
+	"bytes"
 	"context"
+	"fmt"
+	"os/exec"
 
 	"git.sr.ht/~gabrielgio/midr/db"
-	"git.sr.ht/~gabrielgio/midr/yt"
 	work "git.sr.ht/~sircmpwn/dowork"
 )
 
@@ -40,6 +43,30 @@ func NewWorkder() Worker {
 	}
 }
 
+func RunYtDlpProcess(entry *db.Entry) (error, []byte, []byte) {
+	args := []string{entry.Link}
+	var stdout bytes.Buffer
+	var stderr bytes.Buffer
+
+	output_template := fmt.Sprintf("%s/%%(title)s.%%(ext)s", entry.OutputFolder)
+	args = append(args, "-o", output_template)
+
+	downloaded_txt := fmt.Sprintf("%s/downloaded.txt", entry.OutputFolder)
+	args = append(args, "--download-archive", downloaded_txt)
+
+	if len(entry.DateAfter) > 0 {
+		args = append(args, "--dateafter", entry.DateAfter)
+	}
+
+	cmd := exec.Command("yt-dlp", args...)
+	cmd.Stdout = bufio.NewWriter(&stdout)
+	cmd.Stderr = bufio.NewWriter(&stderr)
+
+	err := cmd.Run()
+
+	return err, stdout.Bytes(), stderr.Bytes()
+}
+
 func (w *Worker) CanEnqueue(index uint) bool {
 	v, found := w.jobs[index]
 	return !found || v == statusNotQueued
@@ -49,23 +76,31 @@ func (w *Worker) RemoveJob(id uint) {
 	delete(w.jobs, id)
 }
 
-func (w *Worker) SpawnWorker(entry *db.Entry) {
-
+func (w *Worker) RunYtDlpWorker(entry *db.Entry) <-chan []byte {
 	if !w.CanEnqueue(entry.ID) {
-		return
+		return nil
 	}
 
+	log := make(chan []byte)
+
 	w.c <- command{action: commandEnqueue, index: entry.ID}
 	task := work.NewTask(func(ctx context.Context) error {
 
 		w.c <- command{action: commandStart, index: entry.ID}
-		yt.RunYtDlpProcess(entry)
-		return nil
+		err, stdout, stderr := RunYtDlpProcess(entry)
+
+		log <- stdout
+		log <- stderr
+
+		return err
 	}).After(func(ctx context.Context, task *work.Task) {
 		w.c <- command{action: commandDequeue, index: entry.ID}
+		close(log)
 	})
 
 	work.Enqueue(task)
+
+	return log
 }
 
 func (w *Worker) startReader() {