mineral-server 0.5.10

//! ServerStore 句柄与后端。

use std::sync::Arc;

use mineral_log::{info, warn};
use mineral_model::{Song, SongId, SourceKind};
use rustc_hash::{FxHashMap, FxHashSet};
use sea_orm::sea_query::OnConflict;
use sea_orm::{
    ColumnTrait, DatabaseConnection, EntityTrait, JoinType, PaginatorTrait, QueryFilter,
    QueryOrder, QuerySelect, Set, TransactionTrait,
};

use crate::persistence::db::rows::SongArtistRow;
use crate::persistence::db::schema::ensure_schema;
use crate::persistence::db::{NamespaceStore, SessionStore};
use crate::persistence::entity::{
    playlist_cache, playlist_entries, song_artists, song_favorites, song_meta,
};
use mineral_persist::CacheIndex;

/// 持久化服务句柄。廉价 clone(内部 `Arc`)。
///
/// 打开失败时降级为 [`ServerStore::disabled`]:缓存与收藏写入丢弃、读取返回空；自建歌单写入报不可用,
/// 调用方(channel / server)无需特判,播放照常。
#[derive(Clone)]
pub struct ServerStore {
    /// 内部后端(真实 sqlite 或降级 null)。
    backend: Arc<Backend>,
}

/// 歌单缓存计数(只读统计 / 清理回执)。只读返回 DTO,字段全 `pub`。
pub struct PlaylistCacheStats {
    /// `playlist_cache` 行数(缓存的歌单数)。
    pub playlists: u64,

    /// `playlist_entries` 行数(总曲目 relation 数)。
    pub tracks: u64,
}

/// 内部后端:真实 sqlite 或降级 null。
enum Backend {
    /// 真实 sqlite 连接池。
    Sqlite(DatabaseConnection),

    /// 降级：可重建数据写入丢弃，自建歌单写入报错。
    Disabled,
}

impl ServerStore {
    /// 打开(或创建)数据库文件并建表。
    ///
    /// # Params:
    ///   - `db_path`: sqlite 文件路径(不存在则创建)
    ///
    /// # Return:
    ///   成功返回启用的句柄;失败返回 `Err`(调用方可改用 [`Self::disabled`] 降级)。
    pub async fn open(db_path: &std::path::Path) -> crate::persistence::Result<Self> {
        info!(target: "persist", path = %db_path.display(), "打开 server 数据库");
        let pool = mineral_persist::connect(db_path).await?;
        ensure_schema(&pool).await?;
        Ok(Self {
            backend: Arc::new(Backend::Sqlite(pool)),
        })
    }

    /// 降级句柄：读取为空，自建歌单写入报错，其余写入丢弃。
    ///
    /// # Return:
    ///   不访问磁盘的 [`ServerStore`]。
    pub fn disabled() -> Self {
        warn!(target: "persist", "持久化降级为 no-op(disabled)");
        Self {
            backend: Arc::new(Backend::Disabled),
        }
    }

    /// 取连接池(降级时为 `None`)。
    ///
    /// # Return:
    ///   启用时为底层连接池,降级时为 `None`。
    pub(crate) fn pool(&self) -> Option<&DatabaseConnection> {
        match self.backend.as_ref() {
            Backend::Sqlite(p) => Some(p),
            Backend::Disabled => None,
        }
    }

    /// 取某来源命名空间下的存储视图。
    ///
    /// # Params:
    ///   - `source`: 来源标识(决定 namespace 过滤)
    ///
    /// # Return:
    ///   绑定该 namespace 的 [`NamespaceStore`]。
    pub fn scope(&self, source: SourceKind) -> NamespaceStore {
        NamespaceStore::new(self.clone(), source)
    }

    /// 取全局会话存储。
    ///
    /// # Return:
    ///   [`SessionStore`]。
    pub fn session(&self) -> SessionStore {
        SessionStore::new(self.clone())
    }

