#Netflix 架构

// 视频转码任务调度（简化示例）
@Service
public class TranscodeJobService {
    @Autowired private VideoRepository videoRepository;
    @Autowired private TranscodeWorkerPool workerPool;

    public void submitTranscodeJob(Long videoId, TranscodeProfile profile) {
        Video video = videoRepository.findById(videoId);

        // 创建转码任务
        TranscodeJob job = new TranscodeJob();
        job.setVideoId(videoId);
        job.setSourceFile(video.getOriginalFile());
        job.setTargetProfile(profile);
        job.setPriority(calculatePriority(video));
        job.setStatus(JobStatus.PENDING);
        jobRepository.save(job);

        // 提交到 worker 池
        workerPool.submit(job);
    }

    // 热门内容优先转码，新内容可以排队等待
    private int calculatePriority(Video video) {
        if (video.getPredictedPopularity() > 0.8) {
            return 100; // 高优先级，10 分钟内必须完成
        } else if (video.getPredictedPopularity() > 0.5) {
            return 50;  // 中优先级
        } else {
            return 10;  // 低优先级，可以排队等待
        }
    }
}

// Netflix 的微服务划分示例

// API Gateway：统一入口
// 职责：路由、认证、限流、协议转换
@RestController
@RequestMapping("/api")
public class ApiGateway {

    @Autowired private MemberServiceClient memberClient;
    @Autowired private PlaybackServiceClient playbackClient;
    @Autowired private RecommendationServiceClient recommendationClient;

    // 获取用户推荐列表
    @GetMapping("/recommendations")
    public List<Video> getRecommendations(@RequestHeader("Authorization") String token) {
        // 1. 验证 Token
        String userId = authService.validateToken(token);

        // 2. 从推荐服务获取个性化推荐
        //    走的是 gRPC（Netflix 内部），而非 HTTP
        List<String> videoIds = recommendationClient.getTopN(userId, 20);

        // 3. 从内容服务获取视频元数据
        List<Video> videos = contentService.getVideos(videoIds);

        // 4. 过滤：已观看、去版权限制等
        return filterService.apply(userId, videos);
    }
}

// Member Service：用户账户管理
// 职责：注册、登录、订阅计划、账单
@Service
public class MemberService {
    public Subscription getSubscription(String userId) {
        return subscriptionRepository.findByUserId(userId);
    }

    public void upgradePlan(String userId, SubscriptionPlan newPlan) {
        // 升级订阅计划
        // 异步通知计费系统
        kafkaTemplate.send("subscription-events",
            userId, new SubscriptionUpgradeEvent(userId, newPlan));
    }
}

// Recommendation Service：推荐系统
// 职责：协同过滤、内容推荐、新内容发现
@Service
public class RecommendationService {
    @Autowired private KafkaTemplate kafkaTemplate;

    // 用户行为事件异步处理
    public void onUserAction(UserAction action) {
        // 发送事件到 Kafka，供下游算法计算使用
        kafkaTemplate.send("user-actions",
            action.getUserId(), action);
    }

    // 实时推荐
    public List<String> getTopN(String userId, int n) {
        // 1. 从缓存获取用户的实时特征
        UserFeatures features = userFeatureCache.get(userId);

        // 2. 协同过滤召回
        List<String> cfCandidates = collaborativeFilter召回(features, n * 2);

        // 3. 内容匹配召回
        List<String> contentCandidates = contentFilter召回(features, n * 2);

        // 4. 混合排序
        return ranker.rank(
            mergeAndDeduplicate(cfCandidates, contentCandidates),
            features, n);
    }
}

// 异步事件处理：Kafka 消息消费
@Service
public class UserBehaviorConsumer {

    @KafkaListener(topics = "user-actions", groupId = "recommendation")
    public void onUserAction(ConsumerRecord<String, UserAction> record) {
        UserAction action = record.value();

        // 1. 更新用户观看历史
        viewingHistoryService.record(action);

        // 2. 更新实时推荐特征
        userFeatureService.updateFeatures(action);

        // 3. 如果是付费用户，发送数据给广告系统
        if (subscriptionService.isPaidUser(action.getUserId())) {
            kafkaTemplate.send("paid-user-actions",
                action.getUserId(), action);
        }
    }
}

