// Queue persistence and saved queues live in the content store, which the
// `fs` feature brings (architecture/build-features.md D5). Without it the
// queue is in memory only.
#[cfg(feature = "fs")]
use crate::capture::CaptureError;
#[cfg(feature = "fs")]
use crate::crabidy_store::{self, CrabidyStore, QueueSnapshot};
use crate::{PendingResolve, QueueManager, ResolveKind};
use crate::{PlaybackCommand, PlaybackMessage, ProviderCommand, ProviderMessage};
use audio_player::Player;
use crabidy_core::proto::crabidy::QueueModifiers;
use crabidy_core::proto::crabidy::{
    get_update_stream_response::Update as StreamUpdate, InitResponse, PlayState,
    Queue as ProtoQueue, QueueTrack, Track, TrackPosition,
};
use crabidy_core::ProviderError;
use std::collections::{HashMap, VecDeque};
use std::sync::atomic::{AtomicU64, Ordering};
#[cfg(feature = "fs")]
use std::sync::Arc;
use std::sync::Mutex;
use std::time::Duration;
use tokio::time::timeout;
#[cfg(feature = "fs")]
use tracing::info;
use tracing::{debug, debug_span, error, instrument, trace, warn, Instrument};

pub struct Playback {
    update_tx: tokio::sync::broadcast::Sender<StreamUpdate>,
    provider_tx: flume::Sender<ProviderMessage>,
    pub playback_tx: flume::Sender<PlaybackMessage>,
    playback_rx: flume::Receiver<PlaybackMessage>,
    queue: Mutex<QueueManager>,
    state: Mutex<PlayState>,
    /// In-flight resolve operations by op id. Non-empty means the broadcast
    /// `Queue` snapshots carry `resolving = true`. Only the playback loop
    /// touches this map (same single-writer discipline as `queue`).
    pending: Mutex<HashMap<u64, PendingResolve>>,
    next_op_id: AtomicU64,
    /// `None` when queue persistence is disabled (no usable state
    /// directory) — the queue then lives in memory only.
    #[cfg(feature = "fs")]
    store: Option<Arc<CrabidyStore>>,
    /// Feeds the persister task; latest snapshot wins, so the loop never
    /// waits on disk (architecture/queue-persistence.md D4).
    #[cfg(feature = "fs")]
    persist_tx: tokio::sync::watch::Sender<Option<QueueSnapshot>>,
    pub player: Player,
}

impl Playback {
    pub fn new(
        update_tx: tokio::sync::broadcast::Sender<StreamUpdate>,
        provider_tx: flume::Sender<ProviderMessage>,
        #[cfg(feature = "fs")] store: Option<Arc<CrabidyStore>>,
        audio_device: Option<String>,
    ) -> Self {
        let (playback_tx, playback_rx) = flume::bounded(64);
        let queue = Mutex::new(QueueManager::new());
        let state = Mutex::new(PlayState::Stopped);
        #[cfg(feature = "fs")]
        let (persist_tx, _) = tokio::sync::watch::channel(None);
        let player = Player::new(audio_device);
        Self {
            update_tx,
            provider_tx,
            playback_tx,
            playback_rx,
            queue,
            state,
            pending: Mutex::new(HashMap::new()),
            next_op_id: AtomicU64::new(0),
            #[cfg(feature = "fs")]
            store,
            #[cfg(feature = "fs")]
            persist_tx,
            player,
        }
    }

    /// Reloads the persisted current queue: tracks, position, and
    /// shuffle/repeat. Never starts playback — a restarted server stays
    /// silent. Call before [`Self::run`] so nothing observes the empty
    /// queue first.
    #[cfg(feature = "fs")]
    pub async fn restore_current(&self) {
        let Some(store) = &self.store else {
            return;
        };
        let Some(snapshot) = store.load_current().await else {
            debug!("no persisted queue, starting fresh");
            return;
        };
        let Ok(mut queue) = self.queue.lock() else {
            error!("queue lock poisoned");
            return;
        };
        // No autoplay: the track the replace would start is ignored.
        let _ = queue.replace_with_tracks(&snapshot.tracks);
        queue.repeat = snapshot.repeat;
        // An out-of-range position (edited folder) is refused by
        // `set_current_position` and playback starts at the first track.
        let _ = queue.set_current_position(snapshot.current_position);
        if snapshot.shuffle {
            // The play order is not persisted; restoring shuffle reshuffles
            // around the restored current track.
            queue.shuffle_on();
        }
        info!(
            tracks = snapshot.tracks.len(),
            position = snapshot.current_position,
            "restored the persisted queue"
        );
    }

    pub fn run(self) {
        #[cfg(feature = "fs")]
        if let Some(store) = &self.store {
            crabidy_store::spawn_persister(Arc::clone(store), self.persist_tx.subscribe());
        }
        tokio::spawn(async move {
            while let Ok(PlaybackMessage { span, command }) = self.playback_rx.recv_async().await {
                // Attribute all handler events to a span that is a child of
                // the span that was current when the command was sent.
                let handler_span =
                    debug_span!(parent: &span, "playback_command", command = command.name());
                self.handle_command(command).instrument(handler_span).await;
            }
            warn!("playback message channel closed, loop exiting");
        });
    }

