snapcast-server 0.6.0

#![deny(unsafe_code)]
#![warn(clippy::redundant_closure)]
#![warn(clippy::implicit_clone)]
#![warn(clippy::uninlined_format_args)]
#![warn(missing_docs)]

//! Snapcast server library — embeddable synchronized multiroom audio server.
//!
//! See also: [`snapcast-client`](https://docs.rs/snapcast-client) for the client library.
//! # Architecture
//!
//! The server is built around a channel-based API matching `snapcast-client`:
//!
//! - [`SnapServer`] is the main entry point
//! - [`ServerEvent`] flows from server → consumer (client connected, stream status, custom messages)
//! - [`ServerCommand`] flows from consumer → server (typed mutations, custom messages, stop)
//!
//! # Example
//!
//! ```no_run
//! use snapcast_server::{SnapServer, ServerConfig, ServerEvent, ServerCommand};
//!
//! # async fn example() -> anyhow::Result<()> {
//! let config = ServerConfig::default();
//! let (mut server, mut events) = SnapServer::new(config);
//! let _audio_tx = server.add_stream("default");
//! let cmd = server.command_sender();
//!
//! tokio::spawn(async move {
//!     while let Some(event) = events.recv().await {
//!         match event {
//!             ServerEvent::ClientConnected { id, name, .. } => {
//!                 tracing::info!(id, name, "Client connected");
//!             }
//!             _ => {}
//!         }
//!     }
//! });
//!
//! server.run().await?;
//! # Ok(())
//! # }
//! ```

use std::sync::Arc;

use tokio::sync::{broadcast, mpsc};

// Re-export proto types that embedders need
#[cfg(feature = "custom-protocol")]
pub use snapcast_proto::CustomMessage;
#[cfg(feature = "encryption")]
pub use snapcast_proto::DEFAULT_ENCRYPTION_PSK;
pub use snapcast_proto::SampleFormat;
pub use snapcast_proto::{DEFAULT_SAMPLE_FORMAT, DEFAULT_STREAM_PORT};

const EVENT_CHANNEL_SIZE: usize = 256;
const COMMAND_CHANNEL_SIZE: usize = 64;
const AUDIO_CHANNEL_SIZE: usize = 256;

/// Channel size for F32 embedded sources — backpressure from encoder pacing.
const F32_CHANNEL_SIZE: usize = 1;

/// Audio data pushed by the consumer — either f32 or raw PCM.
#[derive(Debug, Clone)]
pub enum AudioData {
    /// Interleaved f32 samples (from DSP, EQ, AirPlay receivers).
    /// Range: -1.0 to 1.0.
    F32(Vec<f32>),
    /// Raw interleaved PCM bytes at the stream's configured sample format
    /// (from pipe/file/process readers). Byte order: little-endian.
    Pcm(Vec<u8>),
}

/// A timestamped audio frame for server input.
#[derive(Debug, Clone)]
pub struct AudioFrame {
    /// Audio samples.
    pub data: AudioData,
    /// Timestamp in microseconds (server time).
    pub timestamp_usec: i64,
}

/// Buffered sender for F32 audio that handles chunking, timestamping, and gap detection.
///
/// Accumulates variable-size F32 sample buffers and emits fixed-size 20ms chunks
/// with monotonic timestamps. Automatically resets on playback gaps (>500ms).
///
/// Created by [`SnapServer::add_f32_stream`].
pub struct F32AudioSender {
    tx: mpsc::Sender<AudioFrame>,
    buf: Vec<f32>,
    chunk_samples: usize,
    channels: u16,
    sample_rate: u32,
    ts: Option<time::ChunkTimestamper>,
    last_send: std::time::Instant,
}

impl F32AudioSender {
    fn new(tx: mpsc::Sender<AudioFrame>, sample_rate: u32, channels: u16) -> Self {
        let chunk_samples = (sample_rate as usize * 20 / 1000) * channels as usize;
        Self {
            tx,
            buf: Vec::with_capacity(chunk_samples * 2),
            chunk_samples,
            channels,
            sample_rate,
            ts: None,
            last_send: std::time::Instant::now(),
        }
    }

