#LinkedIn 架构

// Kafka 生产者：用户行为事件发布
public class UserActivityProducer {

    private final KafkaProducer<String, UserActivityEvent> producer;

    public void onPageView(Long userId, String pageUrl, long timestamp) {
        UserActivityEvent event = new UserActivityEvent();
        event.setEventType("PAGE_VIEW");
        event.setUserId(userId);
        event.setPageUrl(pageUrl);
        event.setTimestamp(timestamp);
        event.setSessionId(getCurrentSessionId());

        // 发送到 Kafka Topic
        // 同一个用户的事件发送到同一个 Partition（保证顺序）
        producer.send(new ProducerRecord<>(
            "user-activity",
            userId.toString(),  // key：用于 Partition 路由
            event
        ));
    }

    public void onConnectionMade(Long userId, Long connectedUserId) {
        UserActivityEvent event = new UserActivityEvent();
        event.setEventType("CONNECTION_MADE");
        event.setUserId(userId);
        event.setTargetUserId(connectedUserId);
        event.setTimestamp(System.currentTimeMillis());

        producer.send(new ProducerRecord<>(
            "user-activity",
            userId.toString(),
            event
        ));
    }
}

// Kafka 消费者：搜索索引更新
public class SearchIndexConsumer {

    @KafkaListener(topics = "user-activity", groupId = "search-indexer")
    public void onUserActivity(UserActivityEvent event) {
        switch (event.getEventType()) {
            case "PAGE_VIEW":
                // 更新用户搜索偏好索引
                searchIndexService.updateUserPreferences(
                    event.getUserId(), event.getPageUrl());
                break;
            case "CONNECTION_MADE":
                // 重建用户的社交图谱索引
                searchIndexService.rebuildSocialIndex(event.getUserId());
                break;
            case "PROFILE_UPDATE":
                // 重新索引用户简历
                searchIndexService.reindexProfile(event.getUserId());
                break;
        }
    }
}

// Espresso 数据模型：用户简历
public class ProfileStore {

    private final EspressoClient client;

    // 读取用户简历
    public Profile getProfile(Long userId) {
        ReadResult result = client.get(
            "profiles",           // 表名
            userId.toString(),    // 主键
            "id, headline, summary, skills, experiences"
        );

        return deserialize(result.getColumns());
    }

    // 更新用户简历（MVCC 支持幂等更新）
    public void updateHeadline(Long userId, String newHeadline) {
        // 乐观锁：基于 version 字段防止并发写冲突
        client.update("profiles",
            userId.toString(),
            new ColumnUpdate("headline", newHeadline),
            new ColumnUpdate("version", currentVersion + 1)
        );
    }

    // 搜索：按技能关键词查询用户
    public List<Long> searchBySkill(String skill, int limit) {
        // 二级索引查询
        ScanResult result = client.scan(
            "profiles",
            new IndexRangeQuery("skills_index", skill),
            limit
        );

        return result.getKeys().stream()
            .map(Long::parseLong)
            .collect(Collectors.toList());
    }
}

// Samza 作业：实时计算用户「你可能认识的人」
public class PeopleYouMayKnowJob implements StreamTask {

    // 维护用户一度人脉的缓存
    private StateStore<Long, Set<Long>> firstDegreeCache;

    @Override
    public void process(IncomingMessageEnvelope envelope,
                       MessageCollector collector,
                       TaskCoordinator coordinator) {

        ConnectionEvent event = (ConnectionEvent) envelope.getMessage();

        Long userId = event.getUserId();
        Long connectedUserId = event.getConnectedUserId();

        // 1. 获取用户的一度人脉
        Set<Long> userConnections = getFirstDegree(userId);
        Set<Long> connectedUserConnections = getFirstDegree(connectedUserId);

        // 2. 找出「共同好友」
        Set<Long> mutualConnections = new HashSet<>(userConnections);
        mutualConnections.retainAll(connectedUserConnections);

        // 3. 生成推荐：共同好友越多，推荐优先级越高
        for (Long mutual : mutualConnections) {
            int score = scoreConnection(userId, mutual);
            Recommendation rec = new Recommendation(mutual, score);
            collector.send(new ProducerRecordEnvelope<>(
                "recommendations",
                userId.toString(),
                rec
            ));
        }

        // 4. 更新缓存
        updateFirstDegreeCache(userId, connectedUserId);
    }
}