use tokio::net::{TcpListener, TcpStream};
use tokio::sync::broadcast;
use futures::stream::StreamExt;
use bytes::Bytes;
use std::sync::Arc;
use std::collections::HashMap;
use dashmap::DashMap;

struct StreamingServer {
    // 使用DashMap提供執行緒安全的併發存取
    active_streams: Arc<DashMap<String, broadcast::Sender<Bytes>>>,
    stats_collector: Arc<StreamStatsCollector>,
}

impl StreamingServer {
    async fn handle_publisher(&self, stream_id: String, mut tcp_stream: TcpStream) -> Result<(), Box<dyn std::error::Error>> {
        // 建立廣播通道，允許多個訂閱者接收同一串流
        let (tx, _) = broadcast::channel::<Bytes>(1024);
        self.active_streams.insert(stream_id.clone(), tx.clone());
        
        let mut buffer = vec![0u8; 16384]; // 16KB 緩衝區
        
        // 非阻塞讀取串流資料
        loop {
            match tcp_stream.read(&mut buffer).await {
                Ok(0) => break, // 連線關閉
                Ok(n) => {
                    let data = Bytes::copy_from_slice(&buffer[..n]);
                    
                    // 使用Tokio的廣播機制將資料傳送給所有訂閱者
                    if tx.send(data.clone()).is_err() {
                        // 沒有接收者，可能需要停止發布
                        break;
                    }
                    
                    // 收集串流統計資料
                    self.stats_collector.record_bytes_sent(stream_id.clone(), n as u64).await;
                }
                Err(e) => {
                    eprintln!("Error reading from publisher: {}", e);
                    break;
                }
            }
        }
        
        // 發布者斷開連線，移除串流
        self.active_streams.remove(&stream_id);
        Ok(())
    }
    
    async fn handle_subscriber(&self, stream_id: String, mut tcp_stream: TcpStream) -> Result<(), Box<dyn std::error::Error>> {
        // 從活躍串流表中取得廣播傳送者
        let tx = match self.active_streams.get(&stream_id) {
            Some(tx) => tx.clone(),
            None => return Err("Stream not found".into()),
        };
        
        // 建立接收者
        let mut rx = tx.subscribe();
        
        // 非阻塞接收並轉發資料
        while let Ok(data) = rx.recv().await {
            if let Err(e) = tcp_stream.write_all(&data).await {
                eprintln!("Error sending to subscriber: {}", e);
                break;
            }
            
            // 收集統計資料
            self.stats_collector.record_bytes_received(stream_id.clone(), data.len() as u64).await;
        }
        
        Ok(())
    }
}

// 前端WebRTC發布者實作
class WebRTCPublisher {
  constructor(config) {
    this.peerConnection = null;
    this.localStream = null;
    this.signaling = new SignalingClient(config.signalingUrl);
    this.fallbackToQuic = config.enableQuicFallback || false;
    this.statsInterval = null;
  }
  
  async initialize() {
    try {
      // 取得媒體串流（攝影機/麥克風）
      this.localStream = await navigator.mediaDevices.getUserMedia({
        video: {
          width: { ideal: 1920 },
          height: { ideal: 1080 },
          frameRate: { ideal: 30 }
        },
        audio: true
      });
      
      // 建立RTCPeerConnection
      this.peerConnection = new RTCPeerConnection({
        iceServers: [
          { urls: 'stun:stun.l.google.com:19302' },
          {
            urls: 'turn:turn.example.com:3478',
            username: 'username',
            credential: 'credential'
          }
        ],
        // 啟用Google的低延遲策略
        sdpSemantics: 'unified-plan',
        iceTransportPolicy: 'all'
      });
      
      // 監聽ICE連線狀態變化
      this.peerConnection.oniceconnectionstatechange = () => {
        if (this.peerConnection.iceConnectionState === 'failed' && this.fallbackToQuic) {
          this._switchToQuicTransport();
        }
      };
      
      // 新增本地媒體軌道到對等連線
      this.localStream.getTracks().forEach(track => {
        this.peerConnection.addTrack(track, this.localStream);
      });
      
      // 建立SDP提議
      const offer = await this.peerConnection.createOffer();
      await this.peerConnection.setLocalDescription(offer);
      
      // 透過信令伺服器交換SDP
      this.signaling.sendOffer(offer.sdp);
      this.signaling.onAnswer = async (sdp) => {
        await this.peerConnection.setRemoteDescription({
          type: 'answer',
          sdp: sdp
        });
      };
      
      // 啟動統計收集
      this._startStatsCollection();
      
      return true;
    } catch (error) {
      console.error('WebRTC初始化失敗:', error);
      if (this.fallbackToQuic) {
        return this._switchToQuicTransport();
      }
      return false;
    }
  }
  