    /// 音频本体缓存索引(`audio_cache` 表,LRU 驱逐)。播放命中本地副本走它。
    ///
    /// # Params:
    ///   - `root`: 缓存文件根目录(`relpath` 相对它)
    ///   - `capacity`: 容量上限字节(LRU 满了驱逐最旧)
    ///
    /// # Return:
    ///   就绪索引;降级句柄返回 [`CacheIndex::disabled`];建表 / 载入失败返回 `Err`。
    pub async fn audio_cache(
        &self,
        root: std::path::PathBuf,
        capacity: u64,
    ) -> crate::persistence::Result<CacheIndex> {
        match self.pool() {
            Some(pool) => {
                Ok(CacheIndex::open(pool.clone(), "audio_cache", root, Some(capacity)).await?)
            }
            None => Ok(CacheIndex::disabled()),
        }
    }

    /// 全部源的 loved 歌曲(join meta 重建),按 `entered_at` 降序(最新收藏在顶),
    /// 同一导入批次共用时间，以 `(namespace, song_value)` 破平局，顺序稳定不随库文件重排。
    ///
    /// loved 但缺 meta 的行**跳过**——聚合视图与其曲目计数保持同口径,不出现「有行但
    /// 没名字」的占位。缺 meta 是常态:sync 导入的远端红心先只有 id,meta 随浏览补全。
    /// 降级句柄返回空集。
    ///
    /// # Return:
    ///   跨 namespace 的收藏 `Vec<Song>`。
    pub async fn loved_songs(&self) -> crate::persistence::Result<Vec<Song>> {
        let Some(db) = self.pool() else {
            return Ok(Vec::new());
        };
        let rows = song_meta::Entity::find()
            .join(
                JoinType::InnerJoin,
                song_meta::Entity::belongs_to(song_favorites::Entity)
                    .from((song_meta::Column::Namespace, song_meta::Column::SongValue))
                    .to((
                        song_favorites::Column::Namespace,
                        song_favorites::Column::SongValue,
                    ))
                    .into(),
            )
            .order_by_desc(song_favorites::Column::EnteredAt)
            .order_by_asc(song_favorites::Column::Namespace)
            .order_by_asc(song_favorites::Column::SongValue)
            .all(db)
            .await
            .map_err(|source| crate::persistence::Error::Database {
                operation: "查跨源 loved 元数据",
                source,
            })?;
        let artist_rows = song_artists::Entity::find()
            .join(
                JoinType::InnerJoin,
                song_artists::Entity::belongs_to(song_favorites::Entity)
                    .from((
                        song_artists::Column::Namespace,
                        song_artists::Column::SongValue,
                    ))
                    .to((
                        song_favorites::Column::Namespace,
                        song_favorites::Column::SongValue,
                    ))
                    .into(),
            )
            .order_by_asc(song_artists::Column::Position)
            .all(db)
            .await
            .map_err(|source| crate::persistence::Error::Database {
                operation: "查跨源 loved 艺人",
                source,
            })?;
        let mut artists = FxHashMap::<(String, String), Vec<SongArtistRow>>::default();
        for row in artist_rows {
            artists
                .entry((row.namespace, row.song_value))
                .or_default()
                .push(SongArtistRow {
                    artist_id: row.artist_id,
                    artist_name: row.artist_name,
                });
        }
        rows.into_iter()
            .map(|row| {
                let key = (row.namespace.clone(), row.song_value.clone());
                row.into_song(artists.remove(&key).unwrap_or_default())
            })
            .collect()
    }

    /// 跨源 loved 歌曲计数,与 [`Self::loved_songs`] **严格同口径**(只计 join 到 meta 的
    /// 收藏,缺 meta 的行不计)——聚合歌单列表面只要计数时走它,免为拿个数字重建整个
    /// `Vec<Song>`。降级句柄返回 0。
    ///
    /// # Return:
    ///   有 meta 的跨源收藏数。
    pub async fn loved_count(&self) -> crate::persistence::Result<u64> {
        let Some(db) = self.pool() else {
            return Ok(0);
        };
        song_meta::Entity::find()
            .join(
                JoinType::InnerJoin,
                song_meta::Entity::belongs_to(song_favorites::Entity)
                    .from((song_meta::Column::Namespace, song_meta::Column::SongValue))
                    .to((
                        song_favorites::Column::Namespace,
                        song_favorites::Column::SongValue,
                    ))
                    .into(),
            )
            .count(db)
            .await
            .map_err(|source| crate::persistence::Error::Database {
                operation: "统计跨源 loved 计数",
                source,
            })
    }

