feat: update download job

backend/app/jobs/download_from_alioss.go

@@ -2,10 +2,12 @@ package jobs
 
 import (
 	"context"
+	"errors"
 	"os"
 	"path/filepath"
 	"time"
 
+	"quyun/app/models"
 	"quyun/providers/ali"
 
 	. "github.com/riverqueue/river"
@@ -18,7 +20,7 @@ import (
 var _ contracts.JobArgs = (*WechatCallback)(nil)
 
 type DownloadFromAliOSS struct {
-	Path string
+	MediaID int64 `json:"media_id"`
 }
 
 func (s DownloadFromAliOSS) InsertOpts() InsertOpts {
@@ -50,13 +52,18 @@ func (w *DownloadFromAliOSSWorker) Work(ctx context.Context, job *Job[DownloadFr
 	log.Infof("[Start] Working on job with strings: %+v", job.Args)
 	defer log.Infof("[End] Finished %s", job.Args.Kind())
 
-	dst := filepath.Join("/Users/rogee/Projects/self/quyun/backend/fixtures/oss/", job.Args.Path)
+	media, err := models.Medias.GetByID(ctx, job.Args.MediaID)
+	if err != nil {
+		log.Errorf("Error getting media by ID: %v", err)
+		return JobCancel(err)
+	}
+
+	dst := filepath.Join("/Users/rogee/Projects/self/quyun/backend/fixtures/oss/", media.Path)
 
 	// check is path exist
-	_, err := os.Stat(dst)
+	st, err := os.Stat(dst)
 	if os.IsNotExist(err) {
 		log.Infof("File not exists: %s", dst)
-		// midir
 		err := os.MkdirAll(filepath.Dir(dst), os.ModePerm)
 		if err != nil {
 			log.Errorf("Error creating directory: %v", err)
@@ -64,23 +71,23 @@ func (w *DownloadFromAliOSSWorker) Work(ctx context.Context, job *Job[DownloadFr
 		}
 	}
 
-	if err == nil {
-		log.Infof("File already exists: %s", dst)
-		err := os.Remove(dst)
-		if err != nil {
+	if st.Size() == media.Size {
+		return JobCancel(errors.New("file already downloaded"))
+	} else {
+		// remove file
+		if err := os.Remove(dst); err != nil {
 			log.Errorf("Error removing file: %v", err)
 			return err
 		}
-
-		return nil
 	}
 
-	if err := w.oss.Download(ctx, job.Args.Path, dst); err != nil {
+	log.Infof("Starting download for file: %s", media.Path)
+	if err := w.oss.Download(ctx, media.Path, dst); err != nil {
 		log.Errorf("Error downloading file: %v", err)
 		return err
 	}
 
-	log.Infof("Successfully downloaded file: %s", job.Args.Path)
+	log.Infof("Successfully downloaded file: %s", media.Path)
 
 	return nil
 }

backend/app/jobs/download_from_alioss_test.go

@@ -4,6 +4,7 @@ import (
 	"context"
 	"testing"
 
+	"quyun/app/models"
 	"quyun/app/service/testx"
 	"quyun/providers/ali"
 	"quyun/providers/job"
@@ -31,7 +32,7 @@ type DownloadFromAliOSSSuite struct {
 }
 
 func Test_DownloadFromAliOSS(t *testing.T) {
-	providers := testx.Default().With(Provide)
+	providers := testx.Default().With(Provide, models.Provide)
 
 	testx.Serve(providers, t, func(p DownloadFromAliOSSSuiteInjectParams) {
 		suite.Run(t, &DownloadFromAliOSSSuite{DownloadFromAliOSSSuiteInjectParams: p})
@@ -43,7 +44,7 @@ func (t *DownloadFromAliOSSSuite) Test_Work() {
 		Convey("step 1", func() {
 			job := &Job[DownloadFromAliOSS]{
 				Args: DownloadFromAliOSS{
-					Path: "quyun/959e5310105c96e653f10b74e5bdc36b.mp4",
+					MediaID: 3,
 				},
 			}
 

backend/app/models/medias.go

@@ -231,3 +231,21 @@ func (m *mediasModel) Update(ctx context.Context, hash string, model *model.Medi
 	m.log.Infof("media item updated successfully")
 	return nil
 }
+
+// GetByID
+func (m *mediasModel) GetByID(ctx context.Context, id int64) (*model.Medias, error) {
+	tbl := table.Medias
+	stmt := tbl.
+		SELECT(tbl.AllColumns).
+		WHERE(tbl.ID.EQ(Int64(id)))
+	m.log.Infof("sql: %s", stmt.DebugSql())
+
+	var media model.Medias
+	err := stmt.QueryContext(ctx, db, &media)
+	if err != nil {
+		m.log.Errorf("error querying media item by ID: %v", err)
+		return nil, err
+	}
+
+	return &media, nil
+}