diff --git a/api/cmd/api/main.go b/api/cmd/api/main.go index a707e9b..5a30280 100644 --- a/api/cmd/api/main.go +++ b/api/cmd/api/main.go @@ -11,4 +11,5 @@ func main() { router. NewRouter(). Run() + } diff --git a/api/internal/app/router/router.go b/api/internal/app/router/router.go index 7679990..28a855f 100644 --- a/api/internal/app/router/router.go +++ b/api/internal/app/router/router.go @@ -10,6 +10,7 @@ import ( "downloader/internal/core/ports/fileorchestrator" "downloader/internal/core/ports/fileorchestrator/destinationhandler" "downloader/internal/core/ports/fileorchestrator/sourcehandler" + "downloader/internal/core/services/cleanup" "downloader/internal/core/services/download/default" "downloader/internal/core/services/download/handlers" "downloader/internal/core/services/downloader" @@ -78,4 +79,7 @@ func setupDownloadRoute(router *router) { downloadApi := download.New(drService) downloadApi.SetupDownloadApi(router.internalRouter) + + cleanupJob := cleanup.New(drRepository, newLogger) + cleanupJob.RunOnSchedule() } diff --git a/api/internal/core/ports/download_request/repository.go b/api/internal/core/ports/download_request/repository.go index 7a53fec..c2293e9 100644 --- a/api/internal/core/ports/download_request/repository.go +++ b/api/internal/core/ports/download_request/repository.go @@ -10,4 +10,6 @@ type Repository interface { GetById(ctx context.Context, id string) (*entities.Download, error) Update(ctx context.Context, download *entities.Download) error Get(ctx context.Context, active bool) ([]*entities.Download, error) + GetOldOrStuck(ctx context.Context) ([]*entities.Download, error) + BatchDelete(ctx context.Context, requests []*entities.Download) error } diff --git a/api/internal/core/ports/download_request/sql/repository.go b/api/internal/core/ports/download_request/sql/repository.go index be40eb1..2f1e4fd 100644 --- a/api/internal/core/ports/download_request/sql/repository.go +++ b/api/internal/core/ports/download_request/sql/repository.go @@ -3,6 +3,7 @@ package sql import ( "context" "downloader/internal/core/entities" + "downloader/internal/core/ports/download_request" "github.com/uptrace/bun" "go.uber.org/zap" ) @@ -12,7 +13,7 @@ type repository struct { logger *zap.SugaredLogger } -func NewDownloadRequestSqlRepository(db *bun.DB, logger *zap.SugaredLogger) *repository { +func NewDownloadRequestSqlRepository(db *bun.DB, logger *zap.SugaredLogger) download_request.Repository { return &repository{ db: db, logger: logger, @@ -82,6 +83,7 @@ func (r repository) Get(ctx context.Context, active bool) ([]*entities.Download, err := r.db.NewSelect(). Model(&downloads). Column("id", "status", "link"). + Where("status LIKE ?", "in-progress%"). Limit(20). Order("created_at ASC"). Scan(ctx) @@ -100,3 +102,53 @@ func (r repository) Get(ctx context.Context, active bool) ([]*entities.Download, return responseDownloads, nil } + +func (r repository) GetOldOrStuck(ctx context.Context) ([]*entities.Download, error) { + + var downloads []Download + err := r.db.NewSelect(). + Model(&downloads). + Column("id", "status", "link"). + Where("status LIKE ?", "in-progress%"). + Where("updated_at < now() - interval '1 minutes'"). + Limit(20). + 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, + }) + } + if len(responseDownloads) == 0 { + return []*entities.Download{}, nil + } + + return responseDownloads, nil +} + +func (r repository) BatchDelete(ctx context.Context, requests []*entities.Download) error { + var downloads []*Download + for _, request := range requests { + downloads = append(downloads, &Download{ID: request.ID}) + } + if len(downloads) == 0 { + return nil + } + + _, err := r.db.NewDelete(). + Model(&downloads). + WherePK(). + Exec(ctx) + if err != nil { + return err + } + + return nil +} diff --git a/api/internal/core/services/cleanup/cleanup.go b/api/internal/core/services/cleanup/cleanup.go new file mode 100644 index 0000000..03bd007 --- /dev/null +++ b/api/internal/core/services/cleanup/cleanup.go @@ -0,0 +1,35 @@ +package cleanup + +import ( + "context" + "downloader/internal/core/ports/download_request" + "go.uber.org/zap" + "time" +) + +type cleanUp struct { + repository download_request.Repository + logger *zap.SugaredLogger +} + +func New(repository download_request.Repository, logger *zap.SugaredLogger) *cleanUp { + return &cleanUp{repository: repository, logger: logger} +} + +func (c *cleanUp) RunOnSchedule() { + ctx := context.TODO() + 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 { + c.logger.Warn("could not process old or stuck in-progress jobs") + } + + time.Sleep(time.Minute) + } + }() +}