    /// 全部源里 loved 但**缺 meta** 的歌 id(sync 导入的远端红心先只有 id、无 meta)。
    /// 供后台补 meta 任务拉详情回填;补齐后它们就进 [`Self::loved_songs`] 的聚合视图。
    /// 降级句柄返回空。
    ///
    /// # Return:
    ///   跨 namespace 的缺 meta 收藏 id(namespace 从行内 `SourceKind::from_name` 还原)。
    pub async fn missing_meta_loved_ids(&self) -> crate::persistence::Result<Vec<SongId>> {
        let Some(db) = self.pool() else {
            return Ok(Vec::new());
        };
        let rows = song_favorites::Entity::find()
            .join(
                JoinType::LeftJoin,
                song_favorites::Entity::belongs_to(song_meta::Entity)
                    .from((
                        song_favorites::Column::Namespace,
                        song_favorites::Column::SongValue,
                    ))
                    .to((song_meta::Column::Namespace, song_meta::Column::SongValue))
                    .into(),
            )
            .filter(song_meta::Column::SongValue.is_null())
            .all(db)
            .await
            .map_err(|source| crate::persistence::Error::Database {
                operation: "查缺 meta 的 loved 行",
                source,
            })?;
        Ok(rows
            .into_iter()
            .map(|row| SongId::new(SourceKind::from_name(&row.namespace), row.song_value))
            .collect())
    }

    /// 把 remote membership snapshot 中尚未存在的 Song 作为同一 import batch 加入 Favorites。
    ///
    /// 整批只读取一次 `entered_at`；既有 membership 按 conflict no-op，不更新时间。返回顺序按
    /// `(namespace, song_value)` 固定，与输入 HashSet iteration 无关。
    ///
    /// # Params:
    ///   - `ids`: remote snapshot 中的全量 favorite SongId
    ///
    /// # Return:
    ///   本次实际新插入的 SongId；降级句柄返回空。
    pub async fn import_favorites(
        &self,
        ids: &FxHashSet<SongId>,
    ) -> crate::persistence::Result<Vec<SongId>> {
        let Some(pool) = self.pool() else {
            return Ok(Vec::new());
        };
        let mut ordered = ids.iter().cloned().collect::<Vec<SongId>>();
        ordered.sort_by(|left, right| {
            left.namespace()
                .name()
                .cmp(right.namespace().name())
                .then_with(|| left.value().cmp(right.value()))
        });
        let tx = pool
            .begin()
            .await
            .map_err(|source| crate::persistence::Error::Database {
                operation: "开启 Favorites import 事务",
                source,
            })?;
        let entered_at = next_favorite_entry_time(&tx).await?;
        let mut inserted = Vec::<SongId>::new();
        for id in ordered {
            let result = song_favorites::Entity::insert(song_favorites::ActiveModel {
                namespace: Set(id.namespace().name().to_owned()),
                song_value: Set(id.value().to_owned()),
                entered_at: Set(entered_at),
            })
            .on_conflict(
                OnConflict::columns([
                    song_favorites::Column::Namespace,
                    song_favorites::Column::SongValue,
                ])
                .do_nothing()
                .to_owned(),
            )
            .exec_without_returning(&tx)
            .await
            .map_err(|source| crate::persistence::Error::Record {
                operation: "导入 favorite",
                record: id.qualified(),
                source,
            })?;
            if result > 0 {
                inserted.push(id);
            }
        }
        tx.commit()
            .await
            .map_err(|source| crate::persistence::Error::Database {
                operation: "提交 Favorites import 事务",
                source,
            })?;
        Ok(inserted)
    }