  _switchToQuicTransport() {
    console.log('切換到QUIC傳輸...');
    // 實際的QUIC切換邏輯
    // 使用HTTP/3 + QUIC的推流機制
    return new QuicPublisher().initialize(this.localStream);
  }
  
  _startStatsCollection() {
    this.statsInterval = setInterval(async () => {
      if (!this.peerConnection) return;
      
      const stats = await this.peerConnection.getStats();
      let outboundRtpStats = null;
      let candidatePairStats = null;
      
      stats.forEach(report => {
        if (report.type === 'outbound-rtp' && report.kind === 'video') {
          outboundRtpStats = report;
        } else if (report.type === 'candidate-pair' && report.state === 'succeeded') {
          candidatePairStats = report;
        }
      });
      
      if (outboundRtpStats && candidatePairStats) {
        const bitrate = outboundRtpStats.bytesSent * 8 / 1000; // kbps
        const packetLoss = outboundRtpStats.packetsLost || 0;
        const rtt = candidatePairStats.currentRoundTripTime * 1000; // ms
        
        // 將統計資料傳送到伺服器，用於自適應位元率調整
        this.signaling.sendStats({
          bitrate,
          packetLoss,
          rtt,
          resolution: `${outboundRtpStats.frameWidth}x${outboundRtpStats.frameHeight}`,
          frameRate: outboundRtpStats.framesPerSecond
        });
      }
    }, 1000);
  }
}

// 後端Rust實作的混合傳輸協調器
pub struct HybridTransportCoordinator {
    webrtc_server: Arc<WebRTCServer>,
    quic_server: Arc<QuicServer>,
    signaling_server: Arc<SignalingServer>,
    session_manager: Arc<SessionManager>,
}

impl HybridTransportCoordinator {
    pub async fn new() -> Self {
        let session_manager = Arc::new(SessionManager::new());
        
        let webrtc_server = Arc::new(WebRTCServer::new(session_manager.clone()));
        let quic_server = Arc::new(QuicServer::new(session_manager.clone()));
        let signaling_server = Arc::new(SignalingServer::new(session_manager.clone()));
        
        Self {
            webrtc_server,
            quic_server,
            signaling_server,
            session_manager,
        }
    }
    
    pub async fn start(&self, webrtc_addr: SocketAddr, quic_addr: SocketAddr, signaling_addr: SocketAddr) -> Result<(), Box<dyn Error>> {
        // 啟動信令伺服器
        let sig_server = self.signaling_server.clone();
        tokio::spawn(async move {
            sig_server.start(signaling_addr).await.unwrap();
        });
        
        // 啟動WebRTC伺服器
        let webrtc = self.webrtc_server.clone();
        tokio::spawn(async move {
            webrtc.start(webrtc_addr).await.unwrap();
        });
        
        // 啟動QUIC伺服器
        let quic = self.quic_server.clone();
        tokio::spawn(async move {
            quic.start(quic_addr).await.unwrap();
        });
        
        // 設定協定自動切換處理
        self.setup_protocol_switching().await;
        
        Ok(())
    }
    
