refactor(P2): Handler 错误码精确化 — service 层引入哨兵错误，handler 层用 ClassifyHTTPStatus 分类
This commit is contained in:
Sisyphus
@@ -1,220 +1,221 @@
|
||||
package video
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"feedsystem_video_go/internal/middleware/rabbitmq"
|
||||
rediscache "feedsystem_video_go/internal/middleware/redis"
|
||||
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
type VideoService struct {
|
||||
repo *VideoRepository
|
||||
cache *rediscache.Client
|
||||
cacheTTL time.Duration
|
||||
popularityMQ *rabbitmq.PopularityMQ
|
||||
}
|
||||
|
||||
func NewVideoService(repo *VideoRepository, cache *rediscache.Client, popularityMQ *rabbitmq.PopularityMQ) *VideoService {
|
||||
return &VideoService{repo: repo, cache: cache, cacheTTL: 5 * time.Minute, popularityMQ: popularityMQ}
|
||||
}
|
||||
|
||||
func (vs *VideoService) Publish(ctx context.Context, video *Video) error {
|
||||
if video == nil {
|
||||
return errors.New("video is nil")
|
||||
}
|
||||
video.Title = strings.TrimSpace(video.Title)
|
||||
video.PlayURL = strings.TrimSpace(video.PlayURL)
|
||||
video.CoverURL = strings.TrimSpace(video.CoverURL)
|
||||
|
||||
if video.Title == "" {
|
||||
return errors.New("title is required")
|
||||
}
|
||||
if video.PlayURL == "" {
|
||||
return errors.New("play url is required")
|
||||
}
|
||||
if video.CoverURL == "" {
|
||||
return errors.New("cover url is required")
|
||||
}
|
||||
|
||||
//事务保证视频写入库和消息写入本地消息表的一致性
|
||||
err := vs.repo.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
||||
if err := tx.Create(video).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
msg := OutboxMsg{
|
||||
VideoID: video.ID,
|
||||
EventType: "video_published",
|
||||
Status: "pending",
|
||||
CreateTime: video.CreateTime,
|
||||
}
|
||||
|
||||
if err := tx.Create(&msg).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
|
||||
})
|
||||
return err
|
||||
|
||||
}
|
||||
|
||||
func (vs *VideoService) Delete(ctx context.Context, id uint, authorID uint) error {
|
||||
video, err := vs.repo.GetByID(ctx, id)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if video == nil {
|
||||
return errors.New("video not found")
|
||||
}
|
||||
if video.AuthorID != authorID {
|
||||
return errors.New("unauthorized")
|
||||
}
|
||||
if err := vs.repo.DeleteVideo(ctx, id); err != nil {
|
||||
return err
|
||||
}
|
||||
if vs.cache != nil {
|
||||
cacheKey := fmt.Sprintf("video:detail:id=%d", id)
|
||||
_ = vs.cache.Del(context.Background(), cacheKey)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (vs *VideoService) ListByAuthorID(ctx context.Context, authorID uint) ([]Video, error) {
|
||||
videos, err := vs.repo.ListByAuthorID(ctx, int64(authorID))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return videos, nil
|
||||
}
|
||||
|
||||
func (vs *VideoService) GetDetail(ctx context.Context, id uint) (*Video, error) {
|
||||
cacheKey := fmt.Sprintf("video:detail:id=%d", id)
|
||||
|
||||
getCached := func() (*Video, bool) {
|
||||
opCtx, cancel := context.WithTimeout(ctx, 50*time.Millisecond)
|
||||
defer cancel()
|
||||
|
||||
b, err := vs.cache.GetBytes(opCtx, cacheKey)
|
||||
if err != nil {
|
||||
return nil, false
|
||||
}
|
||||
var cached Video
|
||||
if err := json.Unmarshal(b, &cached); err != nil {
|
||||
return nil, false
|
||||
}
|
||||
return &cached, true
|
||||
}
|
||||
|
||||
setCached := func(video *Video) {
|
||||
b, err := json.Marshal(video)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
opCtx, cancel := context.WithTimeout(ctx, 50*time.Millisecond)
|
||||
defer cancel()
|
||||
_ = vs.cache.SetBytes(opCtx, cacheKey, b, vs.cacheTTL)
|
||||
}
|
||||
|
||||
if vs.cache != nil {
|
||||
if v, ok := getCached(); ok {
|
||||
return v, nil
|
||||
}
|
||||
|
||||
opCtx, cancel := context.WithTimeout(ctx, 50*time.Millisecond)
|
||||
b, err := vs.cache.GetBytes(opCtx, cacheKey)
|
||||
cancel()
|
||||
if err == nil {
|
||||
var cached Video
|
||||
if err := json.Unmarshal(b, &cached); err == nil {
|
||||
return &cached, nil
|
||||
}
|
||||
} else if rediscache.IsMiss(err) {
|
||||
lockKey := "lock:" + cacheKey
|
||||
|
||||
lockCtx, lockCancel := context.WithTimeout(ctx, 50*time.Millisecond)
|
||||
token, locked, lockErr := vs.cache.Lock(lockCtx, lockKey, 2*time.Second)
|
||||
lockCancel()
|
||||
|
||||
if lockErr == nil && locked {
|
||||
defer func() { _ = vs.cache.Unlock(context.Background(), lockKey, token) }()
|
||||
|
||||
if v, ok := getCached(); ok {
|
||||
return v, nil
|
||||
}
|
||||
|
||||
video, err := vs.repo.GetByID(ctx, id)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
setCached(video)
|
||||
return video, nil
|
||||
}
|
||||
|
||||
// 没拿到锁：等待别人回填缓存
|
||||
for i := 0; i < 5; i++ {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return nil, ctx.Err()
|
||||
case <-time.After(20 * time.Millisecond):
|
||||
}
|
||||
if v, ok := getCached(); ok {
|
||||
return v, nil
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
video, err := vs.repo.GetByID(ctx, id)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if vs.cache != nil {
|
||||
setCached(video)
|
||||
}
|
||||
return video, nil
|
||||
}
|
||||
|
||||
func (vs *VideoService) UpdateLikesCount(ctx context.Context, id uint, likesCount int64) error {
|
||||
if err := vs.repo.UpdateLikesCount(ctx, id, likesCount); err != nil {
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (vs *VideoService) UpdatePopularity(ctx context.Context, id uint, change int64) error {
|
||||
if err := vs.repo.UpdatePopularity(ctx, id, change); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if vs.popularityMQ != nil {
|
||||
if err := vs.popularityMQ.Update(ctx, id, change); err == nil {
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
if vs.cache != nil {
|
||||
// 1) 详情缓存：直接失效（最简单靠谱）
|
||||
_ = vs.cache.Del(context.Background(), fmt.Sprintf("video:detail:id=%d", id))
|
||||
|
||||
// 2) 热榜：写到“时间窗ZSET”，不要用 detail key
|
||||
now := time.Now().UTC().Truncate(time.Minute)
|
||||
windowKey := "hot:video:1m:" + now.Format("200601021504")
|
||||
member := strconv.FormatUint(uint64(id), 10)
|
||||
|
||||
opCtx, cancel := context.WithTimeout(ctx, 50*time.Millisecond)
|
||||
defer cancel()
|
||||
|
||||
_ = vs.cache.ZincrBy(opCtx, windowKey, member, float64(change))
|
||||
_ = vs.cache.Expire(opCtx, windowKey, 2*time.Hour)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
package video
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"strconv"
|
||||
"strings"
|
||||
"time"
|
||||
|
||||
"feedsystem_video_go/internal/middleware/rabbitmq"
|
||||
rediscache "feedsystem_video_go/internal/middleware/redis"
|
||||
httputil "feedsystem_video_go/internal/http"
|
||||
|
||||
"gorm.io/gorm"
|
||||
)
|
||||
|
||||
type VideoService struct {
|
||||
repo *VideoRepository
|
||||
cache *rediscache.Client
|
||||
cacheTTL time.Duration
|
||||
popularityMQ *rabbitmq.PopularityMQ
|
||||
}
|
||||
|
||||
func NewVideoService(repo *VideoRepository, cache *rediscache.Client, popularityMQ *rabbitmq.PopularityMQ) *VideoService {
|
||||
return &VideoService{repo: repo, cache: cache, cacheTTL: 5 * time.Minute, popularityMQ: popularityMQ}
|
||||
}
|
||||
|
||||
func (vs *VideoService) Publish(ctx context.Context, video *Video) error {
|
||||
if video == nil {
|
||||
return errors.New("video is nil")
|
||||
}
|
||||
video.Title = strings.TrimSpace(video.Title)
|
||||
video.PlayURL = strings.TrimSpace(video.PlayURL)
|
||||
video.CoverURL = strings.TrimSpace(video.CoverURL)
|
||||
|
||||
if video.Title == "" {
|
||||
return errors.New("title is required")
|
||||
}
|
||||
if video.PlayURL == "" {
|
||||
return errors.New("play url is required")
|
||||
}
|
||||
if video.CoverURL == "" {
|
||||
return errors.New("cover url is required")
|
||||
}
|
||||
|
||||
//事务保证视频写入库和消息写入本地消息表的一致性
|
||||
err := vs.repo.db.WithContext(ctx).Transaction(func(tx *gorm.DB) error {
|
||||
if err := tx.Create(video).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
msg := OutboxMsg{
|
||||
VideoID: video.ID,
|
||||
EventType: "video_published",
|
||||
Status: "pending",
|
||||
CreateTime: video.CreateTime,
|
||||
}
|
||||
|
||||
if err := tx.Create(&msg).Error; err != nil {
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
|
||||
})
|
||||
return err
|
||||
|
||||
}
|
||||
|
||||
func (vs *VideoService) Delete(ctx context.Context, id uint, authorID uint) error {
|
||||
video, err := vs.repo.GetByID(ctx, id)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
if video == nil {
|
||||
return errors.New("video not found")
|
||||
}
|
||||
if video.AuthorID != authorID {
|
||||
return httputil.ErrUnauthorized
|
||||
}
|
||||
if err := vs.repo.DeleteVideo(ctx, id); err != nil {
|
||||
return err
|
||||
}
|
||||
if vs.cache != nil {
|
||||
cacheKey := fmt.Sprintf("video:detail:id=%d", id)
|
||||
_ = vs.cache.Del(context.Background(), cacheKey)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (vs *VideoService) ListByAuthorID(ctx context.Context, authorID uint) ([]Video, error) {
|
||||
videos, err := vs.repo.ListByAuthorID(ctx, int64(authorID))
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
return videos, nil
|
||||
}
|
||||
|
||||
func (vs *VideoService) GetDetail(ctx context.Context, id uint) (*Video, error) {
|
||||
cacheKey := fmt.Sprintf("video:detail:id=%d", id)
|
||||
|
||||
getCached := func() (*Video, bool) {
|
||||
opCtx, cancel := context.WithTimeout(ctx, 50*time.Millisecond)
|
||||
defer cancel()
|
||||
|
||||
b, err := vs.cache.GetBytes(opCtx, cacheKey)
|
||||
if err != nil {
|
||||
return nil, false
|
||||
}
|
||||
var cached Video
|
||||
if err := json.Unmarshal(b, &cached); err != nil {
|
||||
return nil, false
|
||||
}
|
||||
return &cached, true
|
||||
}
|
||||
|
||||
setCached := func(video *Video) {
|
||||
b, err := json.Marshal(video)
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
opCtx, cancel := context.WithTimeout(ctx, 50*time.Millisecond)
|
||||
defer cancel()
|
||||
_ = vs.cache.SetBytes(opCtx, cacheKey, b, vs.cacheTTL)
|
||||
}
|
||||
|
||||
if vs.cache != nil {
|
||||
if v, ok := getCached(); ok {
|
||||
return v, nil
|
||||
}
|
||||
|
||||
opCtx, cancel := context.WithTimeout(ctx, 50*time.Millisecond)
|
||||
b, err := vs.cache.GetBytes(opCtx, cacheKey)
|
||||
cancel()
|
||||
if err == nil {
|
||||
var cached Video
|
||||
if err := json.Unmarshal(b, &cached); err == nil {
|
||||
return &cached, nil
|
||||
}
|
||||
} else if rediscache.IsMiss(err) {
|
||||
lockKey := "lock:" + cacheKey
|
||||
|
||||
lockCtx, lockCancel := context.WithTimeout(ctx, 50*time.Millisecond)
|
||||
token, locked, lockErr := vs.cache.Lock(lockCtx, lockKey, 2*time.Second)
|
||||
lockCancel()
|
||||
|
||||
if lockErr == nil && locked {
|
||||
defer func() { _ = vs.cache.Unlock(context.Background(), lockKey, token) }()
|
||||
|
||||
if v, ok := getCached(); ok {
|
||||
return v, nil
|
||||
}
|
||||
|
||||
video, err := vs.repo.GetByID(ctx, id)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
setCached(video)
|
||||
return video, nil
|
||||
}
|
||||
|
||||
// 没拿到锁：等待别人回填缓存
|
||||
for i := 0; i < 5; i++ {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return nil, ctx.Err()
|
||||
case <-time.After(20 * time.Millisecond):
|
||||
}
|
||||
if v, ok := getCached(); ok {
|
||||
return v, nil
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
video, err := vs.repo.GetByID(ctx, id)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if vs.cache != nil {
|
||||
setCached(video)
|
||||
}
|
||||
return video, nil
|
||||
}
|
||||
|
||||
func (vs *VideoService) UpdateLikesCount(ctx context.Context, id uint, likesCount int64) error {
|
||||
if err := vs.repo.UpdateLikesCount(ctx, id, likesCount); err != nil {
|
||||
return err
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (vs *VideoService) UpdatePopularity(ctx context.Context, id uint, change int64) error {
|
||||
if err := vs.repo.UpdatePopularity(ctx, id, change); err != nil {
|
||||
return err
|
||||
}
|
||||
|
||||
if vs.popularityMQ != nil {
|
||||
if err := vs.popularityMQ.Update(ctx, id, change); err == nil {
|
||||
return nil
|
||||
}
|
||||
}
|
||||
|
||||
if vs.cache != nil {
|
||||
// 1) 详情缓存：直接失效（最简单靠谱）
|
||||
_ = vs.cache.Del(context.Background(), fmt.Sprintf("video:detail:id=%d", id))
|
||||
|
||||
// 2) 热榜：写到“时间窗ZSET”，不要用 detail key
|
||||
now := time.Now().UTC().Truncate(time.Minute)
|
||||
windowKey := "hot:video:1m:" + now.Format("200601021504")
|
||||
member := strconv.FormatUint(uint64(id), 10)
|
||||
|
||||
opCtx, cancel := context.WithTimeout(ctx, 50*time.Millisecond)
|
||||
defer cancel()
|
||||
|
||||
_ = vs.cache.ZincrBy(opCtx, windowKey, member, float64(change))
|
||||
_ = vs.cache.Expire(opCtx, windowKey, 2*time.Hour)
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user