04 - Feed 流系统设计

package service

import (
    "database/sql"
)

type FeedService struct {
    db *sql.DB
}

// PublishFeed 发布 Feed（Push 模式）
func (s *FeedService) PublishFeed(userID int64, content string, mediaURLs []string) (string, error) {
    // 1. 生成 Feed ID
    feedID := generateFeedID()

    // 2. 保存 Feed 到 feed 表
    insertFeedSQL := `
        INSERT INTO feed (feed_id, user_id, content, media_urls, created_at)
        VALUES (?, ?, ?, ?, NOW())
    `
    mediaJSON, _ := json.Marshal(mediaURLs)
    if _, err := s.db.Exec(insertFeedSQL, feedID, userID, content, mediaJSON); err != nil {
        return "", err
    }

    // 3. 查询粉丝列表
    followers, err := s.getFollowers(userID)
    if err != nil {
        return "", err
    }

    // 4. 推送到所有粉丝的收件箱（写扩散）
    if err := s.pushToInbox(feedID, userID, followers); err != nil {
        log.Errorf("推送到收件箱失败: %v", err)
        // 不影响 Feed 发布成功，可异步重试
    }

    return feedID, nil
}

// getFollowers 查询粉丝列表
func (s *FeedService) getFollowers(userID int64) ([]int64, error) {
    query := "SELECT follower_id FROM follow_relation WHERE followee_id = ? AND status = 1"
    rows, err := s.db.Query(query, userID)
    if err != nil {
        return nil, err
    }
    defer rows.Close()

    var followers []int64
    for rows.Next() {
        var followerID int64
        rows.Scan(&followerID)
        followers = append(followers, followerID)
    }

    return followers, nil
}

// pushToInbox 推送到粉丝收件箱
func (s *FeedService) pushToInbox(feedID string, authorID int64, followers []int64) error {
    if len(followers) == 0 {
        return nil
    }

    // 批量插入（每次 1000 条）
    batchSize := 1000
    for i := 0; i < len(followers); i += batchSize {
        end := i + batchSize
        if end > len(followers) {
            end = len(followers)
        }

        batch := followers[i:end]
        if err := s.batchInsertInbox(feedID, authorID, batch); err != nil {
            return err
        }
    }

    return nil
}

// batchInsertInbox 批量插入收件箱
func (s *FeedService) batchInsertInbox(feedID string, authorID int64, followers []int64) error {
    // 构造批量插入 SQL
    values := make([]string, len(followers))
    args := make([]interface{}, 0, len(followers)*3)

    for i, followerID := range followers {
        values[i] = "(?, ?, ?)"
        args = append(args, followerID, feedID, authorID)
    }

    sql := fmt.Sprintf(`
        INSERT INTO feed_inbox (user_id, feed_id, author_id, created_at)
        VALUES %s
    `, strings.Join(values, ","))

    // 每条记录的 created_at 使用 NOW()
    sql = strings.ReplaceAll(sql, "?)", "?, NOW())")

    _, err := s.db.Exec(sql, args...)
    return err
}

// GetTimeline 获取用户时间线
func (s *FeedService) GetTimeline(userID int64, cursor string, limit int) ([]*Feed, string, error) {
    // 1. 解析 cursor（格式: timestamp_feedID）
    var lastTimestamp int64
    var lastFeedID string
    if cursor != "" {
        parts := strings.Split(cursor, "_")
        lastTimestamp, _ = strconv.ParseInt(parts[0], 10, 64)
        lastFeedID = parts[1]
    }

    // 2. 查询收件箱
    query := `
        SELECT feed_id, author_id, created_at
        FROM feed_inbox
        WHERE user_id = ?
    `
    args := []interface{}{userID}

    if cursor != "" {
        // 游标分页
        query += " AND (created_at < ? OR (created_at = ? AND feed_id < ?))"
        args = append(args, lastTimestamp, lastTimestamp, lastFeedID)
    }

    query += " ORDER BY created_at DESC, feed_id DESC LIMIT ?"
    args = append(args, limit+1)  // 多查 1 条判断是否还有更多

    rows, err := s.db.Query(query, args...)
    if err != nil {
        return nil, "", err
    }
    defer rows.Close()

    var feedIDs []string
    var inboxItems []struct {
        FeedID    string
        AuthorID  int64
        CreatedAt int64
    }

    for rows.Next() {
        var item struct {
            FeedID    string
            AuthorID  int64
            CreatedAt int64
        }
        rows.Scan(&item.FeedID, &item.AuthorID, &item.CreatedAt)
        feedIDs = append(feedIDs, item.FeedID)
        inboxItems = append(inboxItems, item)
    }

    // 3. 判断是否还有更多
    hasMore := false
    if len(feedIDs) > limit {
        hasMore = true
        feedIDs = feedIDs[:limit]
        inboxItems = inboxItems[:limit]
    }

    // 4. 批量查询 Feed 详情
    feeds, err := s.batchGetFeeds(feedIDs)
    if err != nil {
        return nil, "", err
    }

    // 5. 生成下一页 cursor
    var nextCursor string
    if hasMore && len(inboxItems) > 0 {
        last := inboxItems[len(inboxItems)-1]
        nextCursor = fmt.Sprintf("%d_%s", last.CreatedAt, last.FeedID)
    }

    return feeds, nextCursor, nil
}

