use mineral_log::trace;
use mineral_model::{AlbumId, ArtistId, CollectionIndex, PlaylistId, SongId, SourceKind};
use sea_orm::sea_query::OnConflict;
use sea_orm::{
ColumnTrait, EntityTrait, QueryFilter, QueryOrder, QuerySelect, Set, TransactionTrait,
};
use crate::persistence::ServerStore;
use crate::persistence::entity::{
playlist_cache, playlist_entries, song_artists, song_favorites, song_meta,
};
use crate::persistence::store::next_favorite_entry_time;
use mineral_channel_core::store::{CachedPlaylistEntry, PlaylistCacheEntry};
pub struct NamespaceStore {
persist: ServerStore,
source: SourceKind,
}
impl NamespaceStore {
pub(crate) fn new(persist: ServerStore, source: SourceKind) -> Self {
Self { persist, source }
}
pub fn source(&self) -> SourceKind {
self.source
}
pub(crate) fn check_source(&self, actual: SourceKind) -> crate::persistence::Result<()> {
if actual != self.source {
mineral_log::warn!(target: "persist", expected = self.source.name(), actual = actual.name(), "rejected foreign source identity");
return Err(crate::persistence::Error::NamespaceMismatch {
expected: self.source,
actual,
});
}
Ok(())
}
pub(crate) fn check_song(&self, song: &mineral_model::Song) -> crate::persistence::Result<()> {
self.check_source(song.source())?;
if let Some(album) = &song.album {
self.check_source(album.id.namespace())?;
}
for artist in &song.artists {
self.check_source(artist.id.namespace())?;
}
Ok(())
}
pub(crate) fn pool(&self) -> Option<&sea_orm::DatabaseConnection> {
self.persist.pool()
}
pub(crate) fn namespace(&self) -> &str {
self.source.name()
}
pub async fn album_name(&self, id: &AlbumId) -> crate::persistence::Result<Option<String>> {
self.check_source(id.namespace())?;
let Some(db) = self.pool() else {
return Ok(None);
};
song_meta::Entity::find()
.select_only()
.column(song_meta::Column::AlbumName)
.filter(song_meta::Column::Namespace.eq(self.namespace()))
.filter(song_meta::Column::AlbumId.eq(id.value()))
.filter(song_meta::Column::AlbumName.is_not_null())
.limit( 1)
.into_tuple::<String>()
.one(db)
.await
.map_err(|source| crate::persistence::Error::Record {
operation: "回查专辑名",
record: id.value().to_owned(),
source,
})
}
pub async fn artist_name(&self, id: &ArtistId) -> crate::persistence::Result<Option<String>> {
self.check_source(id.namespace())?;
let Some(db) = self.pool() else {
return Ok(None);
};
song_artists::Entity::find()
.select_only()
.column(song_artists::Column::ArtistName)
.filter(song_artists::Column::Namespace.eq(self.namespace()))
.filter(song_artists::Column::ArtistId.eq(id.value()))
.limit( 1)
.into_tuple::<String>()
.one(db)
.await
.map_err(|source| crate::persistence::Error::Record {
operation: "回查艺名",
record: id.value().to_owned(),
source,
})
}
pub async fn set_loved(&self, id: &SongId, loved: bool) -> crate::persistence::Result<bool> {
self.check_source(id.namespace())?;
let Some(db) = self.pool() else {
return Ok(false);
};
trace!(target: "persist", song = id.value(), loved, "set_loved");
let changed = if loved {
let tx = db
.begin()
.await
.map_err(|source| crate::persistence::Error::Database {
operation: "开启收藏写入事务",
source,
})?;
let entered_at = next_favorite_entry_time(&tx).await?;
let changed = song_favorites::Entity::insert(song_favorites::ActiveModel {
namespace: Set(self.namespace().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: "写收藏",
record: id.value().to_owned(),
source,
})?;
tx.commit()
.await
.map_err(|source| crate::persistence::Error::Database {
operation: "提交收藏写入事务",
source,
})?;
changed
} else {
song_favorites::Entity::delete_by_id((
self.namespace().to_owned(),
id.value().to_owned(),
))
.exec(db)
.await
.map_err(|source| crate::persistence::Error::Record {
operation: "取消收藏",
record: id.value().to_owned(),
source,
})?
.rows_affected
};
Ok(changed != 0)
}
pub async fn is_loved(&self, id: &SongId) -> crate::persistence::Result<bool> {
self.check_source(id.namespace())?;
let Some(db) = self.pool() else {
return Ok(false);
};
Ok(
song_favorites::Entity::find_by_id((
self.namespace().to_owned(),
id.value().to_owned(),
))
.one(db)
.await
.map_err(|source| crate::persistence::Error::Record {
operation: "查收藏",
record: id.value().to_owned(),
source,
})?
.is_some(),
)
}
pub async fn loved_ids(&self) -> crate::persistence::Result<rustc_hash::FxHashSet<SongId>> {
let Some(db) = self.pool() else {
return Ok(rustc_hash::FxHashSet::default());
};
let rows = song_favorites::Entity::find()
.filter(song_favorites::Column::Namespace.eq(self.namespace()))
.all(db)
.await
.map_err(|source| crate::persistence::Error::Database {
operation: "列出收藏身份",
source,
})?;
Ok(rows
.into_iter()
.map(|row| SongId::new(self.source, row.song_value))
.collect())
}
pub async fn put_playlist_cache(
&self,
id: &PlaylistId,
name: Option<&str>,
track_update_time: Option<i64>,
entries: &[CachedPlaylistEntry],
) -> crate::persistence::Result<()> {
self.check_source(id.namespace())?;
let Some(db) = self.pool() else {
return Ok(());
};
let ns = self.namespace();
let pid = id.value();
trace!(target: "persist", playlist = pid, tracks = entries.len(), "put_playlist_cache");
let tx = db
.begin()
.await
.map_err(|source| crate::persistence::Error::Database {
operation: "开启 put_playlist_cache 事务",
source,
})?;
playlist_cache::Entity::insert(playlist_cache::ActiveModel {
namespace: Set(ns.to_owned()),
playlist_id: Set(pid.to_owned()),
name: Set(name.map(str::to_owned)),
fetched_at: Set(crate::persistence::db::time::now_ms()),
track_update_time: Set(track_update_time),
})
.on_conflict(
OnConflict::columns([
playlist_cache::Column::Namespace,
playlist_cache::Column::PlaylistId,
])
.update_columns([
playlist_cache::Column::Name,
playlist_cache::Column::FetchedAt,
playlist_cache::Column::TrackUpdateTime,
])
.to_owned(),
)
.exec_without_returning(&tx)
.await
.map_err(|source| crate::persistence::Error::Record {
operation: "upsert playlist_cache",
record: pid.to_string(),
source,
})?;
playlist_entries::Entity::delete_many()
.filter(playlist_entries::Column::PlaylistNamespace.eq(ns))
.filter(playlist_entries::Column::PlaylistValue.eq(pid))
.exec(&tx)
.await
.map_err(|source| crate::persistence::Error::Record {
operation: "清空 playlist_entries",
record: pid.to_string(),
source,
})?;
for batch in entries.chunks(100) {
let models = batch
.iter()
.map(|entry| {
Ok(playlist_entries::ActiveModel {
playlist_namespace: Set(ns.to_owned()),
playlist_value: Set(pid.to_owned()),
collection_index: Set(i64::try_from(entry.index.get())?),
song_namespace: Set(entry.song_id.namespace().name().to_owned()),
song_value: Set(entry.song_id.value().to_owned()),
})
})
.collect::<crate::persistence::Result<Vec<_>>>()?;
playlist_entries::Entity::insert_many(models)
.exec_without_returning(&tx)
.await
.map_err(|source| crate::persistence::Error::Record {
operation: "批量写入 playlist_entries",
record: pid.to_string(),
source,
})?;
}
tx.commit()
.await
.map_err(|source| crate::persistence::Error::Record {
operation: "提交 put_playlist_cache 事务",
record: pid.to_string(),
source,
})?;
Ok(())
}
pub async fn get_playlist_cache(
&self,
id: &PlaylistId,
) -> crate::persistence::Result<Option<PlaylistCacheEntry>> {
self.check_source(id.namespace())?;
let Some(db) = self.pool() else {
return Ok(None);
};
let ns = self.namespace();
let pid = id.value();
let Some(head) = playlist_cache::Entity::find_by_id((ns.to_owned(), pid.to_owned()))
.one(db)
.await
.map_err(|source| crate::persistence::Error::Record {
operation: "查 playlist_cache",
record: pid.to_string(),
source,
})?
else {
return Ok(None);
};
let rows = playlist_entries::Entity::find()
.filter(playlist_entries::Column::PlaylistNamespace.eq(ns))
.filter(playlist_entries::Column::PlaylistValue.eq(pid))
.order_by_asc(playlist_entries::Column::CollectionIndex)
.all(db)
.await
.map_err(|source| crate::persistence::Error::Record {
operation: "查 playlist_entries",
record: pid.to_string(),
source,
})?;
let entries = rows
.into_iter()
.map(|row| {
Ok(CachedPlaylistEntry {
index: CollectionIndex::new(u64::try_from(row.collection_index)?),
song_id: SongId::new(
SourceKind::from_name(&row.song_namespace),
row.song_value,
),
})
})
.collect::<crate::persistence::Result<Vec<_>>>()?;
Ok(Some(PlaylistCacheEntry {
name: head.name,
fetched_at: head.fetched_at,
track_update_time: head.track_update_time,
entries,
}))
}
}
#[cfg(test)]
mod tests {
use mineral_model::{CollectionIndex, Song, SongId, SourceKind};
use super::CachedPlaylistEntry;
fn cached(index: u64, source: SourceKind, value: &str) -> CachedPlaylistEntry {
CachedPlaylistEntry {
index: CollectionIndex::new(index),
song_id: SongId::new(source, value),
}
}
#[tokio::test]
async fn album_and_artist_name_lookup() -> color_eyre::Result<()> {
use mineral_model::{AlbumId, AlbumRef, ArtistId, ArtistRef};
let dir = tempfile::tempdir()?;
let p = crate::persistence::ServerStore::open(&dir.path().join("t.db")).await?;
let s = p.scope(SourceKind::NETEASE);
let song = Song::builder()
.id(SongId::new(SourceKind::NETEASE, "1"))
.name("稻香".to_owned())
.artists(vec![ArtistRef {
id: ArtistId::new(SourceKind::NETEASE, "jay"),
name: "周杰伦".to_owned(),
}])
.album(Some(AlbumRef {
id: AlbumId::new(SourceKind::NETEASE, "mojito"),
name: "魔杰座".to_owned(),
}))
.build();
s.upsert_meta(&song).await?;
assert_eq!(
s.album_name(&AlbumId::new(SourceKind::NETEASE, "mojito"))
.await?,
Some("魔杰座".to_owned())
);
assert_eq!(
s.artist_name(&ArtistId::new(SourceKind::NETEASE, "jay"))
.await?,
Some("周杰伦".to_owned())
);
assert_eq!(
s.album_name(&AlbumId::new(SourceKind::NETEASE, "absent"))
.await?,
None,
"未命中回落 None"
);
Ok(())
}
#[tokio::test]
async fn love_toggle_and_list() -> color_eyre::Result<()> {
let dir = tempfile::tempdir()?;
let p = crate::persistence::ServerStore::open(&dir.path().join("t.db")).await?;
let s = p.scope(SourceKind::NETEASE);
let id = SongId::new(SourceKind::NETEASE, "123");
assert!(!s.is_loved(&id).await?);
assert!(s.set_loved(&id, true).await?, "false -> true 应创建");
assert!(!s.set_loved(&id, true).await?, "true -> true 应 no-op");
assert!(s.is_loved(&id).await?);
assert!(s.loved_ids().await?.contains(&id));
assert!(s.set_loved(&id, false).await?, "true -> false 应删除");
assert!(!s.set_loved(&id, false).await?, "false -> false 应 no-op");
assert!(!s.is_loved(&id).await?);
assert!(!s.loved_ids().await?.contains(&id));
Ok(())
}
#[tokio::test]
async fn loved_ids_isolated_by_source() -> color_eyre::Result<()> {
let dir = tempfile::tempdir()?;
let p = crate::persistence::ServerStore::open(&dir.path().join("t.db")).await?;
let netease = p.scope(SourceKind::NETEASE);
let local = p.scope(SourceKind::LOCAL);
netease
.set_loved(&SongId::new(SourceKind::NETEASE, "n1"), true)
.await?;
assert_eq!(netease.loved_ids().await?.len(), 1);
assert_eq!(local.loved_ids().await?.len(), 0);
Ok(())
}
#[tokio::test]
async fn playlist_cache_roundtrip() -> color_eyre::Result<()> {
let dir = tempfile::tempdir()?;
let p = crate::persistence::ServerStore::open(&dir.path().join("t.db")).await?;
let s = p.scope(SourceKind::NETEASE);
let pid = mineral_model::PlaylistId::new(SourceKind::NETEASE, "p1");
let entries = vec![
cached(0, SourceKind::NETEASE, "s1"),
cached(3, SourceKind::BILIBILI, "s2"),
cached(9, SourceKind::NETEASE, "s3"),
];
s.put_playlist_cache(&pid, Some("我的歌单"), Some(1_775_781_450_653), &entries)
.await?;
let got = s.get_playlist_cache(&pid).await?;
assert!(got.is_some());
if let Some(g) = got {
assert_eq!(g.name, Some("我的歌单".to_owned()));
assert_eq!(
g.entries, entries,
"index gap 与 mixed-source SongId 原值保留"
);
assert!(g.fetched_at > 0);
assert_eq!(g.track_update_time, Some(1_775_781_450_653)); }
Ok(())
}
#[tokio::test]
async fn playlist_cache_overwrite_clears_old_tracks() -> color_eyre::Result<()> {
let dir = tempfile::tempdir()?;
let p = crate::persistence::ServerStore::open(&dir.path().join("t.db")).await?;
let s = p.scope(SourceKind::NETEASE);
let pid = mineral_model::PlaylistId::new(SourceKind::NETEASE, "p1");
s.put_playlist_cache(
&pid,
Some("v1"),
Some(100),
&[
cached(0, SourceKind::NETEASE, "a"),
cached(1, SourceKind::NETEASE, "b"),
],
)
.await?;
s.put_playlist_cache(
&pid,
Some("v2"),
Some(200),
&[cached(5, SourceKind::BILIBILI, "c")],
)
.await?; let got = s.get_playlist_cache(&pid).await?;
assert!(got.is_some());
if let Some(g) = got {
assert_eq!(g.name, Some("v2".to_owned()));
assert_eq!(
g.entries,
vec![cached(5, SourceKind::BILIBILI, "c")],
"旧 a,b 不残留,explicit index/source 保留"
);
assert_eq!(g.track_update_time, Some(200)); }
Ok(())
}
#[tokio::test]
async fn playlist_cache_miss_returns_none() -> color_eyre::Result<()> {
let dir = tempfile::tempdir()?;
let p = crate::persistence::ServerStore::open(&dir.path().join("t.db")).await?;
let s = p.scope(SourceKind::NETEASE);
let pid = mineral_model::PlaylistId::new(SourceKind::NETEASE, "absent");
assert!(s.get_playlist_cache(&pid).await?.is_none());
Ok(())
}
}