Furumi/furumi_tui
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Activity

Added network catalog slices

Browse Source
This commit is contained in:
Ultradesu
2026-07-25 22:40:27 +03:00
parent b725ca8d05
commit 788bc2e01f
14 changed files with 878 additions and 152 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/federation/catalog.rs
+113 -101
View File
@@ -11,114 +11,22 @@ use std::collections::HashMap;
use std::sync::Arc;
use anyhow::{Context, Result};
pub use music_dht::catalog::{
CATALOG_ALPN, CatalogAppearance, CatalogArtist, CatalogArtistPreview, CatalogRelease,
CatalogTrack,
};
use music_dht::catalog::{CatalogImageHeader as ImageHeader, CatalogRequest, CatalogResponse};
use music_dht::{ByteStream, EndpointId, ItemKind, MusicDhtService, StreamAcceptor};
use serde::{Deserialize, Serialize};
use tokio::io::AsyncReadExt;
use crate::library::Library;
/// ALPN of the catalog protocol.
pub const CATALOG_ALPN: &[u8] = b"furumi-fd/catalog/1";
/// Upper bound for one catalog response (thousands of tracks fit easily).
const MAX_CATALOG_BYTES: u64 = 4 * 1024 * 1024;
#[derive(Debug, Serialize, Deserialize)]
struct CatalogRequest {
/// Artist display name; matched case-insensitively by the owner.
artist: String,
/// What is being asked for: `None`/"catalog" — the JSON catalog;
/// "artist_image" — the artist's image; "release_cover" — the cover of
/// `release`. Image responses are a JSON header line + raw bytes.
#[serde(default)]
want: Option<String>,
#[serde(default)]
release: Option<String>,
}
/// Header line preceding raw image bytes (artist image / release cover).
#[derive(Debug, Default, Serialize, Deserialize)]
struct ImageHeader {
ok: bool,
#[serde(default)]
error: Option<String>,
#[serde(default)]
mime_type: String,
#[serde(default)]
size: u64,
}
/// Images above this size are skipped rather than transferred.
const MAX_IMAGE_BYTES: u64 = 16 * 1024 * 1024;
#[derive(Debug, Default, Serialize, Deserialize)]
struct CatalogResponse {
ok: bool,
#[serde(default)]
error: Option<String>,
#[serde(default)]
artist: Option<CatalogArtist>,
}
/// One peer's library slice for an artist.
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct CatalogArtist {
#[serde(default)]
pub name: String,
#[serde(default)]
pub releases: Vec<CatalogRelease>,
/// Tracks where the requested artist is featured instead of being a
/// release/main artist.
#[serde(default)]
pub appears_on: Vec<CatalogAppearance>,
}
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct CatalogRelease {
#[serde(default)]
pub title: String,
#[serde(default)]
pub release_type: String,
#[serde(default)]
pub year: Option<i32>,
#[serde(default)]
pub tracks: Vec<CatalogTrack>,
}
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct CatalogAppearance {
#[serde(default)]
pub release_title: String,
#[serde(default)]
pub release_type: String,
#[serde(default)]
pub year: Option<i32>,
#[serde(default)]
pub track: CatalogTrack,
}
#[derive(Debug, Clone, Default, Serialize, Deserialize)]
pub struct CatalogTrack {
#[serde(default)]
pub title: String,
#[serde(default)]
pub artists: Vec<String>,
#[serde(default)]
pub featured_artists: Vec<String>,
#[serde(default)]
pub track_number: Option<i32>,
#[serde(default)]
pub disc_number: Option<i32>,
#[serde(default)]
pub duration_seconds: Option<f64>,
/// Stable audio content id (`b3:<64 hex>`) when known.
#[serde(default)]
pub content_id: Option<String>,
/// Hex DHT item id — the key the audio is requested by (FedPlay).
#[serde(default)]
pub item_id: String,
}
// ---------------------------------------------------------------------------
// Serving side
// ---------------------------------------------------------------------------
@@ -165,14 +73,28 @@ async fn serve_one(
);
match request.want.as_deref() {
None | Some("catalog") => {
Some("artists") => {
let cursor = request.cursor.clone();
let limit = request.limit.unwrap_or(64).clamp(1, 200);
let response =
tokio::task::spawn_blocking(move || build_artist_slice(&library, cursor, limit))
.await?
.unwrap_or_else(|err| CatalogResponse {
ok: false,
error: Some(format!("artist slice failed: {err:#}")),
..CatalogResponse::default()
});
let payload = serde_json::to_vec(&response)?;
stream.send.write_all(&payload).await?;
}
None | Some("catalog") | Some("artist") => {
let response =
tokio::task::spawn_blocking(move || build_catalog(&library, own, &request.artist))
.await?
.unwrap_or_else(|err| CatalogResponse {
ok: false,
error: Some(format!("catalog lookup failed: {err:#}")),
artist: None,
..CatalogResponse::default()
});
let payload = serde_json::to_vec(&response)?;
stream.send.write_all(&payload).await?;
@@ -198,7 +120,7 @@ async fn serve_one(
let response = CatalogResponse {
ok: false,
error: Some(format!("unknown request kind '{other}'")),
artist: None,
..CatalogResponse::default()
};
stream
.send
@@ -273,13 +195,41 @@ fn build_catalog(library: &Library, own: EndpointId, artist: &str) -> Result<Cat
return Ok(CatalogResponse {
ok: false,
error: Some("artist not found in the library".to_string()),
artist: None,
..CatalogResponse::default()
});
};
Ok(CatalogResponse {
ok: true,
error: None,
artist: Some(artist),
..CatalogResponse::default()
})
}
fn build_artist_slice(
library: &Library,
cursor: Option<String>,
limit: usize,
) -> Result<CatalogResponse> {
let offset = cursor
.as_deref()
.and_then(|value| value.parse::<usize>().ok())
.unwrap_or(0);
let (artists, next_cursor) = library.artist_preview_slice(offset, limit)?;
Ok(CatalogResponse {
ok: true,
artists: artists
.into_iter()
.map(|artist| CatalogArtistPreview {
artist_key: artist.artist_key,
name: artist.name,
image_path: artist.image_path,
release_count: artist.release_count,
track_count: artist.track_count,
})
.collect(),
next_cursor,
..CatalogResponse::default()
})
}
@@ -377,6 +327,8 @@ pub async fn fetch_catalog(
artist: artist.to_string(),
want: None,
release: None,
cursor: None,
limit: None,
})?;
line.push(b'\n');
stream.send.write_all(&line).await?;
@@ -411,6 +363,64 @@ pub async fn fetch_catalog(
response.artist.context("empty catalog response")
}
/// Fetches a thin top-artist slice from one peer.
pub async fn fetch_artist_slice(
service: &MusicDhtService,
owner: EndpointId,
cursor: Option<&str>,
limit: usize,
transport_stats: &Arc<crate::federation::TransportStats>,
) -> Result<(Vec<CatalogArtistPreview>, Option<String>)> {
let mut stream = service
.open_stream(owner, CATALOG_ALPN)
.await
.map_err(|err| anyhow::anyhow!("cannot reach the peer: {err}"))?;
crate::federation::record_stream_transport(
transport_stats,
"catalog",
"outbound",
"open",
&stream,
);
let mut line = serde_json::to_vec(&CatalogRequest {
artist: String::new(),
want: Some("artists".to_string()),
release: None,
cursor: cursor.map(str::to_string),
limit: Some(limit),
})?;
line.push(b'\n');
stream.send.write_all(&line).await?;
stream.send.finish()?;
let mut payload = Vec::new();
tokio::io::AsyncReadExt::take(StreamReader(&mut stream), MAX_CATALOG_BYTES + 1)
.read_to_end(&mut payload)
.await?;
anyhow::ensure!(
payload.len() as u64 <= MAX_CATALOG_BYTES,
"catalog response exceeds {MAX_CATALOG_BYTES} bytes"
);
let response: CatalogResponse =
serde_json::from_slice(&payload).context("malformed catalog response")?;
crate::federation::record_stream_transport(
transport_stats,
"catalog",
"outbound",
"done",
&stream,
);
if !response.ok {
anyhow::bail!(
"peer refused the artist slice: {}",
response
.error
.unwrap_or_else(|| "unknown error".to_string())
);
}
Ok((response.artists, response.next_cursor))
}
/// Fetches an image (artist image or a release cover) from a peer over the
/// catalog protocol. `release: None` asks for the artist image. Returns the
/// raw bytes and a file extension, or None when the peer has no image.
@@ -440,6 +450,8 @@ pub async fn fetch_image(
"artist_image".to_string()
}),
release: release.map(str::to_string),
cursor: None,
limit: None,
})?;
line.push(b'\n');
stream.send.write_all(&line).await?;
src/federation/mod.rs
+94
View File
@@ -32,6 +32,7 @@ use rusqlite::{Connection, OpenFlags, params};
use serde::{Deserialize, Serialize};
use crate::library::Library;
use crate::library::NetworkArtistPreview;
use crate::library::models::{ArtistRef, TrackItem};
pub use audio::{AUDIO_ALPN, DownloadProgress, StreamingStart, TrackMetadata};
@@ -395,6 +396,12 @@ pub struct FedPlayable {
pub imported: bool,
}
#[derive(Debug, Clone, PartialEq, Eq, Hash)]
pub struct NetworkLibrarySource {
pub endpoint_id: String,
pub kind: &'static str,
}
// ---------------------------------------------------------------------------
// The federation manager (lives in Runtime, not AppState)
// ---------------------------------------------------------------------------
@@ -862,6 +869,93 @@ impl Federation {
status
}
pub async fn network_library_sources(
&self,
mode: crate::config::settings::LibrarySourceMode,
) -> Result<Vec<NetworkLibrarySource>> {
if !mode.includes_network() {
return Ok(Vec::new());
}
let service = self.service().await?;
let own = service.endpoint_id().to_string();
let mut seen = std::collections::HashSet::new();
seen.insert(own);
let mut sources = Vec::new();
for endpoint_id in self.devices.active_remote_endpoint_ids()? {
if seen.insert(endpoint_id.clone()) {
sources.push(NetworkLibrarySource {
endpoint_id,
kind: "personal",
});
}
}
if mode.includes_global_peers() {
for endpoint in service
.connected_peers()
.into_iter()
.map(|peer| peer.to_string())
.chain(
service
.known_peers()
.into_iter()
.map(|peer| peer.peer_id.to_string()),
)
{
if seen.insert(endpoint.clone()) {
sources.push(NetworkLibrarySource {
endpoint_id: endpoint,
kind: "federation",
});
}
if sources.len() >= 32 {
break;
}
}
}
Ok(sources)
}
pub async fn cache_artist_slice_from_source(
&self,
source: NetworkLibrarySource,
cursor: Option<String>,
limit: usize,
) -> Result<(usize, Option<String>)> {
let service = self.service().await?;
let owner = EndpointId::from_str(&source.endpoint_id)
.map_err(|_| anyhow::anyhow!("malformed endpoint id '{}'", source.endpoint_id))?;
let replace_source = cursor.is_none();
let (artists, next_cursor) = catalog::fetch_artist_slice(
&service,
owner,
cursor.as_deref(),
limit,
&self.transport_stats,
)
.await?;
let artists: Vec<NetworkArtistPreview> = artists
.into_iter()
.map(|artist| NetworkArtistPreview {
artist_key: if artist.artist_key.trim().is_empty() {
music_dht::normalize_name(&artist.name)
} else {
artist.artist_key
},
name: artist.name,
image_path: None,
release_count: artist.release_count,
track_count: artist.track_count,
})
.collect();
let count = self.library.replace_network_artist_cache(
&source.endpoint_id,
source.kind,
&artists,
replace_source,
)?;
Ok((count, next_cursor))
}
/// Searches the federated network: matching tracks plus the artists a
/// card can be assembled for (from artist records and from the artist
/// names of matching tracks/releases).