    /// 歌单缓存计数(只读)。供 CLI `cache status` 展示用。
    ///
    /// # Return:
    ///   启用态返回 `playlist_cache` / `playlist_entries` 行数;降级句柄返回全 0。
    pub async fn playlist_cache_stats(&self) -> crate::persistence::Result<PlaylistCacheStats> {
        let Some(db) = self.pool() else {
            return Ok(PlaylistCacheStats {
                playlists: 0,
                tracks: 0,
            });
        };
        let playlists = playlist_cache::Entity::find()
            .count(db)
            .await
            .map_err(|source| crate::persistence::Error::Database {
                operation: "统计 playlist_cache 行数",
                source,
            })?;
        let tracks = playlist_entries::Entity::find()
            .count(db)
            .await
            .map_err(|source| crate::persistence::Error::Database {
                operation: "统计 playlist_entries 行数",
                source,
            })?;
        Ok(PlaylistCacheStats { playlists, tracks })
    }

    /// 清空歌单缓存(`playlist_cache` + `playlist_entries` 全部来源)。
    ///
    /// 只清可重建的歌单缓存，**不动**播放统计 / love / 历史 / 会话 / song_meta。
    /// 降级句柄下 no-op。
    ///
    /// # Return:
    ///   清理成功返回被清掉的计数(清理前 `playlist_cache` / `playlist_entries` 行数);降级返回全 0。
    pub async fn clear_playlist_caches(&self) -> crate::persistence::Result<PlaylistCacheStats> {
        let Some(db) = self.pool() else {
            return Ok(PlaylistCacheStats {
                playlists: 0,
                tracks: 0,
            });
        };
        info!(target: "persist", "清理歌单缓存");
        let tx = db
            .begin()
            .await
            .map_err(|source| crate::persistence::Error::Database {
                operation: "开启 clear_playlist_caches 事务",
                source,
            })?;
        let playlists = playlist_cache::Entity::find()
            .count(&tx)
            .await
            .map_err(|source| crate::persistence::Error::Database {
                operation: "统计 playlist_cache 行数",
                source,
            })?;
        let tracks = playlist_entries::Entity::find()
            .count(&tx)
            .await
            .map_err(|source| crate::persistence::Error::Database {
                operation: "统计 playlist_entries 行数",
                source,
            })?;
        playlist_entries::Entity::delete_many()
            .exec(&tx)
            .await
            .map_err(|source| crate::persistence::Error::Database {
                operation: "清空 playlist_entries",
                source,
            })?;
        playlist_cache::Entity::delete_many()
            .exec(&tx)
            .await
            .map_err(|source| crate::persistence::Error::Database {
                operation: "清空 playlist_cache",
                source,
            })?;
        tx.commit()
            .await
            .map_err(|source| crate::persistence::Error::Database {
                operation: "提交 clear_playlist_caches 事务",
                source,
            })?;
        Ok(PlaylistCacheStats { playlists, tracks })
    }
}

/// 为新收藏或整批导入分配晚于现有记录的时间；调用方在同一事务内完成插入。
/// 同毫秒的连续操作也按加入先后排序，批次内共用返回值。
pub(super) async fn next_favorite_entry_time(
    tx: &sea_orm::DatabaseTransaction,
) -> crate::persistence::Result<i64> {
    let latest = song_favorites::Entity::find()
        .select_only()
        .expr(song_favorites::Column::EnteredAt.max())
        .into_tuple::<Option<i64>>()
        .one(tx)
        .await
        .map_err(|source| crate::persistence::Error::Database {
            operation: "读取 Favorites 最新进入时间",
            source,
        })?
        .flatten();
    let now = crate::persistence::db::time::now_ms();
    Ok(latest
        .map(|value| value.saturating_add(1))
        .map_or(now, |next| next.max(now)))
}

#[cfg(test)]
mod tests {
    use super::ServerStore;
    use crate::persistence::entity::song_favorites;
    use sea_orm::sea_query::{Expr, ExprTrait};
    use sea_orm::{ColumnTrait, EntityTrait, QueryFilter, QuerySelect};

    #[test]
    fn disabled_has_no_pool() {
        assert!(ServerStore::disabled().pool().is_none());
    }

