10 playlist download (#71)

* leveraging message queue for playlist entries DL

* playlist support implemented

It's a little bit slow but solid enough :D

server/internal/common.go

@@ -12,15 +12,16 @@ type DownloadProgress struct {
 
 // Used to deser the yt-dlp -J output
 type DownloadInfo struct {
-	URL        string    `json:"url"`
-	Title      string    `json:"title"`
-	Thumbnail  string    `json:"thumbnail"`
-	Resolution string    `json:"resolution"`
-	Size       int32     `json:"filesize_approx"`
-	VCodec     string    `json:"vcodec"`
-	ACodec     string    `json:"acodec"`
-	Extension  string    `json:"ext"`
-	CreatedAt  time.Time `json:"created_at"`
+	URL         string    `json:"url"`
+	Title       string    `json:"title"`
+	Thumbnail   string    `json:"thumbnail"`
+	Resolution  string    `json:"resolution"`
+	Size        int32     `json:"filesize_approx"`
+	VCodec      string    `json:"vcodec"`
+	ACodec      string    `json:"acodec"`
+	Extension   string    `json:"ext"`
+	OriginalURL string    `json:"original_url"`
+	CreatedAt   time.Time `json:"created_at"`
 }
 
 // Used to deser the formats in the -J output
@@ -64,11 +65,9 @@ type AbortRequest struct {
 
 // struct representing the intent to start a download
 type DownloadRequest struct {
-	Url      string   `json:"url"`
-	Params   []string `json:"params"`
-	RenameTo string   `json:"renameTo"`
-	Id       string
-	URL      string
-	Path     string
-	Rename   string
+	Id     string
+	URL    string
+	Path   string
+	Rename string
+	Params []string
 }

server/internal/memory_db.go

@@ -30,6 +30,7 @@ func (m *MemoryDB) Get(id string) (*Process, error) {
 func (m *MemoryDB) Set(process *Process) string {
 	id := uuid.Must(uuid.NewRandom()).String()
 	m.table.Store(id, process)
+	process.Id = id
 	return id
 }
 
@@ -129,7 +130,6 @@ func (m *MemoryDB) Restore() {
 			Url:      proc.Info.URL,
 			Info:     proc.Info,
 			Progress: proc.Progress,
-			DB:       m,
 		})
 	}
 

server/internal/message_queue.go

@@ -30,7 +30,15 @@ func NewMessageQueue() *MessageQueue {
 
 // Publish a message to the queue and set the task to a peding state.
 func (m *MessageQueue) Publish(p *Process) {
-	go p.SetPending()
+	p.SetPending()
+	go p.SetMetadata()
+	m.producerCh <- p
+}
+
+// Publish a message to the queue and set the task to a peding state.
+// ENSURE P IS PART OF A PLAYLIST
+// Needs a further review
+func (m *MessageQueue) PublishPlaylistEntry(p *Process) {
 	m.producerCh <- p
 }
 
@@ -45,3 +53,13 @@ func (m *MessageQueue) Subscriber() {
 		}(msg)
 	}
 }
+
+// Empties the message queue
+func (m *MessageQueue) Empty() {
+	for range m.producerCh {
+		<-m.producerCh
+	}
+	for range m.consumerCh {
+		<-m.consumerCh
+	}
+}

server/internal/playlist.go

@@ -0,0 +1,81 @@
+package internal
+
+import (
+	"errors"
+	"log"
+	"os/exec"
+	"time"
+
+	"github.com/goccy/go-json"
+	"github.com/marcopeocchi/yt-dlp-web-ui/server/cli"
+)
+
+type metadata struct {
+	Entries []DownloadInfo `json:"entries"`
+	Count   int            `json:"playlist_count"`
+	Type    string         `json:"_type"`
+}
+
+func PlaylistDetect(req DownloadRequest, mq *MessageQueue, db *MemoryDB) error {
+	cmd := exec.Command(cfg.GetConfig().DownloaderPath, req.URL, "-J")
+
+	stdout, err := cmd.StdoutPipe()
+	if err != nil {
+		return err
+	}
+
+	m := metadata{}
+
+	err = cmd.Start()
+	if err != nil {
+		return err
+	}
+
+	log.Println(cli.BgRed, "Decoding metadata", cli.Reset, req.URL)
+
+	err = json.NewDecoder(stdout).Decode(&m)
+	if err != nil {
+		return err
+	}
+
+	log.Println(cli.BgGreen, "Decoded metadata", cli.Reset, req.URL)
+
+	if m.Type == "" {
+		cmd.Wait()
+		return errors.New("probably not a valid URL")
+	}
+
+	if m.Type == "playlist" {
+		log.Println(
+			cli.BgGreen, "Playlist detected", cli.Reset, m.Count, "entries",
+		)
+
+		for _, meta := range m.Entries {
+			proc := &Process{
+				Url:      meta.OriginalURL,
+				Progress: DownloadProgress{},
+				Output:   DownloadOutput{},
+				Info:     meta,
+				Params:   req.Params,
+			}
+
+			proc.Info.URL = meta.OriginalURL
+			proc.Info.CreatedAt = time.Now().Add(time.Second)
+
+			db.Set(proc)
+			proc.SetPending()
+			mq.PublishPlaylistEntry(proc)
+		}
+
+		err = cmd.Wait()
+		return err
+	}
+
+	proc := &Process{Url: req.URL, Params: req.Params}
+
+	mq.Publish(proc)
+	log.Println("Sending new process to message queue", proc.Url)
+
+	err = cmd.Wait()
+	return err
+}

server/internal/process.go

@@ -4,6 +4,7 @@ import (
 	"bufio"
 	"fmt"
 	"regexp"
+	"sync"
 	"syscall"
 
 	"github.com/goccy/go-json"
@@ -50,7 +51,6 @@ type Process struct {
 	Params   []string
 	Info     DownloadInfo
 	Progress DownloadProgress
-	DB       *MemoryDB
 	Output   DownloadOutput
 	proc     *os.Process
 }
@@ -128,7 +128,7 @@ func (p *Process) Start() {
 
 		for scan.Scan() {
 			stdout := ProgressTemplate{}
-			err := json.Unmarshal([]byte(scan.Text()), &stdout)
+			err := json.Unmarshal(scan.Bytes(), &stdout)
 			if err == nil {
 				p.Progress = DownloadProgress{
 					Status:     StatusDownloading,
@@ -175,26 +175,49 @@ func (p *Process) Kill() error {
 		return err
 	}
 
-	p.DB.Delete(p.Id)
 	return nil
 }
 
 // Returns the available format for this URL
 func (p *Process) GetFormatsSync() (DownloadFormats, error) {
 	cmd := exec.Command(cfg.GetConfig().DownloaderPath, p.Url, "-J")
-	stdout, err := cmd.Output()
+	stdout, err := cmd.StdoutPipe()
 
 	if err != nil {
 		return DownloadFormats{}, err
 	}
 
-	cmd.Wait()
-
 	info := DownloadFormats{URL: p.Url}
 	best := Format{}
 
-	json.Unmarshal(stdout, &info)
-	json.Unmarshal(stdout, &best)
+	var (
+		wg            sync.WaitGroup
+		decodingError error
+	)
+
+	wg.Add(2)
+
+	err = cmd.Start()
+	if err != nil {
+		return DownloadFormats{}, err
+	}
+
+	go func() {
+		decodingError = json.NewDecoder(stdout).Decode(&info)
+		wg.Done()
+	}()
+
+	go func() {
+		decodingError = json.NewDecoder(stdout).Decode(&best)
+		wg.Done()
+	}()
+
+	wg.Wait()
+	cmd.Wait()
+
+	if decodingError != nil {
+		return DownloadFormats{}, err
+	}
 
 	info.Best = best
 
@@ -202,14 +225,17 @@ func (p *Process) GetFormatsSync() (DownloadFormats, error) {
 }
 
 func (p *Process) SetPending() {
-	p.Id = p.DB.Set(p)
+	p.Progress.Status = StatusPending
+}
 
+func (p *Process) SetMetadata() error {
 	cmd := exec.Command(cfg.GetConfig().DownloaderPath, p.Url, "-J")
 	cmd.SysProcAttr = &syscall.SysProcAttr{Setpgid: true}
 
-	stdout, err := cmd.Output()
+	stdout, err := cmd.StdoutPipe()
 	if err != nil {
 		log.Println("Cannot retrieve info for", p.Url)
+		return err
 	}
 
 	info := DownloadInfo{
@@ -217,8 +243,20 @@ func (p *Process) SetPending() {
 		CreatedAt: time.Now(),
 	}
 
-	json.Unmarshal(stdout, &info)
-	p.Info = info
+	err = cmd.Start()
+	if err != nil {
+		return err
+	}
 
+	err = json.NewDecoder(stdout).Decode(&info)
+	if err != nil {
+		return err
+	}
+
+	p.Info = info
 	p.Progress.Status = StatusPending
+
+	err = cmd.Wait()
+
+	return err
 }

server/rpc/handlers.go

@@ -7,7 +7,11 @@ import (
 	"github.com/gorilla/websocket"
 )
 
-var upgrader websocket.Upgrader
+var upgrader = websocket.Upgrader{
+	CheckOrigin: func(r *http.Request) bool {
+		return true
+	},
+}
 
 func WebSocket(w http.ResponseWriter, r *http.Request) {
 	c, err := upgrader.Upgrade(w, r, nil)

server/rpc/service.go

@@ -24,14 +24,6 @@ type Args struct {
 	Params []string
 }
 
-type DownloadSpecificArgs struct {
-	Id     string
-	URL    string
-	Path   string
-	Rename string
-	Params []string
-}
-
 // Dependency injection container.
 func Container(db *internal.MemoryDB, mq *internal.MessageQueue) *Service {
 	return &Service{
@@ -42,11 +34,8 @@ func Container(db *internal.MemoryDB, mq *internal.MessageQueue) *Service {
 
 // Exec spawns a Process.
 // The result of the execution is the newly spawned process Id.
-func (s *Service) Exec(args DownloadSpecificArgs, result *string) error {
-	log.Println("Sending new process to message queue", args.URL)
-
+func (s *Service) Exec(args internal.DownloadRequest, result *string) error {
 	p := &internal.Process{
-		DB:     s.db,
 		Url:    args.URL,
 		Params: args.Params,
 		Output: internal.DownloadOutput{
@@ -55,8 +44,22 @@ func (s *Service) Exec(args DownloadSpecificArgs, result *string) error {
 		},
 	}
 
+	s.db.Set(p)
 	s.mq.Publish(p)
+
 	*result = p.Id
+	return nil
+}
+
+// Exec spawns a Process.
+// The result of the execution is the newly spawned process Id.
+func (s *Service) ExecPlaylist(args internal.DownloadRequest, result *string) error {
+	err := internal.PlaylistDetect(args, s.mq, s.db)
+	if err != nil {
+		return err
+	}
+
+	*result = ""
 
 	return nil
 }
@@ -71,7 +74,7 @@ func (s *Service) Progess(args Args, progress *internal.DownloadProgress) error
 	return nil
 }
 
-// Progess retrieves the Progress of a specific Process given its Id
+// Progess retrieves available format for a given resource
 func (s *Service) Formats(args Args, progress *internal.DownloadFormats) error {
 	var err error
 	p := internal.Process{Url: args.URL}
@@ -101,6 +104,7 @@ func (s *Service) Kill(args string, killed *string) error {
 	}
 	if proc != nil {
 		err = proc.Kill()
+		s.db.Delete(proc.Id)
 	}
 
 	s.db.Delete(proc.Id)
@@ -120,8 +124,10 @@ func (s *Service) KillAll(args NoArgs, killed *string) error {
 		}
 		if proc != nil {
 			proc.Kill()
+			s.db.Delete(proc.Id)
 		}
 	}
+	s.mq.Empty()
 	return err
 }
 
@@ -142,7 +148,9 @@ func (s *Service) FreeSpace(args NoArgs, free *uint64) error {
 // Return a flattned tree of the download directory
 func (s *Service) DirectoryTree(args NoArgs, tree *[]string) error {
 	dfsTree, err := sys.DirectoryTree()
-	*tree = *dfsTree
+	if dfsTree != nil {
+		*tree = *dfsTree
+	}
 	return err
 }
 

server/server.go

@@ -14,6 +14,7 @@ import (
 
 	"github.com/go-chi/chi/v5"
 	"github.com/go-chi/chi/v5/middleware"
+	"github.com/go-chi/cors"
 	"github.com/marcopeocchi/yt-dlp-web-ui/server/internal"
 	middlewares "github.com/marcopeocchi/yt-dlp-web-ui/server/middleware"
 	"github.com/marcopeocchi/yt-dlp-web-ui/server/rest"
@@ -54,7 +55,7 @@ func newServer(c serverConfig) *http.Server {
 
 	r := chi.NewRouter()
 
-	r.Use(middlewares.CORS)
+	r.Use(cors.AllowAll().Handler)
 	r.Use(middleware.Logger)
 
 	sh := middlewares.NewSpaHandler("index.html", c.frontend)