From 4f47b5f0901342f2123400c3f4fa849745ccac51 Mon Sep 17 00:00:00 2001 From: Rogee Date: Fri, 17 Jan 2025 19:26:32 +0800 Subject: [PATCH] fix: issues --- backend/app/http/posts/service.go | 120 +++++++++++++++++++ backend/app/jobs/post_delete_assets.go | 42 ------- backend/app/jobs/post_video_cut.go | 26 +++- backend/app/jobs/post_video_extract_audio.go | 24 +++- qvyun.code-workspace | 3 +- 5 files changed, 161 insertions(+), 54 deletions(-) diff --git a/backend/app/http/posts/service.go b/backend/app/http/posts/service.go index b09944d..0dca7cd 100644 --- a/backend/app/http/posts/service.go +++ b/backend/app/http/posts/service.go @@ -245,6 +245,7 @@ func (svc *Service) Create(ctx context.Context, tenant *model.Tenants, user *mod span.SetAttributes(semconv.DBStatementKey.String(stmt.DebugSql())) if _, err := stmt.ExecContext(ctx, svc.db); err != nil { + span.RecordError(err) return err } @@ -268,6 +269,35 @@ func (svc *Service) GetPostByHash(ctx context.Context, tenantID int64, hash stri return svc.GetPostByID(ctx, postId) } +// ForceDelete +func (svc *Service) ForceDelete(ctx context.Context, tenantID, userID, postID int64) error { + _, span := otel.Start(ctx, "users.service.ForceDelete") + defer span.End() + span.SetAttributes( + attribute.Int64("tenant.id", tenantID), + attribute.Int64("user.id", userID), + attribute.Int64("post.id", postID), + ) + tbl := table.Posts + + stmt := tbl. + DELETE(). + WHERE( + tbl.ID.EQ(Int64(postID)).AND( + tbl.TenantID.EQ(Int64(tenantID)).AND( + tbl.UserID.EQ(Int64(userID)), + ), + ), + ) + span.SetAttributes(semconv.DBStatementKey.String(stmt.DebugSql())) + + if _, err := stmt.ExecContext(ctx, svc.db); err != nil { + span.RecordError(err) + return err + } + return nil +} + // Delete func (svc *Service) Delete(ctx context.Context, tenantID, userID, postID int64) error { _, span := otel.Start(ctx, "users.service.Delete") @@ -294,6 +324,7 @@ func (svc *Service) Delete(ctx context.Context, tenantID, userID, postID int64) span.SetAttributes(semconv.DBStatementKey.String(stmt.DebugSql())) if _, err := stmt.ExecContext(ctx, svc.db); err != nil { + span.RecordError(err) return err } @@ -330,6 +361,7 @@ func (svc *Service) Update(ctx context.Context, tenantID, userID, postID int64, span.SetAttributes(semconv.DBStatementKey.String(stmt.DebugSql())) if _, err := stmt.ExecContext(ctx, svc.db); err != nil { + span.RecordError(err) return err } @@ -348,6 +380,7 @@ func (svc *Service) AttachAssets(ctx context.Context, tenantID, userID, postID i post, err := svc.ForceGetPostByID(ctx, postID) if err != nil { + span.RecordError(err) return err } @@ -371,6 +404,7 @@ func (svc *Service) AttachAssets(ctx context.Context, tenantID, userID, postID i span.SetAttributes(semconv.DBStatementKey.String(stmt.DebugSql())) if _, err := stmt.ExecContext(ctx, svc.db); err != nil { + span.RecordError(err) return err } @@ -389,6 +423,7 @@ func (svc *Service) UpdateMeta(ctx context.Context, tenantID, userID, postID int post, err := svc.ForceGetPostByID(ctx, postID) if err != nil { + span.RecordError(err) return err } @@ -411,6 +446,91 @@ func (svc *Service) UpdateMeta(ctx context.Context, tenantID, userID, postID int span.SetAttributes(semconv.DBStatementKey.String(stmt.DebugSql())) if _, err := stmt.ExecContext(ctx, svc.db); err != nil { + span.RecordError(err) + return err + } + + return svc.Update(ctx, tenantID, userID, postID, post) +} + +// UpdateStatus +func (svc *Service) UpdateStatus(ctx context.Context, tenantID, userID, postID int64, status fields.PostStatus) error { + _, span := otel.Start(ctx, "users.service.UpdateStatus") + defer span.End() + span.SetAttributes( + attribute.Int64("tenant.id", tenantID), + attribute.Int64("user.id", userID), + attribute.Int64("post.id", postID), + ) + + post, err := svc.ForceGetPostByID(ctx, postID) + if err != nil { + span.RecordError(err) + return err + } + + post.Status = status + + tbl := table.Posts + stmt := tbl. + UPDATE(tbl.UpdatedAt, tbl.Status). + SET( + tbl.UpdatedAt.SET(TimestampT(time.Now())), + tbl.Status.SET(Int16(int16(status))), + ). + WHERE( + tbl.ID.EQ(Int64(postID)).AND( + tbl.TenantID.EQ(Int64(tenantID)).AND( + tbl.UserID.EQ(Int64(userID)), + ), + ), + ) + span.SetAttributes(semconv.DBStatementKey.String(stmt.DebugSql())) + + if _, err := stmt.ExecContext(ctx, svc.db); err != nil { + span.RecordError(err) + return err + } + + return svc.Update(ctx, tenantID, userID, postID, post) +} + +// UpdateStage +func (svc *Service) UpdateStage(ctx context.Context, tenantID, userID, postID int64, stage fields.PostStage) error { + _, span := otel.Start(ctx, "users.service.UpdateStage") + defer span.End() + span.SetAttributes( + attribute.Int64("tenant.id", tenantID), + attribute.Int64("user.id", userID), + attribute.Int64("post.id", postID), + ) + + post, err := svc.ForceGetPostByID(ctx, postID) + if err != nil { + span.RecordError(err) + return err + } + + post.Stage = stage + + tbl := table.Posts + stmt := tbl. + UPDATE(tbl.UpdatedAt, tbl.Stage). + SET( + tbl.UpdatedAt.SET(TimestampT(time.Now())), + tbl.Stage.SET(Int16(int16(stage))), + ). + WHERE( + tbl.ID.EQ(Int64(postID)).AND( + tbl.TenantID.EQ(Int64(tenantID)).AND( + tbl.UserID.EQ(Int64(userID)), + ), + ), + ) + span.SetAttributes(semconv.DBStatementKey.String(stmt.DebugSql())) + + if _, err := stmt.ExecContext(ctx, svc.db); err != nil { + span.RecordError(err) return err } diff --git a/backend/app/jobs/post_delete_assets.go b/backend/app/jobs/post_delete_assets.go index e715006..94ddce0 100644 --- a/backend/app/jobs/post_delete_assets.go +++ b/backend/app/jobs/post_delete_assets.go @@ -7,14 +7,10 @@ import ( "backend/app/http/medias" "backend/app/http/posts" "backend/app/http/storages" - "backend/database/fields" - "backend/database/models/qvyun_v2/public/model" _ "git.ipao.vip/rogeecn/atom" _ "git.ipao.vip/rogeecn/atom/contracts" - "github.com/pkg/errors" . "github.com/riverqueue/river" - "github.com/samber/lo" "github.com/sirupsen/logrus" ) @@ -65,43 +61,5 @@ func (w *PostDeleteAssetsJobWorker) NextRetry(job *Job[PostDeleteAssetsJob]) tim } func (w *PostDeleteAssetsJobWorker) Work(ctx context.Context, job *Job[PostDeleteAssetsJob]) error { - post, err := w.postSvc.ForceGetPostByID(ctx, job.Args.PostID) - if err != nil { - return errors.Wrapf(err, "failed to get post(%d) by id", job.Args.PostID) - } - - mediaIDs := lo.Map(post.Assets.Data, func(asset fields.MediaAsset, _ int) int64 { - return asset.Media - }) - - medias, err := w.mediaSvc.GetMediasByIDs(ctx, post.TenantID, post.UserID, mediaIDs) - if err != nil { - return errors.Wrapf(err, "failed to get medias by ids(%v)", mediaIDs) - } - - storageIds := lo.Map(medias, func(media *model.Medias, _ int) int64 { return media.StorageID }) - storageMap, err := w.storageSvc.GetFSMapByIDs(ctx, storageIds...) - if err != nil { - return err - } - - // remove assets - for _, media := range medias { - st, ok := storageMap[media.StorageID] - if !ok { - continue - } - - if err := st.Remove(media.Path); err != nil { - return errors.Wrapf(err, "failed to remove media(%d)", media.ID) - } - } - - // delete all assets - ids := lo.Map(medias, func(media *model.Medias, _ int) int64 { return media.ID }) - if err := w.mediaSvc.DeleteByID(ctx, ids...); err != nil { - return errors.Wrapf(err, "failed to delete media(%d)", ids) - } - return nil } diff --git a/backend/app/jobs/post_video_cut.go b/backend/app/jobs/post_video_cut.go index c698be7..f5767af 100644 --- a/backend/app/jobs/post_video_cut.go +++ b/backend/app/jobs/post_video_cut.go @@ -88,9 +88,8 @@ func (w *PostVideoCutJobWorker) Work(ctx context.Context, job *Job[PostVideoCutJ videoPath := media.Path - // 获取全长度的音频 _, ok := lo.Find(post.Assets.Data, func(asset fields.MediaAsset) bool { - return asset.Type == fields.MediaAssetTypeAudio && asset.Mark != nil && *asset.Mark == "audio-preview" + return asset.Type == fields.MediaAssetTypeAudio && asset.Mark != nil && *asset.Mark == "video-preview" }) if ok { return nil @@ -142,9 +141,26 @@ func (w *PostVideoCutJobWorker) Work(ctx context.Context, job *Job[PostVideoCutJ return errors.Wrapf(err, "attach video(%s) to post(%d) failed", videoPath, post.ID) } - post.Meta.WorkerMark = post.Meta.WorkerMark & 1 << 0 - if err := w.postSvc.UpdateMeta(ctx, job.Args.TenantID, job.Args.UserID, post.ID, post.Meta); err != nil { - return errors.Wrapf(err, "update post(%d) meta failed", post.ID) + post, err = w.postSvc.GetPostByID(ctx, job.Args.PostID) + if err != nil { + return errors.Wrapf(err, "get post(%d) failed", job.Args.PostID) + } + + marks := lo.Map(post.Assets.Data, func(asset fields.MediaAsset, _ int) string { + if asset.Mark != nil { + return *asset.Mark + } + return "" + }) + + if items := lo.Intersect([]string{"audio-preview", "video-preview", "video", "audio"}, marks); len(items) == 4 { + if err := w.postSvc.UpdateStatus(ctx, job.Args.TenantID, job.Args.UserID, post.ID, fields.PostStatusVerified); err != nil { + return errors.Wrapf(err, "update post(%d) status failed", post.ID) + } + + if err := w.postSvc.UpdateStage(ctx, job.Args.TenantID, job.Args.UserID, post.ID, fields.PostStageCompleted); err != nil { + return errors.Wrapf(err, "update post(%d) state failed", post.ID) + } } return nil diff --git a/backend/app/jobs/post_video_extract_audio.go b/backend/app/jobs/post_video_extract_audio.go index 662e87e..2117f5e 100644 --- a/backend/app/jobs/post_video_extract_audio.go +++ b/backend/app/jobs/post_video_extract_audio.go @@ -145,14 +145,26 @@ func (w *PostVideoExtractAudioJobWorker) Work(ctx context.Context, job *Job[Post return errors.Wrapf(err, "attach audio(%s) to post(%d) failed", audioPath, post.ID) } - if job.Args.Mark == "audio-preview" { - post.Meta.WorkerMark = post.Meta.WorkerMark & 1 << 1 - } else if job.Args.Mark == "audio" { - post.Meta.WorkerMark = post.Meta.WorkerMark & 1 << 2 + // 检查是否可发布 + post, err = w.postSvc.GetPostByID(ctx, job.Args.PostID) + if err != nil { + return errors.Wrapf(err, "get post(%d) failed", job.Args.PostID) } - if err := w.postSvc.UpdateMeta(ctx, job.Args.TenantID, job.Args.UserID, post.ID, post.Meta); err != nil { - return errors.Wrapf(err, "update post(%d) meta failed", post.ID) + marks := lo.Map(post.Assets.Data, func(asset fields.MediaAsset, _ int) string { + if asset.Mark != nil { + return *asset.Mark + } + return "" + }) + if items := lo.Intersect([]string{"audio-preview", "video-preview", "video", "audio"}, marks); len(items) == 4 { + if err := w.postSvc.UpdateStatus(ctx, job.Args.TenantID, job.Args.UserID, post.ID, fields.PostStatusVerified); err != nil { + return errors.Wrapf(err, "update post(%d) status failed", post.ID) + } + + if err := w.postSvc.UpdateStage(ctx, job.Args.TenantID, job.Args.UserID, post.ID, fields.PostStageCompleted); err != nil { + return errors.Wrapf(err, "update post(%d) state failed", post.ID) + } } return nil diff --git a/qvyun.code-workspace b/qvyun.code-workspace index b03a228..054faec 100644 --- a/qvyun.code-workspace +++ b/qvyun.code-workspace @@ -7,7 +7,8 @@ "path": "frontend" }, { - "path": "backend" + "path": "backend/app", +"name":"BACKEND_APP" } ] }