    /// Push interleaved F32 samples. Variable-size input is accumulated and
    /// emitted as fixed 20ms chunks. Returns when all complete chunks are sent.
    pub async fn send(
        &mut self,
        samples: &[f32],
    ) -> Result<(), mpsc::error::SendError<AudioFrame>> {
        let now = std::time::Instant::now();
        if now.duration_since(self.last_send) > std::time::Duration::from_millis(500) {
            self.ts = None;
            self.buf.clear();
        }
        self.last_send = now;

        self.buf.extend_from_slice(samples);
        let ch = self.channels.max(1) as usize;
        while self.buf.len() >= self.chunk_samples {
            let chunk: Vec<f32> = self.buf.drain(..self.chunk_samples).collect();
            let frames = (self.chunk_samples / ch) as u32;
            let ts = self
                .ts
                .get_or_insert_with(|| time::ChunkTimestamper::new(self.sample_rate));
            let timestamp_usec = ts.next(frames);
            self.tx
                .send(AudioFrame {
                    data: AudioData::F32(chunk),
                    timestamp_usec,
                })
                .await?;
        }
        Ok(())
    }

    /// Flush remaining samples (< 20ms) as a final short chunk.
    /// Call at end-of-track to avoid losing the last few milliseconds.
    pub async fn flush(&mut self) -> Result<(), mpsc::error::SendError<AudioFrame>> {
        if self.buf.is_empty() {
            return Ok(());
        }
        let chunk: Vec<f32> = self.buf.drain(..).collect();
        let ch = self.channels.max(1) as usize;
        let frames = (chunk.len() / ch) as u32;
        let ts = self
            .ts
            .get_or_insert_with(|| time::ChunkTimestamper::new(self.sample_rate));
        let timestamp_usec = ts.next(frames);
        self.tx
            .send(AudioFrame {
                data: AudioData::F32(chunk),
                timestamp_usec,
            })
            .await
    }
}

/// An encoded audio chunk ready to be sent to clients.
#[derive(Debug, Clone)]
pub struct WireChunkData {
    /// Stream ID this chunk belongs to.
    pub stream_id: String,
    /// Server timestamp in microseconds.
    pub timestamp_usec: i64,
    /// Encoded audio data.
    pub data: Vec<u8>,
}

pub mod auth;
#[cfg(feature = "encryption")]
pub(crate) mod crypto;
pub(crate) mod encoder;
#[cfg(feature = "mdns")]
pub(crate) mod mdns;
pub(crate) mod session;
pub(crate) mod state;
pub mod status;
pub(crate) mod stream;
pub mod time;

/// Settings update pushed to a streaming client via binary protocol.
#[derive(Debug, Clone)]
pub struct ClientSettingsUpdate {
    /// Target client ID.
    pub client_id: String,
    /// Buffer size in ms.
    pub buffer_ms: i32,
    /// Latency offset in ms.
    pub latency: i32,
    /// Volume (0–100).
    pub volume: u16,
    /// Mute state.
    pub muted: bool,
}

