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

fed fixes

Browse Source
This commit is contained in:
Ultradesu
2026-07-21 00:12:45 +03:00
parent 151e2cf337
commit da5d2410ac
24 changed files with 2660 additions and 461 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/federation/audio.rs
+101 -19
View File
@@ -2,7 +2,8 @@
//!
//! One byte stream per request: the requester sends one JSON line
//! ([`AudioRequest`]) and receives one JSON line ([`AudioResponseHeader`])
//! followed by the raw file bytes from the requested offset.
//! followed by the raw file bytes from the requested offset, unless the
//! requester asked for metadata only.
use std::path::{Path, PathBuf};
use std::sync::Arc;
@@ -32,6 +33,11 @@ struct AudioRequest {
/// older peers in both directions.
#[serde(default)]
want_cover: bool,
/// Ask only for the response header with metadata and file facts.
/// Older peers ignore the field and may start streaming audio; the
/// requester simply drops the stream after reading the header.
#[serde(default)]
metadata_only: bool,
}
#[derive(Debug, Serialize, Deserialize)]
@@ -115,6 +121,16 @@ pub struct TrackMetadata {
pub track_number: Option<i32>,
#[serde(default)]
pub disc_number: Option<i32>,
#[serde(default)]
pub duration_seconds: Option<f64>,
#[serde(default)]
pub audio_format: Option<String>,
#[serde(default)]
pub audio_bitrate: Option<i32>,
#[serde(default)]
pub audio_sample_rate: Option<i32>,
#[serde(default)]
pub audio_bit_depth: Option<i32>,
}
pub fn hex_encode(bytes: &[u8]) -> String {
@@ -167,6 +183,13 @@ fn extension_for_mime(mime: &str) -> &'static str {
}
}
/// Best-effort audio format label for metadata previews, from the owner's
/// response mime type.
pub fn format_for_mime(mime: &str) -> Option<String> {
let extension = extension_for_mime(mime);
(extension != "bin").then(|| extension.to_string())
}
/// Reads one `\n`-terminated line, bounded by [`MAX_PROTOCOL_LINE`].
pub(super) async fn read_line<R: AsyncRead + Unpin>(reader: &mut R) -> Result<Vec<u8>> {
let mut line = Vec::new();
@@ -186,7 +209,10 @@ pub(super) async fn read_line<R: AsyncRead + Unpin>(reader: &mut R) -> Result<Ve
}
}
async fn write_line<W: AsyncWriteExt + Unpin>(writer: &mut W, value: &impl Serialize) -> Result<()> {
async fn write_line<W: AsyncWriteExt + Unpin>(
writer: &mut W,
value: &impl Serialize,
) -> Result<()> {
let mut line = serde_json::to_vec(value)?;
line.push(b'\n');
writer.write_all(&line).await?;
@@ -208,6 +234,50 @@ pub struct Downloaded {
pub artist_image: Option<(Vec<u8>, &'static str)>,
}
/// Header-only metadata fetched without downloading the audio bytes.
pub struct FetchedMetadata {
pub mime_type: String,
pub total_size: u64,
pub metadata: Option<TrackMetadata>,
}
/// Fetches the owner's response header for a track and closes the stream
/// before audio bytes are read.
pub async fn fetch_metadata(
service: &MusicDhtService,
owner: EndpointId,
item_id_hex: &str,
) -> Result<FetchedMetadata> {
let mut stream = service
.open_stream(owner, AUDIO_ALPN)
.await
.map_err(|err| anyhow::anyhow!("cannot reach the owner peer: {err}"))?;
write_line(
&mut stream.send,
&AudioRequest {
item_id: item_id_hex.to_string(),
offset: 0,
want_cover: false,
metadata_only: true,
},
)
.await?;
stream.send.finish()?;
let header: AudioResponseHeader = serde_json::from_slice(&read_line(&mut stream.recv).await?)
.context("malformed response header")?;
if !header.ok {
anyhow::bail!(
"peer refused the metadata: {}",
header.error.unwrap_or_else(|| "unknown error".to_string())
);
}
Ok(FetchedMetadata {
mime_type: header.mime_type,
total_size: header.total_size,
metadata: header.metadata,
})
}
/// Downloads a whole track (with metadata and cover art) from `owner` into
/// `dir/<stem>.<ext>`. An already complete cached audio file is reused;
/// the metadata and cover still come fresh from the header.
@@ -228,6 +298,7 @@ pub async fn download_track(
item_id: item_id_hex.to_string(),
offset: 0,
want_cover: true,
metadata_only: false,
},
)
.await?;
@@ -243,22 +314,23 @@ pub async fn download_track(
// The image segments precede the audio bytes and are read regardless of
// the cache state — they sit first in the stream.
let mut read_image = async |size: u64, mime: &str, what: &str| -> Result<Option<(Vec<u8>, &'static str)>> {
if size == 0 {
return Ok(None);
}
anyhow::ensure!(
size <= MAX_COVER_BYTES,
"{what} of {size} bytes exceeds the {MAX_COVER_BYTES} byte limit"
);
let mut bytes = vec![0u8; size as usize];
stream
.recv
.read_exact(&mut bytes)
.await
.with_context(|| format!("stream ended inside the {what} segment"))?;
Ok(Some((bytes, image_extension(mime))))
};
let mut read_image =
async |size: u64, mime: &str, what: &str| -> Result<Option<(Vec<u8>, &'static str)>> {
if size == 0 {
return Ok(None);
}
anyhow::ensure!(
size <= MAX_COVER_BYTES,
"{what} of {size} bytes exceeds the {MAX_COVER_BYTES} byte limit"
);
let mut bytes = vec![0u8; size as usize];
stream
.recv
.read_exact(&mut bytes)
.await
.with_context(|| format!("stream ended inside the {what} segment"))?;
Ok(Some((bytes, image_extension(mime))))
};
let cover = read_image(header.cover_size, &header.cover_mime, "cover").await?;
let artist_image = read_image(
header.artist_image_size,
@@ -373,6 +445,11 @@ fn resolve_for_serving(
year: track.release_year,
track_number: track.track_number,
disc_number: track.disc_number,
duration_seconds: Some(track.duration_seconds),
audio_format: track.audio_format.clone(),
audio_bitrate: track.audio_bitrate,
audio_sample_rate: track.audio_sample_rate,
audio_bit_depth: track.audio_bit_depth,
};
let artist_image_path = track
.artists
@@ -441,7 +518,7 @@ async fn serve_one(mut stream: ByteStream, library: Arc<Library>, own: EndpointI
}
// Images ride between the header and the audio, when asked for.
let (cover, artist_image) = if request.want_cover {
let (cover, artist_image) = if request.want_cover && !request.metadata_only {
(
load_cover(served.cover_path.as_deref()).await,
load_cover(served.artist_image_path.as_deref()).await,
@@ -473,6 +550,11 @@ async fn serve_one(mut stream: ByteStream, library: Arc<Library>, own: EndpointI
},
)
.await?;
if request.metadata_only {
stream.send.finish()?;
let _ = stream.send.stopped().await;
return Ok(());
}
if let Some((bytes, _)) = &cover {
stream.send.write_all(bytes).await?;
}
src/federation/catalog.rs
+244 -37
View File
@@ -1,6 +1,6 @@
//! The peer catalog protocol: one peer asks another for its library slice
//! of a single artist (releases with full tracklists), used to assemble a
//! federated artist card.
//! of a single artist (releases with full tracklists plus featured
//! appearances), used to assemble a federated artist card.
//!
//! Wire shape on the `furumi-fd/catalog/1` ALPN: the requester sends one
//! JSON line ([`CatalogRequest`]) and finishes; the owner answers with one
@@ -67,6 +67,10 @@ pub struct CatalogArtist {
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)]
@@ -81,16 +85,35 @@ pub struct CatalogRelease {
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,
@@ -159,7 +182,10 @@ async fn serve_one(mut stream: ByteStream, library: Arc<Library>, own: EndpointI
error: Some(format!("unknown request kind '{other}'")),
artist: None,
};
stream.send.write_all(&serde_json::to_vec(&response)?).await?;
stream
.send
.write_all(&serde_json::to_vec(&response)?)
.await?;
}
}
stream.send.finish()?;
@@ -218,13 +244,29 @@ fn image_mime_by_path(path: &str) -> &'static str {
/// Builds this instance's library slice for `artist`.
fn build_catalog(library: &Library, own: EndpointId, artist: &str) -> Result<CatalogResponse> {
let Some(artist_id) = library.artist_id_by_name(artist)? else {
let Some(artist) = build_catalog_artist(library, own, artist)? else {
return Ok(CatalogResponse {
ok: false,
error: Some("artist not found in the library".to_string()),
artist: None,
});
};
Ok(CatalogResponse {
ok: true,
error: None,
artist: Some(artist),
})
}
/// Builds the successful payload for this instance's library slice.
pub(crate) fn build_catalog_artist(
library: &Library,
own: EndpointId,
artist: &str,
) -> Result<Option<CatalogArtist>> {
let Some(artist_id) = library.artist_id_by_name(artist)? else {
return Ok(None);
};
let detail = library.artist(artist_id)?;
let item_id_of = |track_id: i64| -> String {
super::audio::hex_encode(
@@ -242,25 +284,46 @@ fn build_catalog(library: &Library, own: EndpointId, artist: &str) -> Result<Cat
tracks: release
.tracks
.iter()
.map(|track| CatalogTrack {
title: track.title.clone(),
track_number: track.track_number,
disc_number: track.disc_number,
duration_seconds: (track.duration_seconds > 0.0)
.then_some(track.duration_seconds),
item_id: item_id_of(track.id),
})
.map(|track| catalog_track(track, item_id_of(track.id)))
.collect(),
});
}
Ok(CatalogResponse {
ok: true,
error: None,
artist: Some(CatalogArtist {
name: detail.name,
releases,
}),
})
let mut appears_on = Vec::new();
for track in &detail.featured_tracks {
let release = library.release(track.release_id)?;
appears_on.push(CatalogAppearance {
release_title: track.release_title.clone(),
release_type: release.release_type,
year: track.release_year,
track: catalog_track(track, item_id_of(track.id)),
});
}
Ok(Some(CatalogArtist {
name: detail.name,
releases,
appears_on,
}))
}
fn catalog_track(track: &crate::library::models::TrackItem, item_id: String) -> CatalogTrack {
CatalogTrack {
title: track.title.clone(),
artists: track
.artists
.iter()
.map(|artist| artist.name.clone())
.collect(),
featured_artists: track
.featured_artists
.iter()
.map(|artist| artist.name.clone())
.collect(),
track_number: track.track_number,
disc_number: track.disc_number,
duration_seconds: (track.duration_seconds > 0.0).then_some(track.duration_seconds),
content_id: track.content_id.clone(),
item_id,
}
}
// ---------------------------------------------------------------------------
@@ -383,6 +446,9 @@ impl tokio::io::AsyncRead for StreamReader<'_> {
pub struct FedArtistCard {
#[allow(dead_code, reason = "the open card is keyed by name in AppState")]
pub name: String,
/// This node's endpoint id, used to keep own tracks local when an open
/// card includes the local catalog alongside remote peers.
pub own_owner: Option<String>,
/// Peers whose catalogs contributed to the card.
pub peers: usize,
/// Every contributing peer (hex ids) — where images are fetched from.
@@ -390,6 +456,7 @@ pub struct FedArtistCard {
/// Local cache path of the artist image, streamed from a peer.
pub image_path: Option<String>,
pub releases: Vec<FedRelease>,
pub appears_on: Vec<FedAppearsOn>,
}
#[derive(Debug, Clone, Default)]
@@ -404,12 +471,23 @@ pub struct FedRelease {
pub tracks: Vec<FedCardTrack>,
}
#[derive(Debug, Clone, Default)]
pub struct FedAppearsOn {
pub release_title: String,
pub release_type: String,
pub year: Option<i32>,
pub track: FedCardTrack,
}
#[derive(Debug, Clone, Default)]
pub struct FedCardTrack {
pub title: String,
pub artists: Vec<String>,
pub featured_artists: Vec<String>,
pub track_number: Option<i32>,
pub disc_number: Option<i32>,
pub duration_seconds: Option<f64>,
pub content_id: Option<String>,
/// Every peer that can serve this track: (owner hex, item id hex).
/// Duplicates collapse into one row; all sources stay playable.
pub sources: Vec<(String, String)>,
@@ -422,6 +500,8 @@ pub fn merge_catalogs(name: &str, catalogs: Vec<(String, CatalogArtist)>) -> Fed
let peers = catalogs.len();
let mut releases: Vec<FedRelease> = Vec::new();
let mut release_index: HashMap<String, usize> = HashMap::new();
let mut appears_on: Vec<FedAppearsOn> = Vec::new();
let mut appearance_index: HashMap<String, usize> = HashMap::new();
let mut card_owners: Vec<String> = Vec::new();
for (owner_hex, catalog) in catalogs {
@@ -463,31 +543,67 @@ pub fn merge_catalogs(name: &str, catalogs: Vec<(String, CatalogArtist)>) -> Fed
});
match existing {
Some(t) => {
if t.track_number.is_none() {
t.track_number = track.track_number;
}
if t.duration_seconds.is_none() {
t.duration_seconds = track.duration_seconds;
}
t.sources.push((owner_hex.clone(), track.item_id));
merge_card_track(t, &owner_hex, track);
}
None => merged.tracks.push(FedCardTrack {
title: track.title,
track_number: track.track_number,
disc_number: track.disc_number,
duration_seconds: track.duration_seconds,
sources: vec![(owner_hex.clone(), track.item_id)],
}),
None => merged.tracks.push(card_track(owner_hex.clone(), track)),
}
}
}
for appearance in catalog.appears_on {
if appearance.track.item_id.is_empty() {
continue;
}
let key = format!(
"{}:{}:{:?}",
music_dht::normalize_name(&appearance.release_title),
music_dht::normalize_name(&appearance.track.title),
appearance.track.track_number
);
let slot = *appearance_index.entry(key).or_insert_with(|| {
appears_on.push(FedAppearsOn {
release_title: appearance.release_title.clone(),
release_type: appearance.release_type.clone(),
year: appearance.year,
track: FedCardTrack::default(),
});
appears_on.len() - 1
});
let merged = &mut appears_on[slot];
if merged.release_type.is_empty() {
merged.release_type = appearance.release_type.clone();
}
if merged.year.is_none() {
merged.year = appearance.year;
}
if merged.track.title.is_empty() {
merged.track = card_track(owner_hex.clone(), appearance.track);
} else {
merge_card_track(&mut merged.track, &owner_hex, appearance.track);
}
}
}
for release in &mut releases {
release
.tracks
.sort_by_key(|t| (t.disc_number.unwrap_or(1), t.track_number.unwrap_or(i32::MAX)));
release.tracks.sort_by_key(|t| {
(
t.disc_number.unwrap_or(1),
t.track_number.unwrap_or(i32::MAX),
)
});
}
appears_on.sort_by(|a, b| {
b.year
.unwrap_or(i32::MIN)
.cmp(&a.year.unwrap_or(i32::MIN))
.then_with(|| a.release_title.cmp(&b.release_title))
.then_with(|| {
a.track
.track_number
.unwrap_or(i32::MAX)
.cmp(&b.track.track_number.unwrap_or(i32::MAX))
})
.then_with(|| a.track.title.cmp(&b.track.title))
});
releases.sort_by(|a, b| {
a.year
.unwrap_or(i32::MAX)
@@ -497,10 +613,50 @@ pub fn merge_catalogs(name: &str, catalogs: Vec<(String, CatalogArtist)>) -> Fed
FedArtistCard {
name: name.to_string(),
own_owner: None,
peers,
owners: card_owners,
image_path: None,
releases,
appears_on,
}
}
fn card_track(owner_hex: String, track: CatalogTrack) -> FedCardTrack {
FedCardTrack {
title: track.title,
artists: track.artists,
featured_artists: track.featured_artists,
track_number: track.track_number,
disc_number: track.disc_number,
duration_seconds: track.duration_seconds,
content_id: track.content_id,
sources: vec![(owner_hex, track.item_id)],
}
}
fn merge_card_track(target: &mut FedCardTrack, owner_hex: &str, track: CatalogTrack) {
if target.artists.is_empty() {
target.artists = track.artists;
}
if target.featured_artists.is_empty() {
target.featured_artists = track.featured_artists;
}
if target.track_number.is_none() {
target.track_number = track.track_number;
}
if target.disc_number.is_none() {
target.disc_number = track.disc_number;
}
if target.duration_seconds.is_none() {
target.duration_seconds = track.duration_seconds;
}
if target.content_id.is_none() {
target.content_id = track.content_id;
}
let source = (owner_hex.to_string(), track.item_id);
if !target.sources.contains(&source) {
target.sources.push(source);
}
}
@@ -511,9 +667,12 @@ mod tests {
fn track(title: &str, number: i32, item: &str) -> CatalogTrack {
CatalogTrack {
title: title.into(),
artists: vec!["Metallica".into()],
featured_artists: Vec::new(),
track_number: Some(number),
disc_number: None,
duration_seconds: Some(100.0),
content_id: None,
item_id: item.into(),
}
}
@@ -531,6 +690,7 @@ mod tests {
track("Sad But True", 2, &format!("{item_prefix}2")),
],
}],
appears_on: Vec::new(),
};
let card = merge_catalogs(
"Metallica",
@@ -547,4 +707,51 @@ mod tests {
// Both peers stay as sources of the deduplicated track.
assert_eq!(release.tracks[0].sources.len(), 2);
}
#[test]
fn merges_featured_appearances_across_peers() {
let mut featured = track("Guest Verse", 3, "a1");
featured.artists = vec!["Host".into()];
featured.featured_artists = vec!["Guest".into()];
let mut same_featured = featured.clone();
same_featured.item_id = "b1".into();
let card = merge_catalogs(
"Guest",
vec![
(
"peer-a".to_string(),
CatalogArtist {
name: "Guest".into(),
releases: Vec::new(),
appears_on: vec![CatalogAppearance {
release_title: "Host Album".into(),
release_type: "album".into(),
year: Some(2024),
track: featured,
}],
},
),
(
"peer-b".to_string(),
CatalogArtist {
name: "Guest".into(),
releases: Vec::new(),
appears_on: vec![CatalogAppearance {
release_title: "Host Album".into(),
release_type: "album".into(),
year: Some(2024),
track: same_featured,
}],
},
),
],
);
assert!(card.releases.is_empty());
assert_eq!(card.appears_on.len(), 1);
assert_eq!(card.appears_on[0].track.sources.len(), 2);
assert_eq!(card.appears_on[0].track.artists, vec!["Host"]);
assert_eq!(card.appears_on[0].track.featured_artists, vec!["Guest"]);
}
}
src/federation/mod.rs
+993 -50
View File
File diff suppressed because it is too large Load Diff