    async fn handle_command(&self, command: PlaybackCommand) {
        trace!("handling playback command");
        match command {
            PlaybackCommand::Init { result_tx } => {
                // Asked before the queue lock is taken: that lock is a std
                // `Mutex`, so it must not be held across an await, which is
                // why these three were hardcoded below in the first place.
                let (volume, muted, position) = self.player_snapshot().await;
                let response = {
                    let Ok(queue) = self.queue.lock() else {
                        error!("queue lock poisoned");
                        return;
                    };
                    let queue_track = QueueTrack {
                        queue_position: queue.current_position() as u32,
                        track: queue.current_track(),
                    };
                    let play_state = {
                        let Ok(play_state) = self.state.lock() else {
                            error!("play state lock poisoned");
                            return;
                        };
                        *play_state
                    };
                    InitResponse {
                        // Snapshot with `resolving`: a client connecting
                        // mid-resolve must show the indicator right away.
                        queue: Some(self.queue_snapshot(&queue)),
                        queue_track: Some(queue_track),
                        play_state: play_state as i32,
                        volume,
                        mute: muted,
                        position: Some(position),
                        mods: Some(QueueModifiers {
                            repeat: queue.repeat,
                            shuffle: queue.shuffle,
                        }),
                        // Stamped by the RPC handler, which owns the auth
                        // switch; the playback loop only knows queue state.
                        auth_enabled: false,
                    }
                };
                trace!(?response, "sending init response");
                if let Err(err) = result_tx.send(response) {
                    error!("failed to send init response: {err}");
                }
            }

            PlaybackCommand::Replace { paths } => {
                // A replace obsoletes whatever earlier ops are still
                // resolving; their late chunks must not land in the new
                // queue.
                self.cancel_pending_resolves();
                self.start_resolve(ResolveKind::Replace, paths);
            }

            PlaybackCommand::Queue { paths } => {
                let position = {
                    let Ok(queue) = self.queue.lock() else {
                        error!("queue lock poisoned");
                        return;
                    };
                    queue.current_position() as u32
                };
                // Play-next means "right after the current track".
                self.start_resolve(ResolveKind::InsertAt(position + 1), paths);
            }

            PlaybackCommand::Append { paths } => {
                self.start_resolve(ResolveKind::Append, paths);
            }

            PlaybackCommand::ApplyResolvedChunk { op_id, tracks } => {
                self.apply_resolved_chunk(op_id, tracks).await;
            }

            PlaybackCommand::ResolveFinished { op_id } => {
                self.finish_resolve(op_id);
            }

            PlaybackCommand::Remove { positions } => {
                debug!(?positions, "removing tracks");
                let (track, was_last) = {
                    let Ok(mut queue) = self.queue.lock() else {
                        error!("queue lock poisoned");
                        return;
                    };
                    let was_last = queue.is_last_track();
                    let track = queue.remove_tracks(&positions);
                    self.broadcast_queue(&queue);
                    (track, was_last)
                };
                let state = {
                    let Ok(state) = self.state.lock() else {
                        error!("play state lock poisoned");
                        return;
                    };
                    *state
                };
                if state == PlayState::Playing && track.is_some() {
                    // The playing track was removed: play its successor, or
                    // stop when it was the last one.
                    if was_last {
                        self.stop_player().await;
                    } else {
                        self.play(track).await;
                    }
                }
            }

            PlaybackCommand::Insert { position, paths } => {
                // The RPC's `position` is the index to insert *at*: whatever
                // sits there shifts down, 0 is the front. No offset here.
                self.start_resolve(ResolveKind::InsertAt(position), paths);
            }

            PlaybackCommand::Clear { exclude_current } => {
                debug!(exclude_current, "clearing queue");
                // Chunks still resolving would repopulate the queue the
                // user just emptied.
                self.cancel_pending_resolves();
                let should_stop = {
                    let Ok(mut queue) = self.queue.lock() else {
                        error!("queue lock poisoned");
                        return;
                    };
                    let should_stop = queue.clear(exclude_current);
                    self.broadcast_queue(&queue);
                    should_stop
                };
                if should_stop {
                    self.stop_player().await;
                }
            }

            PlaybackCommand::DedupQueue {
                by_title,
                result_tx,
            } => {
                // Never removes the playing entry (architecture/queue-order.md
                // D4), so unlike `Remove` there is no successor to start.
                let removed = {
                    let Ok(mut queue) = self.queue.lock() else {
                        error!("queue lock poisoned");
                        return;
                    };
                    let identity = if by_title {
                        crate::queue_order::Identity::ArtistTitle
                    } else {
                        crate::queue_order::Identity::ProviderItem
                    };
                    let removed = queue.dedup(identity);
                    if removed > 0 {
                        self.broadcast_queue(&queue);
                    }
                    removed
                };
                debug!(removed, by_title, "de-duplicated the queue");
                if let Err(err) = result_tx.send(removed) {
                    // The caller gave up (client gone); the queue is already
                    // deduped and broadcast, so this is only a lost count.
                    debug!("dedup result receiver gone: {err}");
                }
            }

            PlaybackCommand::SortQueue { sort, descending } => {
                debug!(?sort, descending, "sorting the queue");
                let Ok(mut queue) = self.queue.lock() else {
                    error!("queue lock poisoned");
                    return;
                };
                queue.sort(sort, descending);
                // The playing track moved; the `Queue` snapshot carries
                // `current_position`, so this one broadcast tells clients both
                // the new order and where the cursor now sits — the same way
                // `Remove` announces a shifted position.
                self.broadcast_queue(&queue);
            }

            PlaybackCommand::SetCurrent { position } => {
                debug!(position, "jumping to queue position");
                let track = {
                    let Ok(mut queue) = self.queue.lock() else {
                        error!("queue lock poisoned");
                        return;
                    };
                    queue.set_current_position(position);
                    queue.current_track()
                };
                self.play(track).await;
            }

            #[cfg(feature = "fs")]
            PlaybackCommand::SaveQueue { name, result_tx } => {
                debug!(name, "saving the queue");
                // Snapshot on the loop (single-writer discipline), write on
                // a spawned task — the loop never waits on disk.
                let snapshot = {
                    let Ok(queue) = self.queue.lock() else {
                        error!("queue lock poisoned");
                        return;
                    };
                    let proto: ProtoQueue = queue.clone().into();
                    QueueSnapshot {
                        tracks: proto.tracks,
                        current_position: proto.current_position,
                        repeat: queue.repeat,
                        shuffle: queue.shuffle,
                    }
                };
                let store = self.store.clone();
                tokio::spawn(
                    async move {
                        let result = match &store {
                            Some(store) => store.save_snapshot(&name, &snapshot.tracks).await,
                            None => Err(CaptureError::Disabled),
                        };
                        if let Err(err) = &result {
                            warn!(name, "cannot save queue: {err}");
                        }
                        let _ = result_tx.send_async(result).await;
                    }
                    .in_current_span(),
                );
            }

            PlaybackCommand::ToggleShuffle => {
                let (shuffle, repeat) = {
                    let Ok(mut queue) = self.queue.lock() else {
                        error!("queue lock poisoned");
                        return;
                    };
                    if queue.shuffle {
                        queue.shuffle_off()
                    } else {
                        queue.shuffle_on()
                    }
                    self.send_persist_snapshot(&queue);
                    (queue.shuffle, queue.repeat)
                };
                debug!(shuffle, "toggled shuffle");
                self.broadcast(StreamUpdate::Mods(QueueModifiers { shuffle, repeat }));
            }

            PlaybackCommand::ToggleRepeat => {
                let (shuffle, repeat) = {
                    let Ok(mut queue) = self.queue.lock() else {
                        error!("queue lock poisoned");
                        return;
                    };
                    queue.repeat = !queue.repeat;
                    self.send_persist_snapshot(&queue);
                    (queue.shuffle, queue.repeat)
                };
                debug!(repeat, "toggled repeat");
                self.broadcast(StreamUpdate::Mods(QueueModifiers { shuffle, repeat }));
            }

            PlaybackCommand::TogglePlay => {
                let state = {
                    let Ok(state) = self.state.lock() else {
                        error!("play state lock poisoned");
                        return;
                    };
                    *state
                };
                debug!(?state, "toggling play");
                match state {
                    PlayState::Playing => {
                        if let Err(err) = self.player.pause().await {
                            warn!("pause failed: {err:?}");
                        }
                    }
                    PlayState::Paused => {
                        if let Err(err) = self.player.unpause().await {
                            warn!("unpause failed: {err:?}");
                        }
                    }
                    // Stopped/idle: nothing is loaded in the player (e.g. right
                    // after a restart restored the queue without autoplay), so
                    // there is nothing to unpause. Load and play the current
                    // queue track, exactly as `SetCurrent` does — a no-op when
                    // the queue is empty.
                    _ => {
                        self.play(self.current_track()).await;
                    }
                }
            }

            PlaybackCommand::Stop => {
                debug!("stopping playback");
                self.stop_player().await;
            }

            PlaybackCommand::ChangeVolume { delta } => {
                match self.player.volume().await {
                    Ok(volume) => match self.player.set_volume(volume + delta).await {
                        // Nothing was broadcast here, so no client ever
                        // learned the level had changed. The level sent is
                        // the one the engine actually took, so a client
                        // also sees the clamp at 1.1 rather than the value
                        // it asked for. `set_volume` unmutes as well
                        // ("reaching for the volume is an intent to hear
                        // something"), so the mute indicator needs the
                        // same news or it stays stuck on muted.
                        Ok(volume) => {
                            debug!(volume, delta, "changed volume");
                            self.broadcast(StreamUpdate::Volume(volume));
                            self.broadcast(StreamUpdate::Mute(false));
                        }
                        Err(err) => warn!("set_volume failed: {err:?}"),
                    },
                    Err(err) => warn!("could not read volume: {err:?}"),
                };
            }

            PlaybackCommand::ToggleMute => match self.player.toggle_mute().await {
                Ok(muted) => {
                    debug!(muted, "toggled mute");
                    self.broadcast(StreamUpdate::Mute(muted));
                }
                Err(err) => warn!("toggle_mute failed: {err:?}"),
            },

            PlaybackCommand::Next => {
                let track = {
                    let Ok(mut queue) = self.queue.lock() else {
                        error!("queue lock poisoned");
                        return;
                    };
                    queue.next_track()
                };
                debug!(
                    track = track.as_ref().map(|t| t.path.as_str()),
                    "advancing to next track"
                );
                self.play_or_stop(track).await;
            }

            PlaybackCommand::Prev => {
                let track = {
                    let Ok(mut queue) = self.queue.lock() else {
                        error!("queue lock poisoned");
                        return;
                    };
                    queue.prev_track()
                };
                debug!(
                    track = track.as_ref().map(|t| t.path.as_str()),
                    "going back to previous track"
                );
                self.play_or_stop(track).await;
            }

            PlaybackCommand::StateChanged { state } => {
                let play_state = {
                    let Ok(mut state_lock) = self.state.lock() else {
                        error!("play state lock poisoned");
                        return;
                    };
                    *state_lock = state;
                    state
                };
                debug!(?play_state, "player state changed");
                self.broadcast(StreamUpdate::PlayState(play_state as i32));
            }

            PlaybackCommand::RestartTrack => {
                debug!("restarting current track");
                let state = {
                    let Ok(state) = self.state.lock() else {
                        error!("play state lock poisoned");
                        return;
                    };
                    *state
                };
                // With a track loaded (playing or paused) restart it from the
                // top. When nothing is loaded (queue restored without autoplay)
                // there is no source to restart, so start the current queue
                // track instead.
                if matches!(state, PlayState::Playing | PlayState::Paused) {
                    if let Err(err) = self.player.restart().await {
                        warn!("restart failed: {err:?}");
                    }
                } else {
                    self.play(self.current_track()).await;
                }
            }

            PlaybackCommand::Seek { delta_millis } => {
                debug!(delta_millis, "seeking");
                // The player owns both the live position and the clamping, so
                // this arm only forwards. A source that cannot seek (HLS) or a
                // player with nothing loaded is a warning: clients render the
                // position from the update stream and never moved it
                // themselves, so there is nothing to correct
                // (architecture/seek.md D7).
                match self.player.seek_by(delta_millis).await {
                    Ok(position) => trace!(?position, "seeked"),
                    Err(err) => warn!("seek failed: {err:?}"),
                }
            }

            PlaybackCommand::VolumeChanged { volume } => {
                trace!(volume, "volume changed");
                self.broadcast(StreamUpdate::Volume(volume));
            }

            PlaybackCommand::MuteChanged { muted } => {
                trace!(muted, "mute changed");
                self.broadcast(StreamUpdate::Mute(muted));
            }

            PlaybackCommand::PositionChanged { duration, position } => {
                trace!(duration, position, "position changed");
                self.broadcast(StreamUpdate::Position(TrackPosition { duration, position }));
            }
        }
    }

