From dd47b48473c8fa5e82c91dadc204b9a9956643fb Mon Sep 17 00:00:00 2001 From: Sisyphus Date: Sat, 25 Apr 2026 16:57:02 +0800 Subject: [PATCH] =?UTF-8?q?refactor(P3):=20Redis=20=E7=BC=93=E5=AD=98?= =?UTF-8?q?=E9=94=AE=E7=89=88=E6=9C=AC=E5=8C=96=20=E2=80=94=20Client=20?= =?UTF-8?q?=E5=A2=9E=E5=8A=A0=20Key()=20=E6=96=B9=E6=B3=95=EF=BC=8C?= =?UTF-8?q?=E9=BB=98=E8=AE=A4=E5=89=8D=E7=BC=80=20v1:?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/internal/account/service.go | 311 ++++++++++----------- backend/internal/feed/service.go | 28 +- backend/internal/middleware/jwt/jwt.go | 279 +++++++++--------- backend/internal/middleware/redis/redis.go | 210 +++++++------- backend/internal/video/popularity_cache.go | 57 ++-- backend/internal/video/video_service.go | 9 +- backend/internal/worker/outboxworker.go | 186 ++++++------ 7 files changed, 544 insertions(+), 536 deletions(-) diff --git a/backend/internal/account/service.go b/backend/internal/account/service.go index 17debd4..cab8a3d 100644 --- a/backend/internal/account/service.go +++ b/backend/internal/account/service.go @@ -1,156 +1,155 @@ -package account - -import ( - "context" - "errors" - "feedsystem_video_go/internal/auth" - "fmt" - "log" - "time" - - rediscache "feedsystem_video_go/internal/middleware/redis" - - "github.com/go-sql-driver/mysql" - "golang.org/x/crypto/bcrypt" - "gorm.io/gorm" -) - -type AccountService struct { - accountRepository *AccountRepository - cache *rediscache.Client -} - -var ( - ErrUsernameTaken = errors.New("username already exists") - ErrNewUsernameRequired = errors.New("new_username is required") -) - -func NewAccountService(accountRepository *AccountRepository, cache *rediscache.Client) *AccountService { - return &AccountService{accountRepository: accountRepository, cache: cache} -} - -func (as *AccountService) CreateAccount(ctx context.Context, account *Account) error { - passwordHash, err := bcrypt.GenerateFromPassword([]byte(account.Password), bcrypt.DefaultCost) - if err != nil { - return err - } - account.Password = string(passwordHash) - if err := as.accountRepository.CreateAccount(ctx, account); err != nil { - return err - } - return nil -} - -func (as *AccountService) Rename(ctx context.Context, accountID uint, newUsername string) (string, error) { - if newUsername == "" { - return "", ErrNewUsernameRequired - } - - token, err := auth.GenerateToken(accountID, newUsername) - if err != nil { - return "", err - } - - if err := as.accountRepository.RenameWithToken(ctx, accountID, newUsername, token); err != nil { - var mysqlErr *mysql.MySQLError - if errors.As(err, &mysqlErr) && mysqlErr.Number == 1062 { - return "", ErrUsernameTaken - } - if errors.Is(err, gorm.ErrRecordNotFound) { - return "", err - } - return "", err - } - if as.cache != nil { - cacheCtx, cancel := context.WithTimeout(ctx, 50*time.Millisecond) - defer cancel() - - if err := as.cache.SetBytes(cacheCtx, fmt.Sprintf("account:%d", accountID), []byte(token), 24*time.Hour); err != nil { - log.Printf("failed to set cache: %v", err) - } - } - return token, nil -} - -func (as *AccountService) ChangePassword(ctx context.Context, username, oldPassword, newPassword string) error { - account, err := as.FindByUsername(ctx, username) - if err != nil { - return err - } - if err := bcrypt.CompareHashAndPassword([]byte(account.Password), []byte(oldPassword)); err != nil { - return err - } - passwordHash, err := bcrypt.GenerateFromPassword([]byte(newPassword), bcrypt.DefaultCost) - if err != nil { - return err - } - if err := as.accountRepository.ChangePassword(ctx, account.ID, string(passwordHash)); err != nil { - return err - } - if err := as.Logout(ctx, account.ID); err != nil { - return err - } - return nil -} - -func (as *AccountService) FindByID(ctx context.Context, id uint) (*Account, error) { - if account, err := as.accountRepository.FindByID(ctx, id); err != nil { - return nil, err - } else { - return account, nil - } -} - -func (as *AccountService) FindByUsername(ctx context.Context, username string) (*Account, error) { - if account, err := as.accountRepository.FindByUsername(ctx, username); err != nil { - return nil, err - } else { - return account, nil - } -} - -func (as *AccountService) Login(ctx context.Context, username, password string) (string, error) { - account, err := as.FindByUsername(ctx, username) - if err != nil { - return "", err - } - if err := bcrypt.CompareHashAndPassword([]byte(account.Password), []byte(password)); err != nil { - return "", err - } - // generate token - token, err := auth.GenerateToken(account.ID, account.Username) - if err != nil { - return "", err - } - if err := as.accountRepository.Login(ctx, account.ID, token); err != nil { - return "", err - } - if as.cache != nil { - cacheCtx, cancel := context.WithTimeout(ctx, 50*time.Millisecond) - defer cancel() - - if err := as.cache.SetBytes(cacheCtx, fmt.Sprintf("account:%d", account.ID), []byte(token), 24*time.Hour); err != nil { - log.Printf("failed to set cache: %v", err) - } - } - return token, nil -} - -func (as *AccountService) Logout(ctx context.Context, accountID uint) error { - account, err := as.FindByID(ctx, accountID) - if err != nil { - return err - } - if account.Token == "" { - return nil - } - if as.cache != nil { - cacheCtx, cancel := context.WithTimeout(ctx, 50*time.Millisecond) - defer cancel() - - if err := as.cache.Del(cacheCtx, fmt.Sprintf("account:%d", account.ID)); err != nil { - log.Printf("failed to del cache: %v", err) - } - } - return as.accountRepository.Logout(ctx, account.ID) -} +package account + +import ( + "context" + "errors" + "feedsystem_video_go/internal/auth" + "log" + "time" + + rediscache "feedsystem_video_go/internal/middleware/redis" + + "github.com/go-sql-driver/mysql" + "golang.org/x/crypto/bcrypt" + "gorm.io/gorm" +) + +type AccountService struct { + accountRepository *AccountRepository + cache *rediscache.Client +} + +var ( + ErrUsernameTaken = errors.New("username already exists") + ErrNewUsernameRequired = errors.New("new_username is required") +) + +func NewAccountService(accountRepository *AccountRepository, cache *rediscache.Client) *AccountService { + return &AccountService{accountRepository: accountRepository, cache: cache} +} + +func (as *AccountService) CreateAccount(ctx context.Context, account *Account) error { + passwordHash, err := bcrypt.GenerateFromPassword([]byte(account.Password), bcrypt.DefaultCost) + if err != nil { + return err + } + account.Password = string(passwordHash) + if err := as.accountRepository.CreateAccount(ctx, account); err != nil { + return err + } + return nil +} + +func (as *AccountService) Rename(ctx context.Context, accountID uint, newUsername string) (string, error) { + if newUsername == "" { + return "", ErrNewUsernameRequired + } + + token, err := auth.GenerateToken(accountID, newUsername) + if err != nil { + return "", err + } + + if err := as.accountRepository.RenameWithToken(ctx, accountID, newUsername, token); err != nil { + var mysqlErr *mysql.MySQLError + if errors.As(err, &mysqlErr) && mysqlErr.Number == 1062 { + return "", ErrUsernameTaken + } + if errors.Is(err, gorm.ErrRecordNotFound) { + return "", err + } + return "", err + } + if as.cache != nil { + cacheCtx, cancel := context.WithTimeout(ctx, 50*time.Millisecond) + defer cancel() + + if err := as.cache.SetBytes(cacheCtx, as.cache.Key("account:%d", accountID), []byte(token), 24*time.Hour); err != nil { + log.Printf("failed to set cache: %v", err) + } + } + return token, nil +} + +func (as *AccountService) ChangePassword(ctx context.Context, username, oldPassword, newPassword string) error { + account, err := as.FindByUsername(ctx, username) + if err != nil { + return err + } + if err := bcrypt.CompareHashAndPassword([]byte(account.Password), []byte(oldPassword)); err != nil { + return err + } + passwordHash, err := bcrypt.GenerateFromPassword([]byte(newPassword), bcrypt.DefaultCost) + if err != nil { + return err + } + if err := as.accountRepository.ChangePassword(ctx, account.ID, string(passwordHash)); err != nil { + return err + } + if err := as.Logout(ctx, account.ID); err != nil { + return err + } + return nil +} + +func (as *AccountService) FindByID(ctx context.Context, id uint) (*Account, error) { + if account, err := as.accountRepository.FindByID(ctx, id); err != nil { + return nil, err + } else { + return account, nil + } +} + +func (as *AccountService) FindByUsername(ctx context.Context, username string) (*Account, error) { + if account, err := as.accountRepository.FindByUsername(ctx, username); err != nil { + return nil, err + } else { + return account, nil + } +} + +func (as *AccountService) Login(ctx context.Context, username, password string) (string, error) { + account, err := as.FindByUsername(ctx, username) + if err != nil { + return "", err + } + if err := bcrypt.CompareHashAndPassword([]byte(account.Password), []byte(password)); err != nil { + return "", err + } + // generate token + token, err := auth.GenerateToken(account.ID, account.Username) + if err != nil { + return "", err + } + if err := as.accountRepository.Login(ctx, account.ID, token); err != nil { + return "", err + } + if as.cache != nil { + cacheCtx, cancel := context.WithTimeout(ctx, 50*time.Millisecond) + defer cancel() + + if err := as.cache.SetBytes(cacheCtx, as.cache.Key("account:%d", account.ID), []byte(token), 24*time.Hour); err != nil { + log.Printf("failed to set cache: %v", err) + } + } + return token, nil +} + +func (as *AccountService) Logout(ctx context.Context, accountID uint) error { + account, err := as.FindByID(ctx, accountID) + if err != nil { + return err + } + if account.Token == "" { + return nil + } + if as.cache != nil { + cacheCtx, cancel := context.WithTimeout(ctx, 50*time.Millisecond) + defer cancel() + + if err := as.cache.Del(cacheCtx, as.cache.Key("account:%d", account.ID)); err != nil { + log.Printf("failed to del cache: %v", err) + } + } + return as.accountRepository.Logout(ctx, account.ID) +} diff --git a/backend/internal/feed/service.go b/backend/internal/feed/service.go index 95e6939..4c62032 100644 --- a/backend/internal/feed/service.go +++ b/backend/internal/feed/service.go @@ -44,7 +44,7 @@ func (f *FeedService) GetVideoByIDs(ctx context.Context, videoIDs []uint) ([]*vi //L1:本地缓存 var missedL1 []uint for _, id := range videoIDs { - cacheKey := fmt.Sprintf("video:entity:%d", id) + cacheKey := f.rediscache.Key("video:entity:%d", id) if f.localcache != nil { if v, found := f.localcache.Get(cacheKey); found { if data, ok := v.(video.Video); ok { @@ -66,7 +66,7 @@ func (f *FeedService) GetVideoByIDs(ctx context.Context, videoIDs []uint) ([]*vi if len(missedL1) > 0 { cacheKeys := make([]string, len(missedL1)) for i, id := range missedL1 { - cacheKeys[i] = fmt.Sprintf("video:entity:%d", id) + cacheKeys[i] = f.rediscache.Key("video:entity:%d", id) } cacheCtx, cancel := context.WithTimeout(ctx, 50*time.Millisecond) @@ -109,7 +109,7 @@ func (f *FeedService) GetVideoByIDs(ctx context.Context, videoIDs []uint) ([]*vi wg.Add(1) go func(videoID uint) { defer wg.Done() - sfKey := fmt.Sprintf("sf:entity:%d", videoID) + sfKey := f.rediscache.Key("sf:entity:%d", videoID) v, err, _ := f.requestGroup.Do(sfKey, func() (interface{}, error) { videoList, err := f.repo.GetByIDs(ctx, []uint{videoID}) @@ -119,7 +119,7 @@ func (f *FeedService) GetVideoByIDs(ctx context.Context, videoIDs []uint) ([]*vi } safeCopy := *videoList[0] - cachekey := fmt.Sprintf("video:entity:%d", safeCopy.ID) + cachekey := f.rediscache.Key("video:entity:%d", safeCopy.ID) if b, err := json.Marshal(safeCopy); err == nil { //异步回写redis go func(k string, b []byte) { @@ -137,7 +137,7 @@ func (f *FeedService) GetVideoByIDs(ctx context.Context, videoIDs []uint) ([]*vi mu.Lock() videoMap[id] = &safeCopy mu.Unlock() - f.localcache.Set(fmt.Sprintf("video:entity:%d", safeCopy.ID), safeCopy, 5*time.Second) + f.localcache.Set(f.rediscache.Key("video:entity:%d", safeCopy.ID), safeCopy, 5*time.Second) } }(id) } @@ -148,7 +148,7 @@ func (f *FeedService) GetVideoByIDs(ctx context.Context, videoIDs []uint) ([]*vi // 查询最新视频 (冷热分离 + 游标分页) func (f *FeedService) ListLatest(ctx context.Context, limit int, latestBefore time.Time, viewerAccountID uint) (ListLatestResponse, error) { // 获取 ZSET 中最老的一条数据 - zsetTail, err := f.rediscache.ZRangeWithScores(ctx, "feed:global_timeline", 0, 0) + zsetTail, err := f.rediscache.ZRangeWithScores(ctx, f.rediscache.Key("feed:global_timeline"), 0, 0) if err != nil { return ListLatestResponse{}, err @@ -158,7 +158,7 @@ func (f *FeedService) ListLatest(ctx context.Context, limit int, latestBefore ti if isZsetEmpty { //全局静态锁：无视所有用户的不同时间戳游标 - sfKey := "sf:fallback:global_timeline_rebuild" + sfKey := f.rediscache.Key("sf:fallback:global_timeline_rebuild") v, err, _ := f.requestGroup.Do(sfKey, func() (interface{}, error) { // 无视游标，直接去 MySQL 捞最新的 1000 条 @@ -180,7 +180,7 @@ func (f *FeedService) ListLatest(ctx context.Context, limit int, latestBefore ti Member: fmt.Sprintf("%d", vid.ID), }) } - f.rediscache.ZAdd(bgCtx, "feed:global_timeline", zElements...) + f.rediscache.ZAdd(bgCtx, f.rediscache.Key("feed:global_timeline"), zElements...) return "SUCCESS", nil }) @@ -207,7 +207,7 @@ func (f *FeedService) ListLatest(ctx context.Context, limit int, latestBefore ti //冷数据降级查库 // 针对个别用户的防并发（此时可以用时间戳做锁，因为冷尾流量极小） - sfKey := fmt.Sprintf("sf:cold:listLatest:%d:%d", limit, reqTime) + sfKey := f.rediscache.Key("sf:cold:listLatest:%d:%d", limit, reqTime) v, err, _ := f.requestGroup.Do(sfKey, func() (interface{}, error) { return f.repo.ListLatest(ctx, limit, latestBefore) }) @@ -224,7 +224,7 @@ func (f *FeedService) ListLatest(ctx context.Context, limit int, latestBefore ti maxScore = fmt.Sprintf("%d", reqTime-1) // 防重复 } - videoIDsStr, err := f.rediscache.ZRevRangeByScore(ctx, "feed:global_timeline", maxScore, "-inf", 0, int64(limit)) + videoIDsStr, err := f.rediscache.ZRevRangeByScore(ctx, f.rediscache.Key("feed:global_timeline"), maxScore, "-inf", 0, int64(limit)) if err != nil { return ListLatestResponse{}, err } @@ -254,7 +254,7 @@ func (f *FeedService) ListLatest(ctx context.Context, limit int, latestBefore ti coldCursor = latestBefore } - sfKey := fmt.Sprintf("sf:stitch:listLatest:%d:%d", remainLimit, coldCursor.UnixMilli()) + sfKey := f.rediscache.Key("sf:stitch:listLatest:%d:%d", remainLimit, coldCursor.UnixMilli()) v, err, _ := f.requestGroup.Do(sfKey, func() (interface{}, error) { return f.repo.ListLatest(ctx, remainLimit, coldCursor) }) @@ -343,7 +343,7 @@ func (f *FeedService) ListByFollowing(ctx context.Context, limit int, latestBefo if !latestBefore.IsZero() { before = latestBefore.Unix() } - cacheKey = fmt.Sprintf("feed:listByFollowing:limit=%d:accountID=%d:before=%d", limit, viewerAccountID, before) + cacheKey = f.rediscache.Key("feed:listByFollowing:limit=%d:accountID=%d:before=%d", limit, viewerAccountID, before) cacheCtx, cancel := context.WithTimeout(ctx, 50*time.Millisecond) defer cancel() @@ -413,10 +413,10 @@ func (f *FeedService) ListByPopularity(ctx context.Context, limit int, reqAsOf i const win = 60 keys := make([]string, 0, win) for i := 0; i < win; i++ { - keys = append(keys, "hot:video:1m:"+asOf.Add(-time.Duration(i)*time.Minute).Format("200601021504")) + keys = append(keys, f.rediscache.Key("hot:video:1m:%s", asOf.Add(-time.Duration(i)*time.Minute).Format("200601021504"))) } - dest := "hot:video:merge:1m:" + asOf.Format("200601021504") // 快照key：同一个as_of页内复用 + dest := f.rediscache.Key("hot:video:merge:1m:%s", asOf.Format("200601021504")) // 快照key：同一个as_of页内复用 opCtx, cancel := context.WithTimeout(ctx, 80*time.Millisecond) defer cancel() diff --git a/backend/internal/middleware/jwt/jwt.go b/backend/internal/middleware/jwt/jwt.go index 700448f..a1a389c 100644 --- a/backend/internal/middleware/jwt/jwt.go +++ b/backend/internal/middleware/jwt/jwt.go @@ -1,140 +1,139 @@ -package jwt - -import ( - "context" - "errors" - "fmt" - "log" - "net/http" - "strings" - "time" - - "feedsystem_video_go/internal/account" - "feedsystem_video_go/internal/auth" - rediscache "feedsystem_video_go/internal/middleware/redis" - - "github.com/gin-gonic/gin" -) - -// JWTAuth check jwt token and ensure it matches the currently stored token. -func JWTAuth(accountRepo *account.AccountRepository, cache *rediscache.Client) gin.HandlerFunc { - return func(c *gin.Context) { - authHeader := c.GetHeader("Authorization") - if authHeader == "" { - c.AbortWithStatusJSON(http.StatusUnauthorized, gin.H{"error": "missing authorization header"}) - return - } - - parts := strings.SplitN(authHeader, " ", 2) - if len(parts) != 2 || !strings.EqualFold(parts[0], "Bearer") { - c.AbortWithStatusJSON(http.StatusUnauthorized, gin.H{"error": "invalid authorization header"}) - return - } - - tokenString := parts[1] - - claims, err := auth.ParseToken(tokenString) - if err != nil { - c.AbortWithStatusJSON(http.StatusUnauthorized, gin.H{"error": "invalid or expired token"}) - return - } - check(c, claims, tokenString, accountRepo, cache) - } -} - -func SoftJWTAuth(accountRepo *account.AccountRepository, cache *rediscache.Client) gin.HandlerFunc { - return func(c *gin.Context) { - authHeader := c.GetHeader("Authorization") - if authHeader == "" { - c.Next() - return - } - - parts := strings.SplitN(authHeader, " ", 2) - if len(parts) != 2 || !strings.EqualFold(parts[0], "Bearer") { - c.AbortWithStatusJSON(http.StatusUnauthorized, gin.H{"error": "invalid authorization header"}) - return - } - - tokenString := parts[1] - - claims, err := auth.ParseToken(tokenString) - if err != nil { - c.AbortWithStatusJSON(http.StatusUnauthorized, gin.H{"error": "invalid or expired token"}) - return - } - - check(c, claims, tokenString, accountRepo, cache) - } -} - -func check(c *gin.Context, claims *auth.Claims, tokenString string, accountRepo *account.AccountRepository, cache *rediscache.Client) { - key := fmt.Sprintf("account:%d", claims.AccountID) - - // 先查 Redis - if cache != nil { - cacheCtx, cancel := context.WithTimeout(c.Request.Context(), 50*time.Millisecond) - defer cancel() - - b, err := cache.GetBytes(cacheCtx, key) - if err == nil { - if string(b) != tokenString { - c.AbortWithStatusJSON(http.StatusUnauthorized, gin.H{"error": "token has been revoked"}) - return - } - c.Set("accountID", claims.AccountID) - c.Set("username", claims.Username) - c.Next() - return - } - } - - // Redis 故障/未启用：查 DB 兜底 - accountInfo, err := accountRepo.FindByID(c.Request.Context(), claims.AccountID) - if err != nil || accountInfo.Token == "" || accountInfo.Token != tokenString { - c.AbortWithStatusJSON(http.StatusUnauthorized, gin.H{"error": "token has been revoked"}) - return - } - - if cache != nil { - cacheCtx, cancel := context.WithTimeout(c.Request.Context(), 50*time.Millisecond) - defer cancel() - - if err := cache.SetBytes(cacheCtx, key, []byte(tokenString), 24*time.Hour); err != nil { - log.Printf("failed to set cache: %v", err) - } - } - - c.Set("accountID", claims.AccountID) - c.Set("username", claims.Username) - c.Next() - -} - -func GetAccountID(c *gin.Context) (uint, error) { - uidValue, exists := c.Get("accountID") - if !exists { - return 0, errors.New("accountID not found") - } - - accountID, ok := uidValue.(uint) - if !ok { - return 0, errors.New("accountID has invalid type") - } - - return accountID, nil -} - -func GetUsername(c *gin.Context) (string, error) { - val, exists := c.Get("username") - if !exists { - return "", errors.New("username not found") - } - - username, ok := val.(string) - if !ok { - return "", errors.New("username has invalid type") - } - - return username, nil -} +package jwt + +import ( + "context" + "errors" + "log" + "net/http" + "strings" + "time" + + "feedsystem_video_go/internal/account" + "feedsystem_video_go/internal/auth" + rediscache "feedsystem_video_go/internal/middleware/redis" + + "github.com/gin-gonic/gin" +) + +// JWTAuth check jwt token and ensure it matches the currently stored token. +func JWTAuth(accountRepo *account.AccountRepository, cache *rediscache.Client) gin.HandlerFunc { + return func(c *gin.Context) { + authHeader := c.GetHeader("Authorization") + if authHeader == "" { + c.AbortWithStatusJSON(http.StatusUnauthorized, gin.H{"error": "missing authorization header"}) + return + } + + parts := strings.SplitN(authHeader, " ", 2) + if len(parts) != 2 || !strings.EqualFold(parts[0], "Bearer") { + c.AbortWithStatusJSON(http.StatusUnauthorized, gin.H{"error": "invalid authorization header"}) + return + } + + tokenString := parts[1] + + claims, err := auth.ParseToken(tokenString) + if err != nil { + c.AbortWithStatusJSON(http.StatusUnauthorized, gin.H{"error": "invalid or expired token"}) + return + } + check(c, claims, tokenString, accountRepo, cache) + } +} + +func SoftJWTAuth(accountRepo *account.AccountRepository, cache *rediscache.Client) gin.HandlerFunc { + return func(c *gin.Context) { + authHeader := c.GetHeader("Authorization") + if authHeader == "" { + c.Next() + return + } + + parts := strings.SplitN(authHeader, " ", 2) + if len(parts) != 2 || !strings.EqualFold(parts[0], "Bearer") { + c.AbortWithStatusJSON(http.StatusUnauthorized, gin.H{"error": "invalid authorization header"}) + return + } + + tokenString := parts[1] + + claims, err := auth.ParseToken(tokenString) + if err != nil { + c.AbortWithStatusJSON(http.StatusUnauthorized, gin.H{"error": "invalid or expired token"}) + return + } + + check(c, claims, tokenString, accountRepo, cache) + } +} + +func check(c *gin.Context, claims *auth.Claims, tokenString string, accountRepo *account.AccountRepository, cache *rediscache.Client) { + key := cache.Key("account:%d", claims.AccountID) + + // 先查 Redis + if cache != nil { + cacheCtx, cancel := context.WithTimeout(c.Request.Context(), 50*time.Millisecond) + defer cancel() + + b, err := cache.GetBytes(cacheCtx, key) + if err == nil { + if string(b) != tokenString { + c.AbortWithStatusJSON(http.StatusUnauthorized, gin.H{"error": "token has been revoked"}) + return + } + c.Set("accountID", claims.AccountID) + c.Set("username", claims.Username) + c.Next() + return + } + } + + // Redis 故障/未启用：查 DB 兜底 + accountInfo, err := accountRepo.FindByID(c.Request.Context(), claims.AccountID) + if err != nil || accountInfo.Token == "" || accountInfo.Token != tokenString { + c.AbortWithStatusJSON(http.StatusUnauthorized, gin.H{"error": "token has been revoked"}) + return + } + + if cache != nil { + cacheCtx, cancel := context.WithTimeout(c.Request.Context(), 50*time.Millisecond) + defer cancel() + + if err := cache.SetBytes(cacheCtx, key, []byte(tokenString), 24*time.Hour); err != nil { + log.Printf("failed to set cache: %v", err) + } + } + + c.Set("accountID", claims.AccountID) + c.Set("username", claims.Username) + c.Next() + +} + +func GetAccountID(c *gin.Context) (uint, error) { + uidValue, exists := c.Get("accountID") + if !exists { + return 0, errors.New("accountID not found") + } + + accountID, ok := uidValue.(uint) + if !ok { + return 0, errors.New("accountID has invalid type") + } + + return accountID, nil +} + +func GetUsername(c *gin.Context) (string, error) { + val, exists := c.Get("username") + if !exists { + return "", errors.New("username not found") + } + + username, ok := val.(string) + if !ok { + return "", errors.New("username has invalid type") + } + + return username, nil +} diff --git a/backend/internal/middleware/redis/redis.go b/backend/internal/middleware/redis/redis.go index 074f7de..1f7f011 100644 --- a/backend/internal/middleware/redis/redis.go +++ b/backend/internal/middleware/redis/redis.go @@ -1,99 +1,111 @@ -package redis - -import ( - "context" - "crypto/rand" - "encoding/hex" - "feedsystem_video_go/internal/config" - "strconv" - "time" - - redis "github.com/redis/go-redis/v9" -) - -type Client struct { - rdb *redis.Client -} - -func NewFromEnv(cfg *config.RedisConfig) (*Client, error) { - rdb := redis.NewClient(&redis.Options{ - Addr: cfg.Host + ":" + strconv.Itoa(cfg.Port), - Password: cfg.Password, - DB: cfg.DB, - }) - return &Client{rdb: rdb}, nil -} - -func (c *Client) Close() error { - if c == nil || c.rdb == nil { - return nil - } - return c.rdb.Close() -} - -func (c *Client) Ping(ctx context.Context) error { - if c == nil || c.rdb == nil { - return nil - } - return c.rdb.Ping(ctx).Err() -} - -func IsMiss(err error) bool { - return err == redis.Nil -} - -func randToken(n int) (string, error) { - b := make([]byte, n) - if _, err := rand.Read(b); err != nil { - return "", err - } - return hex.EncodeToString(b), nil -} - -func (c *Client) Lock(ctx context.Context, key string, ttl time.Duration) (token string, ok bool, err error) { - if c == nil || c.rdb == nil { - return "", false, nil - } - token, err = randToken(16) - if err != nil { - return "", false, err - } - ok, err = c.rdb.SetNX(ctx, key, token, ttl).Result() - return token, ok, err -} - -var unlockScript = redis.NewScript(` -if redis.call("GET", KEYS[1]) == ARGV[1] then - return redis.call("DEL", KEYS[1]) -else - return 0 -end -`) - -var incrementWithExpireScript = redis.NewScript(` -local count = redis.call("INCR", KEYS[1]) -if count == 1 then - redis.call("PEXPIRE", KEYS[1], ARGV[1]) -end -return count -`) - -func (c *Client) Unlock(ctx context.Context, key string, token string) error { - if c == nil || c.rdb == nil { - return nil - } - _, err := unlockScript.Run(ctx, c.rdb, []string{key}, token).Result() - return err -} - -func (c *Client) IncrementWithExpire(ctx context.Context, key string, expire time.Duration) (int64, error) { - if c == nil || c.rdb == nil { - return 0, nil - } - return incrementWithExpireScript.Run( - ctx, - c.rdb, - []string{key}, - expire.Milliseconds(), - ).Int64() -} +package redis + +import ( + "context" + "crypto/rand" + "encoding/hex" + "feedsystem_video_go/internal/config" + "fmt" + "strconv" + "time" + + redis "github.com/redis/go-redis/v9" +) + +type Client struct { + rdb *redis.Client + keyPrefix string +} + +const defaultKeyPrefix = "v1:" + +func NewFromEnv(cfg *config.RedisConfig) (*Client, error) { + rdb := redis.NewClient(&redis.Options{ + Addr: cfg.Host + ":" + strconv.Itoa(cfg.Port), + Password: cfg.Password, + DB: cfg.DB, + }) + return &Client{rdb: rdb, keyPrefix: defaultKeyPrefix}, nil +} + +func (c *Client) Close() error { + if c == nil || c.rdb == nil { + return nil + } + return c.rdb.Close() +} + +func (c *Client) Ping(ctx context.Context) error { + if c == nil || c.rdb == nil { + return nil + } + return c.rdb.Ping(ctx).Err() +} + +func IsMiss(err error) bool { + return err == redis.Nil +} + +func (c *Client) Key(format string, args ...any) string { + prefix := "" + if c != nil { + prefix = c.keyPrefix + } + return prefix + fmt.Sprintf(format, args...) +} + +func randToken(n int) (string, error) { + b := make([]byte, n) + if _, err := rand.Read(b); err != nil { + return "", err + } + return hex.EncodeToString(b), nil +} + +func (c *Client) Lock(ctx context.Context, key string, ttl time.Duration) (token string, ok bool, err error) { + if c == nil || c.rdb == nil { + return "", false, nil + } + token, err = randToken(16) + if err != nil { + return "", false, err + } + ok, err = c.rdb.SetNX(ctx, key, token, ttl).Result() + return token, ok, err +} + +var unlockScript = redis.NewScript(` +if redis.call("GET", KEYS[1]) == ARGV[1] then + return redis.call("DEL", KEYS[1]) +else + return 0 +end +`) + +var incrementWithExpireScript = redis.NewScript(` +local count = redis.call("INCR", KEYS[1]) +if count == 1 then + redis.call("PEXPIRE", KEYS[1], ARGV[1]) +end +return count +`) + +func (c *Client) Unlock(ctx context.Context, key string, token string) error { + if c == nil || c.rdb == nil { + return nil + } + _, err := unlockScript.Run(ctx, c.rdb, []string{key}, token).Result() + return err +} + +func (c *Client) IncrementWithExpire(ctx context.Context, key string, expire time.Duration) (int64, error) { + if c == nil || c.rdb == nil { + return 0, nil + } + return incrementWithExpireScript.Run( + ctx, + c.rdb, + []string{key}, + expire.Milliseconds(), + ).Int64() +} diff --git a/backend/internal/video/popularity_cache.go b/backend/internal/video/popularity_cache.go index 0117bc0..62b4dc5 100644 --- a/backend/internal/video/popularity_cache.go +++ b/backend/internal/video/popularity_cache.go @@ -1,29 +1,28 @@ -package video - -import ( - "context" - "fmt" - "strconv" - "time" - - rediscache "feedsystem_video_go/internal/middleware/redis" -) - -// 更新视频流行度缓存 -func UpdatePopularityCache(ctx context.Context, cache *rediscache.Client, id uint, change int64) { - if cache == nil || id == 0 || change == 0 { - return - } - - _ = cache.Del(context.Background(), fmt.Sprintf("video:detail:id=%d", id)) - - 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() - - _ = cache.ZincrBy(opCtx, windowKey, member, float64(change)) - _ = cache.Expire(opCtx, windowKey, 2*time.Hour) -} +package video + +import ( + "context" + "strconv" + "time" + + rediscache "feedsystem_video_go/internal/middleware/redis" +) + +// 更新视频流行度缓存 +func UpdatePopularityCache(ctx context.Context, cache *rediscache.Client, id uint, change int64) { + if cache == nil || id == 0 || change == 0 { + return + } + + _ = cache.Del(context.Background(), cache.Key("video:detail:id=%d", id)) + + now := time.Now().UTC().Truncate(time.Minute) + windowKey := cache.Key("hot:video:1m:%s", now.Format("200601021504")) + member := strconv.FormatUint(uint64(id), 10) + + opCtx, cancel := context.WithTimeout(ctx, 50*time.Millisecond) + defer cancel() + + _ = cache.ZincrBy(opCtx, windowKey, member, float64(change)) + _ = cache.Expire(opCtx, windowKey, 2*time.Hour) +} diff --git a/backend/internal/video/video_service.go b/backend/internal/video/video_service.go index 756cc47..85df436 100644 --- a/backend/internal/video/video_service.go +++ b/backend/internal/video/video_service.go @@ -4,7 +4,6 @@ import ( "context" "encoding/json" "errors" - "fmt" "strconv" "strings" "time" @@ -83,7 +82,7 @@ func (vs *VideoService) Delete(ctx context.Context, id uint, authorID uint) erro return err } if vs.cache != nil { - cacheKey := fmt.Sprintf("video:detail:id=%d", id) + cacheKey := vs.cache.Key("video:detail:id=%d", id) _ = vs.cache.Del(context.Background(), cacheKey) } return nil @@ -98,7 +97,7 @@ func (vs *VideoService) ListByAuthorID(ctx context.Context, authorID uint) ([]Vi } func (vs *VideoService) GetDetail(ctx context.Context, id uint) (*Video, error) { - cacheKey := fmt.Sprintf("video:detail:id=%d", id) + cacheKey := vs.cache.Key("video:detail:id=%d", id) getCached := func() (*Video, bool) { opCtx, cancel := context.WithTimeout(ctx, 50*time.Millisecond) @@ -204,11 +203,11 @@ func (vs *VideoService) UpdatePopularity(ctx context.Context, id uint, change in if vs.cache != nil { // 1) 详情缓存：直接失效（最简单靠谱） - _ = vs.cache.Del(context.Background(), fmt.Sprintf("video:detail:id=%d", id)) + _ = vs.cache.Del(context.Background(), vs.cache.Key("video:detail:id=%d", id)) // 2) 热榜：写到“时间窗ZSET”，不要用 detail key now := time.Now().UTC().Truncate(time.Minute) - windowKey := "hot:video:1m:" + now.Format("200601021504") + windowKey := vs.cache.Key("hot:video:1m:%s", now.Format("200601021504")) member := strconv.FormatUint(uint64(id), 10) opCtx, cancel := context.WithTimeout(ctx, 50*time.Millisecond) diff --git a/backend/internal/worker/outboxworker.go b/backend/internal/worker/outboxworker.go index ff62317..1a45b20 100644 --- a/backend/internal/worker/outboxworker.go +++ b/backend/internal/worker/outboxworker.go @@ -1,93 +1,93 @@ -package worker - -import ( - "context" - "encoding/json" - "feedsystem_video_go/internal/middleware/rabbitmq" - "feedsystem_video_go/internal/middleware/redis" - "feedsystem_video_go/internal/video" - "fmt" - "log" - "time" - - oredis "github.com/redis/go-redis/v9" - "gorm.io/gorm" -) - -func StartOutboxPoller(db *gorm.DB, tmq *rabbitmq.TimelineMQ) { - go func() { - for { - var messages []video.OutboxMsg - - err := db.Where("status = ?", "pending").Order("create_time ASC").Limit(100).Find(&messages).Error - - if err != nil || len(messages) == 0 { - time.Sleep(1 * time.Second) - continue - } - - for _, msg := range messages { - err := tmq.PublishVideo(context.Background(), msg.VideoID, msg.CreateTime) - - if err == nil { - db.Delete(&msg) - } else { - log.Printf("投递MQ失败: VideoID: %d, err: %v", msg.VideoID, err) - } - } - } - }() -} - -func StartConsumer(tmq *rabbitmq.TimelineMQ, queueName string, redisClient *redis.Client) { - msgs, err := tmq.Ch.Consume( - queueName, - "", - false, - false, - false, - false, - nil, - ) - - if err != nil { - log.Printf("注册消费失败") - return - } - - go func() { - for msg := range msgs { - var event rabbitmq.TimelineEvent - err := json.Unmarshal(msg.Body, &event) - - if err != nil { - log.Printf("反序列化失败") - msg.Ack(false) - continue - } - - ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond) - timelineKey := "feed:global_timeline" - err = redisClient.ZAdd(ctx, timelineKey, oredis.Z{ - Score: float64(event.CreateTime), - Member: fmt.Sprintf("%d", event.VideoID), - }) - - if err != nil { - log.Printf("写入Zset失败") - msg.Nack(false, true) - cancel() - continue - } - - err = redisClient.ZRemRangeByRank(ctx, timelineKey, 0, -1001) - - if err != nil { - log.Printf("ZRem失败") - } - - msg.Ack(false) - cancel() - } - }() -} +package worker + +import ( + "context" + "encoding/json" + "feedsystem_video_go/internal/middleware/rabbitmq" + "feedsystem_video_go/internal/middleware/redis" + "feedsystem_video_go/internal/video" + "fmt" + "log" + "time" + + oredis "github.com/redis/go-redis/v9" + "gorm.io/gorm" +) + +func StartOutboxPoller(db *gorm.DB, tmq *rabbitmq.TimelineMQ) { + go func() { + for { + var messages []video.OutboxMsg + + err := db.Where("status = ?", "pending").Order("create_time ASC").Limit(100).Find(&messages).Error + + if err != nil || len(messages) == 0 { + time.Sleep(1 * time.Second) + continue + } + + for _, msg := range messages { + err := tmq.PublishVideo(context.Background(), msg.VideoID, msg.CreateTime) + + if err == nil { + db.Delete(&msg) + } else { + log.Printf("投递MQ失败: VideoID: %d, err: %v", msg.VideoID, err) + } + } + } + }() +} + +func StartConsumer(tmq *rabbitmq.TimelineMQ, queueName string, redisClient *redis.Client) { + msgs, err := tmq.Ch.Consume( + queueName, + "", + false, + false, + false, + false, + nil, + ) + + if err != nil { + log.Printf("注册消费失败") + return + } + + go func() { + for msg := range msgs { + var event rabbitmq.TimelineEvent + err := json.Unmarshal(msg.Body, &event) + + if err != nil { + log.Printf("反序列化失败") + msg.Ack(false) + continue + } + + ctx, cancel := context.WithTimeout(context.Background(), 50*time.Millisecond) + timelineKey := redisClient.Key("feed:global_timeline") + err = redisClient.ZAdd(ctx, timelineKey, oredis.Z{ + Score: float64(event.CreateTime), + Member: fmt.Sprintf("%d", event.VideoID), + }) + + if err != nil { + log.Printf("写入Zset失败") + msg.Nack(false, true) + cancel() + continue + } + + err = redisClient.ZRemRangeByRank(ctx, timelineKey, 0, -1001) + + if err != nil { + log.Printf("ZRem失败") + } + + msg.Ack(false) + cancel() + } + }() +}