Add base site, needs clean-up

api/internal/core/ports/download_request/repository.go

@@ -12,4 +12,5 @@ type Repository interface {
 	Get(ctx context.Context, active bool) ([]*entities.Download, error)
 	GetOldOrStuck(ctx context.Context) ([]*entities.Download, error)
 	BatchDelete(ctx context.Context, requests []*entities.Download) error
+	GetDone(ctx context.Context) ([]*entities.Download, error)
 }

api/internal/core/ports/download_request/sql/repository.go

@@ -80,13 +80,22 @@ func (r repository) Update(ctx context.Context, download *entities.Download) err
 
 func (r repository) Get(ctx context.Context, active bool) ([]*entities.Download, error) {
 	var downloads []Download
-	err := r.db.NewSelect().
+	query := r.db.NewSelect().
 		Model(&downloads).
-		Column("id", "status", "link").
-		Where("status LIKE ?", "in-progress%").
-		Limit(20).
-		Order("created_at ASC").
-		Scan(ctx)
+		Column("id", "status", "link")
+
+	var err error
+	if active {
+		err = query.
+			Where("status LIKE ? OR status = ?", "in-progress%", "scheduled").
+			Order("created_at ASC").
+			Scan(ctx)
+	} else {
+		err = query.
+			Order("created_at ASC").
+			Scan(ctx)
+	}
+
 	if err != nil {
 		return nil, err
 	}
@@ -109,7 +118,7 @@ func (r repository) GetOldOrStuck(ctx context.Context) ([]*entities.Download, er
 	err := r.db.NewSelect().
 		Model(&downloads).
 		Column("id", "status", "link").
-		Where("status LIKE ?", "in-progress%").
+		Where("status LIKE ? OR status = ?", "in-progress%", "scheduled").
 		Where("updated_at < now() - interval '1 minutes'").
 		Limit(20).
 		Order("created_at ASC").
@@ -152,3 +161,28 @@ func (r repository) BatchDelete(ctx context.Context, requests []*entities.Downlo
 
 	return nil
 }
+
+func (r repository) GetDone(ctx context.Context) ([]*entities.Download, error) {
+	var downloads []Download
+	err := r.db.NewSelect().
+		Model(&downloads).
+		Column("id", "status", "link").
+		Where("status = ?", "done").
+		Order("created_at ASC").
+		Scan(ctx)
+
+	if err != nil {
+		return nil, err
+	}
+
+	var responseDownloads []*entities.Download
+	for _, download := range downloads {
+		responseDownloads = append(responseDownloads, &entities.Download{
+			ID:     download.ID,
+			Status: download.Status,
+			Link:   download.Link,
+		})
+	}
+
+	return responseDownloads, nil
+}

api/internal/core/ports/fileorchestrator/sourcehandler/sourcehandler.go

@@ -35,5 +35,8 @@ func (l localSourceHandler) Prepare(link string) (string, error) {
 }
 
 func (l localSourceHandler) CleanUp(sourcePath string) error {
-	return os.Remove(sourcePath)
+	if _, err := os.Stat(sourcePath); os.IsNotExist(err) {
+		return nil
+	}
+	return os.RemoveAll(sourcePath)
 }

api/internal/core/services/cleanup/cleanup.go

@@ -3,6 +3,7 @@ package cleanup
 import (
 	"context"
 	"downloader/internal/core/ports/download_request"
+	"downloader/internal/core/ports/fileorchestrator"
 	"go.uber.org/zap"
 	"time"
 )
@@ -10,10 +11,11 @@ import (
 type cleanUp struct {
 	repository download_request.Repository
 	logger     *zap.SugaredLogger
+	fo         *fileorchestrator.FileOrchestrator
 }
 
-func New(repository download_request.Repository, logger *zap.SugaredLogger) *cleanUp {
-	return &cleanUp{repository: repository, logger: logger}
+func New(repository download_request.Repository, logger *zap.SugaredLogger, fo *fileorchestrator.FileOrchestrator) *cleanUp {
+	return &cleanUp{repository: repository, logger: logger, fo: fo}
 }
 
 func (c *cleanUp) RunOnSchedule() {
@@ -21,14 +23,27 @@ func (c *cleanUp) RunOnSchedule() {
 	go func() {
 		for true {
 			requests, err := c.repository.GetOldOrStuck(ctx)
-			if err == nil {
-				c.logger.Debugw("Cleaning up downloads",
-					"downloads", requests)
-				_ = c.repository.BatchDelete(ctx, requests)
-			} else {
+			if err != nil {
 				c.logger.Warn("could not process old or stuck in-progress jobs")
+				time.Sleep(5 * time.Minute)
+				continue
 			}
 
+			c.logger.Debugw("Cleaning up downloads",
+				"downloads", requests)
+
+			for _, request := range requests {
+				basePath, err := c.fo.Begin(request.Link)
+				if err != nil {
+					c.logger.Warnw("could not process request",
+						"downloadId", request.ID)
+					continue
+				}
+
+				c.fo.CleanUp(basePath)
+			}
+			_ = c.repository.BatchDelete(ctx, requests)
+
 			time.Sleep(time.Minute)
 		}
 	}()

api/internal/core/services/download/default/service.go

@@ -56,3 +56,7 @@ func (l *localService) Get(ctx context.Context, id string) (*entities.Download,
 func (l *localService) GetAll(ctx context.Context, active bool) ([]*entities.Download, error) {
 	return l.repository.Get(ctx, active)
 }
+
+func (l *localService) GetDone(ctx context.Context) ([]*entities.Download, error) {
+	return l.repository.GetDone(ctx)
+}

api/internal/core/services/download/service.go

@@ -9,4 +9,5 @@ type Service interface {
 	Schedule(ctx context.Context, link string) (*entities.Download, error)
 	Get(ctx context.Context, id string) (*entities.Download, error)
 	GetAll(ctx context.Context, active bool) ([]*entities.Download, error)
+	GetDone(ctx context.Context) ([]*entities.Download, error)
 }