    /// The player-owned half of the init snapshot: output level, mute
    /// state, and the position within the current track. All three were
    /// hardcoded to neutral values, so every client showed 0% volume, an
    /// unmuted server and 0:00 until something happened to change them —
    /// and while *paused* nothing does, since the position tick skips a
    /// paused sink.
    ///
    /// Every read is bounded. The engine thread is single-threaded and can
    /// be busy for up to 30 s opening a network stream, and this is the
    /// **connect** path: a client must get a usable snapshot rather than
    /// hang waiting for one. On a timeout each field falls back to what the
    /// engine starts at, and the next update on the stream corrects it. A
    /// late reply lands in a dropped receiver, which the engine only logs.
    async fn player_snapshot(&self) -> (f32, bool, TrackPosition) {
        /// Deliberately short. The engine has no middle gear here: these
        /// are plain field reads, so an idle engine answers in
        /// microseconds, while a busy one is mid-`Play` and will not answer
        /// for up to 30 s no matter how long we wait. So the budget only
        /// decides how long the playback loop — which handles commands one
        /// at a time — stalls before giving up on a busy engine. Volume is
        /// read first, being the field a user actually looks at.
        const READ_TIMEOUT: Duration = Duration::from_secs(1);

        let volume = match timeout(READ_TIMEOUT, self.player.volume()).await {
            Ok(Ok(volume)) => volume,
            Ok(Err(err)) => {
                warn!("could not read volume for init: {err:?}");
                1.0
            }
            Err(_) => {
                warn!("timed out reading volume for init");
                1.0
            }
        };
        let muted = match timeout(READ_TIMEOUT, self.player.is_muted()).await {
            Ok(Ok(muted)) => muted,
            Ok(Err(err)) => {
                warn!("could not read mute state for init: {err:?}");
                false
            }
            Err(_) => {
                warn!("timed out reading mute state for init");
                false
            }
        };
        // Both error when nothing is loaded, which is the ordinary idle
        // case rather than a fault — hence zeros without a warning.
        let position = match timeout(READ_TIMEOUT, self.player_position()).await {
            Ok(position) => position,
            Err(_) => {
                warn!("timed out reading position for init");
                TrackPosition {
                    duration: 0,
                    position: 0,
                }
            }
        };
        (volume, muted, position)
    }