// batchGetFeeds 批量查询 Feed 详情
func (s *FeedService) batchGetFeeds(feedIDs []string) ([]*Feed, error) {
    if len(feedIDs) == 0 {
        return []*Feed{}, nil
    }

    placeholders := strings.Repeat("?,", len(feedIDs))
    placeholders = placeholders[:len(placeholders)-1]

    query := fmt.Sprintf(`
        SELECT feed_id, user_id, content, media_urls, created_at
        FROM feed
        WHERE feed_id IN (%s)
    `, placeholders)

    args := make([]interface{}, len(feedIDs))
    for i, id := range feedIDs {
        args[i] = id
    }

    rows, err := s.db.Query(query, args...)
    if err != nil {
        return nil, err
    }
    defer rows.Close()

    feeds := make([]*Feed, 0, len(feedIDs))
    for rows.Next() {
        var feed Feed
        var mediaJSON []byte
        rows.Scan(&feed.FeedID, &feed.UserID, &feed.Content, &mediaJSON, &feed.CreatedAt)
        json.Unmarshal(mediaJSON, &feed.MediaURLs)
        feeds = append(feeds, &feed)
    }

    return feeds, nil
}

package service

const (
    BIG_V_THRESHOLD = 100000  // 粉丝数超过 10 万为大 V
)

// PublishFeed 发布 Feed（智能选择 Push/Pull）
func (s *FeedService) PublishFeed(userID int64, content string, mediaURLs []string) (string, error) {
    // 1. 保存 Feed
    feedID := generateFeedID()
    if err := s.saveFeed(feedID, userID, content, mediaURLs); err != nil {
        return "", err
    }

    // 2. 判断是否为大 V
    followersCount, _ := s.getFollowersCount(userID)

    if followersCount < BIG_V_THRESHOLD {
        // 普通用户: Push 模式
        go s.pushToFollowers(feedID, userID)
    } else {
        // 大 V: 不推送，粉丝读取时实时拉取
        log.Infof("大 V 用户 %d 发布 Feed %s，不推送", userID, feedID)

        // 只推送给活跃粉丝（最近 7 天活跃）
        go s.pushToActiveFollowers(feedID, userID)
    }

    // 3. 缓存到 Redis（发件箱）
    go s.cacheToRedis(userID, feedID)

    return feedID, nil
}

// pushToActiveFollowers 推送给活跃粉丝
func (s *FeedService) pushToActiveFollowers(feedID string, userID int64) {
    // 查询活跃粉丝（最近 7 天有登录）
    query := `
        SELECT fr.follower_id
        FROM follow_relation fr
        JOIN user_activity ua ON fr.follower_id = ua.user_id
        WHERE fr.followee_id = ?
          AND fr.status = 1
          AND ua.last_login > DATE_SUB(NOW(), INTERVAL 7 DAY)
    `
    rows, _ := s.db.Query(query, userID)
    defer rows.Close()

    var activeFollowers []int64
    for rows.Next() {
        var followerID int64
        rows.Scan(&followerID)
        activeFollowers = append(activeFollowers, followerID)
    }

    log.Infof("大 V %d 有 %d 个活跃粉丝", userID, len(activeFollowers))

    // 推送给活跃粉丝
    s.pushToInbox(feedID, userID, activeFollowers)
}

