ShowenV2/src/plugins/http/mod.rs

//! HttpPlugin — Web UI + REST API
//!
//! 基于 warp 的 HTTP 服务，提供播放控制、配置管理、视频管理等 API。

mod routes;

use crate::core::config::AppConfig;
use crate::core::message::{Envelope, Message};
use crate::core::plugin::{Platform, Plugin, PluginContext, PluginInfo};
use anyhow::{Context, Result};
use serde::Serialize;
use std::sync::atomic::{AtomicBool, Ordering};
use std::sync::{Arc, Condvar, Mutex};
use std::thread::JoinHandle;
use tokio::sync::broadcast;
use tokio::sync::{oneshot, Mutex as AsyncMutex};

#[derive(Serialize)]
struct WsEvent<'a, T> {
    #[serde(rename = "type")]
    event_type: &'a str,
    data: T,
}

fn encode_ws_event<T: Serialize>(event_type: &str, data: T) -> Option<String> {
    match serde_json::to_string(&WsEvent { event_type, data }) {
        Ok(payload) => Some(payload),
        Err(error) => {
            eprintln!("[HttpPlugin] failed to serialize websocket event '{event_type}': {error}");
            None
        }
    }
}

struct PendingWifiResponse {
    version: u64,
    payload: Option<String>,
}

pub(crate) struct HttpState {
    wifi_request_lock: AsyncMutex<()>,
    wifi_response: Mutex<PendingWifiResponse>,
    wifi_response_cv: Condvar,
    last_wifi_result: Mutex<Option<String>>,
    config: Mutex<Arc<AppConfig>>,
    player_status: Mutex<crate::core::message::PlayerStatusData>,
    ble_ready: AtomicBool,
    ws_events: broadcast::Sender<String>,
    /// 动态插件管理状态（由 Custom 消息更新）
    plugin_states: Mutex<Vec<crate::core::service_manager::PluginStateInfo>>,
}

impl HttpState {
    fn new(config: Arc<AppConfig>) -> Self {
        let (ws_events, _) = broadcast::channel(32);
        let player_status = crate::core::message::PlayerStatusData {
            running: false,
            paused: !config.playback.auto_start,
            in_transition: false,
            current_index: 0,
            playlist_length: config.playlist.len(),
            current_video: config.playlist.first().map(|item| item.id.clone()),
        };

        Self {
            wifi_request_lock: AsyncMutex::new(()),
            wifi_response: Mutex::new(PendingWifiResponse {
                version: 0,
                payload: None,
            }),
            wifi_response_cv: Condvar::new(),
            last_wifi_result: Mutex::new(None),
            config: Mutex::new(config),
            player_status: Mutex::new(player_status),
            ble_ready: AtomicBool::new(false),
            ws_events,
            plugin_states: Mutex::new(Vec::new()),
        }
    }

    fn publish_wifi_result(&self, payload: String) {
        if let Ok(mut state) = self.wifi_response.lock() {
            state.version += 1;
            state.payload = Some(payload.clone());
            self.wifi_response_cv.notify_all();
        }

        if let Ok(mut last_wifi_result) = self.last_wifi_result.lock() {
            *last_wifi_result = Some(payload.clone());
        }

        let ws_payload = match serde_json::from_str::<serde_json::Value>(&payload) {
            Ok(value) => encode_ws_event("wifi_update", value),
            Err(_) => encode_ws_event("wifi_update", serde_json::json!({ "raw": payload })),
        };

        if let Some(ws_payload) = ws_payload {
            self.publish_ws(ws_payload);
        }
    }

    pub(crate) fn config(&self) -> Arc<AppConfig> {
        Arc::clone(
            &self
                .config
                .lock()
                .expect("http config state poisoned"),
        )
    }

    fn replace_config(&self, config: Arc<AppConfig>) {
        if let Ok(mut current) = self.config.lock() {
            *current = Arc::clone(&config);
        }

        if let Ok(mut player_status) = self.player_status.lock() {
            player_status.playlist_length = config.playlist.len();
            if player_status.current_video.is_none() {
                player_status.current_video = config.playlist.first().map(|item| item.id.clone());
            }
        }
    }

    pub(crate) fn player_status(&self) -> crate::core::message::PlayerStatusData {
        self.player_status
            .lock()
            .expect("http player status state poisoned")
            .clone()
    }

    fn update_player_status(&self, status: crate::core::message::PlayerStatusData) {
        if let Ok(mut current) = self.player_status.lock() {
            *current = status;
        }
    }

    pub(crate) fn ble_ready(&self) -> bool {
        self.ble_ready.load(Ordering::SeqCst)
    }

    fn publish_ws(&self, payload: String) {
        let _ = self.ws_events.send(payload);
    }

    pub(crate) fn ws_snapshots(&self) -> Vec<String> {
        let mut snapshots = Vec::new();

        if let Some(payload) = encode_ws_event("status_update", self.player_status()) {
            snapshots.push(payload);
        }

        let config = self.config();
        if let Some(payload) = encode_ws_event("config_update", config.as_ref()) {
            snapshots.push(payload);
        }

        if let Some(payload) = encode_ws_event(
            "ble_update",
            serde_json::json!({ "ready": self.ble_ready() }),
        ) {
            snapshots.push(payload);
        }

        if let Ok(last_wifi_result) = self.last_wifi_result.lock() {
            if let Some(raw) = last_wifi_result.as_ref() {
                let payload = match serde_json::from_str::<serde_json::Value>(raw) {
                    Ok(value) => encode_ws_event("wifi_update", value),
                    Err(_) => encode_ws_event("wifi_update", serde_json::json!({ "raw": raw })),
                };

                if let Some(payload) = payload {
                    snapshots.push(payload);
                }
            }
        }

        snapshots
    }

