Refactor everything to use nsq instead of hosting a http server

The point of this is (hopefully) some sort of resiliency
I do not want to lose any messages ever
And I want to be able to kill this process whenever it is misbehaving
Hopefully this achieves that goal

dl/go.mod

@@ -1,3 +1,3 @@
-module main
+module dl
 
 go 1.22.4

dl/sync

@@ -1 +0,0 @@
-main.exe,"C:\Program Files\Git\usr\bin\dl.exe",t

download.go

@@ -0,0 +1,33 @@
+package main
+
+import (
+	"context"
+	"log"
+
+	"github.com/lrstanley/go-ytdlp"
+)
+
+const OUTPUT_DIR = "C:/Users/Administrator/ytdlpVideos"
+
+var dl = ytdlp.New().
+	// FormatSort("bestvideo[ext=mp4]+bestaudio[ext=m4a]").
+	FormatSort("res,ext:mp4:m4a").
+	Output("C:/Users/Administrator/ytdlpVideos/%(uploader)s/%(title)s.%(ext)s").
+	LimitRate("50M").
+	// HTTPChunkSize("20M").
+	MarkWatched().
+	SponsorblockMark("all").
+	PrintJSON().
+	RecodeVideo("mp4").
+	ConcurrentFragments(4)
+
+func Download(event PBEvent, status chan error) {
+	_, err := dl.Run(context.TODO(), event.Record.Link)
+	if err != nil {
+		status <- err
+		return
+	}
+
+	log.Printf("Downloaded %s (%s)", event.Record.Id, event.Record.Link)
+	SetDownloaded(event)
+}

downloader/download.go