// cacheToRedis 缓存到 Redis 发件箱
func (s *FeedService) cacheToRedis(userID int64, feedID string) {
    key := fmt.Sprintf("feed:outbox:%d", userID)
    timestamp := time.Now().UnixMilli()

    // 添加到 Sorted Set
    s.rdb.ZAdd(context.Background(), key, &redis.Z{
        Score:  float64(timestamp),
        Member: feedID,
    })

    // 只保留最新 100 条
    s.rdb.ZRemRangeByRank(context.Background(), key, 0, -101)
}

// GetTimelineHybrid 混合模式获取时间线
func (s *FeedService) GetTimelineHybrid(userID int64, cursor string, limit int) ([]*Feed, string, error) {
    // 1. 查询用户关注列表
    following, err := s.getFollowing(userID)
    if err != nil {
        return nil, "", err
    }

    // 2. 区分普通用户和大 V
    normalUsers := []int64{}
    bigVs := []int64{}

    for _, followeeID := range following {
        if s.isBigV(followeeID) {
            bigVs = append(bigVs, followeeID)
        } else {
            normalUsers = append(normalUsers, followeeID)
        }
    }

    // 3. 读取收件箱（普通用户的 Feed）
    inboxFeeds, _ := s.getInboxFeeds(userID, limit)

    // 4. 实时拉取大 V 的 Feed
    bigVFeeds, _ := s.pullBigVFeeds(bigVs, limit)

    // 5. 合并排序（多路归并）
    allFeeds := s.mergeFeeds(inboxFeeds, bigVFeeds, limit)

    // 6. 批量查询 Feed 详情
    feeds, err := s.batchGetFeedsWithDetails(allFeeds)

    return feeds, "", nil
}

// pullBigVFeeds 拉取大 V 的最新 Feed
func (s *FeedService) pullBigVFeeds(bigVs []int64, limit int) ([]string, error) {
    if len(bigVs) == 0 {
        return []string{}, nil
    }

    var allFeeds []struct {
        FeedID    string
        CreatedAt int64
    }

    // 并发拉取每个大 V 的最新 Feed
    var wg sync.WaitGroup
    var mu sync.Mutex
    feedChan := make(chan []struct {
        FeedID    string
        CreatedAt int64
    }, len(bigVs))

    for _, bigVID := range bigVs {
        wg.Add(1)
        go func(userID int64) {
            defer wg.Done()

            // 从 Redis 读取大 V 的发件箱（最新 10 条）
            key := fmt.Sprintf("feed:outbox:%d", userID)
            result, err := s.rdb.ZRevRangeWithScores(context.Background(), key, 0, 9).Result()
            if err != nil {
                return
            }

            feeds := make([]struct {
                FeedID    string
                CreatedAt int64
            }, len(result))

            for i, z := range result {
                feeds[i] = struct {
                    FeedID    string
                    CreatedAt int64
                }{
                    FeedID:    z.Member.(string),
                    CreatedAt: int64(z.Score),
                }
            }

            feedChan <- feeds
        }(bigVID)
    }

    go func() {
        wg.Wait()
        close(feedChan)
    }()

    // 收集所有大 V 的 Feed
    for feeds := range feedChan {
        mu.Lock()
        allFeeds = append(allFeeds, feeds...)
        mu.Unlock()
    }

    // 按时间排序
    sort.Slice(allFeeds, func(i, j int) bool {
        return allFeeds[i].CreatedAt > allFeeds[j].CreatedAt
    })

    // 取前 N 条
    if len(allFeeds) > limit {
        allFeeds = allFeeds[:limit]
    }

    feedIDs := make([]string, len(allFeeds))
    for i, feed := range allFeeds {
        feedIDs[i] = feed.FeedID
    }

    return feedIDs, nil
}

// mergeFeeds 合并排序（多路归并）
func (s *FeedService) mergeFeeds(inbox, bigV []string, limit int) []string {
    // 简化版：直接合并并去重
    feedMap := make(map[string]bool)
    result := []string{}

    for _, feedID := range inbox {
        if !feedMap[feedID] {
            result = append(result, feedID)
            feedMap[feedID] = true
        }
    }

    for _, feedID := range bigV {
        if !feedMap[feedID] {
            result = append(result, feedID)
            feedMap[feedID] = true
        }
    }

    if len(result) > limit {
        result = result[:limit]
    }

    return result
}