// Eureka 服务注册与发现示例

// 服务提供者启动时注册
@SpringBootApplication
@EnableEurekaClient
public class RecommendationServiceApplication {
    public static void main(String[] args) {
        SpringApplication.run(RecommendationServiceApplication.class, args);
        // 启动后自动向 Eureka Server 注册
        // 服务名（spring.application.name）会被用作注册名
    }
}

// 服务消费者调用
@Service
public class ApiGateway {

    // RestTemplate + @LoadBalanced 实现客户端负载均衡
    // Ribbon 会自动从 Eureka 获取可用实例并轮询分发
    @Autowired
    private RestTemplate restTemplate;

    public VideoDetails getVideoDetails(String videoId) {
        // 不用写死 IP：Netflix 建议服务名大写
        // Ribbon 从 Eureka 获取 recommendation-service 的实例列表
        String url = "http://RECOMMENDATION-SERVICE/api/v1/details/" + videoId;
        return restTemplate.getForObject(url, VideoDetails.class);
    }
}

// Resilience4j 熔断器配置（Netflix 内部已迁移到 Resilience4j）
@Configuration
public class Resilience4jConfig {
    @Bean
    public CircuitBreakerRegistry circuitBreakerRegistry() {
        CircuitBreakerConfig config = CircuitBreakerConfig.custom()
            // 滑动窗口类型：基于调用次数（COUNT_BASED）或时间（WINDOW_BASED）
            .slidingWindowType(SlidingWindowType.COUNT_BASED)
            // 滑动窗口大小：统计最近 10 次调用的失败率
            .slidingWindowSize(10)
            // 失败率阈值：超过 50% 则触发熔断
            .failureRateThreshold(50)
            // 熔断持续时间：熔断后 30 秒内拒绝所有请求
            .waitDurationInOpenState(Duration.ofSeconds(30))
            // 半开状态放行数：熔断结束后，先放 3 个请求试试
            .permittedNumberOfCallsInHalfOpenState(3)
            // 最小调用次数：低于此值不计算失败率
            .minimumNumberOfCalls(5)
            .build();

        return CircuitBreakerRegistry.of(config);
    }
}

// 使用熔断器保护服务调用
@Service
public class RecommendationClient {
    private final CircuitBreaker circuitBreaker;
    private final RestTemplate restTemplate;

    public RecommendationClient(CircuitBreakerRegistry registry) {
        // 创建名为 "recommendationService" 的熔断器
        this.circuitBreaker = registry.circuitBreaker("recommendationService");
    }

    public List<Recommendation> getRecommendations(Long userId) {
        // executeSupplier 会自动执行熔断器逻辑
        return circuitBreaker.executeSupplier(() -> {
            return restTemplate.getForObject(
                "http://recommendation-service/api/v1/" + userId,
                List.class
            );
        });
    }

    // 降级方法：熔断打开时调用的兜底逻辑
    public List<Recommendation> fallback(Long userId, Throwable throwable) {
        log.warn("Recommendation service failed: {}, fallback triggered",
            throwable.getMessage());

        // 策略一：返回全局热门内容（不依赖个性化服务）
        return getGlobalTrending();

        // 策略二：返回用户上次观看的同类型内容
        // return getRecentlyWatchedSimilar(userId);
    }
}

// Zuul 过滤器示例：请求限流
@Component
public class RateLimitFilter extends ZuulFilter {

    private final RateLimiter rateLimiter = RateLimiter.create(10000.0);

    @Override
    public String filterType() {
        return FilterConstants.PRE_TYPE; // 前置过滤器
    }

    @Override
    public int filterOrder() {
        return 0; // 优先级最高
    }

    @Override
    public boolean shouldFilter() {
        return true; // 对所有请求生效
    }

    @Override
    public Object run() {
        RequestContext ctx = RequestContext.getCurrentContext();
        String userId = ctx.getRequest().getHeader("X-User-Id");

        // 令牌桶限流：每秒允许 10000 个请求
        if (!rateLimiter.tryAcquire()) {
            ctx.setSendZuulResponse(false); // 拦截请求
            ctx.setResponseStatusCode(429); // Too Many Requests
            ctx.setResponseBody("Rate limit exceeded");
        }

        return null;
    }
}