//! Persistence abstraction for room data.
//!
//! Defines the [`RoomStore`] trait that encapsulates all SQLite operations
//! behind a clean interface, along with an in-memory implementation for
//! testing and a SQLite-backed implementation for production.
//!
//! The room actor never touches SQL directly — all persistence flows through
//! a [`RoomStore`] implementation.

use std::collections::HashMap;
use std::sync::Mutex;
use std::time::Duration;

use async_trait::async_trait;
use sqlx::{Pool, Sqlite};

use crate::state::{Effect, FinishedTrack, QueuedTrack};
use crate::types::{ChatMessage, ChatMessageId, PlaylistEntryId, TrackId};

/// An entry in a room's perpetual playlist — used for auto-fill when the
/// queue runs dry. Metadata is pre-resolved via the [`MEDIA_CACHE`] table so
/// auto-played tracks never show a "Loading..." placeholder.
///
/// The `title`, `duration`, `thumbnail`, `source`, and `pending` fields are
/// populated at load time via a JOIN with `media_cache`; they are NOT stored
/// independently in the `playlist_entries` table.
#[derive(Debug, Clone, PartialEq, Eq, serde::Serialize)]
pub(crate) struct PlaylistEntry {
    pub id: PlaylistEntryId,
    pub url: String,
    pub title: String,
    pub duration: String,
    pub thumbnail: Option<String>,
    pub source: String,
    pub added_by: Option<String>,
    pub added_at: String,
    /// Derived from `media_cache.state` at load time.
    pub pending: bool,
    /// `Some(reason)` if the URL failed to resolve. Mutually exclusive with
    /// `pending: true` — a failed entry has `pending: false` so the UI can
    /// distinguish it from a still-loading placeholder. The reason is the
    /// same string the state machine wrote into `media_cache.title` as
    /// `"Failed to load: <reason>"`; the prefix is stripped during snapshot
    /// load so consumers see the raw reason.
    pub failed: Option<String>,
}

/// Full snapshot of a room's state loaded from the store.
///
/// Fields are split into the four logical partitions: queue (not yet playing),
/// active (currently playing), history (finished), and chat messages.
#[derive(Debug, Clone)]
#[allow(dead_code)]
pub(crate) struct RoomSnapshot {
    pub room_id: String,
    pub room_name: String,
    /// Tracks waiting to play (not started, not finished).
    pub queue: Vec<QueuedTrack>,
    /// The track currently playing, if any.
    pub active: Option<QueuedTrack>,
    /// Recently finished tracks (newest first).
    pub history: Vec<FinishedTrack>,
    /// Pre-resolved perpetual playlist entries.
    pub playlist: Vec<PlaylistEntry>,
    pub chat_messages: Vec<ChatMessage>,
    /// Room settings: when true, idle rooms auto-fill from the
    /// playlist (also drives the playlist panel visibility in the UI).
    pub autoplay_enabled: bool,
    /// Skip-vote threshold as a percentage of room size (0..=100).
    pub skip_threshold: u8,
}

/// Output returned by [`RoomStore::persist`].
///
/// Carries information the caller needs to react to a persist batch.
#[derive(Debug, Clone, Default)]
pub(crate) struct PersistOutput {
    /// Set when a [`Effect::PersistChat`] was applied.
    pub chat_message_id: Option<ChatMessageId>,
}

/// Abstraction over room persistence.
///
/// All implementors must be [`Send`] + [`Sync`] so they can be shared across
/// async tasks and protected behind an [`Arc`](std::sync::Arc).
#[async_trait]
pub(crate) trait RoomStore: Send + Sync {
    /// Create a new room row.
    async fn create_room(&self, id: &str, name: &str) -> Result<(), sqlx::Error>;

    /// Apply a batch of effects atomically.
    ///
    /// Effects that are not persistence-related (e.g. [`Effect::AbortPipeline`],
    /// [`Effect::PublishSnapshot`]) are silently ignored.
    async fn persist(
        &self,
        room_id: &str,
        effects: &[Effect],
    ) -> Result<PersistOutput, sqlx::Error>;

    /// Load the full snapshot for a room, or [`None`] if it does not exist.
    async fn load(&self, room_id: &str) -> Result<Option<RoomSnapshot>, sqlx::Error>;

    /// Look up a room id by its human-readable name. Returns `None`
    /// when the name is unknown. Used by URL routing: rooms are
    /// addressed by name in URLs and resolved to ids for storage.
    #[allow(dead_code)]
    async fn name_to_id(&self, name: &str) -> Result<Option<String>, sqlx::Error>;

    /// Update the room's settings (autoplay toggle + skip threshold).
    /// Both fields are written in a single UPDATE so the row is never
    /// observed in a half-applied state.
    #[allow(dead_code)]
    async fn update_room_settings(
        &self,
        room_id: &str,
        autoplay_enabled: bool,
        skip_threshold: u8,
    ) -> Result<(), sqlx::Error>;

    /// Retrieve the most recent chat messages for a room, newest first.
    #[allow(dead_code)]
    async fn recent_chat(&self, room_id: &str, limit: i64)
        -> Result<Vec<ChatMessage>, sqlx::Error>;

    /// Delete a room and all its associated data (CASCADE).
    #[allow(dead_code)]
    async fn delete_room(&self, room_id: &str) -> Result<(), sqlx::Error>;
}

/// In-memory variant of [`PlaylistEntry`] that carries extra fields needed
/// for round-trip persistence but not exposed via the public type.
#[cfg_attr(not(test), allow(dead_code))]
#[derive(Debug, Clone)]
struct InMemoryPlaylistEntry {
    id: PlaylistEntryId,
    url: String,
    title: String,
    duration: String,
    thumbnail: Option<String>,
    source: String,
    added_by: Option<String>,
    added_at: String,
    pending: bool,
    /// Newly-pushed entries are never failed; the field is set later by
    /// state-machine effects that mutate the playlist in place. Kept in the
    /// in-memory struct so the public `PlaylistEntry` shape is preserved on
    /// snapshot.
    failed: Option<String>,
}

impl From<InMemoryPlaylistEntry> for PlaylistEntry {
    fn from(e: InMemoryPlaylistEntry) -> Self {
        PlaylistEntry {
            id: e.id,
            url: e.url,
            title: e.title,
            duration: e.duration,
            thumbnail: e.thumbnail,
            source: e.source,
            added_by: e.added_by,
            added_at: e.added_at,
            pending: e.pending,
            failed: e.failed,
        }
    }
}

#[cfg_attr(not(test), allow(dead_code))]
#[derive(Debug)]
struct InMemoryRoom {
    name: String,
    queue: Vec<QueuedTrack>,
    active: Option<QueuedTrack>,
    history: Vec<FinishedTrack>,
    playlist: Vec<InMemoryPlaylistEntry>,
    chat_messages: Vec<ChatMessage>,
    autoplay_enabled: bool,
    skip_threshold: u8,
}

/// Thread-safe in-memory store backed by a [`Mutex`]-protected [`HashMap`].
///
/// Implements [`RoomStore`] without any external dependencies, making it
/// suitable for unit tests and deterministic simulations.
#[cfg_attr(not(test), allow(dead_code))]
#[derive(Debug)]
pub(crate) struct InMemoryStore {
    inner: Mutex<HashMap<String, InMemoryRoom>>,
}

#[cfg_attr(not(test), allow(dead_code))]
impl InMemoryStore {
    pub(crate) fn new() -> Self {
        Self {
            inner: Mutex::new(HashMap::new()),
        }
    }
}

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

#[async_trait]
impl RoomStore for InMemoryStore {
    async fn create_room(&self, id: &str, name: &str) -> Result<(), sqlx::Error> {
        let mut inner = self.inner.lock().expect("InMemoryStore mutex poisoned");
        inner.insert(
            id.to_string(),
            InMemoryRoom {
                name: name.to_string(),
                queue: Vec::new(),
                active: None,
                history: Vec::new(),
                playlist: Vec::new(),
                chat_messages: Vec::new(),
                autoplay_enabled: true,
                skip_threshold: 34,
            },
        );
        Ok(())
    }

