Add a Task Manager service and update downloads to use it

.github/dependabot.yml

@@ -1,6 +1,3 @@
-# Please see the documentation for all configuration options:
-# https://docs.github.com/github/administering-a-repository/configuration-options-for-dependency-updates
-
 version: 2
 updates:
   - package-ecosystem: "gomod" 
@@ -11,3 +8,7 @@ updates:
     directory: "/web"
     schedule:
       interval: "weekly"
+  - package-ecosystem: "github-actions"
+    directory: "/"
+    schedule:
+      interval: "weekly"

.github/workflows/build.yml

@@ -23,11 +23,14 @@ jobs:
 
     - uses: actions/setup-go@v2
       with:
-        go-version: '1.17'
+        go-version: '1.20'
 
-    - name: go Version
+    - name: Go Version
       run: go version
 
+    - name: Go Vet
+      run: go vet ./...
+
     - name: Docker Version
       run: docker version
 
@@ -69,7 +72,7 @@ jobs:
       with:
         sarif_file: 'trivy-results.sarif'
   
-    # Release build
+# Release build
     - uses: docker/login-action@v1
       if: startsWith(github.ref, 'refs/tags/')
       with:

cmd/premiumizearrd/app.go

@@ -18,17 +18,20 @@ type App struct {
 	directoryWatcher   service.DirectoryWatcherService
 	webServer          service.WebServerService
 	arrsManager        service.ArrsManagerService
+	downloadManager    service.DownloadManagerService
+	taskRunner         service.TaskRunnerService
 }
 
 // Makes go vet error - prevents copies
 func (app *App) Lock()   {}
 func (app *App) UnLock() {}
 
