#快手短视频架构

// 视频处理流水线
public class VideoProcessingPipeline {

    // 用户上传完成后，触发处理流水线
    public void onVideoUploaded(VideoUploadEvent event) {
        String videoId = event.getVideoId();
        String rawPath = event.getRawFilePath();

        // 1. 异步转码：生成多种分辨率和码率
        CompletableFuture.runAsync(() -> {
            transcode(videoId, rawPath, "1080p", 5000);
            transcode(videoId, rawPath, "720p", 2500);
            transcode(videoId, rawPath, "480p", 1000);
            transcode(videoId, rawPath, "360p", 500);
        });

        // 2. 截图：生成封面帧
        CompletableFuture.runAsync(() -> {
            String coverPath = ffmpeg.extractCover(rawPath, "00:01");
            videoStore.setCover(videoId, coverPath);
        });

        // 3. 生成视频指纹（去重）
        CompletableFuture.runAsync(() -> {
            String fingerprint = videoFingerprint.compute(rawPath);
            // 如果发现相似视频，标记为疑似搬运
            checkDuplicate(videoId, fingerprint);
        });

        // 4. 审核（异步，不阻塞发布）
        auditQueue.submit(videoId);
    }

    // 转码：使用 FFmpeg
    private void transcode(String videoId, String input, String quality,
                          int bitrate) {
        String output = String.format("/storage/transcode/%s_%s.mp4",
            videoId, quality);

        // FFmpeg 命令行封装
        String cmd = String.format(
            "ffmpeg -i %s -b:v %dk -vf scale=-2:540 %s",
            input, bitrate, output);

        ProcessExecutor.execute(cmd);

        // 5. 上传到 CDN
        cdnUploader.upload(output, "videos/" + videoId + "_" + quality + ".mp4");
    }
}

// 两阶段推荐：召回 + 排序
public class RecommendService {

    @Autowired private CandidateGenerator candidateGenerator;
    @Autowired private Ranker ranker;

    public List<Video> recommend(Long userId, int count) {
        // 1. 召回阶段：从千万级视频库中召回 1000 条候选
        long start = System.nanoTime();
        List<Long> candidateIds = candidateGenerator.generate(userId, 1000);

        // 2. 排序阶段：对 1000 条候选打分，取 Top N
        List<Video> ranked = ranker.rank(userId, candidateIds);

        long latency = (System.nanoTime() - start) / 1_000_000;
        metrics.record("recommend.latency_ms", latency);

        return ranked.subList(0, Math.min(count, ranked.size()));
    }
}

// 召回阶段：多种召回策略并行
@Service
public class CandidateGenerator {

    public List<Long> generate(Long userId, int targetCount) {
        List<Long> candidates = new ArrayList<>();

        // 策略 1：关注人最新视频
        candidates.addAll(followingRecall(userId, targetCount / 5));

        // 策略 2：同类目热门视频
        candidates.addAll(categoryHotRecall(userId, targetCount / 5));

        // 策略 3：协同过滤召回
        candidates.addAll(cfRecall(userId, targetCount / 5));

        // 策略 4：热门视频（全局热度）
        candidates.addAll(globalHotRecall(targetCount / 5));

        // 策略 5：相似内容召回
        candidates.addAll(contentBasedRecall(userId, targetCount / 5));

        // 去重 + 截断
        return candidates.stream()
            .distinct()
            .limit(targetCount)
            .collect(Collectors.toList());
    }
}

// CDN 调度：就近分配
public class CdnSelector {

    // 根据用户位置，分配最优 CDN 节点
    public CdnNode selectCdnNode(Long userId, String videoId) {
        // 1. 获取用户地理位置
        GeoLocation location = geoService.locateUser(userId);

        // 2. 获取视频源站位置（通常在 CDN 源站附近）
        String origin = videoStore.getSource(videoId);

        // 3. 选择最优 CDN 节点
        // 优先：同省份 CDN > 同区域 CDN > 全国 CDN
        List<CdnNode> candidates = cdnRegistry.getNodes();

        return candidates.stream()
            .filter(n -> n.isHealthy() && n.hasVideo(videoId))
            .min(Comparator.comparingDouble(n ->
                distance(n.getLocation(), location)))
            .orElseGet(() -> fallbackToOrigin(origin));
    }
}

// 多目标推荐：帕累托最优
@Service
public class MultiObjectiveRanker {

    // MMOE：Multi-gate Mixture-of-Experts
    // 多个任务共享底层特征，每个任务有独立的门控网络

    public List<VideoScore> rank(Long userId, List<Long> videoIds) {
        // 1. 获取用户特征和视频特征
        UserFeatures userFeatures = featureStore.getUserFeatures(userId);
        List<VideoFeatures> videoFeatures = featureStore.getVideoFeatures(videoIds);

        // 2. 共享底层表达
        List<DenseTensor> sharedEmbeddings = sharedTower.forward(
            userFeatures, videoFeatures);

        // 3. 每个目标独立预测
        double ctrScore = ctrTower.forward(sharedEmbeddings);
        double vtrScore = vtrTower.forward(sharedEmbeddings);
        double likeScore = likeTower.forward(sharedEmbeddings);
        double commentScore = commentTower.forward(sharedEmbeddings);
        double shareScore = shareTower.forward(sharedEmbeddings);
        double followScore = followTower.forward(sharedEmbeddings);

        // 4. 加权融合（权重由线上实验决定）
        double finalScore =
            ctrScore * 0.15 +
            vtrScore * 0.25 +
            likeScore * 0.15 +
            commentScore * 0.10 +
            shareScore * 0.15 +
            followScore * 0.20;

        return IntStream.range(0, videoIds.size())
            .mapToObj(i -> new VideoScore(videoIds.get(i), finalScore))
            .sorted(Comparator.comparingDouble(VideoScore::getScore).reversed())
            .collect(Collectors.toList());
    }
}

// 直播弹幕服务：Kafka + Flink 实时处理
@Service
public class LiveDanmuService {

    // 生产者：用户发送弹幕
    public void sendDanmu(Long userId, Long roomId, String content) {
        Danmu danmu = new Danmu();
        danmu.setUserId(userId);
        danmu.setRoomId(roomId);
        danmu.setContent(content);
        danmu.setTimestamp(System.currentTimeMillis());

        // 发送到 Kafka
        kafkaTemplate.send("live-danmu", roomId.toString(), danmu);
    }

    // 消费者：实时处理弹幕
    @Service
    public class DanmuProcessor {

        @KafkaListener(topics = "live-danmu", groupId = "danmu-processor")
        public void process(Danmu danmu) {
            // 1. 内容安全审核
            if (!contentModerator.isSafe(danmu.getContent())) {
                return;  // 不安全弹幕直接丢弃
            }

            // 2. 过滤违禁词
            String filtered = wordFilter.filter(danmu.getContent());

            // 3. 限流：每个房间每分钟最多显示 N 条弹幕
            if (!rateLimiter.tryAcquire(danmu.getRoomId())) {
                return;  // 超过频率限制，丢弃
            }

            // 4. 广播到在线用户（通过 WebSocket）
            broadcastService.broadcast(danmu.getRoomId(), filtered);
        }
    }
}