    /// 跨源聚合:两源各一首 loved(带 meta)按 entered_at 降序返回;
    /// 无 meta 的 loved 跳过;未 loved 的 meta 不出现。
    #[tokio::test]
    async fn loved_songs_aggregates_across_namespaces() -> color_eyre::Result<()> {
        use mineral_model::{SongId, SourceKind};
        use mineral_test::{song, with_artist, with_name};

        let dir = tempfile::tempdir()?;
        let p = ServerStore::open(&dir.path().join("t.db")).await?;
        let netease = p.scope(SourceKind::NETEASE);
        let bilibili = p.scope(SourceKind::BILIBILI);

        // netease:较早收藏;bilibili:较晚收藏(手动定 entered_at,免同毫秒排序不稳)。
        let n1 = with_artist(with_name(song("n1"), "Palisade"), "Mineral");
        netease.upsert_meta(&n1).await?;
        netease.set_loved(&n1.id, true).await?;
        let b1 = {
            let mut s = with_name(song("b1"), "夜間飛行");
            s.id = SongId::new(SourceKind::BILIBILI, "b1");
            s
        };
        bilibili.upsert_meta(&b1).await?;
        bilibili.set_loved(&b1.id, true).await?;
        // 有 meta 但未 loved:不该出现。
        netease
            .upsert_meta(&with_name(song("n2"), "unloved"))
            .await?;
        // loved 但无 meta:跳过。
        netease
            .set_loved(&SongId::new(SourceKind::NETEASE, "ghost"), true)
            .await?;

        let pool = p
            .pool()
            .ok_or_else(|| color_eyre::eyre::eyre!("测试库应有 pool"))?;
        song_favorites::Entity::update_many()
            .col_expr(song_favorites::Column::EnteredAt, Expr::value(100i64))
            .filter(song_favorites::Column::SongValue.eq("n1"))
            .exec(pool)
            .await?;
        song_favorites::Entity::update_many()
            .col_expr(song_favorites::Column::EnteredAt, Expr::value(200i64))
            .filter(song_favorites::Column::SongValue.eq("b1"))
            .exec(pool)
            .await?;

        let songs = p.loved_songs().await?;
        let names = songs.iter().map(|s| s.name.as_str()).collect::<Vec<&str>>();
        assert_eq!(
            names,
            vec!["夜間飛行", "Palisade"],
            "按 entered_at 降序,缺 meta 的 ghost 跳过,未 loved 的不出现"
        );
        let first = songs
            .first()
            .ok_or_else(|| color_eyre::eyre::eyre!("应有两首"))?;
        assert_eq!(first.source(), SourceKind::BILIBILI, "namespace 还原为原源");
        let second = songs
            .get(1)
            .ok_or_else(|| color_eyre::eyre::eyre!("应有两首"))?;
        assert_eq!(
            second.artists.first().map(|a| a.name.as_str()),
            Some("Mineral"),
            "艺人列表随 meta 还原"
        );
        Ok(())
    }

    /// loved_count 与 loved_songs 同口径:只计 join 到 meta 的收藏(ghost 无 meta 不计,
    /// unloved 的 meta 不计)。
    #[tokio::test]
    async fn loved_count_matches_loved_songs() -> color_eyre::Result<()> {
        use mineral_model::{SongId, SourceKind};
        use mineral_test::{song, with_name};

        let dir = tempfile::tempdir()?;
        let p = ServerStore::open(&dir.path().join("t.db")).await?;
        let netease = p.scope(SourceKind::NETEASE);
        let a = with_name(song("a"), "Alpha");
        netease.upsert_meta(&a).await?;
        netease.set_loved(&a.id, true).await?;
        // 有 meta 未 loved:不计。
        netease
            .upsert_meta(&with_name(song("b"), "unloved"))
            .await?;
        // loved 无 meta(ghost):不计。
        netease
            .set_loved(&SongId::new(SourceKind::NETEASE, "ghost"), true)
            .await?;

        assert_eq!(p.loved_count().await?, 1, "只计有 meta 的收藏");
        assert_eq!(
            p.loved_count().await?,
            u64::try_from(p.loved_songs().await?.len())?,
            "count 与 songs 长度同口径"
        );
        Ok(())
    }

