mineral-server 0.5.10

//! [`Server`]:audio engine + scheduler + PlayerCore + PCM puller 的单进程收纳容器。

use std::sync::Arc;
use std::time::{Duration, Instant};

use crate::persistence::ServerStore;
use mineral_audio::{AudioHandle, AudioMode};
use mineral_channel_core::MusicChannel;
use mineral_playback::PlaybackRegistry;
use mineral_protocol::{Event, PlayMode};
use mineral_task::Scheduler;
use tokio::net::UnixListener;
use tokio::sync::{Notify, broadcast};

use crate::client::ClientHandle;
use crate::config::ServerConfig;
use crate::ipc::{ConnRegistry, SessionServices};
use crate::media_cache::MediaCache;
use crate::pcm_relay::PcmRelay;
use crate::player::PlayerCore;
use crate::publisher::DomainPublishers;

/// Catalog channels and playback providers assembled for one server process.
#[non_exhaustive]
#[derive(typed_builder::TypedBuilder)]
pub struct SourceBackends {
    /// Catalog, library, and user-data connectors.
    pub(crate) channels: Vec<Arc<dyn MusicChannel>>,

    /// Playback resource providers keyed by source identity.
    pub(crate) playback: PlaybackRegistry,
}

impl SourceBackends {
    /// Registered catalog channels, used to seed script capabilities before server startup.
    pub fn channels(&self) -> &[Arc<dyn MusicChannel>] {
        &self.channels
    }
}

/// 后台 server。`spawn` 启动 audio engine + scheduler + PlayerCore + PCM puller,
/// 投递初始任务,对外发 [`ClientHandle`]。
pub struct Server {
    /// PlayerCore — 业务状态 + 自治 auto-next + events 中继。
    player: PlayerCore,

    /// 领域发布器 — 会话订阅的共享快照源。
    publishers: DomainPublishers,

    /// 在线连接注册表,与 accept loop 共享:心跳据此报在线 client 数。
    connections: Arc<ConnRegistry>,

    /// Event 推送 hub:daemon 内生产者经
    /// [`Server::event_sink`] 拿发送端;会话按订阅集过滤下发。
    events: broadcast::Sender<Event>,

    /// daemon 级关停通知:client 经 IPC 发 `Request::Shutdown` 时唤醒,daemon
    /// 入口经 [`Server::shutdown_requested`] 等待。
    shutdown: Arc<Notify>,
}

impl Server {
    /// 启动 audio engine + scheduler + PlayerCore + PCM puller,投递初始任务。
    ///
    /// # Params:
    ///   - `sources`: Catalog channels and playback providers. Empty sets are valid.
    ///   - `audio_mode`: 音频后端选择(env / config resolve 后的最终值);无设备时 `Auto` 降级而非失败。
    ///   - `persist`: 持久化句柄,透传给 `PlayerCore::spawn`(收藏 / 歌曲元数据等经它落库)。
    ///   - `config`: daemon 配置切片(引擎参数 / 音质 / 缓存容量 / 各间隔)。
    ///   - `config_tree`: 已校验的 daemon 私有配置底树，供 session 覆盖与重载使用。
    ///   - `script`: 脚本线程投递句柄(daemon 启用脚本时传 `Some`,其余 `None`)。
    pub async fn spawn(
        sources: SourceBackends,
        audio_mode: AudioMode,
        persist: ServerStore,
        config: ServerConfig,
        config_tree: serde_json::Value,
        script: Option<mineral_script::ScriptSender>,
        stats: crate::StatsRecorder,
    ) -> Result<Self, crate::Error> {
        let channels = &sources.channels;
        mineral_log::debug!(target: "server", channels = channels.len(), "spawning server components");
        let scheduler = Scheduler::new(channels, *config.channel_workers_per());
        let (audio, spectrum_tap) = AudioHandle::spawn(audio_mode, config.engine().clone())?;
        mineral_log::debug!(target: "server", "audio engine ready");
        let media_cache = open_media_cache(&persist, *config.audio_cache_capacity()).await;
        let music_dir = crate::download::open_env(config.download().dir().as_deref());
        // 每个订阅者最多保留 256 条 advisory event；落后的 event_pump 收 Lagged 并跳过旧帧。
        let (events, _) = broadcast::channel::<Event>(/*capacity*/ 256);
        let notify = crate::notify::Notifier::new(events.clone(), script);
        let player = PlayerCore::spawn(
            audio,
            scheduler,
            sources,
            crate::player::PlayerStorage {
                persist,
                media_cache,
                music_dir,
            },
            crate::player::SpawnConfig {
                slices: &config,
                tree: config_tree,
            },
            crate::player::Sinks { notify, stats },
        )?;
        // 读回上次会话:恢复播放模式(其余字段仅打日志,不自动恢复队列/进度)。
        // 同步 await:保证在 serve / 首次 PlayerSync 之前生效,client 一连上看到的就是恢复后的模式。
        let session_restored = restore_last_session(&player).await;
        // 埋点:daemon 启动(app_lifecycle;audio 已 spawn 故后端已定,会话恢复已知)。
        // daemon 总要发,recorder 降级 disabled 时此调用静默 no-op。
        let audio_backend = match player.audio_snapshot().backend {
            mineral_audio::AudioBackend::Device => mineral_stats::AudioBackend::Device,
            mineral_audio::AudioBackend::Null => mineral_stats::AudioBackend::Null,
        };
        player
            .inner
            .stats
            .daemon_started(audio_backend, session_restored);
        // 第一次 initial loads — 为「daemon 起来无 client 也能后台 prefetch」考虑。
        player.refresh_initial_loads();
        let pcm = PcmRelay::spawn(spectrum_tap, player.audio().clone());
        let publishers = crate::publisher::spawn(&player, events.clone(), pcm);
        let connections = Arc::new(ConnRegistry::new());
        tokio::spawn(heartbeat(
            player.clone(),
            Arc::clone(&connections),
            *config.timing().heartbeat_secs(),
        ));
        mineral_log::debug!(target: "server", "server components ready");
        Ok(Self {
            player,
            publishers,
            connections,
            events,
            shutdown: Arc::new(Notify::new()),
        })
    }