    async fn persist(
        &self,
        room_id: &str,
        effects: &[Effect],
    ) -> Result<PersistOutput, sqlx::Error> {
        let mut inner = self.inner.lock().expect("InMemoryStore mutex poisoned");
        let room = match inner.get_mut(room_id) {
            Some(room) => room,
            None => return Ok(PersistOutput::default()),
        };

        let mut output = PersistOutput::default();

        for effect in effects {
            match effect {
                Effect::PersistQueuedTrack(track) => {
                    room.queue.push(track.clone());
                }
                Effect::PersistStarted(id) => {
                    if let Some(pos) = room.queue.iter().position(|t| t.id == *id) {
                        let track = room.queue.remove(pos);
                        room.active = Some(track);
                    }
                }
                Effect::PersistFinished(id) => {
                    if let Some(active) = room.active.take() {
                        if active.id == *id {
                            let now = crate::util::now_iso();
                            room.history.insert(
                                0,
                                FinishedTrack {
                                    id: active.id,
                                    title: active.title,
                                    url: active.url,
                                    duration: active.duration,
                                    thumbnail: active.thumbnail,
                                    played_at: now,
                                },
                            );
                        } else {
                            // ID mismatch — put the active track back.
                            room.active = Some(active);
                        }
                    }
                }
                Effect::PersistMetadata {
                    item_id,
                    url: _,
                    title,
                    duration,
                    thumbnail,
                    source: _,
                } => {
                    if let Some(track) = room.queue.iter_mut().find(|t| t.id == *item_id) {
                        track.title.clone_from(title);
                        track.duration.clone_from(duration);
                        track.thumbnail.clone_from(thumbnail);
                        track.pending = false;
                    }
                    if let Some(ref mut active) = room.active {
                        if active.id == *item_id {
                            active.title.clone_from(title);
                            active.duration.clone_from(duration);
                            active.thumbnail.clone_from(thumbnail);
                            active.pending = false;
                        }
                    }
                }
                Effect::PersistChat {
                    user_name,
                    content,
                    msg_type,
                    created_at,
                } => {
                    let id = ChatMessageId::new();
                    room.chat_messages.push(ChatMessage {
                        id,
                        user_name: user_name.clone(),
                        content: content.clone(),
                        msg_type: msg_type.clone(),
                        created_at: created_at.clone(),
                    });
                    output.chat_message_id = Some(id);
                }
                Effect::AddPlaylistEntry {
                    id,
                    url,
                    title,
                    duration,
                    thumbnail,
                    source,
                } => {
                    room.playlist.push(InMemoryPlaylistEntry {
                        id: *id,
                        url: url.clone(),
                        title: title.clone(),
                        duration: duration.clone(),
                        thumbnail: thumbnail.clone(),
                        source: source.clone(),
                        added_by: None,
                        added_at: crate::util::now_iso(),
                        pending: true,
                        failed: None,
                    });
                }
                Effect::RemovePlaylistEntry(id) => {
                    room.playlist.retain(|e| e.id != *id);
                }
                Effect::ReorderQueue { new_order } => {
                    // The state machine's in-memory `room.queue` is already in
                    // the new order; this arm keeps the effect exhaustively
                    // matched. The SQLite path handles position renumbering.
                    let _ = new_order;
                }
                Effect::RemoveQueuedTrack { track_id } => {
                    room.queue.retain(|t| t.id != *track_id);
                }
                _ => {}
            }
        }

        Ok(output)
    }

    async fn load(&self, room_id: &str) -> Result<Option<RoomSnapshot>, sqlx::Error> {
        let inner = self.inner.lock().expect("InMemoryStore mutex poisoned");
        let room = match inner.get(room_id) {
            Some(room) => room,
            None => return Ok(None),
        };

        Ok(Some(RoomSnapshot {
            room_id: room_id.to_string(),
            room_name: room.name.clone(),
            queue: room.queue.clone(),
            active: room.active.clone(),
            history: room.history.clone(),
            playlist: room.playlist.iter().map(|e| e.clone().into()).collect(),
            chat_messages: room.chat_messages.clone(),
            autoplay_enabled: room.autoplay_enabled,
            skip_threshold: room.skip_threshold,
        }))
    }

    async fn name_to_id(&self, name: &str) -> Result<Option<String>, sqlx::Error> {
        let inner = self.inner.lock().expect("InMemoryStore mutex poisoned");
        Ok(inner
            .iter()
            .find(|(_, r)| r.name == name)
            .map(|(id, _)| id.clone()))
    }

    async fn update_room_settings(
        &self,
        room_id: &str,
        autoplay_enabled: bool,
        skip_threshold: u8,
    ) -> Result<(), sqlx::Error> {
        let mut inner = self.inner.lock().expect("InMemoryStore mutex poisoned");
        if let Some(room) = inner.get_mut(room_id) {
            room.autoplay_enabled = autoplay_enabled;
            room.skip_threshold = skip_threshold;
        }
        Ok(())
    }

    async fn recent_chat(
        &self,
        room_id: &str,
        limit: i64,
    ) -> Result<Vec<ChatMessage>, sqlx::Error> {
        let inner = self.inner.lock().expect("InMemoryStore mutex poisoned");
        let room = match inner.get(room_id) {
            Some(room) => room,
            None => return Ok(Vec::new()),
        };

        let limit = limit.max(0) as usize;
        Ok(room
            .chat_messages
            .iter()
            .rev()
            .take(limit)
            .cloned()
            .collect())
    }

    async fn delete_room(&self, room_id: &str) -> Result<(), sqlx::Error> {
        let mut inner = self.inner.lock().expect("InMemoryStore mutex poisoned");
        inner.remove(room_id);
        Ok(())
    }
}

/// Production [`RoomStore`] backed by an [`sqlx`] SQLite pool.
///
/// Schema is isolated from the legacy `db.rs` tables: this module uses
/// `tracks` (not `queue_items`) and a fresh `rooms` table layout.
#[derive(Debug, Clone)]
pub(crate) struct SqliteStore {
    pool: Pool<Sqlite>,
}

impl SqliteStore {
    /// Open a pool at `path` and run schema migrations.
    pub(crate) async fn connect(path: &std::path::Path) -> Result<Self, sqlx::Error> {
        use sqlx::sqlite::SqlitePoolOptions;

        let conn_str = format!("sqlite:{}?mode=rwc", path.to_string_lossy());
        let pool = SqlitePoolOptions::new()
            .max_connections(8)
            .connect(&conn_str)
            .await?;

        for pragma in [
            "PRAGMA journal_mode=wal",
            "PRAGMA synchronous=NORMAL",
            "PRAGMA foreign_keys=ON",
            "PRAGMA busy_timeout=5000",
        ] {
            sqlx::query(pragma).execute(&pool).await?;
        }

        let store = Self::new(pool).await?;
        Ok(store)
    }

    /// Create a new `SqliteStore` and run schema migrations.
    ///
    /// Idempotent: all tables use `CREATE TABLE IF NOT EXISTS`. New
    /// columns are added via `ALTER TABLE ADD COLUMN` with errors
    /// ignored (so re-running on an already-migrated DB is a no-op).
    pub(crate) async fn new(pool: Pool<Sqlite>) -> Result<Self, sqlx::Error> {
        sqlx::query(
            "CREATE TABLE IF NOT EXISTS rooms (
                id          TEXT PRIMARY KEY,
                name        TEXT NOT NULL DEFAULT '',
                created_at  TEXT NOT NULL,
                updated_at  TEXT NOT NULL
            )",
        )
        .execute(&pool)
        .await?;

        // Migration: add room settings columns. Existing rows pick up
        // the schema defaults. Errors on duplicate-column are ignored.
        for sql in [
            "ALTER TABLE rooms ADD COLUMN autoplay_enabled INTEGER NOT NULL DEFAULT 1",
            "ALTER TABLE rooms ADD COLUMN skip_threshold    INTEGER NOT NULL DEFAULT 34",
        ] {
            if let Err(e) = sqlx::query(sql).execute(&pool).await {
                // "duplicate column name" means the migration already
                // ran. Anything else is a real error.
                if !e.to_string().contains("duplicate column") {
                    return Err(e);
                }
            }
        }