    /// 降级句柄:loved_count 返回 0 不报错。
    #[tokio::test]
    async fn loved_count_disabled_is_zero() -> color_eyre::Result<()> {
        assert_eq!(ServerStore::disabled().loved_count().await?, 0);
        Ok(())
    }

    /// 同毫秒收藏:tiebreaker `(namespace, song_value)` 给出确定顺序,不靠 SQLite 任意 tie 序;
    /// 逆序插入也按 song_value 升序返回。
    #[tokio::test]
    async fn loved_songs_stable_order_on_same_millisecond() -> color_eyre::Result<()> {
        use mineral_model::SourceKind;
        use mineral_test::{song, with_name};

        let dir = tempfile::tempdir()?;
        let p = ServerStore::open(&dir.path().join("t.db")).await?;
        let netease = p.scope(SourceKind::NETEASE);
        // 逆序插入(先 bbb 后 aaa),验证返回顺序由 tiebreaker 而非插入序决定。
        let b = with_name(song("bbb"), "Beta");
        netease.upsert_meta(&b).await?;
        netease.set_loved(&b.id, true).await?;
        let a = with_name(song("aaa"), "Alpha");
        netease.upsert_meta(&a).await?;
        netease.set_loved(&a.id, true).await?;

        let pool = p
            .pool()
            .ok_or_else(|| color_eyre::eyre::eyre!("测试库应有 pool"))?;
        song_favorites::Entity::update_many()
            .col_expr(song_favorites::Column::EnteredAt, Expr::value(500i64))
            .exec(pool)
            .await?;

        let songs = p.loved_songs().await?;
        let names = songs.iter().map(|s| s.name.as_str()).collect::<Vec<&str>>();
        assert_eq!(
            names,
            vec!["Alpha", "Beta"],
            "同 entered_at 下按 song_value 升序(aaa 在 bbb 前),与插入序无关"
        );
        Ok(())
    }

    /// Remote import 一批共用 entered_at、按 id 稳定返回；重复 snapshot 是完整 no-op，
    /// 后续新增 batch 排在旧 batch 前。
    #[tokio::test]
    async fn import_favorites_is_stable_idempotent_and_newest_first() -> color_eyre::Result<()> {
        use mineral_model::{SongId, SourceKind};
        use mineral_test::{song, with_name};
        use rustc_hash::FxHashSet;

        let dir = tempfile::tempdir()?;
        let p = ServerStore::open(&dir.path().join("t.db")).await?;
        let netease = p.scope(SourceKind::NETEASE);
        let bilibili = p.scope(SourceKind::BILIBILI);
        let n = with_name(song("n"), "N");
        netease.upsert_meta(&n).await?;
        let mut b = with_name(song("b"), "B");
        b.id = SongId::new(SourceKind::BILIBILI, "b");
        bilibili.upsert_meta(&b).await?;

        let first = FxHashSet::from_iter([n.id.clone(), b.id.clone()]);
        assert_eq!(
            p.import_favorites(&first).await?,
            vec![b.id.clone(), n.id.clone()],
            "返回顺序由 namespace/value 决定，不依赖 HashSet iteration"
        );
        assert!(
            p.import_favorites(&first).await?.is_empty(),
            "相同 snapshot 第二次应零 insert"
        );
        let pool = p
            .pool()
            .ok_or_else(|| color_eyre::eyre::eyre!("测试库应有 pool"))?;
        let (distinct_times,): (i64,) = song_favorites::Entity::find()
            .select_only()
            .expr(Expr::col(song_favorites::Column::EnteredAt).count_distinct())
            .into_tuple::<(i64,)>()
            .one(pool)
            .await?
            .ok_or_else(|| color_eyre::eyre::eyre!("missing favorite count"))?;
        assert_eq!(distinct_times, 1, "同一 batch 必须共用 entered_at");

        let newer = with_name(song("new"), "New");
        netease.upsert_meta(&newer).await?;
        let second = FxHashSet::from_iter([n.id.clone(), b.id.clone(), newer.id.clone()]);
        assert_eq!(p.import_favorites(&second).await?, vec![newer.id]);
        assert_eq!(
            p.loved_songs()
                .await?
                .into_iter()
                .map(|song| song.name)
                .collect::<Vec<String>>(),
            vec!["New".to_owned(), "B".to_owned(), "N".to_owned()],
            "新 batch 应整体位于旧 batch 前，旧 membership 不重排"
        );
        Ok(())
    }