// isBigV 判断是否为大 V（缓存粉丝数）
func (s *FeedService) isBigV(userID int64) bool {
    key := fmt.Sprintf("user:followers_count:%d", userID)
    count, err := s.rdb.Get(context.Background(), key).Int()
    if err != nil {
        // Redis 未命中，查询 DB
        count, _ = s.getFollowersCount(userID)
        s.rdb.Set(context.Background(), key, count, 1*time.Hour)
    }

    return count >= BIG_V_THRESHOLD
}

// PublishFeed 异步推送
func (s *FeedService) PublishFeed(userID int64, content string) (string, error) {
    // 1. 保存 Feed
    feedID, err := s.saveFeed(userID, content)
    if err != nil {
        return "", err
    }

    // 2. 发送到 Kafka（异步推送）
    event := FeedPublishEvent{
        FeedID:  feedID,
        UserID:  userID,
        Content: content,
    }

    data, _ := json.Marshal(event)
    s.kafka.Publish("feed-publish", data)

    return feedID, nil
}

// Kafka 消费者：推送到粉丝收件箱
type FeedPushConsumer struct {
    db *sql.DB
}

func (c *FeedPushConsumer) Consume(msg *kafka.Message) error {
    var event FeedPublishEvent
    json.Unmarshal(msg.Value, &event)

    // 查询粉丝并推送
    followers, _ := c.getFollowers(event.UserID)

    if len(followers) < BIG_V_THRESHOLD {
        c.pushToInbox(event.FeedID, event.UserID, followers)
    }

    return nil
}

// GetPersonalizedTimeline 个性化时间线
func (s *FeedService) GetPersonalizedTimeline(userID int64, limit int) ([]*Feed, error) {
    // 1. 获取候选 Feed（收件箱 + 大 V 拉取）
    candidateFeeds, _ := s.getCandidateFeeds(userID, limit*3)

    // 2. 调用推荐服务打分排序
    rankedFeeds, _ := s.recommendService.Rank(userID, candidateFeeds)

    // 3. 返回 Top N
    if len(rankedFeeds) > limit {
        rankedFeeds = rankedFeeds[:limit]
    }

    return rankedFeeds, nil
}

// 推荐服务（简化版）
type RecommendService struct{}

func (r *RecommendService) Rank(userID int64, feeds []*Feed) ([]*Feed, error) {
    // 特征提取
    features := r.extractFeatures(userID, feeds)

    // 模型预测（离线训练的模型）
    scores := r.model.Predict(features)

    // 排序
    for i, feed := range feeds {
        feed.Score = scores[i]
    }

    sort.Slice(feeds, func(i, j int) bool {
        return feeds[i].Score > feeds[j].Score
    })

    return feeds, nil
}

func (r *RecommendService) extractFeatures(userID int64, feeds []*Feed) [][]float64 {
    // 特征工程（示例）:
    // - 用户特征: 年龄、性别、兴趣标签
    // - Feed 特征: 作者粉丝数、点赞数、发布时间
    // - 交互特征: 用户是否关注作者、历史互动
    // ...

    features := make([][]float64, len(feeds))
    for i, feed := range feeds {
        features[i] = []float64{
            float64(feed.LikesCount),
            float64(feed.CommentsCount),
            float64(time.Since(feed.CreatedAt).Hours()),
            // ...
        }
    }

    return features
}

// 热点 Feed 检测
func (s *FeedService) detectHotFeed(feedID string) {
    key := fmt.Sprintf("feed:views:%s", feedID)

    // 浏览数 +1
    count, _ := s.rdb.Incr(context.Background(), key).Result()

    // 如果 5 分钟内浏览超过 1000 次，视为热点
    if count == 1 {
        s.rdb.Expire(context.Background(), key, 5*time.Minute)
    }

    if count > 1000 {
        // 标记为热点 Feed
        s.markAsHot(feedID)

        // 缓存到本地内存（CDN 预热）
        s.localCache.Set(feedID, feed, 10*time.Minute)

        log.Infof("检测到热点 Feed: %s, 浏览数: %d", feedID, count)
    }
}

// 读取 Feed（冷热分离）
func (s *FeedService) GetFeed(feedID string) (*Feed, error) {
    // 1. 尝试从 Redis 读取
    cacheKey := fmt.Sprintf("feed:content:%s", feedID)
    cached, err := s.rdb.Get(context.Background(), cacheKey).Result()
    if err == nil {
        var feed Feed
        json.Unmarshal([]byte(cached), &feed)
        return &feed, nil
    }

    // 2. 从 MySQL 读取
    feed, err := s.getFeedFromMySQL(feedID)
    if err == nil {
        // 回写缓存
        data, _ := json.Marshal(feed)
        s.rdb.Set(context.Background(), cacheKey, data, 1*time.Hour)
        return feed, nil
    }

    // 3. 从冷存储读取（HBase）
    feed, err = s.getFeedFromHBase(feedID)
    if err != nil {
        return nil, errors.New("Feed 不存在")
    }

    return feed, nil
}