        // Canonical resolved media keyed by URL. Tracks and playlist entries
        // reference this table to avoid duplicating metadata and downloads.
        sqlx::query(
            "CREATE TABLE IF NOT EXISTS media_cache (
                url         TEXT PRIMARY KEY,
                title       TEXT NOT NULL DEFAULT '',
                duration    TEXT NOT NULL DEFAULT '--:--',
                thumbnail   TEXT,
                source      TEXT NOT NULL DEFAULT '',
                file_path   TEXT NOT NULL DEFAULT '',
                state       TEXT NOT NULL DEFAULT 'pending'
            )",
        )
        .execute(&pool)
        .await?;

        sqlx::query(
            "CREATE TABLE IF NOT EXISTS tracks (
                id            TEXT PRIMARY KEY,
                room_id       TEXT NOT NULL REFERENCES rooms(id) ON DELETE CASCADE,
                url           TEXT NOT NULL,
                position      INTEGER NOT NULL,
                added_at      TEXT NOT NULL,
                played        INTEGER NOT NULL DEFAULT 0,
                played_at     TEXT,
                started_at_ms INTEGER
            )",
        )
        .execute(&pool)
        .await?;

        // Composite index for the common room-scoped queries.
        sqlx::query(
            "CREATE INDEX IF NOT EXISTS idx_tracks_room_lookup
             ON tracks(room_id, played, started_at_ms, position)",
        )
        .execute(&pool)
        .await?;

        sqlx::query(
            "CREATE TABLE IF NOT EXISTS chat_messages (
                id          TEXT PRIMARY KEY,
                room_id     TEXT NOT NULL REFERENCES rooms(id) ON DELETE CASCADE,
                user_name   TEXT NOT NULL,
                content     TEXT NOT NULL,
                msg_type    TEXT NOT NULL DEFAULT 'message',
                created_at  TEXT NOT NULL
            )",
        )
        .execute(&pool)
        .await?;

        sqlx::query(
            "CREATE TABLE IF NOT EXISTS playlist_entries (
                id       TEXT PRIMARY KEY,
                room_id  TEXT NOT NULL REFERENCES rooms(id) ON DELETE CASCADE,
                url      TEXT NOT NULL,
                added_by TEXT,
                added_at TEXT NOT NULL
            )",
        )
        .execute(&pool)
        .await?;

        // Enable foreign key enforcement (per-connection, best-effort).
        sqlx::query("PRAGMA foreign_keys = ON")
            .execute(&pool)
            .await?;

        Ok(Self { pool })
    }

    /// Execute persistence effects inside a single transaction.
    ///
    /// Extracted so [`persist`](RoomStore::persist) can wrap it with a retry
    /// loop for `SQLITE_BUSY`.
    async fn persist_inner(
        &self,
        room_id: &str,
        effects: &[Effect],
    ) -> Result<PersistOutput, sqlx::Error> {
        let mut tx = self.pool.begin().await?;
        let mut output = PersistOutput::default();

        for effect in effects {
            match effect {
                Effect::PersistQueuedTrack(track) => {
                    let position: i64 = sqlx::query_scalar(
                        "SELECT COALESCE(MAX(position), -1) + 1 FROM tracks WHERE room_id = ?",
                    )
                    .bind(room_id)
                    .fetch_one(&mut *tx)
                    .await?;

                    // Ensure a media_cache row exists for deduplication.
                    sqlx::query("INSERT OR IGNORE INTO media_cache (url) VALUES (?1)")
                        .bind(&track.url)
                        .execute(&mut *tx)
                        .await?;

                    sqlx::query(
                        "INSERT INTO tracks (id, room_id, url, position, added_at)
                         VALUES (?1, ?2, ?3, ?4, ?5)",
                    )
                    .bind(track.id.as_str_buf())
                    .bind(room_id)
                    .bind(&track.url)
                    .bind(position)
                    .bind(crate::util::now_iso())
                    .execute(&mut *tx)
                    .await?;
                }
                Effect::PersistStarted(id) => {
                    let now_ms = chrono::Utc::now().timestamp_millis();
                    sqlx::query(
                        "UPDATE tracks SET started_at_ms = ?1 WHERE id = ?2 AND played = 0",
                    )
                    .bind(now_ms)
                    .bind(id.as_str_buf())
                    .execute(&mut *tx)
                    .await?;
                }
                Effect::PersistFinished(id) => {
                    let played_at = crate::util::now_iso();
                    sqlx::query("UPDATE tracks SET played = 1, played_at = ?1 WHERE id = ?2")
                        .bind(&played_at)
                        .bind(id.as_str_buf())
                        .execute(&mut *tx)
                        .await?;
                }
                Effect::PersistMetadata {
                    item_id: _,
                    url,
                    title,
                    duration,
                    thumbnail,
                    source,
                } => {
                    sqlx::query(
                        "UPDATE media_cache
                            SET title = ?1, duration = ?2, thumbnail = ?3,
                                source = ?4, state = 'ready'
                          WHERE url = ?5",
                    )
                    .bind(title.as_str())
                    .bind(duration.as_str())
                    .bind(thumbnail.as_deref())
                    .bind(source.as_str())
                    .bind(url.as_str())
                    .execute(&mut *tx)
                    .await?;
                }
                Effect::PersistChat {
                    user_name,
                    content,
                    msg_type,
                    created_at,
                } => {
                    let id = ChatMessageId::new();
                    sqlx::query(
                        "INSERT INTO chat_messages
                             (id, room_id, user_name, content, msg_type, created_at)
                         VALUES (?1, ?2, ?3, ?4, ?5, ?6)",
                    )
                    .bind(id.as_str_buf())
                    .bind(room_id)
                    .bind(user_name.as_str())
                    .bind(content.as_str())
                    .bind(msg_type.as_str())
                    .bind(created_at.as_str())
                    .execute(&mut *tx)
                    .await?;
                    output.chat_message_id = Some(id);
                }
                Effect::AddPlaylistEntry { id, url, .. } => {
                    // Ensure a media_cache row exists for deduplication.
                    sqlx::query("INSERT OR IGNORE INTO media_cache (url) VALUES (?1)")
                        .bind(url.as_str())
                        .execute(&mut *tx)
                        .await?;

                    sqlx::query(
                        "INSERT INTO playlist_entries (id, room_id, url, added_at)
                         VALUES (?1, ?2, ?3, ?4)",
                    )
                    .bind(id.as_str_buf())
                    .bind(room_id)
                    .bind(url.as_str())
                    .bind(crate::util::now_iso())
                    .execute(&mut *tx)
                    .await?;
                }
                Effect::RemovePlaylistEntry(id) => {
                    sqlx::query("DELETE FROM playlist_entries WHERE id = ?")
                        .bind(id.as_str_buf())
                        .execute(&mut *tx)
                        .await?;
                }
                Effect::ReorderQueue { new_order } => {
                    // Renumber every queued track's position to its index in
                    // new_order. Limit to un-played, un-started rows so we
                    // never touch the active or history rows. Using indexed
                    // UPDATEs (one per row) is fine for queues of typical
                    // size (< 100).
                    for (i, track_id) in new_order.iter().enumerate() {
                        sqlx::query(
                            "UPDATE tracks
                                SET position = ?1
                              WHERE id = ?2
                                AND room_id = ?3
                                AND played = 0
                                AND started_at_ms IS NULL",
                        )
                        .bind(i as i64)
                        .bind(track_id.as_str_buf())
                        .bind(room_id)
                        .execute(&mut *tx)
                        .await?;
                    }
                }
                Effect::RemoveQueuedTrack { track_id } => {
                    // Delete the track from the queue. The reorder effect
                    // that follows renumbers the remaining tracks, but
                    // keeping this as a separate statement means a partial
                    // failure leaves the queue in a recoverable state.
                    sqlx::query(
                        "DELETE FROM tracks
                          WHERE id = ?1
                            AND room_id = ?2
                            AND played = 0
                            AND started_at_ms IS NULL",
                    )
                    .bind(track_id.as_str_buf())
                    .bind(room_id)
                    .execute(&mut *tx)
                    .await?;
                }
                Effect::PersistRoomSettings {
                    autoplay_enabled,
                    skip_threshold,
                } => {
                    sqlx::query(
                        "UPDATE rooms
                            SET autoplay_enabled = ?1, skip_threshold = ?2
                          WHERE id = ?3",
                    )
                    .bind(autoplay_enabled)
                    .bind(i64::from(*skip_threshold))
                    .bind(room_id)
                    .execute(&mut *tx)
                    .await?;
                }
                _ => {}
            }
        }

        tx.commit().await?;
        Ok(output)
    }
}