    /// 用户取消后重新收藏会取得新 entered_at 并回到顶部；同状态 set 不移动。
    #[tokio::test]
    async fn relove_returns_to_top_without_same_state_reorder() -> color_eyre::Result<()> {
        use mineral_model::SourceKind;
        use mineral_test::{song, with_name};

        let dir = tempfile::tempdir()?;
        let p = ServerStore::open(&dir.path().join("t.db")).await?;
        let scope = p.scope(SourceKind::NETEASE);
        let a = with_name(song("a"), "A");
        let b = with_name(song("b"), "B");
        scope.upsert_meta(&a).await?;
        scope.upsert_meta(&b).await?;
        assert!(scope.set_loved(&a.id, true).await?);
        // 让现有时间领先时钟，验证新收藏置顶不依赖两次操作间的毫秒推进。
        let pool = p
            .pool()
            .ok_or_else(|| color_eyre::eyre::eyre!("测试库应有 pool"))?;
        let entered_at = crate::persistence::db::time::now_ms() + 60_000;
        song_favorites::Entity::update_many()
            .col_expr(song_favorites::Column::EnteredAt, Expr::value(entered_at))
            .filter(song_favorites::Column::Namespace.eq(a.id.namespace().name()))
            .filter(song_favorites::Column::SongValue.eq(a.id.value()))
            .exec(pool)
            .await?;
        assert!(scope.set_loved(&b.id, true).await?);
        assert!(
            !scope.set_loved(&a.id, true).await?,
            "true -> true 不应移动 A"
        );
        assert_eq!(
            p.loved_songs()
                .await?
                .into_iter()
                .map(|song| song.name)
                .collect::<Vec<String>>(),
            vec!["B".to_owned(), "A".to_owned()]
        );
        assert!(scope.set_loved(&a.id, false).await?);
        assert!(scope.set_loved(&a.id, true).await?);
        assert_eq!(
            p.loved_songs()
                .await?
                .into_iter()
                .map(|song| song.name)
                .collect::<Vec<String>>(),
            vec!["A".to_owned(), "B".to_owned()]
        );
        Ok(())
    }

    /// Metadata backfill 只让既有 favorite membership 变得可见，不更新时间也不把它冒充新收藏。
    #[tokio::test]
    async fn metadata_backfill_preserves_favorite_order() -> color_eyre::Result<()> {
        use mineral_model::{SongId, SourceKind};
        use mineral_test::{song, with_name};

        let dir = tempfile::tempdir()?;
        let persist = ServerStore::open(&dir.path().join("t.db")).await?;
        let scope = persist.scope(SourceKind::NETEASE);
        let old_id = SongId::new(SourceKind::NETEASE, "old-missing-meta");
        assert!(scope.set_loved(&old_id, true).await?);
        let newer = with_name(song("new-visible"), "New");
        scope.upsert_meta(&newer).await?;
        assert!(scope.set_loved(&newer.id, true).await?);
        let pool = persist
            .pool()
            .ok_or_else(|| color_eyre::eyre::eyre!("测试库应有 pool"))?;
        let before: (i64,) = song_favorites::Entity::find_by_id((
            SourceKind::NETEASE.name().to_owned(),
            old_id.value().to_owned(),
        ))
        .select_only()
        .column(song_favorites::Column::EnteredAt)
        .into_tuple::<(i64,)>()
        .one(pool)
        .await?
        .ok_or_else(|| color_eyre::eyre::eyre!("missing favorite timestamp"))?;

        let old_meta = with_name(song("old-missing-meta"), "Old");
        scope.upsert_meta(&old_meta).await?;

        let after: (i64,) = song_favorites::Entity::find_by_id((
            SourceKind::NETEASE.name().to_owned(),
            old_id.value().to_owned(),
        ))
        .select_only()
        .column(song_favorites::Column::EnteredAt)
        .into_tuple::<(i64,)>()
        .one(pool)
        .await?
        .ok_or_else(|| color_eyre::eyre::eyre!("missing favorite timestamp"))?;
        assert_eq!(after, before, "backfill 不得改 entered_at");
        assert_eq!(
            persist
                .loved_songs()
                .await?
                .into_iter()
                .map(|song| song.name)
                .collect::<Vec<_>>(),
            vec!["New".to_owned(), "Old".to_owned()],
            "旧 membership 补齐 metadata 后按既有 entered_at 插回原位置"
        );
        Ok(())
    }