    async fn setup_protocol_switching(&self) {
        let session_mgr = self.session_manager.clone();
        let webrtc = self.webrtc_server.clone();
        let quic = self.quic_server.clone();
        
        // 監聽WebRTC連線失敗事件
        self.webrtc_server.on_connection_failure(move |session_id| {
            let session_mgr = session_mgr.clone();
            let quic = quic.clone();
            
            async move {
                if let Some(session) = session_mgr.get_session(session_id).await {
                    // 嘗試切換到QUIC傳輸
                    quic.take_over_session(session).await.ok();
                }
            }
        });
        
        // 監聽網路條件變化
        self.session_manager.on_network_condition_change(move |session_id, condition| {
            let webrtc = webrtc.clone();
            let quic = quic.clone();
            
            async move {
                match condition {
                    NetworkCondition::Good => {
                        // 如果網路條件良好與當前使用QUIC，考慮切換回WebRTC
                        if session_id.protocol() == Protocol::Quic {
                            webrtc.attempt_takeover(session_id).await.ok();
                        }
                    },
                    NetworkCondition::Poor => {
                        // 如果網路條件差與當前使用WebRTC，切換到QUIC
                        if session_id.protocol() == Protocol::WebRtc

use actix_web::{web, App, HttpResponse, HttpServer, Responder};
use tokio::sync::Mutex;
use std::sync::Arc;
use std::collections::HashMap;

// 分享伺服器狀態
struct AppState {
    active_streams: Mutex<HashMap<String, StreamInfo>>,
}

struct StreamInfo {
    publisher_id: String,
    viewers: usize,
    start_time: chrono::DateTime<chrono::Utc>,
}

async fn health_check() -> impl Responder {
    HttpResponse::Ok().body("串流伺服器執行中")
}

// 處理RTMP/WebRTC串流接入
async fn ingest_stream(
    stream_id: web::Path<String>,
    state: web::Data<Arc<AppState>>
) -> impl Responder {
    let mut streams = state.active_streams.lock().await;
    
    let stream_info = StreamInfo {
        publisher_id: "user123".to_string(), // 實際應從認證取得
        viewers: 0,
        start_time: chrono::Utc::now(),
    };
    
    streams.insert(stream_id.clone(), stream_info);
    
    // 實際環境中會啟動FFmpeg處理或WebRTC工作階段
    HttpResponse::Ok().json(serde_json::json!({
        "status": "串流接入成功",
        "stream_id": stream_id,
        "server_time": chrono::Utc::now().to_rfc3339()
    }))
}

// 取得串流
async fn get_stream(
    stream_id: web::Path<String>,
    state: web::Data<Arc<AppState>>
) -> impl Responder {
    let mut streams = state.active_streams.lock().await;
    
    if let Some(stream) = streams.get_mut(&stream_id) {
        stream.viewers += 1;
        HttpResponse::Ok().json(serde_json::json!({
            "status": "串流連線成功",
            "stream_id": stream_id,
            "viewer_count": stream.viewers
        }))
    } else {
        HttpResponse::NotFound().json(serde_json::json!({
            "error": "串流不存在"
        }))
    }
}

#[actix_web::main]
async fn main() -> std::io::Result<()> {
    let app_state = Arc::new(AppState {
        active_streams: Mutex::new(HashMap::new()),
    });

    HttpServer::new(move || {
        App::new()
            .app_data(web::Data::new(app_state.clone()))
            .route("/health", web::get().to(health_check))
            .route("/ingest/{stream_id}", web::post().to(ingest_stream))
            .route("/stream/{stream_id}", web::get().to(get_stream))
    })
    .bind("0.0.0.0:8080")?
    .run()
    .await
}

use tokio::process::Command;

async fn transcode_stream(stream_id: &str) -> Result<(), Box<dyn std::error::Error>> {
    let output = Command::new("ffmpeg")
        .args(&[
            "-i", &format!("rtmp://localhost/live/{}", stream_id),
            "-c:v", "libx264", "-preset", "veryfast",
            "-b:v", "3000k", "-maxrate", "3000k", "-bufsize", "6000k",
            "-g", "60", // 關鍵影格間隔，影響串流延遲
            "-c:a", "aac", "-b:a", "128k",
            "-f", "flv", &format!("rtmp://localhost/hls/{}", stream_id)
        ])
        .output()
        .await?;
    
    if !output.status.success() {
        let error = String::from_utf8_lossy(&output.stderr);
        return Err(error.into());
    }
    
    Ok(())
}

use redis::{Client, Commands, Connection};

struct RedisStreamManager {
    conn: Mutex<Connection>,
}

impl RedisStreamManager {
    fn new(redis_url: &str) -> Result<Self, redis::RedisError> {
        let client = Client::open(redis_url)?;
        let conn = client.get_connection()?;
        
        Ok(RedisStreamManager {
            conn: Mutex::new(conn),
        })
    }
    
    async fn register_stream(&self, stream_id: &str, publisher_id: &str) -> redis::RedisResult<()> {
        let mut conn = self.conn.lock().await;
        
        // 使用Hash儲存串流資訊
        conn.hset_multiple(
            &format!("stream:{}", stream_id),
            &[
                ("publisher_id", publisher_id),
                ("viewers", "0"),
                ("start_time", &chrono::Utc::now().timestamp().to_string()),
                ("status", "live"),
            ]
        )?;
        
        // 設定過期時間，避免資源洩漏
        conn.expire(&format!("stream:{}", stream_id), 24 * 60 * 60)?;
        
        // 新增到活躍串流集合
        conn.sadd("active_streams", stream_id)?;
        
        Ok(())
    }
    
    async fn increment_viewers(&self, stream_id: &str) -> redis::RedisResult<usize> {
        let mut conn = self.conn.lock().await;
        let viewers: usize = conn.hincr(&format!("stream:{}", stream_id), "viewers", 1)?;
        Ok(viewers)
    }
}

use actix_web::{web, App, Error, HttpRequest, HttpResponse, HttpServer};
use actix_web_actors::ws;
use actix::{Actor, StreamHandler};

struct WebRtcSession {
    id: String,
}

impl Actor for WebRtcSession {
    type Context = ws::WebsocketContext<Self>;
}

// 處理WebRTC信令
impl StreamHandler<Result<ws::Message, ws::ProtocolError>> for WebRtcSession {
    fn handle(&mut self, msg: Result<ws::Message, ws::ProtocolError>, ctx: &mut Self::Context) {
        match msg {
            Ok(ws::Message::Text(text)) => {
                // 解析信令訊息（SDP或ICE候選項）
                let message = serde_json::from_str::<serde_json::Value>(&text);
                if let Ok(message) = message {
                    // 根據訊息類別處理
                    if let Some(msg_type) = message.get("type").and_then(|t| t.as_str()) {
                        match msg_type {
                            "offer" => {
                                // 處理SDP offer，通常轉發給對方或處理媒體引數
                                println!("收到SDP offer");
                                ctx.text(serde_json::json!({
                                    "type": "answer",
                                    "sdp": "實際SDP應答內容" // 實際應用中需動態生成
                                }).to_string());
                            },
                            "ice-candidate" => {
                                // 處理ICE候選項，用於建立P2P連線
                                println!("收到ICE候選項");
                            },
                            _ => println!("未知訊息類別: {}", msg_type),
                        }
                    }
                }
            },
            Ok(ws::Message::Close(reason)) => {
                println!("WebSocket連線關閉: {:?}", reason);
                ctx.close(reason);
            },
            _ => (),
        }
    }
}

async fn websocket_route(
    req: HttpRequest,
    stream: web::Payload,
    path: web::Path<String>,
) -> Result<HttpResponse, Error> {
    let session_id = path.into_inner();
    ws::start(
        WebRtcSession {
            id: session_id,
        },
        &req,
        stream,
    )
}

#[actix_web::main]
async fn main() -> std::io::Result<()> {
    HttpServer::new(|| {
        App::new()
            .route("/ws/{id}", web::get().to(websocket_route))
    })
    .bind("127.0.0.1:8080")?
    .run()
    .await
}

use quinn::{Endpoint, ServerConfig, ServerConfigBuilder};
use std::net::SocketAddr;
use std::sync::Arc;

async fn setup_quic_server() -> Result<Endpoint, Box<dyn std::error::Error>> {
    // 產生自簽SSL憑證（實際環境應使用正式憑證）
    let cert = rcgen::generate_simple_self_signed(vec!["localhost".into()]).unwrap();
    let cert_der = cert.serialize_der().unwrap();
    let priv_key = cert.serialize_private_key_der();
    
    // 設定QUIC伺服

use tokio::net::TcpListener;
use tokio::io::{AsyncReadExt, AsyncWriteExt};

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let listener = TcpListener::bind("127.0.0.1:8080").await?;
    
    loop {
        let (mut socket, _) = listener.accept().await?;
        
        tokio::spawn(async move {
            let mut buffer = [0; 1024];
            
            loop {
                let n = match socket.read(&mut buffer).await {
                    Ok(n) if n == 0 => return,
                    Ok(n) => n,
                    Err(_) => return,
                };
                
                if let Err(_) = socket.write_all(&buffer[0..n]).await {
                    return;
                }
            }
        });
    }
}

use redis::{Client, Commands, Connection};
use std::sync::Arc;

struct StreamMetadataService {
    redis_client: Arc<Client>,
}

impl StreamMetadataService {
    fn new(redis_url: &str) -> Result<Self, redis::RedisError> {
        let client = Client::open(redis_url)?;
        Ok(Self {
            redis_client: Arc::new(client),
        })
    }
    
    async fn store_metadata(&self, stream_id: &str, resolution: &str) -> redis::RedisResult<()> {
        let mut conn = self.redis_client.get_connection()?;
        conn.set(format!("stream:{}:resolution", stream_id), resolution)
    }
    
    async fn get_resolution(&self, stream_id: &str) -> redis::RedisResult<String> {
        let mut conn = self.redis_client.get_connection()?;
        conn.get(format!("stream:{}:resolution", stream_id))
    }
    
    async fn increment_viewers(&self, stream_id: &str) -> redis::RedisResult<i64> {
        let mut conn = self.redis_client.get_connection()?;
        conn.incr(format!("stream:{}:viewers", stream_id), 1)
    }
}

use webrtc::{api::API, peer_connection::RTCPeerConnection};
use std::sync::Arc;

async fn create_webrtc_connection() -> Result<Arc<RTCPeerConnection>, webrtc::Error> {
    let api = API::new();
    let config = webrtc::peer_connection::RTCConfiguration::default();
    let peer_connection = api.new_peer_connection(config).await?;
    
    // 設定媒體軌道和編碼引數
    // ...
    
    Ok(Arc::new(peer_connection))
}

import tensorflow as tf
import numpy as np

class BitratePredictor:
    def __init__(self):
        self.model = tf.keras.Sequential([
            tf.keras.layers.Dense(128, activation='relu', input_shape=(8,)),
            tf.keras.layers.Dense(64, activation='relu'),
            tf.keras.layers.Dropout(0.2),
            tf.keras.layers.Dense(32, activation='relu'),
            tf.keras.layers.Dense(1, activation='linear')
        ])
        self.model.compile(optimizer='adam', loss='mse')
        
    def predict_optimal_bitrate(self, network_metrics):
        """
        根據網路指標預測最佳位元率
        
        引數:
            network_metrics: 包含當前位元率、網路速度、緩衝區大小、封包丟失率等指標
        """
        input_data = np.array([network_metrics])
        return float(self.model.predict(input_data)[0][0])
    
    def train(self, training_data, target_values, epochs=100):
        self.model.fit(training_data, target_values, epochs=epochs, verbose=0)

// Rust端呼叫Python AI服務的gRPC客戶端
use tonic::{Request, Response};
use bitrate_service::bitrate_client::BitrateClient;
use bitrate_service::{PredictionRequest, PredictionResponse};

async fn get_optimal_bitrate(
    client: &mut BitrateClient<tonic::transport::Channel>,
    network_metrics: Vec<f32>
) -> Result<f32, Box<dyn std::error::Error>> {
    let request = Request::new(PredictionRequest {
        metrics: network_metrics,
    });
    
    let response = client.predict_bitrate(request).await?;
    Ok(response.into_inner().optimal_bitrate)
}

def predict_content_popularity(content_features, region_data):
    """預測內容在特定地區的熱度，指導CDN快取策略"""
    combined_features = np.concatenate([content_features, region_data])
    popularity_score = popularity_model.predict([combined_features])[0]
    return popularity_score

def optimize_edge_caching(content_catalog, edge_nodes):
    """根據預測熱度最佳化邊緣節點的快取分配"""
    allocation_plan = {}
    
    for node in edge_nodes:
        region_data = get_region_data(node.location)
        content_scores = []
        
        for content in content_catalog:
            score = predict_content_popularity(content.features, region_data)
            content_scores.append((content.id, score))
        
        # 依據分數排序並選擇前N個內容快取
        top_content = sorted(content_scores, key=lambda x: x[1], reverse=True)[:node.capacity]
        allocation_plan[node.id] = [content_id for content_id, _ in top_content]
    
    return allocation_plan

use actix_web::{web, App, HttpServer, HttpResponse};
use actix_web_actors::ws;
use serde_json::Value;
use std::sync::Arc;
use tokio::sync::Mutex;

struct NetworkState {
    bandwidth: f64,
    latency: f64,
    jitter: f64,
}

struct WsSession {
    state: Arc<Mutex<NetworkState>>,
}

impl actix_web_actors::ws::WebsocketContext<WsSession> {
    fn update_network_state(&mut self, data: &str) {
        if let Ok(json) = serde_json::from_str::<Value>(data) {
            let mut state = self.state.lock().await;
            state.bandwidth = json["bandwidth"].as_f64().unwrap_or(0.0);
            state.latency = json["latency"].as_f64().unwrap_or(0.0);
            state.jitter = json["jitter"].as_f64().unwrap_or(0.0);
        }
    }
}

async fn start_websocket(ws: web.Payload, state: web.Data<Arc<Mutex<NetworkState>>>) -> HttpResponse {
    ws::start(WsSession { state: state.get_ref().clone() }, &mut ws)
}

#[actix_web::main]
async fn main() -> std::io::Result<()> {
    let state = Arc::new(Mutex::new(NetworkState { bandwidth: 0.0, latency: 0.0, jitter: 0.0 }));

    HttpServer::new(move || {
        App::new()
            .app_data(web.Data::new(state.clone()))
            .route("/network_ws", web.get().to(start_websocket))
    })
    .bind("0.0.0.0:8080")?
    .run()
    .await
}

import tensorflow as tf
import numpy as np

# AI模型定義：CNN + LSTM混合模型
def create_bitrate_model():
    model = tf.keras.Sequential([
        tf.keras.layers.Conv1D(32, kernel_size=3, activation='relu', input_shape=(10, 3)),
        tf.keras.layers.LSTM(64, return_sequences=True),
        tf.keras.layers.LSTM(32),
        tf.keras.layers.Dense(16, activation='relu'),
        tf.keras.layers.Dense(1, activation='linear')
    ])
    model.compile(optimizer='adam', loss='mse')
    return model

# 模擬資料：過去10秒的[頻寬、延遲、抖動]資料
sample_data = np.random.rand(100, 10, 3)  # 100個樣本，每個包含10個時間點，3個特徵
target_bitrate = np.random.rand(100, 1) * 5  # 目標位元率在0-5 Mbps之間

# 訓練模型
model = create_bitrate_model()
model.fit(sample_data, target_bitrate, epochs=10, batch_size=16)

# 預測最佳位元率
new_data = np.random.rand(1, 10, 3)  # 最近10秒的網路資料
predicted_bitrate = model.predict(new_data)[0][0]

print(f"預測的最佳位元率：{predicted_bitrate} Mbps")

use tokio::process::Command;
use std::sync::Arc;
use redis::Commands;

// 根據AI預測的位元率觸發動態轉碼
async fn transcode_video(redis_client: Arc<redis::Client>, stream_id: &str, bitrate: f64) {
    let output = Command::new("ffmpeg")
        .args(&[
            "-i", &format!("/tmp/{}.mp4", stream_id),
            "-b:v", &format!("{}M", bitrate),
            "-c:v", "libx264",
            "-preset", "ultrafast",
            "-f", "mp4",
            &format!("/tmp/{}_adaptive.mp4", stream_id),
        ])
        .output()
        .await
        .expect("轉碼啟動失敗");

    if output.status.success() {
        let mut conn = redis_client.get_connection().expect("連線Redis失敗");
        let _: () = conn.set(format!("stream:{}:bitrate", stream_id), bitrate)
            .expect("儲存轉碼位元率失敗");
    }
}

import json
import requests
from redis import Redis

# 建立Redis連線以取得預測位元率
redis_client = Redis(host='localhost', port=6379, db=0)
stream_key = "stream:123:bitrate"
predicted_bitrate = float(redis_client.get(stream_key))

# 根據AI預測調整WebRTC播放器品質
webrtc_config = {
    "videoBitrate": predicted_bitrate,
    "adaptationPolicy": "AI-driven",
    "bufferThreshold": 1.5,  # 秒數
    "recoveryStrategy": "quick-adjust"
}

# 將設定應用到播放器
player_api = "http://localhost:3000/api/player/config"
response = requests.post(player_api, json=webrtc_config)
print(f"設定更新狀態: {response.status_code}")
print(json.dumps(webrtc_config, indent=4))

[網路監控(Rust)] → [AI位元率預測(Python)] → [動態轉碼(Rust+FFmpeg)] → [自適應播放器(WebRTC)]

use actix_web::{web, App, HttpResponse, HttpServer};
use actix_web_actors::ws;
use serde::{Deserialize, Serialize};
use serde_json::Value;
use std::sync::Arc;
use tokio::sync::Mutex;
use redis::Commands;

#[derive(Serialize, Deserialize, Default)]
struct ViewerMetrics {
    watch_time: f64,
    pauses: usize,
    rewinds: usize,
    drop_off_rate: f64,
    content_type: String,
    device_type: String,
}

struct WebSocketSession {
    state: Arc<Mutex<ViewerMetrics>>,
    redis_client: Arc<redis::Client>,
    user_id: String,
}

impl actix::Actor for WebSocketSession {
    type Context = ws::WebsocketContext<Self>;
    
    fn started(&mut self, _ctx: &mut Self::Context) {
        println!("連線已建立: {}", self.user_id);
    }
}

impl actix::StreamHandler<Result<ws::Message, ws::ProtocolError>> for WebSocketSession {
    fn handle(&mut self, msg: Result<ws::Message, ws::ProtocolError>, ctx: &mut Self::Context) {
        match msg {
            Ok(ws::Message::Text(text)) => {
                self.process_message(text.to_string(), ctx);
            }
            Ok(ws::Message::Ping(msg)) => {
                ctx.pong(&msg);
            }
            _ => ()
        }
    }
}

impl WebSocketSession {
    async fn process_message(&mut self, data: String, _ctx: &mut ws::WebsocketContext<Self>) {
        if let Ok(json) = serde_json::from_str::<Value>(&data) {
            let mut metrics = ViewerMetrics::default();
            metrics.watch_time = json["watch_time"].as_f64().unwrap_or(0.0);
            metrics.pauses = json["pauses"].as_u64().unwrap_or(0) as usize;
            metrics.rewinds = json["rewinds"].as_u64().unwrap_or(0) as usize;
            metrics.drop_off_rate = json["drop_off_rate"].as_f64().unwrap_or(0.0);
            metrics.content_type = json["content_type"].as_str().unwrap_or("unknown").to_string();
            metrics.device_type = json["device_type"].as_str().unwrap_or("desktop").to_string();
            
            let mut conn = self.redis_client.get_connection().expect("無法連線Redis");
            let key = format!("viewer:{}:metrics", self.user_id);
            let _: () = conn.set(&key, serde_json::to_string(&metrics).unwrap())
                .expect("無法儲存觀眾指標");
        }
    }
}

async fn start_websocket(
    req: web::HttpRequest,
    ws: web::Payload,
    redis_client: web::Data<Arc<redis::Client>>
) -> HttpResponse {
    let user_id = req.match_info().get("user_id").unwrap_or("anonymous").to_string();
    let state = Arc::new(Mutex::new(ViewerMetrics::default()));
    
    ws::start(
        WebSocketSession {
            state,
            redis_client: redis_client.get_ref().clone(),
            user_id,
        },
        &req,
        ws,
    )
}

#[actix_web::main]
async fn main() -> std::io::Result<()> {
    let redis_client = Arc::new(redis::Client::open("redis://127.0.0.1/").unwrap());

    HttpServer::new(move || {
        App::new()
            .app_data(web::Data::new(redis_client.clone()))
            .route("/ws/{user_id}", web::get().to(start_websocket))
    })
    .bind("0.0.0.0:8080")?
    .workers(4)
    .run()
    .await
}