/// Helper row type for deserialising tracks JOINed with media_cache.
#[derive(sqlx::FromRow)]
struct TrackRow {
    id: String,
    url: String,
    played_at: Option<String>,
    title: String,
    duration: String,
    thumbnail: Option<String>,
    state: String,
}

/// Helper row type for deserialising playlist_entries JOINed with media_cache.
#[derive(sqlx::FromRow)]
struct PlaylistRow {
    id: String,
    url: String,
    added_by: Option<String>,
    added_at: String,
    title: String,
    duration: String,
    thumbnail: Option<String>,
    source: String,
    state: String,
}

/// Helper row type for deserialising `chat_messages` rows.
#[derive(sqlx::FromRow)]
struct ChatMsgRow {
    id: String,
    user_name: String,
    content: String,
    msg_type: String,
    created_at: String,
}

fn parse_track_id(s: &str) -> TrackId {
    TrackId(uuid::Uuid::parse_str(s).expect("valid TrackId UUID in DB"))
}

fn parse_playlist_id(s: &str) -> PlaylistEntryId {
    PlaylistEntryId(uuid::Uuid::parse_str(s).expect("valid PlaylistEntryId UUID in DB"))
}

fn parse_chat_message_id(s: &str) -> ChatMessageId {
    ChatMessageId(uuid::Uuid::parse_str(s).expect("valid ChatMessageId UUID in DB"))
}

/// Returns `true` when the error is an SQLite `SQLITE_BUSY` that can be
/// retried.
fn is_sqlite_busy(e: &sqlx::Error) -> bool {
    match e {
        sqlx::Error::Database(db_err) => {
            db_err.message().contains("BUSY") || db_err.code().is_some_and(|c| c == "5")
        }
        _ => false,
    }
}

#[async_trait]
impl RoomStore for SqliteStore {
    async fn create_room(&self, id: &str, name: &str) -> Result<(), sqlx::Error> {
        let now = crate::util::now_iso();
        sqlx::query("INSERT INTO rooms (id, name, created_at, updated_at) VALUES (?1, ?2, ?3, ?4)")
            .bind(id)
            .bind(name)
            .bind(&now)
            .bind(&now)
            .execute(&self.pool)
            .await?;
        Ok(())
    }

    async fn persist(
        &self,
        room_id: &str,
        effects: &[Effect],
    ) -> Result<PersistOutput, sqlx::Error> {
        let mut delay = Duration::from_millis(50);

        for attempt in 1..=3 {
            match self.persist_inner(room_id, effects).await {
                Ok(out) => return Ok(out),
                Err(e) if attempt < 3 && is_sqlite_busy(&e) => {
                    tracing::warn!(attempt, error = %e, "SQLITE_BUSY on persist, retrying");
                    tokio::time::sleep(delay).await;
                    delay *= 2;
                }
                Err(e) => return Err(e),
            }
        }

        unreachable!("retry loop always returns or errors on last iteration")
    }

    async fn load(&self, room_id: &str) -> Result<Option<RoomSnapshot>, sqlx::Error> {
        // 1. Room existence check + settings. Reads in a single
        //    SELECT so the existence, name, and settings come from the
        //    same row snapshot.
        let room_row: Option<(String, bool, i64)> = sqlx::query_as(
            "SELECT name, autoplay_enabled, skip_threshold
               FROM rooms WHERE id = ?",
        )
        .bind(room_id)
        .fetch_optional(&self.pool)
        .await?;

        let (room_name, autoplay_enabled, skip_threshold) = match room_row {
            Some((name, ap, st)) => (name, ap, st.clamp(0, 100) as u8),
            None => return Ok(None),
        };

        // 2. Queue: not played, not started, ordered by position.
        //    JOIN media_cache for resolved metadata. NULLIF coerces the
        //    default empty strings (from media_cache's schema defaults)
        //    to NULL so the COALESCE fallback kicks in for pending
        //    entries.
        let queue_rows: Vec<TrackRow> = sqlx::query_as(
            "SELECT t.id, t.url, t.played_at,
                    COALESCE(NULLIF(m.title, ''), 'Loading...') AS title,
                    COALESCE(NULLIF(m.duration, ''), '--:--') AS duration,
                    m.thumbnail, m.state
               FROM tracks t
               LEFT JOIN media_cache m ON m.url = t.url
              WHERE t.room_id = ? AND t.played = 0 AND t.started_at_ms IS NULL
              ORDER BY t.position",
        )
        .bind(room_id)
        .fetch_all(&self.pool)
        .await?;

        // 3. Active: not played, started (at most one).
        let active_row: Option<TrackRow> = sqlx::query_as(
            "SELECT t.id, t.url, t.played_at,
                    COALESCE(NULLIF(m.title, ''), 'Loading...') AS title,
                    COALESCE(NULLIF(m.duration, ''), '--:--') AS duration,
                    m.thumbnail, m.state
               FROM tracks t
               LEFT JOIN media_cache m ON m.url = t.url
              WHERE t.room_id = ? AND t.played = 0 AND t.started_at_ms IS NOT NULL
              ORDER BY t.position
              LIMIT 1",
        )
        .bind(room_id)
        .fetch_optional(&self.pool)
        .await?;

        // 4. History: played, newest first.
        let history_rows: Vec<TrackRow> = sqlx::query_as(
            "SELECT t.id, t.url, t.played_at,
                    COALESCE(m.title, 'Unknown') AS title,
                    COALESCE(m.duration, '--:--') AS duration,
                    m.thumbnail, m.state
               FROM tracks t
               LEFT JOIN media_cache m ON m.url = t.url
              WHERE t.room_id = ? AND t.played = 1
              ORDER BY t.played_at DESC",
        )
        .bind(room_id)
        .fetch_all(&self.pool)
        .await?;

        // 5. Recent chat messages (most recent 50).
        let chat_rows: Vec<ChatMsgRow> = sqlx::query_as(
            "SELECT id, user_name, content, msg_type, created_at
               FROM chat_messages
              WHERE room_id = ?
              ORDER BY id DESC
              LIMIT 50",
        )
        .bind(room_id)
        .fetch_all(&self.pool)
        .await?;

        // 6. Playlist entries with media_cache metadata. NULLIF coerces
        // the default empty strings (from media_cache's schema defaults)
        // to NULL so the COALESCE fallback kicks in for pending entries.
        let playlist_rows: Vec<PlaylistRow> = sqlx::query_as(
            "SELECT p.id, p.url, p.added_by, p.added_at,
                    COALESCE(NULLIF(m.title, ''), 'Loading...') AS title,
                    COALESCE(NULLIF(m.duration, ''), '--:--') AS duration,
                    m.thumbnail, m.source, m.state
               FROM playlist_entries p
               LEFT JOIN media_cache m ON m.url = p.url
              WHERE p.room_id = ?
              ORDER BY p.id",
        )
        .bind(room_id)
        .fetch_all(&self.pool)
        .await?;

        let pending_from_state = |state: &str| state == "pending";
        // The failure state is encoded in the cached title (set by
        // `Effect::PersistMetadata` from `handle_download_failed`). Strip the
        // prefix here so the public `PlaylistEntry.failed` field matches the
        // in-memory shape — consumers shouldn't have to parse the title.
        let failed_from_title = |title: &str| -> Option<String> {
            title
                .strip_prefix("Failed to load: ")
                .map(|rest| rest.to_string())
        };

        let queue: Vec<QueuedTrack> = queue_rows
            .into_iter()
            .map(|r| QueuedTrack {
                id: parse_track_id(&r.id),
                title: r.title,
                url: r.url,
                duration: r.duration,
                thumbnail: r.thumbnail,
                pending: pending_from_state(&r.state),
            })
            .collect();

        let active = active_row.map(|r| QueuedTrack {
            id: parse_track_id(&r.id),
            title: r.title,
            url: r.url,
            duration: r.duration,
            thumbnail: r.thumbnail,
            pending: pending_from_state(&r.state),
        });

        let history: Vec<FinishedTrack> = history_rows
            .into_iter()
            .map(|r| FinishedTrack {
                id: parse_track_id(&r.id),
                title: r.title,
                url: r.url,
                duration: r.duration,
                thumbnail: r.thumbnail,
                played_at: r.played_at.unwrap_or_default(),
            })
            .collect();

        let chat_messages: Vec<ChatMessage> = chat_rows
            .into_iter()
            .map(|r| ChatMessage {
                id: parse_chat_message_id(&r.id),
                user_name: r.user_name,
                content: r.content,
                msg_type: r.msg_type,
                created_at: r.created_at,
            })
            .collect();

        let playlist: Vec<PlaylistEntry> = playlist_rows
            .into_iter()
            .map(|r| {
                let failed = failed_from_title(&r.title);
                PlaylistEntry {
                    id: parse_playlist_id(&r.id),
                    url: r.url,
                    title: r.title,
                    duration: r.duration,
                    thumbnail: r.thumbnail,
                    source: r.source,
                    added_by: r.added_by,
                    added_at: r.added_at,
                    pending: pending_from_state(&r.state),
                    failed,
                }
            })
            .collect();

        Ok(Some(RoomSnapshot {
            room_id: room_id.to_string(),
            room_name,
            queue,
            active,
            history,
            playlist,
            chat_messages,
            autoplay_enabled,
            skip_threshold,
        }))
    }