/// Events emitted by the server to the consumer.
#[derive(Debug)]
#[non_exhaustive]
pub enum ServerEvent {
    /// A client connected via the binary protocol.
    ClientConnected {
        /// Unique client identifier.
        id: String,
        /// Client hostname.
        name: String,
        /// Client MAC address.
        mac: String,
    },
    /// A client disconnected.
    ClientDisconnected {
        /// Unique client identifier.
        id: String,
    },
    /// A client's volume changed.
    ClientVolumeChanged {
        /// Client ID.
        client_id: String,
        /// New volume (0–100).
        volume: u16,
        /// Mute state.
        muted: bool,
    },
    /// A client's latency changed.
    ClientLatencyChanged {
        /// Client ID.
        client_id: String,
        /// New latency in ms.
        latency: i32,
    },
    /// A client's name changed.
    ClientNameChanged {
        /// Client ID.
        client_id: String,
        /// New name.
        name: String,
    },
    /// A group's stream assignment changed.
    GroupStreamChanged {
        /// Group ID.
        group_id: String,
        /// New stream ID.
        stream_id: String,
    },
    /// A group's mute state changed.
    GroupMuteChanged {
        /// Group ID.
        group_id: String,
        /// Mute state.
        muted: bool,
    },
    /// A stream's status changed (playing, idle, unknown).
    StreamStatus {
        /// Stream identifier.
        stream_id: String,
        /// New status.
        status: String,
    },
    /// Stream metadata/properties changed.
    StreamMetaChanged {
        /// Stream identifier.
        stream_id: String,
        /// Updated properties.
        metadata: std::collections::HashMap<String, serde_json::Value>,
    },
    /// A group's name changed.
    GroupNameChanged {
        /// Group ID.
        group_id: String,
        /// New name.
        name: String,
    },
    /// Server state changed — groups were reorganized (created, deleted, merged).
    ///
    /// Emitted after structural changes like `SetGroupClients` or `DeleteClient`
    /// when the group topology changes. Mirrors `Server.OnUpdate` in the C++ snapserver.
    /// The consumer should re-read server status via `GetStatus`.
    ServerUpdated,
    /// A stream control command was received (play, pause, next, seek, etc.).
    ///
    /// The library forwards this to the embedder since it doesn't own stream readers.
    StreamControl {
        /// Stream ID.
        stream_id: String,
        /// Command name.
        command: String,
        /// Optional parameters.
        params: serde_json::Value,
    },
    /// Custom binary protocol message from a streaming client.
    #[cfg(feature = "custom-protocol")]
    CustomMessage {
        /// Client ID.
        client_id: String,
        /// The custom message.
        message: snapcast_proto::CustomMessage,
    },
}

/// Commands the consumer sends to the server.
#[derive(Debug)]
#[non_exhaustive]
pub enum ServerCommand {
    /// Set a client's volume.
    SetClientVolume {
        /// Client ID.
        client_id: String,
        /// Volume (0–100).
        volume: u16,
        /// Mute state.
        muted: bool,
    },
    /// Set a client's latency offset.
    SetClientLatency {
        /// Client ID.
        client_id: String,
        /// Latency in milliseconds.
        latency: i32,
    },
    /// Set a client's display name.
    SetClientName {
        /// Client ID.
        client_id: String,
        /// New name.
        name: String,
    },
    /// Assign a stream to a group.
    SetGroupStream {
        /// Group ID.
        group_id: String,
        /// Stream ID.
        stream_id: String,
    },
    /// Mute/unmute a group.
    SetGroupMute {
        /// Group ID.
        group_id: String,
        /// Mute state.
        muted: bool,
    },
    /// Set a group's display name.
    SetGroupName {
        /// Group ID.
        group_id: String,
        /// New name.
        name: String,
    },
    /// Move clients to a group.
    SetGroupClients {
        /// Group ID.
        group_id: String,
        /// Client IDs.
        clients: Vec<String>,
    },
    /// Delete a client from the server.
    DeleteClient {
        /// Client ID.
        client_id: String,
    },
    /// Set stream metadata (artist, title, album, etc.).
    SetStreamMeta {
        /// Stream ID.
        stream_id: String,
        /// Metadata key-value pairs.
        metadata: std::collections::HashMap<String, serde_json::Value>,
    },
    /// Dynamically add a stream source.
    AddStream {
        /// Stream source URI (e.g. `pipe:///tmp/snapfifo?name=default`).
        uri: String,
        /// Response: the stream ID assigned.
        response_tx: tokio::sync::oneshot::Sender<Result<String, String>>,
    },
    /// Remove a stream source.
    RemoveStream {
        /// Stream ID to remove.
        stream_id: String,
    },
    /// Forward a control command to a stream (play, pause, next, etc.).
    StreamControl {
        /// Stream ID.
        stream_id: String,
        /// Command name (e.g. "next", "previous", "pause", "seek").
        command: String,
        /// Optional command parameter (e.g. seek position).
        params: serde_json::Value,
    },
    /// Get full server status.
    GetStatus {
        /// Response channel.
        response_tx: tokio::sync::oneshot::Sender<status::ServerStatus>,
    },
    /// Send a custom binary protocol message to a streaming client.
    #[cfg(feature = "custom-protocol")]
    SendToClient {
        /// Target client ID.
        client_id: String,
        /// The custom message.
        message: snapcast_proto::CustomMessage,
    },
    /// Stop the server gracefully.
    Stop,
}