@@ -2,6 +2,7 @@ package main
 
 import (
 	"context"
+	"fmt"
 	"log"
 	"sync"
 
@@ -11,11 +12,6 @@ import (
 
 const OUTPUT_DIR = "C:/Users/Administrator/ytdlpVideos"
 
-type DownloadWorker struct {
-	id int
-	input chan *DownloadTask
-}
-
 var ongoingDownloads = make(map[string]struct{})
 var ongoingDownloadsMutex = &sync.Mutex{}
 
@@ -23,49 +19,42 @@ var dl = ytdlp.New().
 	// FormatSort("bestvideo[ext=mp4]+bestaudio[ext=m4a]").
 	FormatSort("res,ext:mp4:m4a").
 	Output("C:/Users/Administrator/ytdlpVideos/%(uploader)s/%(title)s.%(ext)s").
-	LimitRate("10M").
+	LimitRate(fmt.Sprintf("%dM", 100/DOWNLOAD_WORKERS)).
 	// HTTPChunkSize("20M").
 	MarkWatched().
 	SponsorblockMark("all").
 	RecodeVideo("mp4").
 	ConcurrentFragments(6)
 
-func (w *DownloadWorker) Run() {
+func Download(url string) error {
-	for {
+	_, ongoing := ongoingDownloads[url]
-		task, ok := <-w.input
-		if !ok {
-			log.Printf("DownloadWorker %d: input channel closed, exiting", w.id)
-			return
-		}
-		_, ongoing := ongoingDownloads[task.Url]
 	if ongoing {
-			log.Printf("DownloadWorker %d: Download %s is already ongoing", w.id, task.Url)
+		// return fmt.Errorf("Download %s is already ongoing", url)
-			continue
+		return nil
 	}
 	ongoingDownloadsMutex.Lock()
-		ongoingDownloads[task.Url] = struct{}{}
+	ongoingDownloads[url] = struct{}{}
 	ongoingDownloadsMutex.Unlock()
 
-		log.Printf("DownloadWorker %d: Downloading %s", w.id, task.Url)
+	log.Printf("Downloading %s", url)
 
 	err := beeep.Beep(beeep.DefaultFreq, beeep.DefaultDuration)
 	if err != nil {
-			log.Printf("Failed beeping with %+v", err)
+		Warning.Printf("Failed beeping with %+v", err)
 	}
-		err = beeep.Alert("Download Started", task.Url, "assets/information.png")
+	err = beeep.Alert("Download Started", url, "assets/information.png")
 	if err != nil {
-			log.Printf("Failed alerting with %+v", err)
+		Warning.Printf("Failed alerting with %+v", err)
 	}
 
-		_, err = dl.Run(context.TODO(), task.Url)
+	_, err = dl.Run(context.TODO(), url)
 	if err != nil {
-			log.Printf("DownloadWorker %d: Failed downloading %s with %+v", w.id, task.Url, err)
+		return fmt.Errorf("failed downloading %s with %+v", url, err)
-			continue
 	}
 
-		log.Printf("DownloadWorker %d: Downloaded %s", w.id, task.Url)
+	log.Printf("Downloaded %s", url)
 	ongoingDownloadsMutex.Lock()
-		delete(ongoingDownloads, task.Url)
+	delete(ongoingDownloads, url)
 	ongoingDownloadsMutex.Unlock()
-	}
+	return nil
 }

downloader/go.mod

@@ -7,6 +7,7 @@ require (
 	github.com/cloudflare/circl v1.3.7 // indirect
 	github.com/go-toast/toast v0.0.0-20190211030409-01e6764cf0a4 // indirect
 	github.com/godbus/dbus/v5 v5.1.0 // indirect
+	github.com/golang/snappy v0.0.1 // indirect
 	github.com/nu7hatch/gouuid v0.0.0-20131221200532-179d4d0c4d8d // indirect
 	github.com/tadvi/systray v0.0.0-20190226123456-11a2b8fa57af // indirect
 	golang.org/x/crypto v0.21.0 // indirect
@@ -17,4 +18,5 @@ require (
 	github.com/gen2brain/beeep v0.0.0-20240516210008-9c006672e7f4
 	github.com/gorilla/websocket v1.5.3
 	github.com/lrstanley/go-ytdlp v0.0.0-20240616011628-f35a10876c99
+	github.com/nsqio/go-nsq v1.1.0
 )

downloader/go.sum

@@ -10,10 +10,14 @@ github.com/go-toast/toast v0.0.0-20190211030409-01e6764cf0a4 h1:qZNfIGkIANxGv/Oq
 github.com/go-toast/toast v0.0.0-20190211030409-01e6764cf0a4/go.mod h1:kW3HQ4UdaAyrUCSSDR4xUzBKW6O2iA4uHhk7AtyYp10=
 github.com/godbus/dbus/v5 v5.1.0 h1:4KLkAxT3aOY8Li4FRJe/KvhoNFFxo0m6fNuFUO8QJUk=
 github.com/godbus/dbus/v5 v5.1.0/go.mod h1:xhWf0FNVPg57R7Z0UbKHbJfkEywrmjJnf7w5xrFpKfA=
+github.com/golang/snappy v0.0.1 h1:Qgr9rKW7uDUkrbSmQeiDsGa8SjGyCOGtuasMWwvp2P4=
+github.com/golang/snappy v0.0.1/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
 github.com/gorilla/websocket v1.5.3 h1:saDtZ6Pbx/0u+bgYQ3q96pZgCzfhKXGPqt7kZ72aNNg=
 github.com/gorilla/websocket v1.5.3/go.mod h1:YR8l580nyteQvAITg2hZ9XVh4b55+EU/adAjf1fMHhE=
 github.com/lrstanley/go-ytdlp v0.0.0-20240616011628-f35a10876c99 h1:ZAo7qJht9PqefOD7C0ZKQ8dEkpJeM955sYw0FtQnzvo=
 github.com/lrstanley/go-ytdlp v0.0.0-20240616011628-f35a10876c99/go.mod h1:75ujbafjqiJugIGw4K6o52/p8C0m/kt+DrYwgClXYT4=
+github.com/nsqio/go-nsq v1.1.0 h1:PQg+xxiUjA7V+TLdXw7nVrJ5Jbl3sN86EhGCQj4+FYE=
+github.com/nsqio/go-nsq v1.1.0/go.mod h1:vKq36oyeVXgsS5Q8YEO7WghqidAVXQlcFxzQbQTuDEY=
 github.com/nu7hatch/gouuid v0.0.0-20131221200532-179d4d0c4d8d h1:VhgPp6v9qf9Agr/56bj7Y/xa04UccTW04VP0Qed4vnQ=
 github.com/nu7hatch/gouuid v0.0.0-20131221200532-179d4d0c4d8d/go.mod h1:YUTz3bUH2ZwIWBy3CJBeOBEugqcmXREj14T+iG/4k4U=
 github.com/tadvi/systray v0.0.0-20190226123456-11a2b8fa57af h1:6yITBqGTE2lEeTPG04SN9W+iWHCRyHqlVYILiSXziwk=

downloader/main.go

@@ -2,12 +2,20 @@ package main
 
 import (
 	"encoding/json"
+	"fmt"
 	"io"
 	"log"
-	"net/http"
 	"os"
+	"os/signal"
+	"syscall"
+	"time"
+
+	"github.com/nsqio/go-nsq"
 )
 
+var Error *log.Logger
+var Warning *log.Logger
+
 func init() {
 	log.SetFlags(log.Lmicroseconds | log.Lshortfile)
 	logFile, err := os.Create("main.log")
@@ -17,33 +25,66 @@ func init() {
 	}
 	logger := io.MultiWriter(os.Stdout, logFile)
 	log.SetOutput(logger)
+
+	Error = log.New(io.MultiWriter(logFile, os.Stderr, os.Stdout),
+		fmt.Sprintf("%sERROR:%s ", "\033[0;101m", "\033[0m"),
+		log.Lmicroseconds|log.Lshortfile)
+	Warning = log.New(io.MultiWriter(logFile, os.Stdout),
+		fmt.Sprintf("%sWarning:%s ", "\033[0;93m", "\033[0m"),
+		log.Lmicroseconds|log.Lshortfile)
 }
 
-const DOWNLOAD_WORKERS = 10
+const DOWNLOAD_WORKERS = 4
+
+type DLHandler struct{}
+
+func (*DLHandler) HandleMessage(message *nsq.Message) error {
+	log.Printf("Received message '%s' with %d attempts", message.Body, message.Attempts)
+	data := DownloadRequest{}
+	err := json.Unmarshal(message.Body, &data)
+	if err != nil {
+		Error.Printf("Error unmarshalling message: %v", err)
+		return err
+	}
+	log.Printf("Downloading %s", data.Link)
+	err = Download(data.Link)
+	if err != nil {
+		Error.Printf("Error downloading %s: %v", data.Link, err)
+		return err
+	}
+	message.Finish()
+	return nil
+}
 
 func main() {
-	downloadQueue := make(chan *DownloadTask, 100)
+	config := nsq.NewConfig()
+	config.MaxAttempts = 5
+	config.MaxInFlight = DOWNLOAD_WORKERS
+	config.MsgTimeout = 10 * time.Second
+
+	consumer, err := nsq.NewConsumer("ytdqueue", "dl", config)
+	if err != nil {
+		Error.Printf("Error creating consumer: %v", err)
+		return
+	}
 	for i := 0; i < DOWNLOAD_WORKERS; i++ {
-		worker := &DownloadWorker{id: i, input: downloadQueue}
+		consumer.AddHandler(&DLHandler{})
-		go worker.Run()
 	}
 
-	http.HandleFunc("/download", func(responseWriter http.ResponseWriter, request *http.Request) {
+	err = consumer.ConnectToNSQD("nsq.site.quack-lab.dev:41505")
-		defer request.Body.Close()
-
-		req := DownloadRequest{}
-		err := json.NewDecoder(request.Body).Decode(&req)
 	if err != nil {
-			log.Printf("Error parsing JSON: %v", err)
+		Error.Printf("Error connecting to nsqlookupd: %v", err)
-			http.Error(responseWriter, "Error parsing JSON", http.StatusBadRequest)
 		return
 	}
 
-		downloadQueue <- &DownloadTask{Url: req.Link}
+	sigChan := make(chan os.Signal, 1)
-	})
+	signal.Notify(sigChan, syscall.SIGINT, syscall.SIGTERM)
-	log.Println("Server starting on :5000")
+
-	err := http.ListenAndServe(":5000", nil)
+	<-sigChan
-	if err != nil {
+	log.Println("Received signal to terminate. Initiating graceful shutdown...")
-		log.Println("Error starting server:", err)
+
-	}
+	consumer.Stop()
+	<-consumer.StopChan
+
+	log.Println("Graceful shutdown completed.")
 }

downloader/types.go

@@ -1,8 +1,5 @@
 package main
 
-type DownloadTask struct {
-	Url string
-}
 type DownloadRequest struct {
 	Link string `json:"link"`
 }

extension/background.js

@@ -0,0 +1,10 @@
+browser.menus.create({
+	id: "youtube-download",
+	title: "Download",
+	contexts: ["all"],
+	documentUrlPatterns: ["*://*.youtube.com/*"],
+});
+
+browser.menus.onShown.addListener((info, tab) => {
+	browser.tabs.sendMessage(tab.id, { action: "check-element", info: info });
+});

extension/content.js

@@ -0,0 +1,34 @@
+const URL = `http://localhost:5000/download`;
+let lastRightClickCoords = { x: 0, y: 0 };
+
+document.addEventListener("contextmenu", (event) => {
+	lastRightClickCoords = { x: event.clientX, y: event.clientY };
+	console.log("Right-click coordinates:", lastRightClickCoords);
+});
+
+browser.runtime.onMessage.addListener((message, sender, sendResponse) => {
+	const { x, y } = lastRightClickCoords;
+	let element = document.elementFromPoint(x, y);
+
+	while (element && element.tagName != "YTD-RICH-ITEM-RENDERER") {
+		element = element.parentElement;
+	}
+	if (!element.tagName == "YTD-RICH-ITEM-RENDERER") {
+		console.error("No video element found.");
+		return;
+	}
+
+	const link = element.querySelector("a#video-title-link").href;
+	fetch(URL, {
+		method: "POST",
+		headers: {
+			"Content-Type": "application/json",
+		},
+		body: JSON.stringify({
+			link: link,
+		}),
+	}).then((res) => {
+		console.log(res);
+		res.json().then((data) => console.log(data));
+	});
+});

extension/manifest.json

@@ -0,0 +1,20 @@
+{
+	"manifest_version": 2,
+	"name": "Youtube Downloader",
+	"version": "1.0",
+	"permissions": [
+		"menus",
+		"activeTab",
+		"tabs",
+		"http://localhost:5000/*"
+	],
+	"background": {
+		"scripts": ["background.js"]
+	},
+	"content_scripts": [
+		{
+			"matches": ["*://*.youtube.com/*"],
+			"js": ["content.js"]
+		}
+	]
+}

hotkey.js

@@ -0,0 +1,41 @@
+nodes = document.querySelectorAll(":hover");
+i = 1;
+console.log(nodes);
+titleNode = nodes[nodes.length - i];
+
+selector = "a#video-title-link";
+invidious = false;
+if (window.location.href.includes("invidious.site")) {
+	selector = "a";
+	invidious = true;
+}
+
+while (titleNode && !titleNode.matches(selector)) {
+	titleNode = titleNode.parentElement;
+	console.log(titleNode);
+}
+if (!(titleNode && titleNode.matches(selector))) {
+	console.error("No video element found.");
+} else {
+	link = titleNode.href;
+	if (link.startsWith("/")) {
+		link = window.location.origin + link;
+	}
+	console.log(link);
+	fetch("http://localhost:5000/download", {
+		method: "POST",
+		headers: {
+			"Content-Type": "application/json",
+		},
+		body: JSON.stringify({
+			link: link,
+		}),
+	}).then((res) => {
+		textNode = titleNode.querySelector("yt-formatted-string");
+		if (invidious) {
+			textNode = titleNode.querySelector("p");
+		}
+		textNode.style.setProperty("color", "green", "important");
+		res.json().then((data) => console.log(data));
+	});
+}