package merge

import (
    "container/heap"
)

// FeedWithSource Feed 来源
type FeedWithSource struct {
    Feed     *Feed
    SourceID int  // 来源 ID（用于区分不同关注者）
}

// MinHeap 最小堆（按时间倒序）
type MinHeap []FeedWithSource

func (h MinHeap) Len() int           { return len(h) }
func (h MinHeap) Less(i, j int) bool { return h[i].Feed.CreatedAt > h[j].Feed.CreatedAt }
func (h MinHeap) Swap(i, j int)      { h[i], h[j] = h[j], h[i] }

func (h *MinHeap) Push(x interface{}) {
    *h = append(*h, x.(FeedWithSource))
}

func (h *MinHeap) Pop() interface{} {
    old := *h
    n := len(old)
    x := old[n-1]
    *h = old[0 : n-1]
    return x
}

// MergeSortedFeeds 多路归并排序
func MergeSortedFeeds(sources [][]FeedWithSource, limit int) []*Feed {
    h := &MinHeap{}
    heap.Init(h)

    // 初始化堆（每个来源的第一个元素）
    indices := make([]int, len(sources))
    for i, source := range sources {
        if len(source) > 0 {
            heap.Push(h, source[0])
            indices[i] = 1
        }
    }

    // 依次取出最大的元素
    result := make([]*Feed, 0, limit)
    for h.Len() > 0 && len(result) < limit {
        top := heap.Pop(h).(FeedWithSource)
        result = append(result, top.Feed)

        // 从同一来源取下一个元素
        sourceID := top.SourceID
        if indices[sourceID] < len(sources[sourceID]) {
            next := sources[sourceID][indices[sourceID]]
            indices[sourceID]++
            heap.Push(h, next)
        }
    }

    return result
}

package filter

import (
    "github.com/bits-and-blooms/bloom/v3"
)

type FeedFilter struct {
    bloom *bloom.BloomFilter
}

func NewFeedFilter(expectedFeeds uint) *FeedFilter {
    return &FeedFilter{
        bloom: bloom.NewWithEstimates(expectedFeeds, 0.01),
    }
}

// IsSeen 检查 Feed 是否已看过
func (f *FeedFilter) IsSeen(feedID string) bool {
    return f.bloom.TestString(feedID)
}

// MarkAsSeen 标记 Feed 为已看过
func (f *FeedFilter) MarkAsSeen(feedID string) {
    f.bloom.AddString(feedID)
}

// 发布 Feed 限流（用户维度）
func (s *FeedService) PublishFeed(userID int64, content string) (string, error) {
    // 限流检查：每分钟最多发布 5 条
    key := fmt.Sprintf("ratelimit:publish:%d", userID)
    count, _ := s.rdb.Incr(context.Background(), key).Result()

    if count == 1 {
        s.rdb.Expire(context.Background(), key, 1*time.Minute)
    }

    if count > 5 {
        return "", errors.New("发布过于频繁，请稍后再试")
    }

    // 发布 Feed...
}

// 批量查询 Feed 详情（包含用户信息、统计数据）
func (s *FeedService) BatchGetFeedDetails(feedIDs []string) ([]*FeedDetail, error) {
    var wg sync.WaitGroup
    wg.Add(3)

    var feeds []*Feed
    var users map[int64]*User
    var stats map[string]*FeedStats

    // 并发查询
    go func() {
        defer wg.Done()
        feeds, _ = s.batchGetFeeds(feedIDs)
    }()

    go func() {
        defer wg.Done()
        userIDs := s.extractUserIDs(feeds)
        users, _ = s.batchGetUsers(userIDs)
    }()

    go func() {
        defer wg.Done()
        stats, _ = s.batchGetStats(feedIDs)
    }()

    wg.Wait()

    // 组装数据
    details := make([]*FeedDetail, len(feeds))
    for i, feed := range feeds {
        details[i] = &FeedDetail{
            Feed:  feed,
            User:  users[feed.UserID],
            Stats: stats[feed.FeedID],
        }
    }

    return details, nil
}