    async fn name_to_id(&self, name: &str) -> Result<Option<String>, sqlx::Error> {
        let id: Option<String> = sqlx::query_scalar("SELECT id FROM rooms WHERE name = ?")
            .bind(name)
            .fetch_optional(&self.pool)
            .await?;
        Ok(id)
    }

    async fn update_room_settings(
        &self,
        room_id: &str,
        autoplay_enabled: bool,
        skip_threshold: u8,
    ) -> Result<(), sqlx::Error> {
        sqlx::query(
            "UPDATE rooms
                SET autoplay_enabled = ?1, skip_threshold = ?2
              WHERE id = ?3",
        )
        .bind(autoplay_enabled)
        .bind(skip_threshold as i64)
        .bind(room_id)
        .execute(&self.pool)
        .await?;
        Ok(())
    }

    async fn recent_chat(
        &self,
        room_id: &str,
        limit: i64,
    ) -> Result<Vec<ChatMessage>, sqlx::Error> {
        let rows: Vec<ChatMsgRow> = sqlx::query_as(
            "SELECT id, user_name, content, msg_type, created_at
               FROM chat_messages
              WHERE room_id = ?
              ORDER BY id DESC
              LIMIT ?",
        )
        .bind(room_id)
        .bind(limit)
        .fetch_all(&self.pool)
        .await?;

        Ok(rows
            .into_iter()
            .map(|r| ChatMessage {
                id: parse_chat_message_id(&r.id),
                user_name: r.user_name,
                content: r.content,
                msg_type: r.msg_type,
                created_at: r.created_at,
            })
            .collect())
    }