    /// The current position, as milliseconds for the wire. Zeros when
    /// nothing is playing.
    async fn player_position(&self) -> TrackPosition {
        let position = self
            .player
            .elapsed()
            .await
            .map(|elapsed| elapsed.as_millis().min(u32::MAX.into()) as u32)
            .unwrap_or(0);
        let duration = self
            .player
            .duration()
            .await
            .map(|duration| duration.as_millis().min(u32::MAX.into()) as u32)
            .unwrap_or(0);
        TrackPosition { duration, position }
    }

    /// Sends an update to all connected clients. Having no subscribers is
    /// normal and not an error.
    fn broadcast(&self, update: StreamUpdate) {
        if let Err(err) = self.update_tx.send(update) {
            trace!("no update stream subscribers: {err}");
        }
    }

    /// Registers a pending resolve operation and spawns its forwarder task.
    ///
    /// The forwarder resolves `paths` with an *exponential read-ahead*: a
    /// concurrency window that starts at 1 and doubles (1, 2, 4, 8, 16, then
    /// steady 16) after each path completes. It keeps up to `window` paths
    /// resolving at once (each via `ProviderCommand::ResolveTracks` on its own
    /// bounded chunk channel) but forwards chunks to the playback loop as
    /// `PlaybackCommand::ApplyResolvedChunk` in strict path order, so the queue
    /// keeps the requested order and the first chunk still lands — and starts
    /// playback — as fast as a single resolve. The first track therefore plays
    /// as soon as one resolve yields it, while the window fills the queue far
    /// ahead of playback: a very short or skipped leading track cannot outrun
    /// resolution and cause a silent stall (architecture/progressive-queueing.md D8).
    ///
    /// After the last path it sends `ResolveFinished`. When the op's
    /// cancellation flag is set, the forwarder stops launching resolves and
    /// drops its chunk receivers — each provider's next send fails and the
    /// fetch stops. Queue state is never touched here: mutations happen only
    /// when the loop processes the forwarded commands. An immediate `Queue`
    /// broadcast (unchanged tracks, `resolving = true`) gives clients instant
    /// feedback.
    fn start_resolve(&self, kind: ResolveKind, paths: Vec<String>) {
        let op = PendingResolve::new(kind);
        let cancelled = op.cancel_flag();
        let op_id = self.next_op_id.fetch_add(1, Ordering::Relaxed);
        {
            let Ok(mut pending) = self.pending.lock() else {
                error!("pending ops lock poisoned");
                return;
            };
            pending.insert(op_id, op);
        }
        debug!(op_id, ?paths, "starting queue resolve");
        {
            let Ok(queue) = self.queue.lock() else {
                error!("queue lock poisoned");
                return;
            };
            // Instant feedback: clients see resolving = true before the
            // first chunk exists.
            self.broadcast_queue(&queue);
        }
        let provider_tx = self.provider_tx.clone();
        let playback_tx = self.playback_tx.clone();
        tokio::spawn(
            async move {
                // Exponential read-ahead window: 1, 2, 4, 8, 16, then steady.
                const MAX_WINDOW: usize = 16;
                let mut window = 1usize;
                let mut next = 0usize;
                let mut inflight: VecDeque<(String, flume::Receiver<Vec<Track>>)> = VecDeque::new();
                'resolve: loop {
                    if cancelled.load(Ordering::Relaxed) {
                        break;
                    }
                    // Top the window up with fresh concurrent resolves. Each
                    // resolves in the background into its own bounded channel
                    // while we drain the oldest one below.
                    while inflight.len() < window && next < paths.len() {
                        let path = paths[next].clone();
                        next += 1;
                        let (chunk_tx, chunk_rx) = flume::bounded(4);
                        let message = ProviderMessage::new(ProviderCommand::ResolveTracks {
                            path: path.clone(),
                            chunk_tx,
                        });
                        if provider_tx.send_async(message).await.is_err() {
                            error!("provider channel closed");
                            break 'resolve;
                        }
                        inflight.push_back((path, chunk_rx));
                    }
                    // Drain the oldest (lowest path index) resolve to
                    // completion before the next, so chunks reach the loop in
                    // request order even though resolution ran concurrently.
                    let Some((path, chunk_rx)) = inflight.pop_front() else {
                        break;
                    };
                    let mut forwarded = 0usize;
                    while let Ok(tracks) = chunk_rx.recv_async().await {
                        // On cancellation this drops chunk_rx (and, on the next
                        // iteration, the rest of `inflight`); each provider's
                        // next send then fails and its fetch stops.
                        if cancelled.load(Ordering::Relaxed) {
                            break 'resolve;
                        }
                        forwarded += tracks.len();
                        let apply = PlaybackCommand::ApplyResolvedChunk { op_id, tracks };
                        if playback_tx
                            .send_async(PlaybackMessage::new(apply))
                            .await
                            .is_err()
                        {
                            error!("playback channel closed");
                            return;
                        }
                    }
                    if forwarded == 0 && !cancelled.load(Ordering::Relaxed) {
                        warn!(path, "path resolved to no playable tracks");
                    }
                    window = (window * 2).min(MAX_WINDOW);
                }
                // Always reported — also for cancelled or empty ops — so
                // the pending map can never leak a stuck resolving flag.
                let finished = PlaybackCommand::ResolveFinished { op_id };
                let _ = playback_tx.send_async(PlaybackMessage::new(finished)).await;
            }
            .in_current_span(),
        );
    }

    /// Applies one chunk to the queue for pending op `op_id`, broadcasts
    /// the grown queue, and starts playback when the chunk made a track
    /// current — or when this op still wants to start but its earlier attempt
    /// ran the player dry on skipped/short head tracks before more resolved.
    /// Chunks for an unknown op id (finished or cancelled) are dropped
    /// silently.
    async fn apply_resolved_chunk(&self, op_id: u64, tracks: Vec<Track>) {
        let attempt = {
            let Ok(mut queue) = self.queue.lock() else {
                error!("queue lock poisoned");
                return;
            };
            let (designated, wants_start) = {
                let Ok(mut pending) = self.pending.lock() else {
                    error!("pending ops lock poisoned");
                    return;
                };
                let Some(op) = pending.get_mut(&op_id) else {
                    trace!(op_id, "dropping chunk for a finished or cancelled op");
                    return;
                };
                let designated = op.apply_chunk(&mut queue, &tracks);
                (designated, op.wants_start())
            };
            self.broadcast_queue(&queue);
            // Start on the track this chunk made current; failing that, if the
            // op still wants to start (a prior attempt found nothing playable
            // yet), retry from the current position — `next_playable_urls`
            // advances past skipped/unplayable heads to the first track that
            // has now resolved.
            designated.or_else(|| {
                if wants_start {
                    queue.current_track()
                } else {
                    None
                }
            })
        };
        // Confirming the start clears the op's `wants_start`, so later chunks
        // only extend the queue and never restart the playing track.
        if self.play(attempt).await {
            if let Ok(mut pending) = self.pending.lock() {
                if let Some(op) = pending.get_mut(&op_id) {
                    op.mark_started();
                }
            }
        }
    }

    /// Removes the finished op and broadcasts the final `Queue` snapshot
    /// (clearing `resolving` once no ops remain). An op already removed by
    /// cancellation needs no broadcast — the cancelling command mutates
    /// the queue and broadcasts itself.
    fn finish_resolve(&self, op_id: u64) {
        let removed = {
            let Ok(mut pending) = self.pending.lock() else {
                error!("pending ops lock poisoned");
                return;
            };
            pending.remove(&op_id)
        };
        let Some(op) = removed else {
            trace!(op_id, "resolve finished for a cancelled op");
            return;
        };
        debug!(op_id, tracks = op.applied(), "queue resolve finished");
        let Ok(queue) = self.queue.lock() else {
            error!("queue lock poisoned");
            return;
        };
        self.broadcast_queue(&queue);
    }

    /// Cancels every in-flight resolve op (used by `Replace` and `Clear`).
    /// The forwarders see the flag, drop their chunk receivers (stopping
    /// the fetches) and still report `ResolveFinished`, which is dropped as
    /// unknown here.
    fn cancel_pending_resolves(&self) {
        let Ok(mut pending) = self.pending.lock() else {
            error!("pending ops lock poisoned");
            return;
        };
        for (op_id, op) in pending.drain() {
            debug!(op_id, "cancelling in-flight resolve");
            op.cancel();
        }
    }

    /// A wire snapshot of the queue with the `resolving` flag set from the
    /// pending-op map. Callers must not hold the `pending` lock (`queue` is
    /// fine — the lock order is queue, then pending).
    fn queue_snapshot(&self, queue: &QueueManager) -> ProtoQueue {
        let resolving = self
            .pending
            .lock()
            .map(|pending| !pending.is_empty())
            .unwrap_or(false);
        let mut snapshot: ProtoQueue = queue.clone().into();
        snapshot.resolving = resolving;
        snapshot
    }

    /// Broadcasts the current queue snapshot. All queue broadcasts go
    /// through here so the `resolving` flag can never be forgotten — and
    /// every queue-content change reaches the persister the same way.
    fn broadcast_queue(&self, queue: &QueueManager) {
        self.send_persist_snapshot(queue);
        self.broadcast(StreamUpdate::Queue(self.queue_snapshot(queue)));
    }

    /// Hands the queue's persistable state to the persister task; a no-op
    /// when persistence is disabled. Latest snapshot wins, so calling this
    /// on every mutation is free of backpressure (the persister skips
    /// writes for unchanged snapshots).
    #[cfg(not(feature = "fs"))]
    fn send_persist_snapshot(&self, _queue: &QueueManager) {}

    /// Hands the queue's persistable state to the persister task; a no-op
    /// when persistence is disabled.
    #[cfg(feature = "fs")]
    fn send_persist_snapshot(&self, queue: &QueueManager) {
        if self.store.is_none() {
            return;
        }
        let proto: ProtoQueue = queue.clone().into();
        self.persist_tx.send_replace(Some(QueueSnapshot {
            tracks: proto.tracks,
            current_position: proto.current_position,
            repeat: queue.repeat,
            shuffle: queue.shuffle,
        }));
    }

    #[instrument(skip(self))]
    async fn get_urls_for_track(&self, path: &str) -> Result<Vec<String>, ProviderError> {
        let (result_tx, result_rx) = flume::bounded(1);
        let message = ProviderMessage::new(ProviderCommand::GetTrackUrls {
            path: path.to_string(),
            result_tx,
        });
        self.provider_tx
            .send_async(message)
            .await
            .map_err(|_| ProviderError::InternalError)?;
        result_rx
            .recv_async()
            .await
            .map_err(|_| ProviderError::InternalError)?
    }

    async fn stop_player(&self) {
        if let Err(err) = self.player.stop().await {
            debug!("stop had no effect: {err:?}");
        }
    }

    /// The queue's current track, or `None` when the queue is empty (or its
    /// lock is poisoned). The resume-style controls (`TogglePlay`,
    /// `RestartTrack`) use it to load what a restored/idle queue points at.
    fn current_track(&self) -> Option<Track> {
        match self.queue.lock() {
            Ok(queue) => queue.current_track(),
            Err(_) => {
                error!("queue lock poisoned");
                None
            }
        }
    }

    /// Plays the given track if there is one, otherwise stops the player.
    #[instrument(skip(self, track), fields(track = track.as_ref().map(|t| t.path.as_str())))]
    async fn play_or_stop(&self, track: Option<Track>) {
        if track.is_some() {
            self.play(track).await;
        } else {
            self.stop_player().await;
        }
    }

    /// Finds the stream URLs of the first playable track, starting at
    /// `track` and advancing the queue past unplayable ones. Tracks marked
    /// `is_skipped` (captures recorded their source as uncapturable) are
    /// skipped without a provider round trip; tracks whose stream URLs
    /// fail to resolve are skipped with a warning. Bounded by the queue
    /// length at entry — one full pass at most — so an all-skipped queue
    /// with repeat on returns `None` instead of spinning
    /// (architecture/incremental-captures.md D3).
    async fn next_playable_urls(&self, mut track: Track) -> Option<Vec<String>> {
        let mut attempts_left = {
            let Ok(queue) = self.queue.lock() else {
                error!("queue lock poisoned");
                return None;
            };
            queue.len()
        };
        loop {
            let path = track.path.as_str();
            if track.is_skipped {
                debug!(path, "track is marked skipped, skipping");
            } else {
                match self.get_urls_for_track(path).await {
                    Ok(urls) if !urls.is_empty() => return Some(urls),
                    Ok(_) => warn!(path, "provider returned no stream urls, skipping track"),
                    Err(err) => warn!(path, "failed to fetch stream urls ({err}), skipping track"),
                }
            }
            attempts_left = attempts_left.saturating_sub(1);
            if attempts_left == 0 {
                warn!("no playable track in the queue after a full pass");
                return None;
            }
            let next = {
                let Ok(mut queue) = self.queue.lock() else {
                    error!("queue lock poisoned");
                    return None;
                };
                queue.next_track()
            };
            match next {
                Some(next_track) => track = next_track,
                None => {
                    debug!("reached the end of the queue without a playable track");
                    return None;
                }
            }
        }
    }

    /// Starts playback of the given track, skipping past unplayable ones
    /// (see [`Self::next_playable_urls`]); stops the player when nothing
    /// in the queue is playable.
    ///
    /// Returns `true` when a playable track was found and handed to the
    /// player. A device error while starting is logged but still counts as
    /// started — retrying the same track would not help. Returns `false` when
    /// there was nothing to play or nothing playable remained (the player is
    /// then stopped), so the caller can retry once more tracks resolve.
    #[instrument(skip(self, track), fields(track = track.as_ref().map(|t| t.path.as_str())))]
    async fn play(&self, track: Option<Track>) -> bool {
        let Some(track) = track else {
            debug!("nothing to play");
            return false;
        };
        let Some(urls) = self.next_playable_urls(track).await else {
            self.stop_player().await;
            return false;
        };
        {
            let Ok(queue) = self.queue.lock() else {
                error!("queue lock poisoned");
                return false;
            };
            // Current-track moves (Next/Prev/SetCurrent/skips) change the
            // persisted position without a queue broadcast.
            self.send_persist_snapshot(&queue);
            self.broadcast(StreamUpdate::QueueTrack(QueueTrack {
                queue_position: queue.current_position() as u32,
                track: queue.current_track(),
            }));
        }
        debug!(url_count = urls.len(), "starting player");
        if let Err(err) = self.player.play(&urls[0]).await {
            error!("player failed to start track: {err:?}");
        }
        true
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    #[cfg(feature = "fs")]
    use tempfile::TempDir;

    fn track(i: usize) -> Track {
        Track {
            path: format!("/tidal/playlists/p/{i}"),
            artist: "artist".to_string(),
            title: format!("track {i}"),
            duration: None,
            album: None,
            is_skipped: false,
            provider_item_id: String::new(),
            is_captured: false,
        }
    }

    #[cfg(feature = "fs")]
    async fn store_in(dir: &TempDir) -> Arc<CrabidyStore> {
        Arc::new(
            CrabidyStore::open(dir.path().join("state"), dir.path().join("store"))
                .await
                .expect("open store"),
        )
    }

    fn playback_with(#[cfg(feature = "fs")] store: Option<Arc<CrabidyStore>>) -> Playback {
        let (update_tx, _) = tokio::sync::broadcast::channel(64);
        let (provider_tx, _provider_rx) = flume::bounded(16);
        Playback::new(
            update_tx,
            provider_tx,
            #[cfg(feature = "fs")]
            store,
            None,
        )
    }

    /// Queue titles in queue order, for order assertions.
    fn queue_titles(playback: &Playback) -> Vec<String> {
        let proto: ProtoQueue = playback.queue.lock().expect("queue lock").clone().into();
        proto.tracks.iter().map(|t| t.title.clone()).collect()
    }

    fn fill_queue(playback: &Playback, n: usize) {
        let tracks: Vec<Track> = (0..n).map(track).collect();
        let mut queue = playback.queue.lock().expect("queue lock");
        let _ = queue.replace_with_tracks(&tracks);
    }

    #[cfg(feature = "fs")]
    #[tokio::test]
    async fn restore_fills_the_queue_without_starting_playback() {
        let dir = TempDir::new().expect("tempdir");
        let store = store_in(&dir).await;
        store
            .persist_current(&QueueSnapshot {
                tracks: (0..3).map(track).collect(),
                current_position: 1,
                repeat: true,
                shuffle: false,
            })
            .await
            .expect("persist");

        let playback = playback_with(Some(store));
        playback.restore_current().await;

        let queue = playback.queue.lock().expect("queue lock");
        let snapshot: ProtoQueue = queue.clone().into();
        assert_eq!(snapshot.tracks.len(), 3);
        assert_eq!(queue.current_position(), 1);
        assert!(queue.repeat);
        // A restarted server stays silent: restoring must not play.
        assert_eq!(
            *playback.state.lock().expect("state lock"),
            PlayState::Stopped
        );
    }

    #[cfg(feature = "fs")]
    #[tokio::test]
    async fn restore_survives_an_out_of_range_position() {
        let dir = TempDir::new().expect("tempdir");
        let store = store_in(&dir).await;
        store
            .persist_current(&QueueSnapshot {
                tracks: vec![track(0)],
                current_position: 99, // hand-edited folder
                repeat: false,
                shuffle: false,
            })
            .await
            .expect("persist");
        let playback = playback_with(Some(store));
        playback.restore_current().await;
        let queue = playback.queue.lock().expect("queue lock");
        assert_eq!(queue.current_position(), 0);
    }

    #[cfg(feature = "fs")]
    #[tokio::test]
    async fn save_queue_command_snapshots_the_live_queue() {
        let dir = TempDir::new().expect("tempdir");
        let store = store_in(&dir).await;
        let playback = playback_with(Some(Arc::clone(&store)));
        fill_queue(&playback, 2);

        let (result_tx, result_rx) = flume::bounded(1);
        playback
            .handle_command(PlaybackCommand::SaveQueue {
                name: "road trip".to_string(),
                result_tx,
            })
            .await;
        result_rx
            .recv_async()
            .await
            .expect("reply")
            .expect("save succeeds");

        let entries = std::fs::read_dir(store.tree_dir().join("road trip"))
            .expect("saved queue folder")
            .filter(|e| {
                !e.as_ref()
                    .expect("entry")
                    .file_name()
                    .to_string_lossy()
                    .starts_with('.')
            })
            .count();
        assert_eq!(entries, 2);
    }

    #[cfg(feature = "fs")]
    #[tokio::test]
    async fn save_queue_rejects_an_empty_queue() {
        let dir = TempDir::new().expect("tempdir");
        let playback = playback_with(Some(store_in(&dir).await));
        let (result_tx, result_rx) = flume::bounded(1);
        playback
            .handle_command(PlaybackCommand::SaveQueue {
                name: "empty".to_string(),
                result_tx,
            })
            .await;
        let result = result_rx.recv_async().await.expect("reply");
        assert!(matches!(result, Err(CaptureError::BadSource(_))));
    }

    #[tokio::test]
    async fn skipped_tracks_are_skipped_with_a_bounded_pass() {
        // All tracks are marked skipped and repeat is on: `next_track`
        // cycles forever, so only the one-full-pass bound ends the loop
        // with `None`. The marked tracks are skipped without any provider
        // round trip (the provider channel is closed — a call would fail,
        // not hang).
        let playback = playback_with(
            #[cfg(feature = "fs")]
            None,
        );
        let tracks: Vec<Track> = (0..3)
            .map(|i| Track {
                is_skipped: true,
                ..track(i)
            })
            .collect();
        let first = {
            let mut queue = playback.queue.lock().expect("queue lock");
            let first = queue.replace_with_tracks(&tracks);
            queue.repeat = true;
            first
        };
        let urls = playback
            .next_playable_urls(first.expect("first track"))
            .await;
        assert!(urls.is_none(), "an all-skipped queue has nothing playable");
    }

    #[cfg(feature = "fs")]
    #[tokio::test]
    async fn queue_mutations_reach_the_persist_channel() {
        let dir = TempDir::new().expect("tempdir");
        let playback = playback_with(Some(store_in(&dir).await));
        fill_queue(&playback, 2);
        let rx = playback.persist_tx.subscribe();

        playback
            .handle_command(PlaybackCommand::Remove { positions: vec![1] })
            .await;

        let snapshot = rx.borrow().clone().expect("snapshot sent");
        assert_eq!(snapshot.tracks.len(), 1);
    }

    /// The queue-order commands are command-level wiring: the ordering itself
    /// is pinned in `queue_order` and `QueueManager`, so what these check is
    /// that the loop performs them, answers, and broadcasts.
    #[tokio::test]
    async fn dedup_command_removes_duplicates_and_replies_with_the_count() {
        let playback = playback_with(
            #[cfg(feature = "fs")]
            None,
        );
        {
            let mut queue = playback.queue.lock().expect("queue lock");
            let _ = queue.replace_with_tracks(&[track(0), track(1), track(0)]);
        }
        let mut updates = playback.update_tx.subscribe();

        let (result_tx, result_rx) = flume::bounded(1);
        playback
            .handle_command(PlaybackCommand::DedupQueue {
                by_title: false,
                result_tx,
            })
            .await;

        assert_eq!(result_rx.recv_async().await.expect("a reply"), 1);
        assert_eq!(queue_titles(&playback), vec!["track 0", "track 1"]);
        // And the new queue went out on the stream, not just into memory.
        match updates.recv().await.expect("a queue update") {
            StreamUpdate::Queue(queue) => assert_eq!(queue.tracks.len(), 2),
            other => panic!("expected a queue update, got {other:?}"),
        }
    }

    /// A dedup with nothing to do still answers — 0 is the reply that tells a
    /// client "no duplicates" (architecture/queue-order.md D2) — and does not
    /// spend a broadcast on an unchanged queue.
    #[tokio::test]
    async fn dedup_command_answers_zero_without_broadcasting() {
        let playback = playback_with(
            #[cfg(feature = "fs")]
            None,
        );
        fill_queue(&playback, 3);
        let mut updates = playback.update_tx.subscribe();

        let (result_tx, result_rx) = flume::bounded(1);
        playback
            .handle_command(PlaybackCommand::DedupQueue {
                by_title: false,
                result_tx,
            })
            .await;

        assert_eq!(result_rx.recv_async().await.expect("a reply"), 0);
        assert!(
            updates.try_recv().is_err(),
            "an unchanged queue is not worth a broadcast"
        );
    }

    /// A dedup whose caller vanished must not stall or panic the loop: the
    /// queue is already changed by the time the count is sent.
    #[tokio::test]
    async fn dedup_command_survives_a_dropped_result_channel() {
        let playback = playback_with(
            #[cfg(feature = "fs")]
            None,
        );
        {
            let mut queue = playback.queue.lock().expect("queue lock");
            let _ = queue.replace_with_tracks(&[track(0), track(0)]);
        }
        let (result_tx, result_rx) = flume::bounded(1);
        drop(result_rx);
        playback
            .handle_command(PlaybackCommand::DedupQueue {
                by_title: false,
                result_tx,
            })
            .await;
        assert_eq!(queue_titles(&playback), vec!["track 0"]);
    }

    #[tokio::test]
    async fn sort_command_reorders_the_queue_and_broadcasts_it() {
        let playback = playback_with(
            #[cfg(feature = "fs")]
            None,
        );
        fill_queue(&playback, 3); // titles ascend: track 0, 1, 2
        let mut updates = playback.update_tx.subscribe();

        playback
            .handle_command(PlaybackCommand::SortQueue {
                sort: crabidy_core::proto::crabidy::QueueSort::Reverse,
                descending: false,
            })
            .await;

        assert_eq!(
            queue_titles(&playback),
            vec!["track 2", "track 1", "track 0"]
        );
        match updates.recv().await.expect("a queue update") {
            StreamUpdate::Queue(queue) => {
                // The playing track moved with the sort, and the snapshot says
                // where it went.
                assert_eq!(queue.current_position, 2);
                assert_eq!(queue.tracks.len(), 3);
            }
            other => panic!("expected a queue update, got {other:?}"),
        }
    }

    /// Both operations are queue mutations, so both must reach the persister
    /// (architecture/queue-order.md D12) — the same check
    /// `queue_mutations_reach_the_persist_channel` makes for `Remove`.
    #[cfg(feature = "fs")]
    #[tokio::test]
    async fn queue_order_commands_reach_the_persist_channel() {
        let dir = TempDir::new().expect("tempdir");
        let playback = playback_with(Some(store_in(&dir).await));
        {
            let mut queue = playback.queue.lock().expect("queue lock");
            let _ = queue.replace_with_tracks(&[track(0), track(1), track(0)]);
        }
        let rx = playback.persist_tx.subscribe();

        let (result_tx, _result_rx) = flume::bounded(1);
        playback
            .handle_command(PlaybackCommand::DedupQueue {
                by_title: false,
                result_tx,
            })
            .await;
        assert_eq!(
            rx.borrow().clone().expect("snapshot sent").tracks.len(),
            2,
            "a dedup must be persisted"
        );

        playback
            .handle_command(PlaybackCommand::SortQueue {
                sort: crabidy_core::proto::crabidy::QueueSort::Reverse,
                descending: false,
            })
            .await;
        let snapshot = rx.borrow().clone().expect("snapshot sent");
        assert_eq!(
            snapshot.tracks.first().map(|t| t.title.clone()),
            Some("track 1".to_string()),
            "a sort must be persisted in its new order"
        );
    }

    /// The gap that let the paste bug through: every previous test drove
    /// either `insert_tracks` or `PendingResolve` directly, so none of them
    /// pinned what the **`Insert` command** does with its position. It is an
    /// index — the row there shifts down — and 0 is the front.
    #[tokio::test]
    async fn insert_command_places_tracks_at_the_given_index() {
        let playback = playback_with(
            #[cfg(feature = "fs")]
            None,
        );
        fill_queue(&playback, 3);

        // The first op registered gets id 0; the resolve itself goes nowhere
        // in tests (no provider), so the chunk is applied by hand.
        playback
            .handle_command(PlaybackCommand::Insert {
                position: 1,
                paths: vec!["/x".to_string()],
            })
            .await;
        playback
            .handle_command(PlaybackCommand::ApplyResolvedChunk {
                op_id: 0,
                tracks: vec![track(9)],
            })
            .await;

        assert_eq!(
            queue_titles(&playback),
            vec!["track 0", "track 9", "track 1", "track 2"],
            "position 1 must push the row that was there down"
        );
    }

    /// Play-next (`L`) must still land right *after* the current track, not
    /// on top of it — the same command-level check for the other caller of
    /// the insert op.
    #[tokio::test]
    async fn queue_command_places_tracks_after_the_current_track() {
        let playback = playback_with(
            #[cfg(feature = "fs")]
            None,
        );
        fill_queue(&playback, 3); // playing track 0
        playback
            .handle_command(PlaybackCommand::Queue {
                paths: vec!["/x".to_string()],
            })
            .await;
        playback
            .handle_command(PlaybackCommand::ApplyResolvedChunk {
                op_id: 0,
                tracks: vec![track(9)],
            })
            .await;
        assert_eq!(
            queue_titles(&playback),
            vec!["track 0", "track 9", "track 1", "track 2"]
        );
    }

    // Seek is not covered here on purpose. The handler's only job is to call
    // `player.seek_by`, and every test in this module builds a real `Player`
    // whose engine thread opens an audio device — so a test that *awaits* a
    // player reply passes or hangs depending on whether the machine running it
    // has working audio output, which is not a property of this code. The
    // clamping arithmetic is tested as a pure function in `audio-player`, and
    // the mapping from the RPC to the command (the layer the paste bug lived
    // in) is tested in `rpc.rs`.

    /// And position 0 reaches the very front — what `P` on the first row needs.
    #[tokio::test]
    async fn insert_command_at_zero_reaches_the_front() {
        let playback = playback_with(
            #[cfg(feature = "fs")]
            None,
        );
        fill_queue(&playback, 2);
        playback
            .handle_command(PlaybackCommand::Insert {
                position: 0,
                paths: vec!["/x".to_string()],
            })
            .await;
        playback
            .handle_command(PlaybackCommand::ApplyResolvedChunk {
                op_id: 0,
                tracks: vec![track(9)],
            })
            .await;
        assert_eq!(
            queue_titles(&playback),
            vec!["track 9", "track 0", "track 1"]
        );
    }
}