    /// Event 推送 hub 的发送端(clone 廉价)。生产者(脚本运行时 / 测试)往里
    /// send,所有已连接且订阅了对应类别的 client 都会收到;无 client 时 send
    /// 返回 `Err`,按 advisory 语义丢弃即可。
    pub fn event_sink(&self) -> broadcast::Sender<Event> {
        self.events.clone()
    }

    /// 接上脚本双泵(命令 → player 执行面;推送 → event hub)。daemon 入口在
    /// [`Server::spawn`] 之后调一次;无脚本时泵闲置无害。
    ///
    /// # Params:
    ///   - `pumps`: [`crate::ScriptParts::spawn_runtime`] 拆出的泵接线件
    pub fn attach_script_pumps(&self, pumps: crate::ScriptPumps) -> crate::ScriptReloadParts {
        pumps.start(self.player.clone(), self.events.clone())
    }

    /// 拿一个 client handle。clone 廉价(全 Arc 内部),可任意复制给多处调用。
    pub fn client(&self) -> ClientHandle {
        ClientHandle::new(self.player.clone())
    }

    /// 接入系统媒体服务(Linux MPRIS):上报当前播放、响应媒体键 / 桌面控件。
    ///
    /// 仅 daemon 模式调用 —— 控制的是常驻播放。注册失败(无 D-Bus session 等)
    /// 返回 `Err`,调用方应降级而非中止 daemon。
    pub fn start_media_service(&self) -> Result<(), crate::Error> {
        crate::media::start(self.player.clone()).map_err(crate::Error::from)
    }

    /// 消费 server 句柄。daemon 关停路径在 accept loop 结束后调用,
    /// 随后由 Tokio runtime 退出终止其余后台任务。
    pub async fn shutdown(self) {
        self.player.shutdown_downloads().await;
        drop(self);
    }

    /// 等待 client 经 IPC 发来的 daemon 关停请求([`mineral_protocol::Request::Shutdown`])。
    /// daemon 入口把它与 accept loop / 信号一起 select,任一唤醒即走 graceful 收尾。
    /// `Notify` 带 permit:请求先于本调用到达也不丢。
    pub async fn shutdown_requested(&self) {
        self.shutdown.notified().await;
    }

    /// IPC accept loop:每条新 connection 走会话握手 + 结构化消息管线(请求合批
    /// 配对 + 按订阅主题的状态发布)。多 client:连接数不设限,每条独立 task /
    /// writer / 订阅泵。
    ///
    /// 每条新 connection 接受后,内部重跑 `PlayerCore::refresh_initial_loads`
    /// ——数据现状已由订阅初始快照兜底,这里是连接时顺手保鲜远端数据。
    pub async fn serve(&self, listener: UnixListener) -> Result<(), crate::Error> {
        let player = self.player.clone();
        let on_connect = move || player.refresh_initial_loads();
        let services = Arc::new(SessionServices {
            client: self.client(),
            registry: Arc::clone(&self.connections),
            publishers: self.publishers.clone(),
            shutdown: Arc::clone(&self.shutdown),
            on_connect: Arc::new(on_connect),
        });
        crate::ipc::run(listener, services).await
    }
}