    pub(crate) fn ws_subscribe(&self) -> broadcast::Receiver<String> {
        self.ws_events.subscribe()
    }

    fn set_ble_ready(&self, ready: bool) {
        self.ble_ready.store(ready, Ordering::SeqCst);
        if let Some(payload) = encode_ws_event("ble_update", serde_json::json!({ "ready": ready }))
        {
            self.publish_ws(payload);
        }
    }

    pub(crate) fn plugin_states(&self) -> Vec<crate::core::service_manager::PluginStateInfo> {
        self.plugin_states
            .lock()
            .map(|s| s.clone())
            .unwrap_or_default()
    }

    fn update_plugin_states(&self, json: &str) {
        if let Ok(states) = serde_json::from_str::<Vec<crate::core::service_manager::PluginStateInfo>>(json) {
            if let Ok(mut current) = self.plugin_states.lock() {
                *current = states;
            }
        }
    }
}

pub struct HttpPlugin {
    ctx: Option<PluginContext>,
    state: Option<Arc<HttpState>>,
    shutdown_tx: Option<oneshot::Sender<()>>,
    server_thread: Option<JoinHandle<()>>,
}

impl HttpPlugin {
    pub fn new() -> Self {
        Self {
            ctx: None,
            state: None,
            shutdown_tx: None,
            server_thread: None,
        }
    }
}

impl Default for HttpPlugin {
    fn default() -> Self {
        Self::new()
    }
}

impl Plugin for HttpPlugin {
    fn id(&self) -> &str {
        "http"
    }

    fn info(&self) -> PluginInfo {
        PluginInfo {
            name: "HTTP API".to_string(),
            version: "0.2.0".to_string(),
            description: "Web UI + REST API (warp)".to_string(),
            platform: Platform::Any,
        }
    }

    fn dependencies(&self) -> Vec<String> {
        vec!["video".to_string()]
    }

    fn init(&mut self, ctx: PluginContext) -> Result<()> {
        self.state = Some(Arc::new(HttpState::new(Arc::clone(&ctx.config))));
        self.ctx = Some(ctx);
        Ok(())
    }

    fn start(&mut self) -> Result<()> {
        self.stop()?;

        let ctx = self
            .ctx
            .as_ref()
            .context("http plugin context is not initialized")?;

        if !ctx.config.remote_control.enabled {
            println!("[HttpPlugin] Remote control disabled, skip HTTP server startup");
            return Ok(());
        }

        let host = ctx.config.remote_control.host.clone();
        let port = ctx.config.remote_control.port;
        let tx = ctx.tx.clone();
        let state = Arc::clone(
            self.state
                .as_ref()
                .context("http plugin state is not initialized")?,
        );

        let (shutdown_tx, shutdown_rx) = oneshot::channel();

        let server_thread = std::thread::spawn(move || {
            let runtime = match tokio::runtime::Builder::new_multi_thread()
                .enable_all()
                .build()
            {
                Ok(runtime) => runtime,
                Err(error) => {
                    eprintln!("[HttpPlugin] failed to create tokio runtime: {error}");
                    return;
                }
            };

            runtime.block_on(async move {
                let routes = routes::build_routes(tx.clone(), state);
                let addr: std::net::SocketAddr = match format!("{host}:{port}").parse() {
                    Ok(addr) => addr,
                    Err(error) => {
                        eprintln!("[HttpPlugin] invalid listen address {host}:{port}: {error}");
                        return;
                    }
                };

                if let Err(error) = tx.send(Envelope {
                    from: "http".to_string(),
                    to: crate::core::message::Destination::Manager,
                    message: Message::PluginReady("http".to_string()),
                }) {
                    eprintln!("[HttpPlugin] failed to report ready state: {error}");
                }

                println!("[HttpPlugin] listening on http://{addr}");
                warp::serve(routes)
                    .bind_with_graceful_shutdown(addr, async move {
                        let _ = shutdown_rx.await;
                    })
                    .1
                    .await;
            });
        });

        self.shutdown_tx = Some(shutdown_tx);
        self.server_thread = Some(server_thread);

        Ok(())
    }

    fn handle_message(&mut self, msg: Message) -> Result<()> {
        let state = match self.state.as_ref() {
            Some(state) => state,
            None => return Ok(()),
        };

        match msg {
            Message::WifiResult(payload) => state.publish_wifi_result(payload),
            Message::PlayerStatus(status) => {
                state.update_player_status(status.clone());
                if let Some(payload) = encode_ws_event("status_update", &status) {
                    state.publish_ws(payload);
                }
            }
            Message::ConfigReloaded(config) => {
                state.replace_config(Arc::new(config.clone()));
                if let Some(payload) = encode_ws_event("config_update", &config) {
                    state.publish_ws(payload);
                }
            }
            Message::StateChanged {
                old_state,
                new_state,
            } => {
                if let Some(payload) = encode_ws_event(
                    "state_update",
                    serde_json::json!({ "old_state": old_state, "new_state": new_state }),
                ) {
                    state.publish_ws(payload);
                }
            }
            Message::PluginReady(ref id) if id == "ble" => state.set_ble_ready(true),
            Message::Shutdown => state.set_ble_ready(false),
            Message::Custom { ref kind, ref payload } if kind == "plugin_states" => {
                state.update_plugin_states(payload);
            }
            _ => {}
        }

        Ok(())
    }

    fn stop(&mut self) -> Result<()> {
        if let Some(shutdown_tx) = self.shutdown_tx.take() {
            let _ = shutdown_tx.send(());
        }

        if let Some(server_thread) = self.server_thread.take() {
            server_thread
                .join()
                .map_err(|_| anyhow::anyhow!("http server thread panicked"))?;
        }

        Ok(())
    }
}