    async fn delete_room(&self, room_id: &str) -> Result<(), sqlx::Error> {
        sqlx::query("DELETE FROM rooms WHERE id = ?")
            .bind(room_id)
            .execute(&self.pool)
            .await?;
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    fn tid(n: u128) -> TrackId {
        TrackId(uuid::Uuid::from_u128(n))
    }

    #[tokio::test]
    async fn test_create_and_load_roundtrip() {
        let store = InMemoryStore::new();
        store.create_room("test", "Test Room").await.unwrap();

        let snapshot = store.load("test").await.unwrap().unwrap();
        assert_eq!(snapshot.room_id, "test");
        assert_eq!(snapshot.room_name, "Test Room");
        assert!(snapshot.queue.is_empty());
        assert!(snapshot.active.is_none());
        assert!(snapshot.history.is_empty());
        assert!(snapshot.chat_messages.is_empty());
    }

    #[tokio::test]
    async fn test_load_nonexistent_room() {
        let store = InMemoryStore::new();
        let snapshot = store.load("nonexistent").await.unwrap();
        assert!(snapshot.is_none());
    }

    #[tokio::test]
    async fn test_delete_room() {
        let store = InMemoryStore::new();
        store.create_room("test", "Test Room").await.unwrap();

        store.delete_room("test").await.unwrap();
        assert!(store.load("test").await.unwrap().is_none());
    }

    #[tokio::test]
    async fn test_delete_nonexistent_room_is_noop() {
        let store = InMemoryStore::new();
        store.delete_room("nonexistent").await.unwrap();
    }

    #[tokio::test]
    async fn test_recent_chat_empty_room() {
        let store = InMemoryStore::new();
        store.create_room("test", "Test").await.unwrap();

        let msgs = store.recent_chat("test", 10).await.unwrap();
        assert!(msgs.is_empty());
    }

    #[tokio::test]
    async fn test_recent_chat_nonexistent_room() {
        let store = InMemoryStore::new();
        let msgs = store.recent_chat("nobody", 10).await.unwrap();
        assert!(msgs.is_empty());
    }

    #[tokio::test]
    async fn test_persist_preserves_room_order() {
        let store = InMemoryStore::new();
        store.create_room("a", "A").await.unwrap();
        store.create_room("b", "B").await.unwrap();

        let snap_a = store.load("a").await.unwrap().unwrap();
        let snap_b = store.load("b").await.unwrap().unwrap();
        assert_eq!(snap_a.room_name, "A");
        assert_eq!(snap_b.room_name, "B");
    }

    #[tokio::test]
    async fn test_persist_queued_track() {
        let store = InMemoryStore::new();
        store.create_room("r", "R").await.unwrap();

        let track = QueuedTrack {
            id: tid(1),
            title: "Loading...".into(),
            url: "https://example.com/t".into(),
            duration: "--:--".into(),
            thumbnail: None,
            pending: true,
        };

        store
            .persist("r", &[Effect::PersistQueuedTrack(track)])
            .await
            .unwrap();

        let snap = store.load("r").await.unwrap().unwrap();
        assert_eq!(snap.queue.len(), 1);
        assert_eq!(snap.queue[0].id, tid(1));
        assert!(snap.queue[0].pending);
        assert!(snap.active.is_none());
        assert!(snap.history.is_empty());
    }

    #[tokio::test]
    async fn test_persist_started_moves_track_to_active() {
        let store = InMemoryStore::new();
        store.create_room("r", "R").await.unwrap();

        let track = QueuedTrack {
            id: tid(1),
            title: "Track A".into(),
            url: "https://example.com/a".into(),
            duration: "3:45".into(),
            thumbnail: None,
            pending: false,
        };

        store
            .persist("r", &[Effect::PersistQueuedTrack(track)])
            .await
            .unwrap();
        store
            .persist("r", &[Effect::PersistStarted(tid(1))])
            .await
            .unwrap();

        let snap = store.load("r").await.unwrap().unwrap();
        assert!(snap.queue.is_empty());
        assert!(snap.active.is_some());
        assert_eq!(snap.active.unwrap().id, tid(1));
    }

    #[tokio::test]
    async fn test_persist_started_nonexistent_track_is_noop() {
        let store = InMemoryStore::new();
        store.create_room("r", "R").await.unwrap();

        store
            .persist("r", &[Effect::PersistStarted(tid(999))])
            .await
            .unwrap();
        let snap = store.load("r").await.unwrap().unwrap();
        assert!(snap.active.is_none());
    }

    #[tokio::test]
    async fn test_persist_finished_moves_active_to_history() {
        let store = InMemoryStore::new();
        store.create_room("r", "R").await.unwrap();

        store
            .persist(
                "r",
                &[
                    Effect::PersistQueuedTrack(QueuedTrack {
                        id: tid(1),
                        title: "A".into(),
                        url: "https://e.com/a".into(),
                        duration: "3:00".into(),
                        thumbnail: None,
                        pending: false,
                    }),
                    Effect::PersistStarted(tid(1)),
                ],
            )
            .await
            .unwrap();
        store
            .persist("r", &[Effect::PersistFinished(tid(1))])
            .await
            .unwrap();

        let snap = store.load("r").await.unwrap().unwrap();
        assert!(snap.active.is_none());
        assert_eq!(snap.history.len(), 1);
        assert_eq!(snap.history[0].id, tid(1));
        assert!(!snap.history[0].played_at.is_empty());
    }

    #[tokio::test]
    async fn test_persist_finished_without_active_is_noop() {
        let store = InMemoryStore::new();
        store.create_room("r", "R").await.unwrap();

        store
            .persist("r", &[Effect::PersistFinished(tid(1))])
            .await
            .unwrap();
        let snap = store.load("r").await.unwrap().unwrap();
        assert!(snap.history.is_empty());
    }

    #[tokio::test]
    async fn test_persist_metadata_updates_queued_track() {
        let store = InMemoryStore::new();
        store.create_room("r", "R").await.unwrap();

        let track = QueuedTrack {
            id: tid(1),
            title: "Loading...".into(),
            url: "https://e.com/t".into(),
            duration: "--:--".into(),
            thumbnail: None,
            pending: true,
        };

        store
            .persist("r", &[Effect::PersistQueuedTrack(track)])
            .await
            .unwrap();
        store
            .persist(
                "r",
                &[Effect::PersistMetadata {
                    item_id: tid(1),
                    url: "https://e.com/t".into(),
                    title: "Real Title".into(),
                    duration: "4:20".into(),
                    thumbnail: Some("https://img.example/t.jpg".into()),
                    source: "ytdlp".into(),
                }],
            )
            .await
            .unwrap();

        let snap = store.load("r").await.unwrap().unwrap();
        assert_eq!(snap.queue[0].title, "Real Title");
        assert_eq!(snap.queue[0].duration, "4:20");
        assert_eq!(
            snap.queue[0].thumbnail,
            Some("https://img.example/t.jpg".into())
        );
        assert!(!snap.queue[0].pending);
    }

    #[tokio::test]
    async fn test_persist_metadata_updates_active_track() {
        let store = InMemoryStore::new();
        store.create_room("r", "R").await.unwrap();

        store
            .persist(
                "r",
                &[
                    Effect::PersistQueuedTrack(QueuedTrack {
                        id: tid(1),
                        title: "Loading...".into(),
                        url: "https://e.com/t".into(),
                        duration: "--:--".into(),
                        thumbnail: None,
                        pending: true,
                    }),
                    Effect::PersistStarted(tid(1)),
                ],
            )
            .await
            .unwrap();
        store
            .persist(
                "r",
                &[Effect::PersistMetadata {
                    item_id: tid(1),
                    url: "https://e.com/t".into(),
                    title: "Real Title".into(),
                    duration: "4:20".into(),
                    thumbnail: None,
                    source: "direct".into(),
                }],
            )
            .await
            .unwrap();

        let snap = store.load("r").await.unwrap().unwrap();
        assert!(snap.queue.is_empty());
        let active = snap.active.unwrap();
        assert_eq!(active.title, "Real Title");
        assert!(!active.pending);
    }

    #[tokio::test]
    async fn test_persist_chat() {
        let store = InMemoryStore::new();
        store.create_room("r", "R").await.unwrap();

        let output = store
            .persist(
                "r",
                &[Effect::PersistChat {
                    user_name: "alice".into(),
                    content: "hello".into(),
                    msg_type: "message".into(),
                    created_at: "2025-01-01T00:00:00.000Z".into(),
                }],
            )
            .await
            .unwrap();

        assert!(output.chat_message_id.is_some());

        let snap = store.load("r").await.unwrap().unwrap();
        assert_eq!(snap.chat_messages.len(), 1);
        assert_eq!(snap.chat_messages[0].user_name, "alice");
        assert_eq!(snap.chat_messages[0].content, "hello");
    }

    #[tokio::test]
    async fn test_persist_chat_multiple_increments_id() {
        let store = InMemoryStore::new();
        store.create_room("r", "R").await.unwrap();

        let o1 = store
            .persist(
                "r",
                &[Effect::PersistChat {
                    user_name: "a".into(),
                    content: "1".into(),
                    msg_type: "message".into(),
                    created_at: "t1".into(),
                }],
            )
            .await
            .unwrap();
        let o2 = store
            .persist(
                "r",
                &[Effect::PersistChat {
                    user_name: "b".into(),
                    content: "2".into(),
                    msg_type: "message".into(),
                    created_at: "t2".into(),
                }],
            )
            .await
            .unwrap();

        assert!(o1.chat_message_id.is_some());
        assert!(o2.chat_message_id.is_some());
        assert_ne!(o1.chat_message_id, o2.chat_message_id);

        let snap = store.load("r").await.unwrap().unwrap();
        assert_eq!(snap.chat_messages.len(), 2);
    }

    #[tokio::test]
    async fn test_persist_to_nonexistent_room_is_noop() {
        let store = InMemoryStore::new();
        let output = store
            .persist(
                "ghost",
                &[Effect::PersistChat {
                    user_name: "x".into(),
                    content: "x".into(),
                    msg_type: "message".into(),
                    created_at: "x".into(),
                }],
            )
            .await
            .unwrap();
        assert!(output.chat_message_id.is_none());
    }

    #[tokio::test]
    async fn test_persist_recent_chat_ordering() {
        let store = InMemoryStore::new();
        store.create_room("r", "R").await.unwrap();

        for i in 0..5 {
            store
                .persist(
                    "r",
                    &[Effect::PersistChat {
                        user_name: "user".into(),
                        content: format!("msg {i}"),
                        msg_type: "message".into(),
                        created_at: format!("t{i}"),
                    }],
                )
                .await
                .unwrap();
        }

        let recent = store.recent_chat("r", 3).await.unwrap();
        assert_eq!(recent.len(), 3);
        assert_eq!(recent[0].content, "msg 4");
        assert_eq!(recent[2].content, "msg 2");
    }

    // ============================================================
    // SqliteStore tests. These exercise the actual SQL paths for the
    // newer effects (ReorderQueue, RemoveQueuedTrack, AddPlaylistEntry,
    // RemovePlaylistEntry) and verify the on-disk state survives a
    // re-load. InMemoryStore tests above already cover the queue/active
    // /history round-trip; SqliteStore catches SQL bugs that the
    // in-memory impl can't (scoping to played=0, started_at_ms IS NULL,
    // referential integrity via foreign keys, etc.).
    // ============================================================

    async fn sqlite_test_store() -> SqliteStore {
        use sqlx::sqlite::SqlitePoolOptions;
        let pool = SqlitePoolOptions::new()
            .max_connections(1)
            .connect("sqlite::memory:")
            .await
            .unwrap();
        SqliteStore::new(pool).await.unwrap()
    }

    fn queued_track(id: u128, url: &str, title: &str) -> (QueuedTrack, Effect) {
        let t = QueuedTrack {
            id: tid(id),
            title: title.into(),
            url: url.into(),
            duration: "5:00".into(),
            thumbnail: None,
            pending: false,
        };
        (t.clone(), Effect::PersistQueuedTrack(t))
    }

    #[tokio::test]
    async fn sqlite_reorder_queue_renumbers_positions() {
        let store = sqlite_test_store().await;
        store.create_room("r", "R").await.unwrap();

        for (i, url) in ["a", "b", "c", "d"].iter().enumerate() {
            let (_, eff) = queued_track(i as u128 + 1, &format!("u/{url}"), url);
            store.persist("r", &[eff]).await.unwrap();
        }
        // Sanity: initial load returns them in insertion order.
        let snap = store.load("r").await.unwrap().unwrap();
        let ids: Vec<_> = snap.queue.iter().map(|t| t.id).collect();
        assert_eq!(ids, vec![tid(1), tid(2), tid(3), tid(4)]);

        // Move first to last (c, d, a, b).
        store
            .persist(
                "r",
                &[Effect::ReorderQueue {
                    new_order: vec![tid(3), tid(4), tid(1), tid(2)],
                }],
            )
            .await
            .unwrap();

        let snap = store.load("r").await.unwrap().unwrap();
        let ids: Vec<_> = snap.queue.iter().map(|t| t.id).collect();
        assert_eq!(
            ids,
            vec![tid(3), tid(4), tid(1), tid(2)],
            "ReorderQueue must renumber so load returns new_order"
        );
    }

    #[tokio::test]
    async fn sqlite_reorder_queue_does_not_touch_active_or_history() {
        let store = sqlite_test_store().await;
        store.create_room("r", "R").await.unwrap();

        // Queue then start one track to make it active, then queue three more.
        let (_, a) = queued_track(100, "u/active", "active");
        store.persist("r", &[a]).await.unwrap();
        store
            .persist("r", &[Effect::PersistStarted(tid(100))])
            .await
            .unwrap();
        for (i, url) in ["x", "y", "z"].iter().enumerate() {
            let (_, q) = queued_track(i as u128 + 1, &format!("u/{url}"), url);
            store.persist("r", &[q]).await.unwrap();
        }

        // Reorder the queue. The active track (id 100) should be left alone.
        store
            .persist(
                "r",
                &[Effect::ReorderQueue {
                    new_order: vec![tid(3), tid(1), tid(2)],
                }],
            )
            .await
            .unwrap();

        let snap = store.load("r").await.unwrap().unwrap();
        let active_id = snap.active.as_ref().map(|t| t.id);
        assert_eq!(active_id, Some(tid(100)), "active track must be unchanged");
        let queue_ids: Vec<_> = snap.queue.iter().map(|t| t.id).collect();
        assert_eq!(queue_ids, vec![tid(3), tid(1), tid(2)]);
    }

    #[tokio::test]
    async fn sqlite_reorder_queue_with_id_not_in_room_is_a_noop() {
        // The store's UPDATE is scoped to the current room; a stale
        // track id from a different room should be silently skipped.
        let store = sqlite_test_store().await;
        store.create_room("r1", "R1").await.unwrap();
        store.create_room("r2", "R2").await.unwrap();

        let (_, a) = queued_track(1, "u/a", "a");
        store.persist("r1", &[a]).await.unwrap();
        let (_, b) = queued_track(2, "u/b", "b");
        store.persist("r2", &[b]).await.unwrap();

        // Apply a reorder on r1 that includes r2's track id. r1's
        // queue should stay intact.
        store
            .persist(
                "r1",
                &[Effect::ReorderQueue {
                    new_order: vec![tid(2), tid(1)],
                }],
            )
            .await
            .unwrap();

        let snap1 = store.load("r1").await.unwrap().unwrap();
        assert_eq!(snap1.queue.len(), 1);
        assert_eq!(snap1.queue[0].id, tid(1));
        // r2 untouched.
        let snap2 = store.load("r2").await.unwrap().unwrap();
        assert_eq!(snap2.queue[0].id, tid(2));
    }

    #[tokio::test]
    async fn sqlite_remove_queued_track_deletes_row() {
        let store = sqlite_test_store().await;
        store.create_room("r", "R").await.unwrap();

        for (i, url) in ["a", "b", "c"].iter().enumerate() {
            let (_, q) = queued_track(i as u128 + 1, &format!("u/{url}"), url);
            store.persist("r", &[q]).await.unwrap();
        }

        store
            .persist(
                "r",
                &[
                    Effect::RemoveQueuedTrack { track_id: tid(2) },
                    Effect::ReorderQueue {
                        new_order: vec![tid(1), tid(3)],
                    },
                ],
            )
            .await
            .unwrap();

        let snap = store.load("r").await.unwrap().unwrap();
        let ids: Vec<_> = snap.queue.iter().map(|t| t.id).collect();
        assert_eq!(ids, vec![tid(1), tid(3)]);
    }

    #[tokio::test]
    async fn sqlite_remove_queued_track_does_not_touch_active() {
        let store = sqlite_test_store().await;
        store.create_room("r", "R").await.unwrap();

        let (_, a) = queued_track(100, "u/active", "active");
        store.persist("r", &[a]).await.unwrap();
        store
            .persist("r", &[Effect::PersistStarted(tid(100))])
            .await
            .unwrap();
        let (_, q) = queued_track(1, "u/q", "q");
        store.persist("r", &[q]).await.unwrap();

        // Try to remove the active track id via RemoveQueuedTrack —
        // should be a no-op because the SQL scopes to queued rows.
        store
            .persist("r", &[Effect::RemoveQueuedTrack { track_id: tid(100) }])
            .await
            .unwrap();

        let snap = store.load("r").await.unwrap().unwrap();
        assert_eq!(snap.active.as_ref().unwrap().id, tid(100));
        assert_eq!(snap.queue.len(), 1);
    }

    #[tokio::test]
    async fn sqlite_add_playlist_entry_persists_and_appears_in_load() {
        let store = sqlite_test_store().await;
        store.create_room("r", "R").await.unwrap();

        let entry_id = PlaylistEntryId(uuid::Uuid::from_u128(1));
        store
            .persist(
                "r",
                &[Effect::AddPlaylistEntry {
                    id: entry_id,
                    url: "https://example.com/x".into(),
                    title: "Loading...".into(),
                    duration: "--:--".into(),
                    thumbnail: None,
                    source: "ytdlp".into(),
                }],
            )
            .await
            .unwrap();

        let snap = store.load("r").await.unwrap().unwrap();
        assert_eq!(snap.playlist.len(), 1);
        assert_eq!(snap.playlist[0].id, entry_id);
        assert_eq!(snap.playlist[0].url, "https://example.com/x");
        // Without a matching media_cache row, the load shows
        // 'Loading...' / pending=true via the LEFT JOIN.
        assert_eq!(snap.playlist[0].title, "Loading...");
        assert!(snap.playlist[0].pending);
    }

    #[tokio::test]
    async fn sqlite_add_playlist_entry_picks_up_resolved_metadata() {
        // Add the playlist entry first (this seeds media_cache with
        // defaults), then PersistMetadata updates the same row to the
        // resolved state. The load should pick up the resolved title.
        let store = sqlite_test_store().await;
        store.create_room("r", "R").await.unwrap();

        let entry_id = PlaylistEntryId(uuid::Uuid::from_u128(1));
        store
            .persist(
                "r",
                &[Effect::AddPlaylistEntry {
                    id: entry_id,
                    url: "https://example.com/x".into(),
                    title: "Loading...".into(),
                    duration: "--:--".into(),
                    thumbnail: None,
                    source: "ytdlp".into(),
                }],
            )
            .await
            .unwrap();

        store
            .persist(
                "r",
                &[Effect::PersistMetadata {
                    item_id: TrackId::nil(),
                    url: "https://example.com/x".into(),
                    title: "Real Title".into(),
                    duration: "4:20".into(),
                    thumbnail: Some("https://img/x.jpg".into()),
                    source: "ytdlp".into(),
                }],
            )
            .await
            .unwrap();

        let snap = store.load("r").await.unwrap().unwrap();
        assert_eq!(snap.playlist.len(), 1);
        let entry = &snap.playlist[0];
        assert_eq!(entry.title, "Real Title");
        assert_eq!(entry.duration, "4:20");
        assert_eq!(entry.thumbnail, Some("https://img/x.jpg".into()));
        assert!(!entry.pending);
    }

    #[tokio::test]
    async fn sqlite_failed_playlist_entry_rehydrates_with_failed_field() {
        // A playlist entry whose media_cache row was rewritten to a
        // "Failed to load: <err>" title (via Effect::PersistMetadata from
        // handle_download_failed) must rehydrate with `failed = Some(err)`.
        // The DB schema doesn't have a `failed` column — the title prefix is
        // the only signal — so the snapshot loader is the place this
        // contract is enforced.
        let store = sqlite_test_store().await;
        store.create_room("r", "R").await.unwrap();

        let entry_id = PlaylistEntryId(uuid::Uuid::from_u128(1));
        store
            .persist(
                "r",
                &[Effect::AddPlaylistEntry {
                    id: entry_id,
                    url: "https://example.com/dead".into(),
                    title: "Loading...".into(),
                    duration: "--:--".into(),
                    thumbnail: None,
                    source: "ytdlp".into(),
                }],
            )
            .await
            .unwrap();

        store
            .persist(
                "r",
                &[Effect::PersistMetadata {
                    item_id: TrackId::nil(),
                    url: "https://example.com/dead".into(),
                    title: "Failed to load: yt-dlp returned 404".into(),
                    duration: "--:--".into(),
                    thumbnail: None,
                    source: "ytdlp".into(),
                }],
            )
            .await
            .unwrap();

        let snap = store.load("r").await.unwrap().unwrap();
        assert_eq!(snap.playlist.len(), 1);
        let entry = &snap.playlist[0];
        assert_eq!(entry.title, "Failed to load: yt-dlp returned 404");
        assert!(!entry.pending);
        assert_eq!(
            entry.failed.as_deref(),
            Some("yt-dlp returned 404"),
            "snapshot loader must strip the title prefix into the `failed` field",
        );
    }

    #[tokio::test]
    async fn sqlite_resolved_playlist_entry_rehydrates_with_failed_none() {
        // A playlist entry with a non-failure title must rehydrate with
        // `failed = None`. The snapshot loader is prefix-match based, so
        // a title that doesn't start with "Failed to load: " must not
        // surface a stale `failed` reason.
        let store = sqlite_test_store().await;
        store.create_room("r", "R").await.unwrap();

        let entry_id = PlaylistEntryId(uuid::Uuid::from_u128(1));
        store
            .persist(
                "r",
                &[Effect::AddPlaylistEntry {
                    id: entry_id,
                    url: "https://example.com/ok".into(),
                    title: "Loading...".into(),
                    duration: "--:--".into(),
                    thumbnail: None,
                    source: "ytdlp".into(),
                }],
            )
            .await
            .unwrap();

        store
            .persist(
                "r",
                &[Effect::PersistMetadata {
                    item_id: TrackId::nil(),
                    url: "https://example.com/ok".into(),
                    title: "Resolved Title".into(),
                    duration: "1:23".into(),
                    thumbnail: None,
                    source: "ytdlp".into(),
                }],
            )
            .await
            .unwrap();

        let snap = store.load("r").await.unwrap().unwrap();
        let entry = &snap.playlist[0];
        assert_eq!(entry.title, "Resolved Title");
        assert!(!entry.pending);
        assert!(
            entry.failed.is_none(),
            "resolved entry must not have a `failed` reason: {entry:?}",
        );
    }

    #[tokio::test]
    async fn sqlite_remove_playlist_entry_deletes_row() {
        let store = sqlite_test_store().await;
        store.create_room("r", "R").await.unwrap();

        let id_a = PlaylistEntryId(uuid::Uuid::from_u128(1));
        let id_b = PlaylistEntryId(uuid::Uuid::from_u128(2));
        for id in [id_a, id_b] {
            store
                .persist(
                    "r",
                    &[Effect::AddPlaylistEntry {
                        id,
                        url: format!("u/{id}", id = id.as_str_buf()),
                        title: "t".into(),
                        duration: "--:--".into(),
                        thumbnail: None,
                        source: "ytdlp".into(),
                    }],
                )
                .await
                .unwrap();
        }
        assert_eq!(store.load("r").await.unwrap().unwrap().playlist.len(), 2);

        store
            .persist("r", &[Effect::RemovePlaylistEntry(id_a)])
            .await
            .unwrap();

        let snap = store.load("r").await.unwrap().unwrap();
        assert_eq!(snap.playlist.len(), 1);
        assert_eq!(snap.playlist[0].id, id_b);
    }

    #[tokio::test]
    async fn sqlite_cascade_delete_room_removes_tracks_and_playlist() {
        // foreign_keys=ON, ON DELETE CASCADE — deleting a room must
        // remove its tracks, playlist, and chat messages.
        let store = sqlite_test_store().await;
        store.create_room("r", "R").await.unwrap();
        let (_, q) = queued_track(1, "u/a", "a");
        store.persist("r", &[q]).await.unwrap();
        let entry_id = PlaylistEntryId(uuid::Uuid::from_u128(1));
        store
            .persist(
                "r",
                &[Effect::AddPlaylistEntry {
                    id: entry_id,
                    url: "u/a".into(),
                    title: "t".into(),
                    duration: "--:--".into(),
                    thumbnail: None,
                    source: "ytdlp".into(),
                }],
            )
            .await
            .unwrap();

        store.delete_room("r").await.unwrap();
        assert!(store.load("r").await.unwrap().is_none());
    }

    #[tokio::test]
    async fn sqlite_queued_track_loads_placeholder_when_unresolved() {
        // Verifies the NULLIF/COALESCE fix: a queued track whose URL
        // has a default media_cache row (title='', state='pending')
        // should load as 'Loading...' / '--:--' / pending=true. Before
        // the fix, this returned '' / '' / pending=true.
        let store = sqlite_test_store().await;
        store.create_room("r", "R").await.unwrap();
        let (_, q) = queued_track(1, "https://example.com/new", "Loading...");
        store.persist("r", &[q]).await.unwrap();

        let snap = store.load("r").await.unwrap().unwrap();
        assert_eq!(snap.queue.len(), 1);
        let t = &snap.queue[0];
        assert_eq!(t.title, "Loading...");
        assert_eq!(t.duration, "--:--");
        assert!(t.pending);
    }

    #[tokio::test]
    async fn sqlite_new_room_uses_default_room_settings() {
        // A freshly created room picks up the column defaults
        // (autoplay=true, skip_threshold=34) until explicitly changed.
        let store = sqlite_test_store().await;
        store.create_room("r", "R").await.unwrap();
        let snap = store.load("r").await.unwrap().unwrap();
        assert!(snap.autoplay_enabled);
        assert_eq!(snap.skip_threshold, 34);
    }

    #[tokio::test]
    async fn sqlite_update_room_settings_persists_and_loads() {
        let store = sqlite_test_store().await;
        store.create_room("r", "R").await.unwrap();

        store.update_room_settings("r", false, 75).await.unwrap();

        let snap = store.load("r").await.unwrap().unwrap();
        assert!(!snap.autoplay_enabled);
        assert_eq!(snap.skip_threshold, 75);
    }

    #[tokio::test]
    async fn sqlite_persist_room_settings_effect_round_trips() {
        // The effect-driven path: actor sends
        // Effect::PersistRoomSettings, store applies it, load reflects
        // the new values.
        let store = sqlite_test_store().await;
        store.create_room("r", "R").await.unwrap();

        store
            .persist(
                "r",
                &[Effect::PersistRoomSettings {
                    autoplay_enabled: false,
                    skip_threshold: 50,
                }],
            )
            .await
            .unwrap();

        let snap = store.load("r").await.unwrap().unwrap();
        assert!(!snap.autoplay_enabled);
        assert_eq!(snap.skip_threshold, 50);
    }

    #[tokio::test]
    async fn sqlite_name_to_id_resolves_existing_room() {
        let store = sqlite_test_store().await;
        store
            .create_room("uuid-abc", "happy-otters-dance")
            .await
            .unwrap();
        let resolved = store.name_to_id("happy-otters-dance").await.unwrap();
        assert_eq!(resolved, Some("uuid-abc".to_string()));
    }

    #[tokio::test]
    async fn sqlite_name_to_id_returns_none_for_unknown_name() {
        let store = sqlite_test_store().await;
        store
            .create_room("uuid-abc", "happy-otters-dance")
            .await
            .unwrap();
        let resolved = store.name_to_id("sad-otters-mope").await.unwrap();
        assert!(resolved.is_none());
    }

    #[tokio::test]
    async fn sqlite_load_clamps_out_of_range_skip_threshold() {
        // Defensive: if a future bug writes 200 into skip_threshold,
        // the load should clamp to 100 rather than blow up the
        // frontend's u8 JSON deserialization.
        let store = sqlite_test_store().await;
        store.create_room("r", "R").await.unwrap();
        // Bypass the public API to plant a bogus value.
        let SqliteStore { pool } = &store;
        let raw_pool = pool.clone();
        sqlx::query("UPDATE rooms SET skip_threshold = 200 WHERE id = 'r'")
            .execute(&raw_pool)
            .await
            .unwrap();

        let snap = store.load("r").await.unwrap().unwrap();
        assert_eq!(snap.skip_threshold, 100);
    }
}