/// 打开音频本体缓存(`audio_cache` 表落 `persist` 的 `mineral.db`);目录解析 / open 失败时
/// `warn` 并降级到 [`MediaCache::disabled`](不阻断 daemon 启动)。
///
/// # Params:
///   - `persist`: 持久化句柄
///   - `capacity`: 缓存容量上限(字节,配置 `cache.audio_capacity`)
async fn open_media_cache(persist: &ServerStore, capacity: u64) -> MediaCache {
    let dir = match mineral_paths::audio_cache_dir() {
        Ok(d) => d,
        Err(e) => {
            mineral_log::warn!(target: "server", error = mineral_log::chain(&e), "音频缓存目录解析失败,降级禁用");
            return MediaCache::disabled();
        }
    };
    match MediaCache::open(persist, dir, capacity).await {
        Ok(cache) => cache,
        Err(e) => {
            mineral_log::warn!(target: "server", error = mineral_log::chain(&e), "音频缓存打开失败,降级禁用");
            MediaCache::disabled()
        }
    }
}

/// 启动时读回上次会话:**恢复播放模式**,其余字段(队列 / 进度 / 音量)仅打日志不应用
/// (不自动恢复播放)。模式名解析不出(脏数据)warn 后保持默认;读不到走 debug;
/// 读出错仅 warn,不影响 daemon 启动。
///
/// # Return:
///   是否读到并采纳了上次会话(app_lifecycle.session_restored 用):`Ok(Some)` 且模式
///   可解析为 `true`;无历史 / 脏模式 / 读出错为 `false`(均视为「未恢复」)。
async fn restore_last_session(player: &PlayerCore) -> bool {
    match player.load_session().await {
        Ok(Some(snap)) => {
            let restored = match PlayMode::from_name(&snap.play_mode) {
                Some(mode) => {
                    player.restore_play_mode(mode);
                    true
                }
                None => {
                    mineral_log::warn!(
                        target: "session",
                        play_mode = %snap.play_mode,
                        "上次会话播放模式无法解析,保持默认"
                    );
                    false
                }
            };
            mineral_log::info!(
                target: "session",
                queue_len = snap.queue.len(),
                position_ms = snap.position_ms,
                play_mode = %snap.play_mode,
                restored_play_mode = restored,
                "读到上次会话"
            );
            restored
        }
        Ok(None) => {
            mineral_log::debug!(target: "session", "无历史会话");
            false
        }
        Err(e) => {
            mineral_log::warn!(target: "session", error = mineral_log::chain(&e), "读取上次会话失败");
            false
        }
    }
}

/// 心跳:每 `interval_secs` 秒把内部状态快照打一条 info,供事后回溯间歇性问题
/// (出问题时往往没提前开 debug,有心跳就能看到那个时间点系统在干嘛)。
///
/// # Params:
///   - `interval_secs`: 心跳间隔(秒,配置 `heartbeat_secs`)
async fn heartbeat(player: PlayerCore, connections: Arc<ConnRegistry>, interval_secs: u64) {
    let start = Instant::now();
    let mut tick = tokio::time::interval(Duration::from_secs(interval_secs));
    loop {
        tick.tick().await;
        // in-process 直读 State 摘日志字段,不走含整队列 clone 的全量快照。
        let (song_id, play_mode, queue_len, cursor, format, bitrate_kbps, lyrics_loaded) = player
            .with_state(|st| {
                (
                    st.current_song
                        .as_ref()
                        .map_or_else(|| "-".to_owned(), |s| s.id.as_str().to_owned()),
                    st.play_mode,
                    st.queue.len(),
                    st.cursor,
                    st.media_info
                        .as_ref()
                        .and_then(|info| info.format.as_ref())
                        .map_or_else(|| "-".to_owned(), |f| f.as_str().to_owned()),
                    st.media_info
                        .as_ref()
                        .and_then(|info| info.bitrate_bps)
                        .map_or_else(|| "-".to_owned(), |b| (b / 1000).to_string()),
                    st.current_lyrics.is_some(),
                )
            });
        let audio = player.audio().snapshot();
        let tasks = player.task_snapshot();
        mineral_log::info!(
            target: "heartbeat",
            uptime_s = start.elapsed().as_secs(),
            clients_online = connections.online(),
            playing = audio.playing,
            position_ms = audio.position_ms,
            duration_ms = audio.duration_ms,
            volume_pct = audio.volume_pct,
            song_id,
            play_mode = ?play_mode,
            queue_len,
            cursor = ?cursor,
            format,
            bitrate_kbps,
            lyrics_loaded,
            prefetched = player.prefetched_ready(),
            tasks_running = tasks.running,
            tasks_by_kind = ?tasks.by_kind,
            "status"
        );
    }
}