#Twitter 流架构

// Fanout 推送服务：写入时将推文推送到粉丝的时间线
public class FanoutService {

    @Autowired private RedisTemplate redisTemplate;

    // 用户发推时触发 Fanout
    public void fanoutTweet(Long authorId, Long tweetId) {
        // 1. 获取作者的粉丝列表
        List<Long> followerIds = followService.getFollowers(authorId);

        // 2. 将推文 ID 写入每个粉丝的时间线缓存
        // 时间线只存推文 ID，不存内容，节省内存
        for (Long followerId : followerIds) {
            String timelineKey = "timeline:" + followerId;

            // LPUSH：头部插入，最新的在最前面
            // LTRIM：只保留最近 800 条，防止内存溢出
            redisTemplate.opsForList().leftPush(timelineKey, tweetId.toString());
            redisTemplate.opsForList().trim(timelineKey, 0, 799);
        }

        // 3. 明星账号特殊处理（避免 Fanout 耗时过长）
        if (followerIds.size() > 100000) {
            // 异步 Fanout：后台慢慢推送，不阻塞发推响应
            asyncFanoutQueue.submit(() -> fanoutTweetAsync(authorId, tweetId, followerIds));
        }
    }

    // 读取时间线
    public List<Tweet> getTimeline(Long userId, int count) {
        // 1. 从 Redis 获取推文 ID 列表
        String timelineKey = "timeline:" + userId;
        List<String> tweetIds = redisTemplate.opsForList()
            .range(timelineKey, 0, count - 1);

        if (tweetIds.isEmpty()) {
            return Collections.emptyList();
        }

        // 2. 批量查询推文详情（使用 Redis 缓存推文内容）
        return tweetService.getTweetsByIds(tweetIds);
    }
}

// Manhattan 数据模型

// 推文表（时间序）
// Key: userId + timestamp，Value: tweet JSON
ManhattanTable tweets = manhattan.open("tweets");

// 写入推文
tweets.set(userId, tweetId, tweet.toJson());

// 按时间范围查询某用户的所有推文
Iterable<Entry> userTweets = tweets.range(
    userId + ":" + startTimestamp,
    userId + ":" + endTimestamp
);

// 关注关系表（关系序）
ManhattanTable follows = manhattan.open("follows");

// 查出所有关注某用户的人（反向索引）
Iterable<Entry> followers = follows.reverseIndex("followee", userId);

// Timeline Service 的核心数据结构
// 每个用户的时间线是一个 Redis Sorted Set
// Score = 时间戳（用于排序），Member = 推文 ID

public class TimelineService {

    private static final int MAX_TIMELINE_SIZE = 800;

    // Push 时间线：写入时推送（适合普通用户）
    public void pushTimeline(Long userId, Long tweetId, long timestamp) {
        String key = "timeline:" + userId;

        redisTemplate.opsForZSet().add(key, tweetId.toString(), timestamp);
        // 只保留最近的 800 条
        if (redisTemplate.opsForZSet().zCard(key) > MAX_TIMELINE_SIZE) {
            redisTemplate.opsForZSet().removeRange(key, 0,
                MAX_TIMELINE_SIZE - 1);
        }
    }

    // Pull 时间线：读取时聚合（适合明星账号）
    // 明星账号不 Fanout，用户读取时动态聚合
    public List<Tweet> pullTimeline(Long userId, int count) {
        // 1. 获取用户关注的账号列表
        List<Long> following = followService.getFollowing(userId);

        // 2. 并行从 Manhattan 获取每个账号的最新推文
        List<List<Tweet>> tweetsPerUser = following.parallelStream()
            .map(user -> tweetService.getRecentTweets(user, 10))
            .collect(Collectors.toList());

        // 3. 合并排序（按时间倒序）
        List<Tweet> merged = tweetsPerUser.stream()
            .flatMap(List::stream)
            .sorted(Comparator.comparing(Tweet::getCreatedAt).reversed())
            .limit(count)
            .collect(Collectors.toList());

        return merged;
    }

    // 混合策略：小账号 Push，大账号 Pull
    // 阈值：粉丝数 < 10000 用 Push，> 10000 用 Pull
    public void fanout(Tweet tweet) {
        long followerCount = followService.getFollowerCount(tweet.getAuthorId());

        if (followerCount < 10000) {
            pushTimelineForAllFollowers(tweet);
        } else {
            // 明星账号：只写自己的时间线，读者 Pull 时聚合
            pushTimelineForSelf(tweet.getAuthorId(), tweet.getId());
        }
    }
}

// Heron 拓扑：实时处理用户行为事件
public class UserActivityTopology {

    public static void main(String[] args) {
        TopologyBuilder builder = new TopologyBuilder();

        // Spout：从 Kafka 消费用户行为事件
        builder.setSpout("kafka-spout",
            new KafkaSpout<>(kafkaConfig), 4);

        // Bolt 1：解析事件
        builder.setBolt("parser-bolt",
            new EventParserBolt(), 8)
            .shuffleGrouping("kafka-spout");

        // Bolt 2：更新用户特征（写入 Redis）
        builder.setBolt("feature-bolt",
            new FeatureUpdateBolt(), 16)
            .shuffleGrouping("parser-bolt");

        // Bolt 3：触发推荐重新计算
        builder.setBolt("recommend-bolt",
            new RecommendationBolt(), 8)
            .shuffleGrouping("feature-bolt");

        // 配置回压机制：当消费延迟 > 1 分钟时触发
        Config config = new Config();
        config.setMaxSpoutPending(10000);
        config.setTopologyBackpressure.enable(true);

        StormSubmitter.submitTopology("user-activity-topology", config,
            builder.createTopology());
    }
}

// Earlybird 实时索引：推文发出后几秒内可被搜索
public class EarlybirdIndexer {

    // 1. 接收新推文
    public void onNewTweet(Tweet tweet) {
        // 2. 文本分析：分词、提取实体（@提及、#话题、URL）
        ParsedTweet parsed = analyzer.parse(tweet);

        // 3. 构建倒排索引条目
        IndexDocument doc = new IndexDocument(tweet.getId());
        doc.addField("text", parsed.getTokens());      // 文本内容
        doc.addField("author", tweet.getAuthorId());  // 作者
        doc.addField("hashtag", parsed.getHashtags()); // 话题
        doc.addField("created_at", tweet.getTimestamp()); // 时间

        // 4. 写入索引（增量更新，不影响读请求）
        earlybirdWriter.addDocument(doc);
    }

    // 搜索：几毫秒内返回结果
    public SearchResult search(String query, int count) {
        Query parsedQuery = queryParser.parse(query);

        // 过滤：只看最近 7 天的推文
        parsedQuery.setTimeRange(now() - 7 * 24 * 3600, now());

        // 排序：时间相关性 + 互动量
        return earlybirdSearcher.search(parsedQuery, count);
    }
}