/// Default codec based on compiled features.
fn default_codec() -> &'static str {
    #[cfg(feature = "flac")]
    return "flac";
    #[cfg(all(feature = "f32lz4", not(feature = "flac")))]
    return "f32lz4";
    #[cfg(not(any(feature = "flac", feature = "f32lz4")))]
    return "pcm";
}

/// Server configuration for the embeddable library.
pub struct ServerConfig {
    /// TCP port for binary protocol (client connections). Default: 1704.
    pub stream_port: u16,
    /// Audio buffer size in milliseconds. Default: 1000.
    pub buffer_ms: u32,
    /// Default codec: "f32lz4", "pcm", "opus", "ogg". Default: "f32lz4".
    pub codec: String,
    /// Default sample format. Default: 48000:16:2.
    pub sample_format: String,
    /// mDNS service type. Default: "_snapcast._tcp.local.".
    #[cfg(feature = "mdns")]
    pub mdns_service_type: String,
    /// Enable mDNS advertisement. Default: true (when mdns feature is compiled in).
    #[cfg(feature = "mdns")]
    pub mdns_enabled: bool,
    /// mDNS service name. Default: "Snapserver".
    #[cfg(feature = "mdns")]
    pub mdns_name: String,
    /// Auth validator for streaming clients. `None` = no auth required.
    pub auth: Option<std::sync::Arc<dyn auth::AuthValidator>>,
    /// Pre-shared key for f32lz4 encryption. `None` = no encryption.
    #[cfg(feature = "encryption")]
    pub encryption_psk: Option<String>,
    /// Path to persist server state (clients, groups). `None` = no persistence.
    pub state_file: Option<std::path::PathBuf>,
    /// Send audio data to muted clients. Default: false (skip muted, saves bandwidth).
    pub send_audio_to_muted: bool,
}

impl Default for ServerConfig {
    fn default() -> Self {
        Self {
            stream_port: snapcast_proto::DEFAULT_STREAM_PORT,
            buffer_ms: 1000,
            codec: default_codec().into(),
            sample_format: "48000:16:2".into(),
            #[cfg(feature = "mdns")]
            mdns_service_type: "_snapcast._tcp.local.".into(),
            #[cfg(feature = "mdns")]
            mdns_enabled: true,
            #[cfg(feature = "mdns")]
            mdns_name: "Snapserver".into(),
            auth: None,
            #[cfg(feature = "encryption")]
            encryption_psk: None,
            state_file: None,
            send_audio_to_muted: false,
        }
    }
}

/// Per-stream configuration. If `None`, inherits from [`ServerConfig`].
#[derive(Debug, Clone, Default)]
pub struct StreamConfig {
    /// Codec override (e.g. "flac", "f32lz4", "opus", "ogg", "pcm").
    pub codec: Option<String>,
    /// Sample format override (e.g. "48000:16:2").
    pub sample_format: Option<String>,
}

/// The embeddable Snapcast server.
pub struct SnapServer {
    config: ServerConfig,
    event_tx: mpsc::Sender<ServerEvent>,
    command_tx: mpsc::Sender<ServerCommand>,
    command_rx: Option<mpsc::Receiver<ServerCommand>>,
    /// Named audio streams — each gets its own encoder at run().
    streams: Vec<(String, StreamConfig, mpsc::Receiver<AudioFrame>)>,
    /// Broadcast channel for encoded chunks → sessions.
    chunk_tx: broadcast::Sender<WireChunkData>,
}

