From cf83e117788ad89c0b37a549511e7fec44759071 Mon Sep 17 00:00:00 2001 From: Amnesia <7299924@qq.com> Date: Fri, 13 Mar 2026 15:32:48 +0800 Subject: [PATCH 1/8] =?UTF-8?q?feat:=E5=AE=8C=E5=96=84redis=E5=B0=81?= =?UTF-8?q?=E8=A3=85?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/internal/middleware/redis/cache.go | 6 ++++++ backend/internal/middleware/redis/zset.go | 22 ++++++++++++++++++++++ 2 files changed, 28 insertions(+) diff --git a/backend/internal/middleware/redis/cache.go b/backend/internal/middleware/redis/cache.go index 71aed68..d8bae7b 100644 --- a/backend/internal/middleware/redis/cache.go +++ b/backend/internal/middleware/redis/cache.go @@ -3,6 +3,8 @@ package redis import ( "context" "time" + + "github.com/redis/go-redis/v9" ) func (c *Client) GetBytes(ctx context.Context, key string) ([]byte, error) { @@ -16,3 +18,7 @@ func (c *Client) SetBytes(ctx context.Context, key string, value []byte, ttl tim func (c *Client) Del(ctx context.Context, key string) error { return c.rdb.Del(ctx, key).Err() } + +func (c *Client) MGet(cacheCtx context.Context, cacheKeys ...string) *redis.SliceCmd { + return c.rdb.MGet(cacheCtx, cacheKeys...) +} diff --git a/backend/internal/middleware/redis/zset.go b/backend/internal/middleware/redis/zset.go index 9a4b6b4..6ec69eb 100644 --- a/backend/internal/middleware/redis/zset.go +++ b/backend/internal/middleware/redis/zset.go @@ -2,6 +2,7 @@ package redis import ( "context" + "errors" "time" redis "github.com/redis/go-redis/v9" @@ -14,6 +15,27 @@ func (c *Client) ZincrBy(ctx context.Context, key string, member string, score f return c.rdb.ZIncrBy(ctx, key, score, member).Err() } +func (c *Client) ZAdd(ctx context.Context, key string, members ...redis.Z) error { + if c == nil || c.rdb == nil { + return nil + } + return c.rdb.ZAdd(ctx, key, members...).Err() +} + +func (c *Client) ZRemRangeByRank(ctx context.Context, key string, start int64, stop int64) error { + if c == nil || c.rdb == nil { + return nil + } + return c.rdb.ZRemRangeByRank(ctx, key, start, stop).Err() +} + +func (c *Client) ZRangeWithScores(ctx context.Context, key string, start int64, stop int64) ([]redis.Z, error) { + if c == nil || c.rdb == nil { + return nil, errors.New("redis client not initialized") + } + return c.rdb.ZRangeWithScores(ctx, key, start, stop).Result() +} + func (c *Client) Expire(ctx context.Context, key string, ttl time.Duration) error { if c == nil || c.rdb == nil { return nil From 7be6e8bdb5caa68c390c852feb165cae2a594bcc Mon Sep 17 00:00:00 2001 From: Amnesia <7299924@qq.com> Date: Fri, 13 Mar 2026 16:00:17 +0800 Subject: [PATCH 2/8] =?UTF-8?q?feat:=E6=97=B6=E9=97=B4=E7=BA=BF=E7=9B=B8?= =?UTF-8?q?=E5=85=B3rabbitMQ?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../internal/middleware/rabbitmq/rabbitMQ.go | 28 +++++----- .../middleware/rabbitmq/timelineMQ.go | 55 +++++++++++++++++++ backend/internal/video/video_entity.go | 8 +++ 3 files changed, 77 insertions(+), 14 deletions(-) create mode 100644 backend/internal/middleware/rabbitmq/timelineMQ.go diff --git a/backend/internal/middleware/rabbitmq/rabbitMQ.go b/backend/internal/middleware/rabbitmq/rabbitMQ.go index 4dd0d1a..5647835 100644 --- a/backend/internal/middleware/rabbitmq/rabbitMQ.go +++ b/backend/internal/middleware/rabbitmq/rabbitMQ.go @@ -14,8 +14,8 @@ import ( ) type RabbitMQ struct { - conn *amqp.Connection - ch *amqp.Channel + Conn *amqp.Connection + Ch *amqp.Channel } func NewRabbitMQ(cfg *config.RabbitMQConfig) (*RabbitMQ, error) { @@ -23,39 +23,39 @@ func NewRabbitMQ(cfg *config.RabbitMQConfig) (*RabbitMQ, error) { return nil, errors.New("rabbitmq config is nil") } url := "amqp://" + cfg.Username + ":" + cfg.Password + "@" + cfg.Host + ":" + strconv.Itoa(cfg.Port) + "/" - conn, err := amqp.Dial(url) + Conn, err := amqp.Dial(url) if err != nil { return nil, err } - ch, err := conn.Channel() + Ch, err := Conn.Channel() if err != nil { return nil, err } - return &RabbitMQ{conn: conn, ch: ch}, nil + return &RabbitMQ{Conn: Conn, Ch: Ch}, nil } func (r *RabbitMQ) Close() error { - if r == nil || r.ch == nil || r.conn == nil { + if r == nil || r.Ch == nil || r.Conn == nil { return nil } - if err := r.ch.Close(); err != nil { + if err := r.Ch.Close(); err != nil { return err } - if err := r.conn.Close(); err != nil { + if err := r.Conn.Close(); err != nil { return err } return nil } func (r *RabbitMQ) DeclareTopic(exchange string, queue string, bindingKey string) error { - if r == nil || r.ch == nil { + if r == nil || r.Ch == nil { return errors.New("rabbitmq is not initialized") } if exchange == "" || queue == "" || bindingKey == "" { return errors.New("exchange/queue/bindingKey is required") } - if err := r.ch.ExchangeDeclare( + if err := r.Ch.ExchangeDeclare( exchange, "topic", true, @@ -67,7 +67,7 @@ func (r *RabbitMQ) DeclareTopic(exchange string, queue string, bindingKey string return err } - q, err := r.ch.QueueDeclare( + q, err := r.Ch.QueueDeclare( queue, true, false, @@ -79,7 +79,7 @@ func (r *RabbitMQ) DeclareTopic(exchange string, queue string, bindingKey string return err } - return r.ch.QueueBind( + return r.Ch.QueueBind( q.Name, bindingKey, exchange, @@ -89,7 +89,7 @@ func (r *RabbitMQ) DeclareTopic(exchange string, queue string, bindingKey string } func (r *RabbitMQ) PublishJSON(ctx context.Context, exchange string, routingKey string, payload any) error { - if r == nil || r.ch == nil { + if r == nil || r.Ch == nil { return errors.New("rabbitmq is not initialized") } if exchange == "" || routingKey == "" { @@ -99,7 +99,7 @@ func (r *RabbitMQ) PublishJSON(ctx context.Context, exchange string, routingKey if err != nil { return err } - return r.ch.PublishWithContext(ctx, exchange, routingKey, false, false, amqp.Publishing{ + return r.Ch.PublishWithContext(ctx, exchange, routingKey, false, false, amqp.Publishing{ ContentType: "application/json", DeliveryMode: amqp.Persistent, Timestamp: time.Now(), diff --git a/backend/internal/middleware/rabbitmq/timelineMQ.go b/backend/internal/middleware/rabbitmq/timelineMQ.go new file mode 100644 index 0000000..f3d1fdc --- /dev/null +++ b/backend/internal/middleware/rabbitmq/timelineMQ.go @@ -0,0 +1,55 @@ +package rabbitmq + +import ( + "context" + "errors" + "time" +) + +type TimelineMQ struct { + *RabbitMQ +} + +const ( + timelineExchange = "video.timeline.events" + timelineQueue = "video.timeline.update.queue" + timelineBindingKey = "video.timeline.*" + timelinePublishRK = "video.timeline.publish" +) + +type TimelineEvent struct { + EventID string `json:"event_id"` + ViedoID uint `json:"video_id"` + CreateTime int64 `json:"create_time"` + OccurredAt time.Time `json:"occurred_at"` +} + +func NewTimelineMQ(base *RabbitMQ) (*TimelineMQ, error) { + if base == nil { + return nil, errors.New("rabbitmq base is nil") + } + if err := base.DeclareTopic(timelineExchange, timelineQueue, timelineBindingKey); err != nil { + return nil, err + } + return &TimelineMQ{RabbitMQ: base}, nil +} + +func (t *TimelineMQ) PublishVideo(ctx context.Context, videoID uint, createTime time.Time) error { + if t == nil || t.RabbitMQ == nil { + return errors.New("like mq is not initialized") + } + if videoID == 0 { + return errors.New("videoID are required") + } + id, err := newEventID(16) + if err != nil { + return err + } + timeline := TimelineEvent{ + EventID: id, + ViedoID: videoID, + CreateTime: createTime.UnixMilli(), + OccurredAt: time.Now(), + } + return t.PublishJSON(ctx, timelineExchange, timelinePublishRK, timeline) +} diff --git a/backend/internal/video/video_entity.go b/backend/internal/video/video_entity.go index 9259540..9487427 100644 --- a/backend/internal/video/video_entity.go +++ b/backend/internal/video/video_entity.go @@ -38,3 +38,11 @@ type UpdateLikesCountRequest struct { ID uint `json:"id"` LikesCount int64 `json:"likes_count"` } + +type OutboxMsg struct { + ID uint `gorm:"primaryKey"` + VideoID uint `gorm:"index"` + EventType string `gorm:"type:varchar(50)"` + CreateTime time.Time `gorm:"autoCreateTime"` + Status string `gorm:"type:varchar(50);index"` +} From 0ef79f54cbb2bbca78105cc0e6984dadbce41dd4 Mon Sep 17 00:00:00 2001 From: Amnesia <7299924@qq.com> Date: Fri, 13 Mar 2026 16:07:34 +0800 Subject: [PATCH 3/8] =?UTF-8?q?feat:=E6=97=B6=E9=97=B4=E7=BA=BFMQ=E5=8F=91?= =?UTF-8?q?=E9=80=81=E7=AB=AF=E5=92=8C=E6=B6=88=E8=B4=B9=E7=AB=AF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/internal/db/db.go | 2 +- backend/internal/http/router.go | 5 ++ backend/internal/worker/outboxworker.go | 92 +++++++++++++++++++++++++ 3 files changed, 98 insertions(+), 1 deletion(-) create mode 100644 backend/internal/worker/outboxworker.go diff --git a/backend/internal/db/db.go b/backend/internal/db/db.go index f082f97..23593eb 100644 --- a/backend/internal/db/db.go +++ b/backend/internal/db/db.go @@ -24,7 +24,7 @@ func NewDB(dbcfg config.DatabaseConfig) (*gorm.DB, error) { } func AutoMigrate(db *gorm.DB) error { - return db.AutoMigrate(&account.Account{}, &video.Video{}, &video.Like{}, &video.Comment{}, &social.Social{}) + return db.AutoMigrate(&account.Account{}, &video.Video{}, &video.Like{}, &video.Comment{}, &social.Social{}, &video.OutboxMsg{}) } func CloseDB(db *gorm.DB) error { diff --git a/backend/internal/http/router.go b/backend/internal/http/router.go index 8581eb6..7f65f84 100644 --- a/backend/internal/http/router.go +++ b/backend/internal/http/router.go @@ -8,6 +8,7 @@ import ( rediscache "feedsystem_video_go/internal/middleware/redis" "feedsystem_video_go/internal/social" "feedsystem_video_go/internal/video" + "feedsystem_video_go/internal/worker" "log" "github.com/gin-gonic/gin" @@ -127,5 +128,9 @@ func SetRouter(db *gorm.DB, cache *rediscache.Client, rmq *rabbitmq.RabbitMQ) *g { protectedFeedGroup.POST("/listByFollowing", feedHandler.ListByFollowing) } + //worker + timelineMQ, err := rabbitmq.NewTimelineMQ(rmq) + worker.StartOutboxPoller(db, timelineMQ) + worker.StartConsumer(timelineMQ, "video.timeline.update.queue", cache) return r } diff --git a/backend/internal/worker/outboxworker.go b/backend/internal/worker/outboxworker.go new file mode 100644 index 0000000..662b275 --- /dev/null +++ b/backend/internal/worker/outboxworker.go @@ -0,0 +1,92 @@ +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("注册消费失败") + } + + 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.ViedoID), + }) + + 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() + } + }() +} From be48ffced5c357ee738c8d78969b8fd81d737d57 Mon Sep 17 00:00:00 2001 From: Amnesia <7299924@qq.com> Date: Fri, 13 Mar 2026 16:15:44 +0800 Subject: [PATCH 4/8] =?UTF-8?q?perf:=E6=94=B9=E5=8F=98=E5=86=99=E5=85=A5?= =?UTF-8?q?=E9=80=BB=E8=BE=91?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/internal/video/video_repo.go | 7 +++++++ backend/internal/video/video_service.go | 28 +++++++++++++++++++++---- 2 files changed, 31 insertions(+), 4 deletions(-) diff --git a/backend/internal/video/video_repo.go b/backend/internal/video/video_repo.go index c3dd663..d8ee81a 100644 --- a/backend/internal/video/video_repo.go +++ b/backend/internal/video/video_repo.go @@ -22,6 +22,13 @@ func (vr *VideoRepository) CreateVideo(ctx context.Context, video *Video) error return nil } +func (vr *VideoRepository) CreateMsg(ctx context.Context, Msg *OutboxMsg) error { + if err := vr.db.WithContext(ctx).Create(Msg).Error; err != nil { + return err + } + return nil +} + func (vr *VideoRepository) DeleteVideo(ctx context.Context, id uint) error { if err := vr.db.WithContext(ctx).Delete(&Video{}, id).Error; err != nil { return err diff --git a/backend/internal/video/video_service.go b/backend/internal/video/video_service.go index af03ef3..b9be9ef 100644 --- a/backend/internal/video/video_service.go +++ b/backend/internal/video/video_service.go @@ -11,6 +11,8 @@ import ( "feedsystem_video_go/internal/middleware/rabbitmq" rediscache "feedsystem_video_go/internal/middleware/redis" + + "gorm.io/gorm" ) type VideoService struct { @@ -41,10 +43,28 @@ func (vs *VideoService) Publish(ctx context.Context, video *Video) error { if video.CoverURL == "" { return errors.New("cover url is required") } - if err := vs.repo.CreateVideo(ctx, video); err != nil { - return err - } - return nil + + //事务保证视频写入库和消息写入本地消息表的一致性 + 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 { From e3af802784b0a96fb43cf369cfca023933e07ee8 Mon Sep 17 00:00:00 2001 From: Amnesia <7299924@qq.com> Date: Fri, 13 Mar 2026 18:10:46 +0800 Subject: [PATCH 5/8] =?UTF-8?q?perf:=E4=BC=98=E5=8C=96=E4=BA=86ListLatest?= =?UTF-8?q?=E5=9C=A8=E5=A4=A7=E8=A7=84=E6=A8=A1=E8=A7=86=E9=A2=91=E6=95=B0?= =?UTF-8?q?=E6=8D=AE=E4=B8=8B=E7=9A=84=E5=93=8D=E5=BA=94=E6=97=B6=E9=97=B4?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/go.mod | 1 + backend/go.sum | 2 + backend/internal/feed/service.go | 372 +++++++++++++++++++++++-------- 3 files changed, 279 insertions(+), 96 deletions(-) diff --git a/backend/go.mod b/backend/go.mod index f8d5d70..fd2b43e 100644 --- a/backend/go.mod +++ b/backend/go.mod @@ -36,6 +36,7 @@ require ( github.com/mattn/go-isatty v0.0.20 // indirect github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421 // indirect github.com/modern-go/reflect2 v1.0.2 // indirect + github.com/patrickmn/go-cache v2.1.0+incompatible github.com/pelletier/go-toml/v2 v2.2.4 // indirect github.com/quic-go/qpack v0.5.1 // indirect github.com/quic-go/quic-go v0.54.0 // indirect diff --git a/backend/go.sum b/backend/go.sum index a94cb7d..8063d6f 100644 --- a/backend/go.sum +++ b/backend/go.sum @@ -58,6 +58,8 @@ github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421 h1:ZqeYNhU3OH github.com/modern-go/concurrent v0.0.0-20180228061459-e0a39a4cb421/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q= github.com/modern-go/reflect2 v1.0.2 h1:xBagoLtFs94CBntxluKeaWgTMpvLxC4ur3nMaC9Gz0M= github.com/modern-go/reflect2 v1.0.2/go.mod h1:yWuevngMOJpCy52FWWMvUC8ws7m/LJsjYzDa0/r8luk= +github.com/patrickmn/go-cache v2.1.0+incompatible h1:HRMgzkcYKYpi3C8ajMPV8OFXaaRUnok+kx1WdO15EQc= +github.com/patrickmn/go-cache v2.1.0+incompatible/go.mod h1:3Qf8kWWT7OJRJbdiICTKqZju1ZixQ/KpMGzzAfe6+WQ= github.com/pelletier/go-toml/v2 v2.2.4 h1:mye9XuhQ6gvn5h28+VilKrrPoQVanw5PMw/TB0t5Ec4= github.com/pelletier/go-toml/v2 v2.2.4/go.mod h1:2gIqNv+qfxSVS7cM2xJQKtLSTLUE9V8t9Stt+h56mCY= github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM= diff --git a/backend/internal/feed/service.go b/backend/internal/feed/service.go index 6577138..e6da21b 100644 --- a/backend/internal/feed/service.go +++ b/backend/internal/feed/service.go @@ -6,113 +6,285 @@ import ( rediscache "feedsystem_video_go/internal/middleware/redis" "feedsystem_video_go/internal/video" "fmt" + "log" "strconv" + "sync" "time" + + "github.com/patrickmn/go-cache" + redis "github.com/redis/go-redis/v9" + "golang.org/x/sync/singleflight" ) type FeedService struct { - repo *FeedRepository - likeRepo *video.LikeRepository - cache *rediscache.Client - cacheTTL time.Duration + repo *FeedRepository + likeRepo *video.LikeRepository + rediscache *rediscache.Client + localcache *cache.Cache + cacheTTL time.Duration + requestGroup singleflight.Group } -func NewFeedService(repo *FeedRepository, likeRepo *video.LikeRepository, cache *rediscache.Client) *FeedService { - return &FeedService{repo: repo, likeRepo: likeRepo, cache: cache, cacheTTL: 5 * time.Second} +type CachedFeedData struct { + PublicVideos []video.Video `json:"pubilc_videos"` } -// 查询最新视频 -func (f *FeedService) ListLatest(ctx context.Context, limit int, latestBefore time.Time, viewerAccountID uint) (ListLatestResponse, error) { - // 从数据库中查询最新视频 - doListLatestFromDB := func() (ListLatestResponse, error) { - videos, err := f.repo.ListLatest(ctx, limit, latestBefore) - if err != nil { - return ListLatestResponse{}, err - } - var nextTime int64 - if len(videos) > 0 { - nextTime = videos[len(videos)-1].CreateTime.Unix() - } else { - nextTime = 0 - } - hasMore := len(videos) == limit - feedVideos, err := f.buildFeedVideos(ctx, videos, viewerAccountID) - if err != nil { - return ListLatestResponse{}, err - } - resp := ListLatestResponse{ - VideoList: feedVideos, - NextTime: nextTime, - HasMore: hasMore, - } - return resp, nil +func NewFeedService(repo *FeedRepository, likeRepo *video.LikeRepository, rediscache *rediscache.Client) *FeedService { + return &FeedService{repo: repo, likeRepo: likeRepo, rediscache: rediscache, localcache: cache.New(3*time.Second, 5*time.Second), cacheTTL: 24 * time.Hour} +} + +func (f *FeedService) GetVideoByIDs(ctx context.Context, videoIDs []uint) ([]*video.Video, error) { + // GetVideoByIDs 批量获取视频信息 + // 采用 L1(本地缓存) -> L2(Redis) -> L3(MySQL) 三级架构 + if len(videoIDs) == 0 { + return []*video.Video{}, nil } - // 先从缓存中查询 - var cacheKey string - if viewerAccountID == 0 && f.cache != nil { - before := int64(0) - if !latestBefore.IsZero() { - before = latestBefore.Unix() + + videoMap := make(map[uint]*video.Video) + //L1:本地缓存 + var missedL1 []uint + for _, id := range videoIDs { + cacheKey := fmt.Sprintf("video:entity:%d", id) + if f.localcache != nil { + if v, found := f.localcache.Get(cacheKey); found { + if data, ok := v.(video.Video); ok { + videoMap[id] = &data + continue + } + } + } + // 记录未命中的 ID，准备进入下一级缓存 + missedL1 = append(missedL1, id) + } + + if len(missedL1) == 0 { + return buildOrderedResult(videoIDs, videoMap), nil + } + + //L2:redis + var missedL2 []uint + if len(missedL1) > 0 { + cacheKeys := make([]string, len(missedL1)) + for i, id := range missedL1 { + cacheKeys[i] = fmt.Sprintf("video:entity:%d", id) } - cacheKey = fmt.Sprintf("feed:listLatest:limit=%d:before=%d", limit, before) cacheCtx, cancel := context.WithTimeout(ctx, 50*time.Millisecond) - defer cancel() + results, err := f.rediscache.MGet(cacheCtx, cacheKeys...).Result() + cancel() - b, err := f.cache.GetBytes(cacheCtx, cacheKey) if err == nil { - var cached ListLatestResponse - if err := json.Unmarshal(b, &cached); err == nil { - return cached, nil - } - } else if rediscache.IsMiss(err) { // 缓存未命中 - lockKey := "lock:" + cacheKey - // 缓存未命中，尝试加锁 - token, locked, _ := f.cache.Lock(cacheCtx, lockKey, 500*time.Millisecond) - if locked { - defer func() { _ = f.cache.Unlock(context.Background(), lockKey, token) }() - if b, err := f.cache.GetBytes(cacheCtx, cacheKey); err == nil { - var cached ListLatestResponse - if err := json.Unmarshal(b, &cached); err == nil { - return cached, nil - } - } else { // 缓存未命中，从数据库中查询 - resp, err := doListLatestFromDB() - if err != nil { - return ListLatestResponse{}, err - } - if b, err := json.Marshal(resp); err == nil { - _ = f.cache.SetBytes(cacheCtx, cacheKey, b, f.cacheTTL) - } - return resp, nil - } - } else { // 缓存未命中，其他goroutine正在查询，等待 - for i := 0; i < 5; i++ { - time.Sleep(20 * time.Millisecond) - if b, err := f.cache.GetBytes(cacheCtx, cacheKey); err == nil { - var cached ListLatestResponse - if err := json.Unmarshal(b, &cached); err == nil { - return cached, nil + for i, res := range results { + id := missedL1[i] + if res != nil { + if str, ok := res.(string); ok { + var v video.Video + if err := json.Unmarshal([]byte(str), &v); err == nil { + videoMap[id] = &v + // 回写更新 L1 本地缓存 + if f.localcache != nil { + f.localcache.Set(cacheKeys[i], v, 5*time.Second) + } + continue } } } + missedL2 = append(missedL2, id) + } + } else { + // 如果 Redis 挂了或者超时了，全部降级到 L3 + missedL2 = missedL1 + log.Printf("L2 Redis MGet 失败，全部降级到 MySQL: %v", err) + } + } + + if len(missedL2) == 0 { + return buildOrderedResult(videoIDs, videoMap), nil + } + + //L3:MySQL + var wg sync.WaitGroup + var mu sync.Mutex + for _, id := range videoIDs { + wg.Add(1) + go func(videoID uint) { + defer wg.Done() + sfKey := fmt.Sprintf("sf:entity:%d", videoID) + + v, err, _ := f.requestGroup.Do(sfKey, func() (interface{}, error) { + videoList, err := f.repo.GetByIDs(ctx, []uint{videoID}) + + if err != nil { + return nil, err + } + + safeCopy := *videoList[0] + cachekey := fmt.Sprintf("video:entity:%d", safeCopy.ID) + if b, err := json.Marshal(safeCopy); err == nil { + //异步回写redis + go func(k string, b []byte) { + setCtx, setCancel := context.WithTimeout(context.Background(), 50*time.Millisecond) + defer setCancel() + + f.rediscache.SetBytes(setCtx, k, b, time.Hour) + }(cachekey, b) + } + return videoList[0], err + }) + + if err == nil && v != nil { + safeCopy := *(v.(*video.Video)) + mu.Lock() + videoMap[id] = &safeCopy + mu.Unlock() + f.localcache.Set(fmt.Sprintf("video:entity:%d", safeCopy.ID), safeCopy, 5*time.Second) + } + }(id) + } + wg.Wait() + return buildOrderedResult(videoIDs, videoMap), nil +} + +// 查询最新视频 (冷热分离 + 游标分页) +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) + isZsetEmpty := len(zsetTail) == 0 + + if isZsetEmpty { + //全局静态锁：无视所有用户的不同时间戳游标 + sfKey := "sf:fallback:global_timeline_rebuild" + + v, err, _ := f.requestGroup.Do(sfKey, func() (interface{}, error) { + // 无视游标，直接去 MySQL 捞最新的 1000 条 + dbVideos, err := f.repo.ListLatest(ctx, 1000, time.Time{}) + if err != nil { + return nil, err + } + if len(dbVideos) == 0 { + return "EMPTY_DB", nil // 防无限递归 + } + + // 重建 ZSET + bgCtx, cancel := context.WithTimeout(context.Background(), 2*time.Second) + defer cancel() + var zElements []redis.Z + for _, vid := range dbVideos { + zElements = append(zElements, redis.Z{ + Score: float64(vid.CreateTime.UnixMilli()), + Member: fmt.Sprintf("%d", vid.ID), + }) + } + f.rediscache.ZAdd(bgCtx, "feed:global_timeline", zElements...) + return "SUCCESS", nil + }) + + if err != nil { + return ListLatestResponse{}, err + } + if v == "EMPTY_DB" { + return ListLatestResponse{HasMore: false}, nil + } + + // 让所有被阻塞的请求重新查一遍 + return f.ListLatest(ctx, limit, latestBefore, viewerAccountID) + } + + watermark := int64(zsetTail[0].Score) + reqTime := time.Now().UnixMilli() + if !latestBefore.IsZero() { + reqTime = latestBefore.UnixMilli() + } + + var baseVideos []*video.Video + + if reqTime <= watermark { + //冷数据降级查库 + + // 针对个别用户的防并发（此时可以用时间戳做锁，因为冷尾流量极小） + sfKey := fmt.Sprintf("sf:cold:listLatest:%d:%d", limit, reqTime) + v, err, _ := f.requestGroup.Do(sfKey, func() (interface{}, error) { + return f.repo.ListLatest(ctx, limit, latestBefore) + }) + if err != nil { + return ListLatestResponse{}, err + } + baseVideos = v.([]*video.Video) + // 不回写 ZSET，防止冷数据污染热点时间线 + + } else { + // 热数据直接查redis + maxScore := "+inf" + if !latestBefore.IsZero() { + maxScore = fmt.Sprintf("%d", reqTime-1) // 防重复 + } + + videoIDsStr, err := f.rediscache.ZRevRangeByScore(ctx, "feed:global_timeline", maxScore, "-inf", 0, int64(limit)) + if err != nil { + return ListLatestResponse{}, err + } + + var videoIDs []uint + for _, idStr := range videoIDsStr { + if id, err := strconv.ParseUint(idStr, 10, 64); err == nil { + videoIDs = append(videoIDs, uint(id)) + } + } + + if len(videoIDs) > 0 { + baseVideos, err = f.GetVideoByIDs(ctx, videoIDs) + if err != nil { + return ListLatestResponse{}, err + } + } + + // 刚好击穿了冷热边界 + if len(baseVideos) < limit { + remainLimit := limit - len(baseVideos) // 计算还差几个 + + var coldCursor time.Time + if len(baseVideos) > 0 { + coldCursor = baseVideos[len(baseVideos)-1].CreateTime + } else { + coldCursor = latestBefore + } + + sfKey := fmt.Sprintf("sf:stitch:listLatest:%d:%d", remainLimit, coldCursor.UnixMilli()) + v, err, _ := f.requestGroup.Do(sfKey, func() (interface{}, error) { + return f.repo.ListLatest(ctx, remainLimit, coldCursor) + }) + + if err == nil { + coldVideos := v.([]*video.Video) + baseVideos = append(baseVideos, coldVideos...) } } } - // 缓存中没有查询到结果，从数据库中查询 - resp, err := doListLatestFromDB() + + var nextTime int64 + if len(baseVideos) > 0 { + // 将本页最后一条视频的时间作为下一次请求的游标 + nextTime = baseVideos[len(baseVideos)-1].CreateTime.UnixMilli() + } + var hasMore bool + + if reqTime <= watermark { + hasMore = len(baseVideos) == limit + } else { + hasMore = true + } + + feedVideos, err := f.buildFeedVideos(ctx, baseVideos, viewerAccountID) if err != nil { return ListLatestResponse{}, err } - // 缓存查询结果 - if cacheKey != "" { - if b, err := json.Marshal(resp); err == nil { - cacheCtx, cancel := context.WithTimeout(ctx, 50*time.Millisecond) - defer cancel() - _ = f.cache.SetBytes(cacheCtx, cacheKey, b, f.cacheTTL) - } - } - return resp, nil + + return ListLatestResponse{ + VideoList: feedVideos, + NextTime: nextTime, + HasMore: hasMore, + }, nil } // 按照点赞数查询视频 @@ -166,7 +338,7 @@ func (f *FeedService) ListByFollowing(ctx context.Context, limit int, latestBefo return resp, nil } var cacheKey string - if viewerAccountID != 0 && f.cache != nil { + if viewerAccountID != 0 && f.rediscache != nil { before := int64(0) if !latestBefore.IsZero() { before = latestBefore.Unix() @@ -175,7 +347,7 @@ func (f *FeedService) ListByFollowing(ctx context.Context, limit int, latestBefo cacheCtx, cancel := context.WithTimeout(ctx, 50*time.Millisecond) defer cancel() - b, err := f.cache.GetBytes(cacheCtx, cacheKey) + b, err := f.rediscache.GetBytes(cacheCtx, cacheKey) if err == nil { var cached ListByFollowingResponse if err := json.Unmarshal(b, &cached); err == nil { @@ -184,10 +356,10 @@ func (f *FeedService) ListByFollowing(ctx context.Context, limit int, latestBefo } else if rediscache.IsMiss(err) { // 缓存未命中 lockKey := "lock:" + cacheKey // 缓存未命中，尝试加锁 - token, locked, _ := f.cache.Lock(cacheCtx, lockKey, 500*time.Millisecond) + token, locked, _ := f.rediscache.Lock(cacheCtx, lockKey, 500*time.Millisecond) if locked { - defer func() { _ = f.cache.Unlock(context.Background(), lockKey, token) }() - if b, err := f.cache.GetBytes(cacheCtx, cacheKey); err == nil { + defer func() { _ = f.rediscache.Unlock(context.Background(), lockKey, token) }() + if b, err := f.rediscache.GetBytes(cacheCtx, cacheKey); err == nil { var cached ListByFollowingResponse if err := json.Unmarshal(b, &cached); err == nil { return cached, nil @@ -198,14 +370,14 @@ func (f *FeedService) ListByFollowing(ctx context.Context, limit int, latestBefo return ListByFollowingResponse{}, err } if b, err := json.Marshal(resp); err == nil { - _ = f.cache.SetBytes(cacheCtx, cacheKey, b, f.cacheTTL) + _ = f.rediscache.SetBytes(cacheCtx, cacheKey, b, f.cacheTTL) } return resp, nil } } else { for i := 0; i < 5; i++ { time.Sleep(20 * time.Millisecond) - if b, err := f.cache.GetBytes(cacheCtx, cacheKey); err == nil { + if b, err := f.rediscache.GetBytes(cacheCtx, cacheKey); err == nil { var cached ListByFollowingResponse if err := json.Unmarshal(b, &cached); err == nil { return cached, nil @@ -224,7 +396,7 @@ func (f *FeedService) ListByFollowing(ctx context.Context, limit int, latestBefo if b, err := json.Marshal(resp); err == nil { cacheCtx, cancel := context.WithTimeout(ctx, 50*time.Millisecond) defer cancel() - _ = f.cache.SetBytes(cacheCtx, cacheKey, b, f.cacheTTL) + _ = f.rediscache.SetBytes(cacheCtx, cacheKey, b, f.cacheTTL) } } return resp, nil @@ -232,7 +404,7 @@ func (f *FeedService) ListByFollowing(ctx context.Context, limit int, latestBefo func (f *FeedService) ListByPopularity(ctx context.Context, limit int, reqAsOf int64, offset int, viewerAccountID uint, latestPopularity int64, latestBefore time.Time, latestIDBefore uint) (ListByPopularityResponse, error) { // Redis 热榜（稳定分页：as_of + offset） - if f.cache != nil { + if f.rediscache != nil { asOf := time.Now().UTC().Truncate(time.Minute) if reqAsOf > 0 { asOf = time.Unix(reqAsOf, 0).UTC().Truncate(time.Minute) @@ -248,15 +420,15 @@ func (f *FeedService) ListByPopularity(ctx context.Context, limit int, reqAsOf i opCtx, cancel := context.WithTimeout(ctx, 80*time.Millisecond) defer cancel() - exists, _ := f.cache.Exists(opCtx, dest) + exists, _ := f.rediscache.Exists(opCtx, dest) if !exists { - _ = f.cache.ZUnionStore(opCtx, dest, keys, "SUM") - _ = f.cache.Expire(opCtx, dest, 2*time.Minute) // 给翻页留时间 + _ = f.rediscache.ZUnionStore(opCtx, dest, keys, "SUM") + _ = f.rediscache.Expire(opCtx, dest, 2*time.Minute) // 给翻页留时间 } start := int64(offset) stop := start + int64(limit) - 1 - members, err := f.cache.ZRevRange(opCtx, dest, start, stop) + members, err := f.rediscache.ZRevRange(opCtx, dest, start, stop) if err == nil && len(members) == 0 { if offset > 0 { return ListByPopularityResponse{ @@ -363,3 +535,11 @@ func (f *FeedService) buildFeedVideos(ctx context.Context, videos []*video.Video } return feedVideos, nil } + +func buildOrderedResult(orderedIDs []uint, dataMap map[uint]*video.Video) []*video.Video { + var res []*video.Video + for _, id := range orderedIDs { + res = append(res, dataMap[id]) + } + return res +} From 39243c14d7fd26196017e59c3054758f5190d76e Mon Sep 17 00:00:00 2001 From: Amnesia <7299924@qq.com> Date: Sat, 14 Mar 2026 10:05:58 +0800 Subject: [PATCH 6/8] =?UTF-8?q?fix:=E4=BF=AE=E5=A4=8D=E6=97=B6=E9=97=B4?= =?UTF-8?q?=E6=88=B3=E8=BD=AC=E6=8D=A2=E9=94=99=E8=AF=AF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/internal/feed/handler.go | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/backend/internal/feed/handler.go b/backend/internal/feed/handler.go index 588490b..8c4378c 100644 --- a/backend/internal/feed/handler.go +++ b/backend/internal/feed/handler.go @@ -26,7 +26,7 @@ func (f *FeedHandler) ListLatest(c *gin.Context) { } var latestTime time.Time if req.LatestTime > 0 { - latestTime = time.Unix(req.LatestTime, 0) + latestTime = time.UnixMilli(req.LatestTime) } viewerAccountID, err := jwt.GetAccountID(c) if err != nil { From 9eed15efd5213fb1f088115f81eef6f7dc8e4754 Mon Sep 17 00:00:00 2001 From: Amnesia <7299924@qq.com> Date: Sat, 14 Mar 2026 18:57:02 +0800 Subject: [PATCH 7/8] =?UTF-8?q?fix:=E5=A4=84=E7=90=86=E4=BA=86=E8=BE=B9?= =?UTF-8?q?=E7=95=8C=E6=83=85=E5=86=B5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/internal/feed/service.go | 26 ++++++++++--------- backend/internal/http/router.go | 4 +++ .../middleware/rabbitmq/timelineMQ.go | 2 +- backend/internal/middleware/redis/cache.go | 6 ++--- backend/internal/worker/outboxworker.go | 3 ++- 5 files changed, 23 insertions(+), 18 deletions(-) diff --git a/backend/internal/feed/service.go b/backend/internal/feed/service.go index e6da21b..24aac3c 100644 --- a/backend/internal/feed/service.go +++ b/backend/internal/feed/service.go @@ -26,7 +26,7 @@ type FeedService struct { } type CachedFeedData struct { - PublicVideos []video.Video `json:"pubilc_videos"` + PublicVideos []video.Video `json:"public_videos"` } func NewFeedService(repo *FeedRepository, likeRepo *video.LikeRepository, rediscache *rediscache.Client) *FeedService { @@ -70,7 +70,7 @@ func (f *FeedService) GetVideoByIDs(ctx context.Context, videoIDs []uint) ([]*vi } cacheCtx, cancel := context.WithTimeout(ctx, 50*time.Millisecond) - results, err := f.rediscache.MGet(cacheCtx, cacheKeys...).Result() + results, err := f.rediscache.MGet(cacheCtx, cacheKeys...) cancel() if err == nil { @@ -105,7 +105,7 @@ func (f *FeedService) GetVideoByIDs(ctx context.Context, videoIDs []uint) ([]*vi //L3:MySQL var wg sync.WaitGroup var mu sync.Mutex - for _, id := range videoIDs { + for _, id := range missedL2 { wg.Add(1) go func(videoID uint) { defer wg.Done() @@ -114,7 +114,7 @@ func (f *FeedService) GetVideoByIDs(ctx context.Context, videoIDs []uint) ([]*vi v, err, _ := f.requestGroup.Do(sfKey, func() (interface{}, error) { videoList, err := f.repo.GetByIDs(ctx, []uint{videoID}) - if err != nil { + if err != nil || len(videoList) == 0 { return nil, err } @@ -147,9 +147,13 @@ 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) + + if err != nil { + return ListLatestResponse{}, err + } + isZsetEmpty := len(zsetTail) == 0 if isZsetEmpty { @@ -269,11 +273,7 @@ func (f *FeedService) ListLatest(ctx context.Context, limit int, latestBefore ti } var hasMore bool - if reqTime <= watermark { - hasMore = len(baseVideos) == limit - } else { - hasMore = true - } + hasMore = len(baseVideos) == limit feedVideos, err := f.buildFeedVideos(ctx, baseVideos, viewerAccountID) if err != nil { @@ -537,9 +537,11 @@ func (f *FeedService) buildFeedVideos(ctx context.Context, videos []*video.Video } func buildOrderedResult(orderedIDs []uint, dataMap map[uint]*video.Video) []*video.Video { - var res []*video.Video + res := make([]*video.Video, 0, len(orderedIDs)) for _, id := range orderedIDs { - res = append(res, dataMap[id]) + if v, exit := dataMap[id]; exit && v != nil { + res = append(res, v) + } } return res } diff --git a/backend/internal/http/router.go b/backend/internal/http/router.go index 7f65f84..c0a767d 100644 --- a/backend/internal/http/router.go +++ b/backend/internal/http/router.go @@ -130,6 +130,10 @@ func SetRouter(db *gorm.DB, cache *rediscache.Client, rmq *rabbitmq.RabbitMQ) *g } //worker timelineMQ, err := rabbitmq.NewTimelineMQ(rmq) + if err != nil { + log.Printf("timelineMQ init failed (mq disabled): %v", err) + socialMQ = nil + } worker.StartOutboxPoller(db, timelineMQ) worker.StartConsumer(timelineMQ, "video.timeline.update.queue", cache) return r diff --git a/backend/internal/middleware/rabbitmq/timelineMQ.go b/backend/internal/middleware/rabbitmq/timelineMQ.go index f3d1fdc..6060623 100644 --- a/backend/internal/middleware/rabbitmq/timelineMQ.go +++ b/backend/internal/middleware/rabbitmq/timelineMQ.go @@ -36,7 +36,7 @@ func NewTimelineMQ(base *RabbitMQ) (*TimelineMQ, error) { func (t *TimelineMQ) PublishVideo(ctx context.Context, videoID uint, createTime time.Time) error { if t == nil || t.RabbitMQ == nil { - return errors.New("like mq is not initialized") + return errors.New("timelike mq is not initialized") } if videoID == 0 { return errors.New("videoID are required") diff --git a/backend/internal/middleware/redis/cache.go b/backend/internal/middleware/redis/cache.go index d8bae7b..270cda3 100644 --- a/backend/internal/middleware/redis/cache.go +++ b/backend/internal/middleware/redis/cache.go @@ -3,8 +3,6 @@ package redis import ( "context" "time" - - "github.com/redis/go-redis/v9" ) func (c *Client) GetBytes(ctx context.Context, key string) ([]byte, error) { @@ -19,6 +17,6 @@ func (c *Client) Del(ctx context.Context, key string) error { return c.rdb.Del(ctx, key).Err() } -func (c *Client) MGet(cacheCtx context.Context, cacheKeys ...string) *redis.SliceCmd { - return c.rdb.MGet(cacheCtx, cacheKeys...) +func (c *Client) MGet(cacheCtx context.Context, cacheKeys ...string) ([]interface{}, error) { + return c.rdb.MGet(cacheCtx, cacheKeys...).Result() } diff --git a/backend/internal/worker/outboxworker.go b/backend/internal/worker/outboxworker.go index 662b275..ff62317 100644 --- a/backend/internal/worker/outboxworker.go +++ b/backend/internal/worker/outboxworker.go @@ -52,6 +52,7 @@ func StartConsumer(tmq *rabbitmq.TimelineMQ, queueName string, redisClient *redi if err != nil { log.Printf("注册消费失败") + return } go func() { @@ -69,7 +70,7 @@ func StartConsumer(tmq *rabbitmq.TimelineMQ, queueName string, redisClient *redi timelineKey := "feed:global_timeline" err = redisClient.ZAdd(ctx, timelineKey, oredis.Z{ Score: float64(event.CreateTime), - Member: fmt.Sprintf("%d", event.ViedoID), + Member: fmt.Sprintf("%d", event.VideoID), }) if err != nil { From 05c846c295738cea4f7465a10cd6f16f779e4805 Mon Sep 17 00:00:00 2001 From: Amnesia <7299924@qq.com> Date: Sat, 14 Mar 2026 19:56:31 +0800 Subject: [PATCH 8/8] =?UTF-8?q?fix:=E4=BF=AE=E5=A4=8D=E5=91=BD=E5=90=8D?= =?UTF-8?q?=E9=94=99=E8=AF=AF?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/internal/feed/service.go | 2 +- backend/internal/middleware/rabbitmq/rabbitMQ.go | 6 +++--- backend/internal/middleware/rabbitmq/timelineMQ.go | 6 +++--- backend/internal/video/video_service.go | 4 ++-- 4 files changed, 9 insertions(+), 9 deletions(-) diff --git a/backend/internal/feed/service.go b/backend/internal/feed/service.go index 24aac3c..95e6939 100644 --- a/backend/internal/feed/service.go +++ b/backend/internal/feed/service.go @@ -539,7 +539,7 @@ func (f *FeedService) buildFeedVideos(ctx context.Context, videos []*video.Video func buildOrderedResult(orderedIDs []uint, dataMap map[uint]*video.Video) []*video.Video { res := make([]*video.Video, 0, len(orderedIDs)) for _, id := range orderedIDs { - if v, exit := dataMap[id]; exit && v != nil { + if v, exits := dataMap[id]; exits && v != nil { res = append(res, v) } } diff --git a/backend/internal/middleware/rabbitmq/rabbitMQ.go b/backend/internal/middleware/rabbitmq/rabbitMQ.go index 5647835..b4a715e 100644 --- a/backend/internal/middleware/rabbitmq/rabbitMQ.go +++ b/backend/internal/middleware/rabbitmq/rabbitMQ.go @@ -23,15 +23,15 @@ func NewRabbitMQ(cfg *config.RabbitMQConfig) (*RabbitMQ, error) { return nil, errors.New("rabbitmq config is nil") } url := "amqp://" + cfg.Username + ":" + cfg.Password + "@" + cfg.Host + ":" + strconv.Itoa(cfg.Port) + "/" - Conn, err := amqp.Dial(url) + conn, err := amqp.Dial(url) if err != nil { return nil, err } - Ch, err := Conn.Channel() + ch, err := conn.Channel() if err != nil { return nil, err } - return &RabbitMQ{Conn: Conn, Ch: Ch}, nil + return &RabbitMQ{Conn: conn, Ch: ch}, nil } func (r *RabbitMQ) Close() error { diff --git a/backend/internal/middleware/rabbitmq/timelineMQ.go b/backend/internal/middleware/rabbitmq/timelineMQ.go index 6060623..898ed0a 100644 --- a/backend/internal/middleware/rabbitmq/timelineMQ.go +++ b/backend/internal/middleware/rabbitmq/timelineMQ.go @@ -19,7 +19,7 @@ const ( type TimelineEvent struct { EventID string `json:"event_id"` - ViedoID uint `json:"video_id"` + VideoID uint `json:"video_id"` CreateTime int64 `json:"create_time"` OccurredAt time.Time `json:"occurred_at"` } @@ -36,7 +36,7 @@ func NewTimelineMQ(base *RabbitMQ) (*TimelineMQ, error) { func (t *TimelineMQ) PublishVideo(ctx context.Context, videoID uint, createTime time.Time) error { if t == nil || t.RabbitMQ == nil { - return errors.New("timelike mq is not initialized") + return errors.New("timeline mq is not initialized") } if videoID == 0 { return errors.New("videoID are required") @@ -47,7 +47,7 @@ func (t *TimelineMQ) PublishVideo(ctx context.Context, videoID uint, createTime } timeline := TimelineEvent{ EventID: id, - ViedoID: videoID, + VideoID: videoID, CreateTime: createTime.UnixMilli(), OccurredAt: time.Now(), } diff --git a/backend/internal/video/video_service.go b/backend/internal/video/video_service.go index b9be9ef..92eba9a 100644 --- a/backend/internal/video/video_service.go +++ b/backend/internal/video/video_service.go @@ -50,14 +50,14 @@ func (vs *VideoService) Publish(ctx context.Context, video *Video) error { return err } - Msg := OutboxMsg{ + msg := OutboxMsg{ VideoID: video.ID, EventType: "video_published", Status: "pending", CreateTime: video.CreateTime, } - if err := tx.Create(Msg).Error; err != nil { + if err := tx.Create(&msg).Error; err != nil { return err } return nil