    /// missing_meta_loved_ids:只列 loved 且缺 meta 的行(有 meta 的 loved 不列,unloved 不列)。
    #[tokio::test]
    async fn missing_meta_loved_ids_lists_only_meta_less_loved() -> color_eyre::Result<()> {
        use mineral_model::{SongId, SourceKind};
        use mineral_test::{song, with_name};

        let dir = tempfile::tempdir()?;
        let p = ServerStore::open(&dir.path().join("t.db")).await?;
        let netease = p.scope(SourceKind::NETEASE);
        // 有 meta + loved:不列。
        let a = with_name(song("a"), "Alpha");
        netease.upsert_meta(&a).await?;
        netease.set_loved(&a.id, true).await?;
        // loved 无 meta:列。
        let ghost = SongId::new(SourceKind::NETEASE, "ghost");
        netease.set_loved(&ghost, true).await?;
        // 有 meta 未 loved:不列。
        netease
            .upsert_meta(&with_name(song("b"), "unloved"))
            .await?;

        let ids = p.missing_meta_loved_ids().await?;
        assert_eq!(ids, vec![ghost], "只列 loved 且缺 meta 的");
        Ok(())
    }

    /// 降级句柄:loved_songs 返回空集不报错。
    #[tokio::test]
    async fn loved_songs_disabled_is_empty() -> color_eyre::Result<()> {
        assert!(ServerStore::disabled().loved_songs().await?.is_empty());
        Ok(())
    }

    #[tokio::test]
    async fn clear_playlist_caches_keeps_user_data() -> color_eyre::Result<()> {
        use mineral_model::{CollectionIndex, PlaylistId, SongId, SourceKind};

        use mineral_channel_core::store::CachedPlaylistEntry;
        let dir = tempfile::tempdir()?;
        let p = ServerStore::open(&dir.path().join("t.db")).await?;
        let s = p.scope(SourceKind::NETEASE);
        // 写入歌单缓存和用户状态；清理只应删除前者。
        let pid = PlaylistId::new(SourceKind::NETEASE, "p1");
        let song = SongId::new(SourceKind::NETEASE, "s1");
        s.put_playlist_cache(
            &pid,
            Some("歌单"),
            Some(1),
            &[CachedPlaylistEntry {
                index: CollectionIndex::new(0),
                song_id: song.clone(),
            }],
        )
        .await?;
        s.set_loved(&song, true).await?;
        s.set_rating(&song, Some(4)).await?;

        // 清理前计数:1 个歌单、1 条曲目。
        let before = p.playlist_cache_stats().await?;
        assert_eq!(before.playlists, 1);
        assert_eq!(before.tracks, 1);

        // 清理回执 = 清理前计数。
        let removed = p.clear_playlist_caches().await?;
        assert_eq!(removed.playlists, 1);
        assert_eq!(removed.tracks, 1);

        // 歌单缓存没了
        assert!(s.get_playlist_cache(&pid).await?.is_none());
        // 清后计数归零
        let after = p.playlist_cache_stats().await?;
        assert_eq!(after.playlists, 0);
        assert_eq!(after.tracks, 0);
        // 收藏与评分仍在。
        assert!(s.is_loved(&song).await?);
        assert_eq!(s.query_rating(&song).await?, Some(4));
        Ok(())
    }
}