/// Spawn a per-stream encode loop on a dedicated thread.
///
/// Receives `AudioFrame`, passes `AudioData` directly to the encoder,
/// and broadcasts encoded `WireChunkData` to sessions.
fn spawn_stream_encoder(
    stream_id: String,
    mut rx: mpsc::Receiver<AudioFrame>,
    mut enc: Box<dyn encoder::Encoder>,
    chunk_tx: broadcast::Sender<WireChunkData>,
    sample_rate: u32,
    channels: u16,
) {
    std::thread::spawn(move || {
        let rt = tokio::runtime::Builder::new_current_thread()
            .enable_time()
            .build()
            .expect("encoder runtime");

        rt.block_on(async {
            let mut next_tick: Option<tokio::time::Instant> = None;
            while let Some(frame) = rx.recv().await {
                // Pace F32 sources to realtime (pipe sources pace naturally via blocking read)
                if let AudioData::F32(ref samples) = frame.data {
                    let num_frames = samples.len() / channels.max(1) as usize;
                    let chunk_dur = std::time::Duration::from_micros(
                        (num_frames as u64 * 1_000_000) / sample_rate as u64,
                    );
                    let now = tokio::time::Instant::now();
                    let tick = next_tick.get_or_insert(now);
                    // Reset on gap (>500ms behind wall clock)
                    if now.checked_duration_since(*tick + chunk_dur)
                        > Some(std::time::Duration::from_millis(500))
                    {
                        *tick = now;
                    }
                    *tick += chunk_dur;
                    tokio::time::sleep_until(*tick).await;
                }
                match enc.encode(&frame.data) {
                    Ok(encoded) if !encoded.data.is_empty() => {
                        let _ = chunk_tx.send(WireChunkData {
                            stream_id: stream_id.clone(),
                            timestamp_usec: frame.timestamp_usec,
                            data: encoded.data,
                        });
                    }
                    Err(e) => {
                        tracing::warn!(stream = %stream_id, error = %e, "Encode failed");
                    }
                    _ => {} // encoder buffering
                }
            }
        });
    });
}

/// Convert f32 samples to PCM bytes at the given bit depth.
impl SnapServer {
    /// Create a new server. Returns the server and event receiver.
    pub fn new(config: ServerConfig) -> (Self, mpsc::Receiver<ServerEvent>) {
        let (event_tx, event_rx) = mpsc::channel(EVENT_CHANNEL_SIZE);
        let (command_tx, command_rx) = mpsc::channel(COMMAND_CHANNEL_SIZE);
        let (chunk_tx, _) = broadcast::channel(256);
        let server = Self {
            config,
            event_tx,
            command_tx,
            command_rx: Some(command_rx),
            streams: Vec::new(),
            chunk_tx,
        };
        (server, event_rx)
    }

    /// Add a named audio stream. Returns a sender for pushing audio frames.
    ///
    /// Uses the server's default codec and sample format.
    pub fn add_stream(&mut self, name: &str) -> mpsc::Sender<AudioFrame> {
        self.add_stream_with_config(name, StreamConfig::default())
    }

    /// Add a named F32 audio stream with automatic chunking and timestamping.
    ///
    /// Returns an [`F32AudioSender`] that accepts variable-size F32 sample buffers
    /// and handles 20ms chunking, monotonic timestamps, and gap detection internally.
    ///
    /// # Errors
    /// Returns an error if the server's `sample_format` cannot be parsed.
    pub fn add_f32_stream(&mut self, name: &str) -> Result<F32AudioSender, String> {
        let sf: SampleFormat =
            self.config.sample_format.parse().map_err(|e| {
                format!("invalid sample_format '{}': {e}", self.config.sample_format)
            })?;
        let (tx, rx) = mpsc::channel(F32_CHANNEL_SIZE);
        self.streams
            .push((name.to_string(), StreamConfig::default(), rx));
        Ok(F32AudioSender::new(tx, sf.rate(), sf.channels()))
    }