+// Start
 func (app *App) Start(logLevel string, configFile string, loggingDirectory string) error {
 	//Setup static login
 	lvl, err := log.ParseLevel(logLevel)
 	if err != nil {
-		log.Errorf("Error flag not recognized, defaulting to Info!!", err)
+		log.Errorf("Error flag not recognized, defaulting to Info!! %v", err)
 		lvl = log.InfoLevel
 	}
 	log.SetLevel(lvl)
@@ -87,10 +90,16 @@ func (app *App) Start(logLevel string, configFile string, loggingDirectory strin
 	app.directoryWatcher = service.DirectoryWatcherService{}.New()
 	app.webServer = service.WebServerService{}.New()
 	app.arrsManager = service.ArrsManagerService{}.New()
+	app.downloadManager = service.DownloadManagerService{}.New()
+	app.taskRunner = service.TaskRunnerService{}.New()
 
 	// Initialise Services
+	app.taskRunner.Init(&app.config)
+
+	// Must come after taskRunner initialised
 	app.arrsManager.Init(&app.config)
 	app.directoryWatcher.Init(&app.premiumizemeClient, &app.config)
+	app.downloadManager.Init(&app.premiumizemeClient, &app.taskRunner, &app.config)
 
 	// Must come after arrsManager
 	app.transferManager.Init(&app.premiumizemeClient, &app.arrsManager, &app.config)
@@ -100,6 +109,7 @@ func (app *App) Start(logLevel string, configFile string, loggingDirectory strin
 	app.arrsManager.Start()
 	app.webServer.Start()
 	app.directoryWatcher.Start()
+	app.taskRunner.Start()
 	//Block until the program is terminated
 	app.transferManager.Run(15 * time.Second)
 

config.yaml

@@ -1,17 +1,20 @@
 PremiumizemeAPIKey: xxxxxxxxx
 Arrs:
-- Name: ""
+- Name: Sonarr
   URL: http://localhost:8989
   APIKey: xxxxxxxxx
   Type: Sonarr
-- Name: ""
+- Name: Radarr
   URL: http://localhost:7878
   APIKey: xxxxxxxxx
   Type: Radarr
 BlackholeDirectory: ""
+PollBlackholeDirectory: false
+PollBlackholeIntervalMinutes: 10
 DownloadsDirectory: ""
 UnzipDirectory: ""
 bindIP: 0.0.0.0
 bindPort: "8182"
 WebRoot: ""
 SimultaneousDownloads: 5
+ArrHistoryUpdateIntervalSeconds: 20

go.mod

@@ -1,6 +1,6 @@
 module github.com/jackdallas/premiumizearr
 
-go 1.17
+go 1.20
 
 require (
 	github.com/dustin/go-humanize v1.0.1

internal/arr/sonarr.go

@@ -69,7 +69,7 @@ func (arr *SonarrArr) HistoryContains(name string) (int64, bool) {
 			return item.ID, true
 		}
 	}
-	log.Tracef("Sonarr [%s]: %s Not in History", name)
+	log.Tracef("Sonarr [%s]: %s Not in History", arr.Name, name)
 
 	return -1, false
 }

internal/arr/types.go

@@ -14,11 +14,11 @@ import (
 
 func CompareFileNamesFuzzy(a, b string) bool {
 	//Strip file extension
-	a = utils.StripDownloadTypesExtention(a)
-	b = utils.StripDownloadTypesExtention(b)
+	a = utils.StripDownloadTypesExtension(a)
+	b = utils.StripDownloadTypesExtension(b)
 	//Strip media type extension
-	a = utils.StripMediaTypesExtention(a)
-	b = utils.StripMediaTypesExtention(b)
+	a = utils.StripMediaTypesExtension(a)
+	b = utils.StripMediaTypesExtension(b)
 	//Strip Spaces
 	a = strings.ReplaceAll(a, " ", "")
 	b = strings.ReplaceAll(b, " ", "")

internal/service/arrs_manager_service.go

@@ -53,7 +53,7 @@ func (am *ArrsManagerService) Start() {
 			am.arrs = append(am.arrs, &wrapper)
 			log.Tracef("Added Radarr arr: %s", arr_config.Name)
 		default:
-			log.Error("Unknown arr type: %s, not adding Arr %s", arr_config.Type, arr_config.Name)
+			log.Errorf("Unknown arr type: %s, not adding Arr %s", arr_config.Type, arr_config.Name)
 		}
 	}
 	log.Debugf("Created %d Arrs", len(am.arrs))

internal/service/directory_watcher_service.go

@@ -1,7 +1,7 @@
 package service
 
 import (
-	"io/ioutil"
+	"io/fs"
 	"os"
 	"path"
 	"path/filepath"
@@ -62,7 +62,7 @@ func (dw *DirectoryWatcherService) GetStatus() string {
 	return dw.status
 }
 
-//Start: This is the entrypoint for the directory watcher
+// Start: This is the entrypoint for the directory watcher
 func (dw *DirectoryWatcherService) Start() {
 	log.Info("Starting directory watcher...")
 
@@ -112,14 +112,14 @@ func (dw *DirectoryWatcherService) Start() {
 
 func (dw *DirectoryWatcherService) directoryScan(p string) {
 	log.Trace("Running directory scan")
-	files, err := ioutil.ReadDir(p)
+	files, err := os.ReadDir(p)
 	if err != nil {
 		log.Errorf("Error with directory scan %+v", err)
 		return
 	}
 
 	for _, file := range files {
-		go func(file os.FileInfo) {
+		go func(file fs.DirEntry) {
 			file_path := path.Join(p, file.Name())
 			if dw.checkFile(file_path) {
 				dw.addFileToQueue(file_path)
@@ -182,7 +182,7 @@ func (dw *DirectoryWatcherService) processUploads() {
 					log.Trace("File already uploaded, removing from Disk")
 					os.Remove(filePath)
 				default:
-					log.Error("Error creating transfer: %s", err)
+					log.Errorf("Error creating transfer: %s", err)
 				}
 			} else {
 				dw.status = "Okay"
@@ -194,7 +194,7 @@ func (dw *DirectoryWatcherService) processUploads() {
 			}
 			time.Sleep(time.Second * time.Duration(sleepTimeSeconds))
 		} else {
-			log.Errorf("Received %s from blackhole Queue. Appears to be an empty path.")
+			log.Error("Received blank string from blackhole Queue.")
 		}
 	}
 }

internal/service/download_manager_service.go

@@ -0,0 +1,166 @@
+package service
+
+import (
+	"os"
+	"path"
+	"strings"
+	"time"
+
+	"github.com/jackdallas/premiumizearr/internal/config"
+	"github.com/jackdallas/premiumizearr/internal/utils"
+
+	"github.com/jackdallas/premiumizearr/pkg/downloadmanager"
+	"github.com/jackdallas/premiumizearr/pkg/premiumizeme"
+	log "github.com/sirupsen/logrus"
+)
+
+type DownloadManagerService struct {
+	downloadManager    *downloadmanager.DownloadManager
+	taskRunner         *TaskRunnerService
+	premiumizemeClient *premiumizeme.Premiumizeme
+	config             *config.Config
+	downloadingIDs     map[string]bool
+
+	downloadsFolderID string
+}
+
+func (DownloadManagerService) New() DownloadManagerService {
+	return DownloadManagerService{
+		downloadsFolderID: "",
+		downloadManager:   &downloadmanager.DownloadManager{},
+		downloadingIDs:    make(map[string]bool),
+	}
+}
+
+func (manager *DownloadManagerService) Init(_premiumizemeClient *premiumizeme.Premiumizeme, taskRunner *TaskRunnerService, _config *config.Config) {
+	manager.premiumizemeClient = _premiumizemeClient
+	manager.taskRunner = taskRunner
+	manager.config = _config
+
+	manager.downloadsFolderID = utils.GetDownloadsFolderIDFromPremiumizeme(manager.premiumizemeClient)
+	manager.CleanUpUnzipDir()
+
+	log.Info("Starting download manager thread")
+	go manager.downloadManager.Run()
+	log.Info("Creating check premiumize downloads folder task")
+	manager.taskRunner.AddTask("Check Premiumize Downloads Folder", 20*time.Second, manager.TaskCheckPremiumizeDownloadsFolder)
+}
+
+func (manager *DownloadManagerService) CleanUpUnzipDir() {
+	log.Info("Cleaning unzip directory")
+
+	unzipBase, err := manager.config.GetUnzipBaseLocation()
+	if err != nil {
+		log.Errorf("Error getting unzip base location: %s", err.Error())
+		return
+	}
+
+	err = utils.RemoveContents(unzipBase)
+	if err != nil {
+		log.Errorf("Error cleaning unzip directory: %s", err.Error())
+		return
+	}
+
+}
+
+func (manager *DownloadManagerService) ConfigUpdatedCallback(currentConfig config.Config, newConfig config.Config) {
+	if currentConfig.UnzipDirectory != newConfig.UnzipDirectory {
+		manager.CleanUpUnzipDir()
+	}
+}
+
+func (manager *DownloadManagerService) TaskCheckPremiumizeDownloadsFolder() {
+	log.Debug("Running Task CheckPremiumizeDownloadsFolder")
+
+	items, err := manager.premiumizemeClient.ListFolder(manager.downloadsFolderID)
+	if err != nil {
+		log.Errorf("Error listing downloads folder: %s", err.Error())
+		return
+	}
+
+	for _, item := range items {
+		if _, ok := manager.downloadingIDs[item.ID]; ok {
+			continue
+		}
+
+		manager.downloadingIDs[item.ID] = true
+		manager.downloadFinishedTransfer(item, manager.config.DownloadsDirectory)
+	}
+
+}
+
+func (manager *DownloadManagerService) downloadFinishedTransfer(item premiumizeme.Item, downloadDirectory string) {
+	log.Debug("Downloading: ", item.Name)
+	log.Tracef("%+v", item)
+	var link string
+	var err error
+	if item.Type == "file" {
+		link, err = manager.premiumizemeClient.GenerateZippedFileLink(item.ID)
+	} else if item.Type == "folder" {
+		link, err = manager.premiumizemeClient.GenerateZippedFolderLink(item.ID)
+	} else {
+		log.Errorf("Item is not of type 'file' or 'folder' !! Can't download %s", item.Name)
+		return
+	}
+	if err != nil {
+		log.Errorf("Error generating download link: %s", err)
+		return
+	}
+	log.Trace("Downloading from: ", link)
+
+	tempDir, err := manager.config.GetNewUnzipLocation()
+	if err != nil {
+		log.Errorf("Could not create temp dir: %s", err)
+		return
+	}
+
+	splitString := strings.Split(link, "/")
+	savePath := path.Join(tempDir, splitString[len(splitString)-1])
+	log.Trace("Downloading to: ", savePath)
+
+	out, err := os.Create(savePath)
+	if err != nil {
+		log.Errorf("Could not create save path: %s", err)
+		return
+	}
+	defer out.Close()
+
+	transfer, err := manager.downloadManager.AddTransfer(link, savePath)
+	if err != nil {
+		log.Errorf("Could not add transfer: %s", err)
+		return
+	}
+
+	go func() {
+		<-transfer.Finished
+
+		if transfer.GetStatus() == downloadmanager.STATUS_ERROR || transfer.GetStatus() == downloadmanager.STATUS_CANCELED {
+			log.Errorf("Could not download file: %s", strings.Join(transfer.GetErrorStrings(), ", "))
+			return
+		}
+
+		unzipped := true
+		log.Tracef("Unzipping %s to %s", savePath, downloadDirectory)
+		err = utils.Unzip(savePath, downloadDirectory)
+		if err != nil {
+			log.Errorf("Could not unzip file: %s", err)
+			unzipped = false
+		}
+
+		log.Tracef("Removing zip %s from system", savePath)
+		err = os.RemoveAll(savePath)
+		if err != nil {
+			log.Errorf("Could not remove zip: %s", err)
+			return
+		}
+
+		if unzipped {
+			err = manager.premiumizemeClient.DeleteFolder(item.ID)
+			if err != nil {
+				log.Errorf("Error deleting folder on premiumize.me: %s", err)
+				return
+			}
+		}
+
+	}()
+}

internal/service/task_runner_service.go

@@ -0,0 +1,68 @@
+package service
+
+import (
+	"sync"
+	"time"
+
+	"github.com/jackdallas/premiumizearr/internal/config"
+)
+
+type ServiceTask struct {
+	TaskName      string        `json:"task_name"`
+	LastCompleted time.Time     `json:"last_completed"`
+	Interval      time.Duration `json:"interval"`
+	IsRunning     bool          `json:"is_running"`
+	function      func()
+}
+
+type TaskRunnerService struct {
+	tasks      []ServiceTask
+	tasksMutex *sync.RWMutex
+	config     *config.Config
+}
+
+func (TaskRunnerService) New() TaskRunnerService {
+	return TaskRunnerService{
+		tasks:      []ServiceTask{},
+		tasksMutex: &sync.RWMutex{},
+	}
+}
+
+func (manager *TaskRunnerService) Init(config *config.Config) {
+	manager.config = config
+}
+
+func (manager *TaskRunnerService) AddTask(taskName string, interval time.Duration, function func()) {
+	manager.tasksMutex.Lock()
+	defer manager.tasksMutex.Unlock()
+	manager.tasks = append(manager.tasks, ServiceTask{
+		TaskName:      taskName,
+		LastCompleted: time.Time{},
+		Interval:      interval,
+		IsRunning:     false,
+		function:      function,
+	})
+}
+
+func (manager *TaskRunnerService) Start() {
+	go func() {
+		for {
+			manager.tasksMutex.Lock()
+			for _, task := range manager.tasks {
+				if task.IsRunning {
+					continue
+				}
+				if time.Since(task.LastCompleted) > task.Interval {
+					task.IsRunning = true
+					go func(task ServiceTask) {
+						task.function()
+						task.LastCompleted = time.Now()
+						task.IsRunning = false
+					}(task)
+				}
+			}
+			manager.tasksMutex.Unlock()
+			time.Sleep(time.Millisecond * 50)
+		}
+	}()
+}

internal/service/transfer_manager_service.go

@@ -1,25 +1,15 @@
 package service
 
 import (
-	"os"
-	"path"
-	"strings"
-	"sync"
+	"fmt"
 	"time"
 
 	"github.com/jackdallas/premiumizearr/internal/config"
-	"github.com/jackdallas/premiumizearr/internal/progress_downloader"
 	"github.com/jackdallas/premiumizearr/internal/utils"
 	"github.com/jackdallas/premiumizearr/pkg/premiumizeme"
 	log "github.com/sirupsen/logrus"
 )
 
-type DownloadDetails struct {
-	Added              time.Time
-	Name               string
-	ProgressDownloader *progress_downloader.WriteCounter
-}
-
 type TransferManagerService struct {
 	premiumizemeClient *premiumizeme.Premiumizeme
 	arrsManager        *ArrsManagerService
@@ -27,10 +17,7 @@ type TransferManagerService struct {
 	lastUpdated        int64
 	transfers          []premiumizeme.Transfer
 	runningTask        bool
-	downloadListMutex  *sync.Mutex
-	downloadList       map[string]*DownloadDetails
 	status             string
-	downloadsFolderID  string
 }
 
 // Handle
@@ -41,10 +28,7 @@ func (t TransferManagerService) New() TransferManagerService {
 	t.lastUpdated = time.Now().Unix()
 	t.transfers = make([]premiumizeme.Transfer, 0)
 	t.runningTask = false
-	t.downloadListMutex = &sync.Mutex{}
-	t.downloadList = make(map[string]*DownloadDetails, 0)
 	t.status = ""
-	t.downloadsFolderID = ""
 	return t
 }
 
@@ -73,27 +57,19 @@ func (t *TransferManagerService) CleanUpUnzipDir() {
 }
 
 func (manager *TransferManagerService) ConfigUpdatedCallback(currentConfig config.Config, newConfig config.Config) {
-	if currentConfig.UnzipDirectory != newConfig.UnzipDirectory {
-		manager.CleanUpUnzipDir()
-	}
+	//NOOP
 }
 
 func (manager *TransferManagerService) Run(interval time.Duration) {
-	manager.downloadsFolderID = utils.GetDownloadsFolderIDFromPremiumizeme(manager.premiumizemeClient)
 	for {
 		manager.runningTask = true
 		manager.TaskUpdateTransfersList()
-		manager.TaskCheckPremiumizeDownloadsFolder()
 		manager.runningTask = false
 		manager.lastUpdated = time.Now().Unix()
 		time.Sleep(interval)
 	}
 }
 
-func (manager *TransferManagerService) GetDownloads() map[string]*DownloadDetails {
-	return manager.downloadList
-}
-
 func (manager *TransferManagerService) GetTransfers() *[]premiumizeme.Transfer {
 	return &manager.transfers
 }
@@ -101,6 +77,10 @@ func (manager *TransferManagerService) GetStatus() string {
 	return manager.status
 }
 
+func (manager *TransferManagerService) updateTransfers(transfers []premiumizeme.Transfer) {
+	manager.transfers = transfers
+}
+
 func (manager *TransferManagerService) TaskUpdateTransfersList() {
 	log.Debug("Running Task UpdateTransfersList")
 	transfers, err := manager.premiumizemeClient.GetTransfers()
@@ -111,6 +91,31 @@ func (manager *TransferManagerService) TaskUpdateTransfersList() {
 	manager.updateTransfers(transfers)
 
 	log.Tracef("Checking %d transfers against %d Arr clients", len(transfers), len(manager.arrsManager.GetArrs()))
+	earlyReturn := false
+
+	if len(transfers) == 0 {
+		manager.status = "No transfers"
+		earlyReturn = true
+	} else {
+		manager.status = fmt.Sprintf("Got %d transfers", len(transfers))
+	}
+
+	if len(manager.arrsManager.GetArrs()) == 0 {
+		manager.status = fmt.Sprintf("%s, no ARRs available", manager.status)
+		earlyReturn = true
+	}
+	//else {
+	// 	//TODO: Test
+	// 	// if manager.status[len(manager.status)-19:] == ", no ARRs available" {
+	// 	// 	manager.status = manager.status[:len(manager.status)-19]
+	// 	// }
+	// 	fmt.Print(manager.status)
+	// }
+
+	if earlyReturn {
+		return
+	}
+
 	for _, transfer := range transfers {
 		found := false
 		for _, arr := range manager.arrsManager.GetArrs() {
@@ -133,149 +138,3 @@ func (manager *TransferManagerService) TaskUpdateTransfersList() {
 		}
 	}
 }
-
-func (manager *TransferManagerService) TaskCheckPremiumizeDownloadsFolder() {
-	log.Debug("Running Task CheckPremiumizeDownloadsFolder")
-
-	items, err := manager.premiumizemeClient.ListFolder(manager.downloadsFolderID)
-	if err != nil {
-		log.Errorf("Error listing downloads folder: %s", err.Error())
-		return
-	}
-
-	for _, item := range items {
-		if manager.countDownloads() < manager.config.SimultaneousDownloads {
-			log.Debugf("Processing completed item: %s", item.Name)
-			manager.HandleFinishedItem(item, manager.config.DownloadsDirectory)
-		} else {
-			log.Debugf("Not processing any more transfers, %d are running and cap is %d", manager.countDownloads(), manager.config.SimultaneousDownloads)
-			break
-		}
-	}
-}
-
-func (manager *TransferManagerService) updateTransfers(transfers []premiumizeme.Transfer) {
-	manager.transfers = transfers
-}
-
-func (manager *TransferManagerService) addDownload(item *premiumizeme.Item) {
-	manager.downloadListMutex.Lock()
-	defer manager.downloadListMutex.Unlock()
-
-	manager.downloadList[item.Name] = &DownloadDetails{
-		Added:              time.Now(),
-		Name:               item.Name,
-		ProgressDownloader: progress_downloader.NewWriteCounter(),
-	}
-}
-
-func (manager *TransferManagerService) countDownloads() int {
-	manager.downloadListMutex.Lock()
-	defer manager.downloadListMutex.Unlock()
-
-	return len(manager.downloadList)
-}
-
-func (manager *TransferManagerService) removeDownload(name string) {
-	manager.downloadListMutex.Lock()
-	defer manager.downloadListMutex.Unlock()
-
-	delete(manager.downloadList, name)
-}
-
-func (manager *TransferManagerService) downloadExists(itemName string) bool {
-	manager.downloadListMutex.Lock()
-	defer manager.downloadListMutex.Unlock()
-
-	for _, dl := range manager.downloadList {
-		if dl.Name == itemName {
-			return true
-		}
-	}
-
-	return false
-}
-
-// Returns when the download has been added to the list
-func (manager *TransferManagerService) HandleFinishedItem(item premiumizeme.Item, downloadDirectory string) {
-	if manager.downloadExists(item.Name) {
-		log.Tracef("Transfer %s is already downloading", item.Name)
-		return
-	}
-
-	manager.addDownload(&item)
-
-	go func() {
-		log.Debug("Downloading: ", item.Name)
-		log.Tracef("%+v", item)
-		var link string
-		var err error
-		if item.Type == "file" {
-			link, err = manager.premiumizemeClient.GenerateZippedFileLink(item.ID)
-		} else if item.Type == "folder" {
-			link, err = manager.premiumizemeClient.GenerateZippedFolderLink(item.ID)
-		} else {
-			log.Errorf("Item is not of type 'file' or 'folder' !! Can't download %s", item.Name)
-			return
-		}
-		if err != nil {
-			log.Error("Error generating download link: %s", err)
-			manager.removeDownload(item.Name)
-			return
-		}
-		log.Trace("Downloading from: ", link)
-
-		tempDir, err := manager.config.GetNewUnzipLocation()
-		if err != nil {
-			log.Errorf("Could not create temp dir: %s", err)
-			manager.removeDownload(item.Name)
-			return
-		}
-
-		splitString := strings.Split(link, "/")
-		savePath := path.Join(tempDir, splitString[len(splitString)-1])
-		log.Trace("Downloading to: ", savePath)
-
-		out, err := os.Create(savePath)
-		if err != nil {
-			log.Errorf("Could not create save path: %s", err)
-			manager.removeDownload(item.Name)
-			return
-		}
-		defer out.Close()
-
-		err = progress_downloader.DownloadFile(link, savePath, manager.downloadList[item.Name].ProgressDownloader)
-
-		if err != nil {
-			log.Errorf("Could not download file: %s", err)
-			manager.removeDownload(item.Name)
-			return
-		}
-
-		log.Tracef("Unzipping %s to %s", savePath, downloadDirectory)
-		err = utils.Unzip(savePath, downloadDirectory)
-		if err != nil {
-			log.Errorf("Could not unzip file: %s", err)
-			manager.removeDownload(item.Name)
-			return
-		}
-
-		log.Tracef("Removing zip %s from system", savePath)
-		err = os.RemoveAll(savePath)
-		if err != nil {
-			manager.removeDownload(item.Name)
-			log.Errorf("Could not remove zip: %s", err)
-			return
-		}
-
-		err = manager.premiumizemeClient.DeleteFolder(item.ID)
-		if err != nil {
-			manager.removeDownload(item.Name)
-			log.Error("Error deleting folder on premiumize.me: %s", err)
-			return
-		}
-
-		//Remove download entry from downloads map
-		manager.removeDownload(item.Name)
-	}()
-}

internal/service/web_service_handlers.go

@@ -37,6 +37,7 @@ type BlackholeResponse struct {
 }
 
 type Download struct {
+	ID       int64  `json:"id"`
 	Added    int64  `json:"added"`
 	Name     string `json:"name"`
 	Progress string `json:"progress"`
@@ -53,14 +54,15 @@ func (s *WebServerService) DownloadsHandler(w http.ResponseWriter, r *http.Reque
 	if s.transferManager == nil {
 		resp.Status = "Not Initialized"
 	} else {
-		for _, v := range s.transferManager.GetDownloads() {
-			resp.Downloads = append(resp.Downloads, Download{
-				Added:    v.Added.Unix(),
-				Name:     v.Name,
-				Progress: v.ProgressDownloader.GetProgress(),
-				Speed:    v.ProgressDownloader.GetSpeed(),
-			})
-		}
+		// for _, v := range s.transferManager.GetDownloads() {
+		// 	resp.Downloads = append(resp.Downloads, Download{
+		// 		ID:       v.ID,
+		// 		Added:    v.Added.Unix(),
+		// 		Name:     v.Name,
+		// 		Progress: v.ProgressDownloader.GetProgress(),
+		// 		Speed:    v.ProgressDownloader.GetSpeed(),
+		// 	})
+		// }
 		resp.Status = ""
 	}
 

internal/utils/utils.go

@@ -12,7 +12,7 @@ import (
 	log "github.com/sirupsen/logrus"
 )
 
-func StripDownloadTypesExtention(fileName string) string {
+func StripDownloadTypesExtension(fileName string) string {
 	var exts = [...]string{".nzb", ".magnet"}
 	for _, ext := range exts {
 		fileName = strings.TrimSuffix(fileName, ext)
@@ -21,7 +21,7 @@ func StripDownloadTypesExtention(fileName string) string {
 	return fileName
 }
 
-func StripMediaTypesExtention(fileName string) string {
+func StripMediaTypesExtension(fileName string) string {
 	var exts = [...]string{".mkv", ".mp4", ".avi", ".mov", ".flv", ".wmv", ".mpg", ".mpeg", ".m4v", ".3gp", ".3g2", ".m2ts", ".mts", ".ts", ".webm", ".m4a", ".m4b", ".m4p", ".m4r", ".m4v"}
 	for _, ext := range exts {
 		fileName = strings.TrimSuffix(fileName, ext)
@@ -93,11 +93,11 @@ func StringInSlice(a string, list []string) int {
 
 func GetDownloadsFolderIDFromPremiumizeme(premiumizemeClient *premiumizeme.Premiumizeme) string {
 	var downloadsFolderID string
+
 	folders, err := premiumizemeClient.GetFolders()
 	if err != nil {
 		log.Errorf("Error getting folders: %s", err)
-		log.Errorf("Cannot read folders from premiumize.me, application will not run!")
-		return ""
+		log.Fatal("Cannot read folders from premiumize.me, application will not run!")
 	}
 
 	const folderName = "arrDownloads"
@@ -143,18 +143,18 @@ func IsRunningInDockerContainer() bool {
 
 func IsDirectoryWriteable(path string) bool {
 	if _, err := os.Stat(path); os.IsNotExist(err) {
-		log.Errorf("Directory does not exist: ", path)
+		log.Errorf("Directory does not exist: %s", path)
 		return false
 	}
 
 	if _, err := os.Create(path + "/test.txt"); err != nil {
-		log.Errorf("Cannot write test.txt to directory: ", path)
+		log.Errorf("Cannot write test.txt to directory: %s", path)
 		return false
 	}
 
 	// Delete test file
 	if err := os.Remove(path + "/test.txt"); err != nil {
-		log.Errorf("Cannot delete test.txt file in: ", path)
+		log.Errorf("Cannot delete test.txt file in: %s", path)
 		return false
 	}
 

pkg/clouddownloader/interfaces.go

@@ -1,6 +0,0 @@
-package clouddownloader
-
-// Interface for the CloudDownloader interface
-type CloudDownloaderInterface interface {
-	GetTransfers() []Transfer
-}

pkg/clouddownloader/types.go

@@ -1,4 +0,0 @@
-package clouddownloader
-
-type Transfer struct {
-}

pkg/downloadmanager/downloadmanager.go

@@ -0,0 +1,86 @@
+package downloadmanager
+
+import (
+	"time"
+
+	log "github.com/sirupsen/logrus"
+)
+
+func (d *DownloadManager) Run() {
+	for {
+		select {
+		case <-d.CancelChannel:
+			return
+		default:
+			time.Sleep(time.Millisecond * 100)
+			for i := 0; i < len(d.transfers); i++ {
+				t := &d.transfers[i]
+				switch t.GetStatus() {
+				case STATUS_QUEUED:
+					if d.GetActiveTransferCount() < d.MaxSimultaneousDownloads {
+						if err := t.Download(); err != nil {
+							log.Errorf("Error downloading: %s", err)
+						}
+					} else {
+						log.Debugf("Too many active transfers, skipping %d", t.GetID())
+					}
+					return
+				}
+			}
+		}
+	}
+}
+
+func (d *DownloadManager) GetTransfers() []Transfer {
+	d.transfersLock.Lock()
+	defer d.transfersLock.Unlock()
+	return d.transfers
+}
+
+func (d *DownloadManager) GetTransfer(id int64) (*Transfer, error) {
+	d.transfersLock.Lock()
+	defer d.transfersLock.Unlock()
+	for i := 0; i < len(d.transfers); i++ {
+		if d.transfers[i].GetID() == id {
+			return &d.transfers[i], nil
+		}
+	}
+	return nil, ErrorNoTransferWithID
+}
+
+func (d *DownloadManager) AddTransfer(url string, savePath string) (*Transfer, error) {
+	d.transfersLock.Lock()
+	defer d.transfersLock.Unlock()
+
+	nextID := d.IdCounter.Add(1)
+
+	d.transfers = append(d.transfers, NewTransfer(nextID, url, savePath))
+
+	log.Debugf("Added transfer %d", nextID)
+	return d.GetTransfer(nextID)
+}
+
+func (d *DownloadManager) GetActiveTransferCount() int {
+	c := 0
+
+	for i := 0; i < len(d.transfers); i++ {
+		if d.transfers[i].GetStatus() == STATUS_DOWNLOADING {
+			c++
+		}
+	}
+
+	return c
+}
+
+func (d *DownloadManager) RemoveTransfer(id int64) error {
+	d.transfersLock.Lock()
+	defer d.transfersLock.Unlock()
+
+	for i := range d.transfers {
+		if d.transfers[i].GetID() == id {
+			return d.transfers[i].Cancel()
+		}
+	}
+
+	return ErrorNoTransferWithID
+}

pkg/downloadmanager/progress_downloader.go

@@ -1,4 +1,4 @@
-package progress_downloader
+package downloadmanager
 
 // https://golangcode.com/download-a-file-with-progress/
 
@@ -19,6 +19,7 @@ type WriteCounter struct {
 	LastUpdate time.Time
 	LastAmount uint64
 	Total      uint64
+	Closing    bool
 }
 
 func NewWriteCounter() *WriteCounter {
@@ -41,6 +42,9 @@ func (wc *WriteCounter) GetSpeed() string {
 }
 
 func (wc *WriteCounter) Write(p []byte) (int, error) {
+	if wc.Closing {
+		return 0, io.ErrClosedPipe
+	}
 	n := len(p)
 	wc.LastAmount = wc.Total
 	wc.Total += uint64(n)
@@ -52,9 +56,6 @@ func (wc WriteCounter) GetProgress() string {
 	return fmt.Sprintf("%s complete", humanize.Bytes(wc.Total))
 }
 
-// DownloadFile will download a url to a local file. It's efficient because it will
-// write as it downloads and not load the whole file into memory. We pass an io.TeeReader
-// into Copy() to report progress on the download.
 func DownloadFile(url string, filepath string, counter *WriteCounter) error {
 
 	// Create the file, but give it a tmp file extension, this means we won't overwrite a
@@ -72,6 +73,7 @@ func DownloadFile(url string, filepath string, counter *WriteCounter) error {
 	}
 	defer resp.Body.Close()
 
+	// resp.Body.
 	if _, err = io.Copy(out, io.TeeReader(resp.Body, counter)); err != nil {
 		out.Close()
 		return err

pkg/downloadmanager/transfer.go

@@ -0,0 +1,208 @@
+package downloadmanager
+
+import (
+	"errors"
+	"fmt"
+	"io"
+	"net/http"
+	"net/url"
+	"os"
+	"strings"
+	"sync"
+	"sync/atomic"
+
+	log "github.com/sirupsen/logrus"
+)
+
+var (
+	ErrorNoTransferWithID = errors.New("no transfer with id")
+)
+
+type transferStatus int
+
+const (
+	STATUS_QUEUED transferStatus = iota
+	STATUS_DOWNLOADING
+	STATUS_PAUSED
+	STATUS_COMPLETED
+	STATUS_CANCELED
+	STATUS_ERROR
+)
+
+type Transfer struct {
+	id               int64
+	totalSize        atomic.Int64
+	downloaded       atomic.Int64
+	savePath         string
+	url              string
+	urlLock          sync.Mutex
+	status           transferStatus
+	statusLock       sync.Mutex
+	errorStrings     []string
+	errorStringsLock sync.Mutex
+	tempFileName     string
+	Finished         chan bool
+}
+
+func NewTransfer(id int64, url string, savePath string) Transfer {
+	return Transfer{
+		id:               id,
+		totalSize:        atomic.Int64{},
+		downloaded:       atomic.Int64{},
+		savePath:         savePath,
+		url:              url,
+		urlLock:          sync.Mutex{},
+		status:           STATUS_QUEUED,
+		statusLock:       sync.Mutex{},
+		errorStrings:     make([]string, 0),
+		errorStringsLock: sync.Mutex{},
+		tempFileName:     "",
+	}
+}
+
+func (t *Transfer) SetID(id int64) {
+	atomic.StoreInt64(&t.id, id)
+}
+
+func (t *Transfer) GetID() int64 {
+	return atomic.LoadInt64(&t.id)
+}
+
+func (t *Transfer) SetTotalSize(size int64) {
+	t.totalSize.Store(size)
+}
+
+func (t *Transfer) GetTotalSize() int64 {
+	return t.totalSize.Load()
+}
+
+func (t *Transfer) SetDownloaded(size int64) {
+	t.downloaded.Store(size)
+}
+
+func (t *Transfer) GetDownloaded() int64 {
+	return t.downloaded.Load()
+}
+
+func (t *Transfer) SetURL(url string) {
+	t.urlLock.Lock()
+	t.url = url
+	t.urlLock.Unlock()
+}
+
+func (t *Transfer) GetURL() string {
+	t.urlLock.Lock()
+	defer t.urlLock.Unlock()
+	return t.url
+}
+
+func (t *Transfer) SetStatus(status transferStatus) {
+	t.statusLock.Lock()
+	t.status = status
+	t.statusLock.Unlock()
+}
+
+func (t *Transfer) GetStatus() transferStatus {
+	t.statusLock.Lock()
+	defer t.statusLock.Unlock()
+	return t.status
+}
+
+func (t *Transfer) AddErrorString(str string) {
+	t.errorStringsLock.Lock()
+	t.errorStrings = append(t.errorStrings, str)
+	t.errorStringsLock.Unlock()
+}
+
+func (t *Transfer) GetErrorStrings() []string {
+	t.errorStringsLock.Lock()
+	defer t.errorStringsLock.Unlock()
+	return t.errorStrings
+}
+
+func (t *Transfer) GetTempFilePath() string {
+	if t.tempFileName == "" {
+		url, err := url.Parse(t.GetURL())
+		if err != nil {
+			t.tempFileName = fmt.Sprintf("download-%d", t.GetID())
+		} else {
+			finalPath := strings.Split(url.Path, "/")[len(strings.Split(url.Path, "/"))-1]
+			t.tempFileName = fmt.Sprintf("download-%d-%s", t.GetID(), finalPath)
+		}
+	}
+
+	return t.tempFileName
+}
+
+func (t *Transfer) Write(p []byte) (int, error) {
+	if t.GetStatus() == STATUS_CANCELED || t.GetStatus() == STATUS_PAUSED {
+		return 0, io.EOF
+	}
+	t.SetDownloaded(t.GetDownloaded() + int64(len(p)))
+	return len(p), nil
+}
+
+func (t *Transfer) Pause() error {
+	t.SetStatus(STATUS_PAUSED)
+	return nil
+}
+
+func (t *Transfer) Cancel() error {
+	t.SetStatus(STATUS_CANCELED)
+	t.Finished <- true
+	return nil
+}
+
+func (t *Transfer) Resume() error {
+	return t.Download()
+}
+
+func (t *Transfer) Download() error {
+	client := &http.Client{}
+
+	//Built http get request with a content range header
+	req, err := http.NewRequest("GET", t.GetURL(), nil)
+	if err != nil {
+		return err
+	}
+	if t.GetDownloaded() > 0 {
+		req.Header.Set("Range", fmt.Sprintf("bytes=%d-", t.GetDownloaded()))
+	}
+	resp, err := client.Do(req)
+	if err != nil {
+		return err
+	}
+	defer resp.Body.Close()
+
+	t.SetTotalSize(resp.ContentLength)
+	var out *os.File
+
+	if t.GetDownloaded() > 0 {
+		out, err = os.Open(t.GetTempFilePath())
+	} else {
+		out, err = os.Create(t.GetTempFilePath())
+	}
+
+	if err != nil {
+		return err
+	}
+
+	t.SetStatus(STATUS_DOWNLOADING)
+	go func() {
+		defer out.Close()
+
+		if _, err := io.Copy(out, io.TeeReader(resp.Body, t)); err != nil {
+			t.AddErrorString(err.Error())
+			t.SetStatus(STATUS_ERROR)
+			log.Error(err)
+		}
+		t.SetStatus(STATUS_COMPLETED)
+		t.Finished <- true
+	}()
+
+	return nil
+}
+
+func Start() {
+
+}

pkg/downloadmanager/types.go

@@ -0,0 +1,25 @@
+package downloadmanager
+
+import (
+	"sync"
+	"sync/atomic"
+)
+
+// type DownloadManager interface {
+// 	GetTransfers() []Transfer
+
+// 	GetTransfer(id int64) (*Transfer, error)
+// 	AddTransfer(url string) (*Transfer, error)
+// 	RemoveTransfer(id int64) error
+// }
+
+type DownloadManager struct {
+	MaxSimultaneousDownloads int
+
+	transfers     []Transfer
+	transfersLock sync.Mutex
+
+	IdCounter atomic.Int64
+
+	CancelChannel chan bool
+}

pkg/premiumizeme/premiumizeme.go

@@ -135,27 +135,39 @@ func (pm *Premiumizeme) GetFolders() ([]Item, error) {
 	}
 
 	var ret []Item
-	req, _ := http.NewRequest("GET", url.String(), nil)
+	req, err := http.NewRequest("GET", url.String(), nil)
+	if err != nil {
+		return ret, err
+	}
 
 	resp, err := http.DefaultClient.Do(req)
 	if err != nil {
 		return ret, err
 	}
 
-	defer resp.Body.Close()
-	res := ListFoldersResponse{}
-	err = json.NewDecoder(resp.Body).Decode(&res)
+	if resp.StatusCode != 200 {
+		return ret, fmt.Errorf("error listing folder: %s (%d)", resp.Status, resp.StatusCode)
+	}
 
-	if res.Status != "success" {
-		return ret, fmt.Errorf("%s", res.Status)
+	defer resp.Body.Close()
+	list_folders_res := ListFoldersResponse{}
+	err = json.NewDecoder(resp.Body).Decode(&list_folders_res)
+	if err != nil {
+		return ret, err
+	}
+
+	if list_folders_res.Status != "success" {
+		fmt.Printf("%+v\n", resp)
+		fmt.Printf("%+v\n", list_folders_res)
+		return ret, fmt.Errorf(list_folders_res.Message)
 	}
 
 	if err != nil {
 		return ret, err
 	}
 
-	log.Tracef("Received %d Folders", len(res.Content))
-	return res.Content, nil
+	log.Tracef("Received %d Folders", len(list_folders_res.Content))
+	return list_folders_res.Content, nil
 }
 
 func (pm *Premiumizeme) CreateTransfer(filePath string, parentID string) error {

pkg/premiumizeme/types.go

@@ -49,7 +49,7 @@ type Item struct {
 }
 type FolderItems struct {
 	Status   string `json:"status"`
-	Contant  []Item `json:"content"`
+	Content  []Item `json:"content"`
 	Name     string `json:"name"`
 	ParentID string `json:"parent_id"`
 	FolderID string `json:"folder_id"`

web/src/components/APITable.svelte

@@ -2,6 +2,10 @@
   import { DataTable, InlineLoading } from "carbon-components-svelte";
   import { CalculateAPIPath } from "../Utilities/web_root";
 
+  export let sortable = true;
+  export let sortKey;
+  export let sortOrder;
+
   export let totalName = "";
   export let headers = {};
   export let updateTimeSeconds = 10;
@@ -10,7 +14,7 @@
     if (!data) return [];
     return data;
   };
-  
+
   let updating = false;
   let status = "";
   let rows = [];
@@ -56,6 +60,6 @@
     Message: {status}
   </p>
   <p>
-    <DataTable sortable {headers} {rows} />
+    <DataTable {sortKey} {sortOrder} {sortable} {headers} {rows} />
   </p>
 </main>

web/src/pages/Info.svelte

@@ -112,11 +112,15 @@
         <h3>Downloads</h3>
         <APITable
           headers={[
+            { key : "id", value : "ID" },
             { key: "added", value: "Added" },
             { key: "name", value: "Name" },
             { key: "progress", value: "Progress" },
             { key: "speed", value: "Speed" },
           ]}
+          sortable={false}
+          sortKey={"id"}
+          sortOrder={"desc"}
           updateTimeSeconds={2}
           APIpath="api/downloads"
           zebra={true}