    /// Add a named audio stream with per-stream codec/format overrides.
    pub fn add_stream_with_config(
        &mut self,
        name: &str,
        config: StreamConfig,
    ) -> mpsc::Sender<AudioFrame> {
        let (tx, rx) = mpsc::channel(AUDIO_CHANNEL_SIZE);
        self.streams.push((name.to_string(), config, rx));
        tx
    }

    /// Get a cloneable command sender.
    pub fn command_sender(&self) -> mpsc::Sender<ServerCommand> {
        self.command_tx.clone()
    }

    /// Access the server configuration.
    pub fn config(&self) -> &ServerConfig {
        &self.config
    }

    /// Run the server. Blocks until stopped or a fatal error occurs.
    pub async fn run(&mut self) -> anyhow::Result<()> {
        let mut command_rx = self
            .command_rx
            .take()
            .ok_or_else(|| anyhow::anyhow!("run() already called"))?;

        let event_tx = self.event_tx.clone();

        let sample_format: snapcast_proto::SampleFormat = self
            .config
            .sample_format
            .parse()
            .unwrap_or(snapcast_proto::DEFAULT_SAMPLE_FORMAT);

        anyhow::ensure!(
            !self.streams.is_empty(),
            "No streams configured — call add_stream() before run()"
        );

        tracing::info!(stream_port = self.config.stream_port, "Snapserver starting");

        // Advertise via mDNS (protocol-level discovery)
        #[cfg(feature = "mdns")]
        let _mdns = if self.config.mdns_enabled {
            mdns::MdnsAdvertiser::new(
                self.config.stream_port,
                &self.config.mdns_service_type,
                &self.config.mdns_name,
            )
            .map_err(|e| tracing::warn!(error = %e, "mDNS advertisement failed"))
            .ok()
        } else {
            None
        };

        // Create default encoder — used for codec header and first default stream
        let default_enc_config = encoder::EncoderConfig {
            codec: self.config.codec.clone(),
            format: sample_format,
            options: String::new(),
            #[cfg(feature = "encryption")]
            encryption_psk: self.config.encryption_psk.clone(),
        };
        let default_enc = encoder::create(&default_enc_config)?;

        // Spawn per-stream encode loops — reuse default_enc for first default stream
        let chunk_tx = self.chunk_tx.clone();
        let streams = std::mem::take(&mut self.streams);
        let mut default_enc = Some(default_enc);

        // Shared state for command handlers
        let initial_state = self
            .config
            .state_file
            .as_ref()
            .map(|p| state::ServerState::load(p))
            .unwrap_or_default();
        let shared_state = Arc::new(tokio::sync::Mutex::new(initial_state));

        // Create session server before stream registration
        // (first_stream_name set in loop below, but SessionServer only needs it for default routing)
        let first_name = streams
            .first()
            .map(|(n, _, _)| n.clone())
            .unwrap_or_default();
        let session_srv = Arc::new(session::SessionServer::new(
            self.config.stream_port,
            self.config.buffer_ms as i32,
            self.config.auth.clone(),
            Arc::clone(&shared_state),
            first_name.clone(),
            self.config.send_audio_to_muted,
        ));

        for (name, stream_cfg, rx) in streams {
            {
                let mut s = shared_state.lock().await;
                s.streams.push(state::StreamInfo {
                    id: name.clone(),
                    status: "idle".into(),
                    uri: String::new(),
                    properties: Default::default(),
                });
            }
            let enc = if stream_cfg.codec.is_none() && stream_cfg.sample_format.is_none() {
                if let Some(enc) = default_enc.take() {
                    enc
                } else {
                    encoder::create(&default_enc_config)?
                }
            } else {
                let stream_codec = stream_cfg.codec.as_deref().unwrap_or(&self.config.codec);
                let stream_format: snapcast_proto::SampleFormat = stream_cfg
                    .sample_format
                    .as_deref()
                    .and_then(|s| s.parse().ok())
                    .unwrap_or(sample_format);
                encoder::create(&encoder::EncoderConfig {
                    codec: stream_codec.to_string(),
                    format: stream_format,
                    options: String::new(),
                    #[cfg(feature = "encryption")]
                    encryption_psk: self.config.encryption_psk.clone(),
                })?
            };
            tracing::info!(stream = %name, codec = enc.name(), %sample_format, "Stream registered");
            session_srv
                .register_stream_codec(&name, enc.name(), enc.header())
                .await;
            spawn_stream_encoder(
                name,
                rx,
                enc,
                chunk_tx.clone(),
                sample_format.rate(),
                sample_format.channels(),
            );
        }

        let session_for_run = Arc::clone(&session_srv);
        let session_event_tx = event_tx.clone();
        let session_chunk_tx = self.chunk_tx.clone();
        let session_handle = tokio::spawn(async move {
            if let Err(e) = session_for_run
                .run(session_chunk_tx, session_event_tx)
                .await
            {
                tracing::error!(error = %e, "Session server error");
            }
        });

        let state_file = self.config.state_file.clone();
        let save_state = |s: &state::ServerState| {
            if let Some(ref path) = state_file {
                let _ = s
                    .save(path)
                    .map_err(|e| tracing::warn!(error = %e, "Failed to save state"));
            }
        };

        // Main loop
        loop {
            tokio::select! {
                cmd = command_rx.recv() => {
                    match cmd {
                        Some(ServerCommand::Stop) | None => {
                            tracing::info!("Server stopped");
                            session_handle.abort();
                            return Ok(());
                        }
                        Some(ServerCommand::SetClientVolume { client_id, volume, muted }) => {
                            let mut s = shared_state.lock().await;
                            if let Some(c) = s.clients.get_mut(&client_id) {
                                c.config.volume.percent = volume;
                                c.config.volume.muted = muted;
                            }
                            let latency = s.clients.get(&client_id).map(|c| c.config.latency).unwrap_or(0);
                            save_state(&s);
                            drop(s);
                            session_srv.push_settings(ClientSettingsUpdate {
                                client_id: client_id.clone(),
                                buffer_ms: self.config.buffer_ms as i32,
                                latency, volume, muted,
                            }).await;
                            let _ = event_tx.try_send(ServerEvent::ClientVolumeChanged { client_id: client_id.clone(), volume, muted });
                            session_srv.update_routing_for_client(&client_id).await;
                        }
                        Some(ServerCommand::SetClientLatency { client_id, latency }) => {
                            let mut s = shared_state.lock().await;
                            if let Some(c) = s.clients.get_mut(&client_id) {
                                c.config.latency = latency;
                                session_srv.push_settings(ClientSettingsUpdate {
                                    client_id: client_id.clone(),
                                    buffer_ms: self.config.buffer_ms as i32,
                                    latency,
                                    volume: c.config.volume.percent,
                                    muted: c.config.volume.muted,
                                }).await;
                            }
                            save_state(&s);
                            drop(s);
                            let _ = event_tx.try_send(ServerEvent::ClientLatencyChanged { client_id, latency });
                        }
                        Some(ServerCommand::SetClientName { client_id, name }) => {
                            let mut s = shared_state.lock().await;
                            if let Some(c) = s.clients.get_mut(&client_id) {
                                c.config.name = name.clone();
                            }
                            save_state(&s);
                            drop(s);
                            let _ = event_tx.try_send(ServerEvent::ClientNameChanged { client_id, name });
                        }
                        Some(ServerCommand::SetGroupStream { group_id, stream_id }) => {
                            let mut s = shared_state.lock().await;
                            s.set_group_stream(&group_id, &stream_id);
                            save_state(&s);
                            drop(s);
                            let _ = event_tx.try_send(ServerEvent::GroupStreamChanged { group_id: group_id.clone(), stream_id });
                            session_srv.update_routing_for_group(&group_id).await;
                        }
                        Some(ServerCommand::SetGroupMute { group_id, muted }) => {
                            let mut s = shared_state.lock().await;
                            if let Some(g) = s.groups.iter_mut().find(|g| g.id == group_id) {
                                g.muted = muted;
                            }
                            save_state(&s);
                            drop(s);
                            let _ = event_tx.try_send(ServerEvent::GroupMuteChanged { group_id: group_id.clone(), muted });
                            session_srv.update_routing_for_group(&group_id).await;
                        }
                        Some(ServerCommand::SetGroupName { group_id, name }) => {
                            let mut s = shared_state.lock().await;
                            if let Some(g) = s.groups.iter_mut().find(|g| g.id == group_id) {
                                g.name = name.clone();
                            }
                            save_state(&s);
                            drop(s);
                            let _ = event_tx.try_send(ServerEvent::GroupNameChanged { group_id, name });
                        }
                        Some(ServerCommand::SetGroupClients { group_id, clients }) => {
                            let mut s = shared_state.lock().await;
                            s.set_group_clients(&group_id, &clients);
                            save_state(&s);
                            drop(s);
                            // Structural change — mirrors Server.OnUpdate in C++ snapserver
                            let _ = event_tx.try_send(ServerEvent::ServerUpdated);
                            session_srv.update_routing_all().await;
                        }
                        Some(ServerCommand::DeleteClient { client_id }) => {
                            let mut s = shared_state.lock().await;
                            s.remove_client_from_groups(&client_id);
                            s.clients.remove(&client_id);
                            save_state(&s);
                            drop(s);
                            let _ = event_tx.try_send(ServerEvent::ServerUpdated);
                            session_srv.update_routing_all().await;
                        }
                        Some(ServerCommand::SetStreamMeta { stream_id, metadata }) => {
                            let mut s = shared_state.lock().await;
                            if let Some(stream) = s.streams.iter_mut().find(|st| st.id == stream_id) {
                                stream.properties = metadata.clone();
                            }
                            drop(s);
                            let _ = event_tx.try_send(ServerEvent::StreamMetaChanged { stream_id, metadata });
                        }
                        Some(ServerCommand::AddStream { uri, response_tx }) => {
                            // Parse stream name from URI query param, or use the URI as ID
                            let name = uri.split("name=").nth(1)
                                .and_then(|s| s.split('&').next())
                                .unwrap_or("dynamic")
                                .to_string();
                            let mut s = shared_state.lock().await;
                            if s.streams.iter().any(|st| st.id == name) {
                                let _ = response_tx.send(Err(format!("Stream '{name}' already exists")));
                            } else {
                                s.streams.push(state::StreamInfo {
                                    id: name.clone(),
                                    status: "idle".into(),
                                    uri: uri.clone(),
                                    properties: Default::default(),
                                });
                                save_state(&s);
                                drop(s);
                                let _ = event_tx.try_send(ServerEvent::ServerUpdated);
                                let _ = response_tx.send(Ok(name));
                            }
                        }
                        Some(ServerCommand::RemoveStream { stream_id }) => {
                            let mut s = shared_state.lock().await;
                            s.streams.retain(|st| st.id != stream_id);
                            // Clear stream_id on groups that referenced this stream
                            for g in &mut s.groups {
                                if g.stream_id == stream_id {
                                    g.stream_id.clear();
                                }
                            }
                            save_state(&s);
                            drop(s);
                            let _ = event_tx.try_send(ServerEvent::ServerUpdated);
                            session_srv.update_routing_all().await;
                        }
                        Some(ServerCommand::StreamControl { stream_id, command, params }) => {
                            tracing::debug!(stream_id, command, ?params, "Stream control forwarded");
                            // Forward to embedder via event — the library doesn't own stream readers
                            let _ = event_tx.try_send(ServerEvent::StreamControl { stream_id, command, params });
                        }
                        Some(ServerCommand::GetStatus { response_tx }) => {
                            let s = shared_state.lock().await;
                            let _ = response_tx.send(s.to_status());
                        }
                        #[cfg(feature = "custom-protocol")]
                        Some(ServerCommand::SendToClient { client_id, message }) => {
                            session_srv.send_custom(&client_id, message.type_id, message.payload).await;
                        }
                    }
                }
            }
        }
    }
}