init federation
This commit is contained in:
Ultradesu
@@ -0,0 +1,311 @@
|
||||
//! The peer-to-peer audio protocol, wire compatible with furumi-fd.
|
||||
//!
|
||||
//! 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.
|
||||
|
||||
use std::path::{Path, PathBuf};
|
||||
use std::sync::Arc;
|
||||
|
||||
use anyhow::{Context, Result};
|
||||
use music_dht::{ByteStream, EndpointId, ItemId, ItemKind, MusicDhtService, StreamAcceptor};
|
||||
use serde::{Deserialize, Serialize};
|
||||
use tokio::io::{AsyncRead, AsyncReadExt, AsyncSeekExt, AsyncWriteExt};
|
||||
|
||||
use crate::library::Library;
|
||||
|
||||
/// ALPN of the audio streaming protocol (shared with furumi-fd).
|
||||
pub const AUDIO_ALPN: &[u8] = b"furumi-fd/audio/1";
|
||||
|
||||
/// Maximum size of a JSON protocol line (request or response header).
|
||||
const MAX_PROTOCOL_LINE: usize = 4096;
|
||||
|
||||
#[derive(Debug, Serialize, Deserialize)]
|
||||
struct AudioRequest {
|
||||
/// Hex-encoded [`ItemId`] of the track.
|
||||
item_id: String,
|
||||
/// Byte offset to start streaming from.
|
||||
offset: u64,
|
||||
}
|
||||
|
||||
#[derive(Debug, Serialize, Deserialize)]
|
||||
struct AudioResponseHeader {
|
||||
ok: bool,
|
||||
#[serde(default)]
|
||||
error: Option<String>,
|
||||
#[serde(default)]
|
||||
mime_type: String,
|
||||
#[serde(default)]
|
||||
total_size: u64,
|
||||
#[serde(default)]
|
||||
offset: u64,
|
||||
}
|
||||
|
||||
pub fn hex_encode(bytes: &[u8]) -> String {
|
||||
bytes.iter().map(|b| format!("{b:02x}")).collect()
|
||||
}
|
||||
|
||||
pub fn hex_decode_item_id(value: &str) -> Option<ItemId> {
|
||||
if value.len() != 64 {
|
||||
return None;
|
||||
}
|
||||
let mut bytes = [0u8; 32];
|
||||
for (i, byte) in bytes.iter_mut().enumerate() {
|
||||
*byte = u8::from_str_radix(&value[i * 2..i * 2 + 2], 16).ok()?;
|
||||
}
|
||||
Some(ItemId::from_bytes(bytes))
|
||||
}
|
||||
|
||||
fn guess_mime(path: &Path) -> &'static str {
|
||||
match path
|
||||
.extension()
|
||||
.and_then(|e| e.to_str())
|
||||
.unwrap_or_default()
|
||||
.to_ascii_lowercase()
|
||||
.as_str()
|
||||
{
|
||||
"mp3" => "audio/mpeg",
|
||||
"flac" => "audio/flac",
|
||||
"ogg" | "oga" => "audio/ogg",
|
||||
"opus" => "audio/opus",
|
||||
"wav" => "audio/wav",
|
||||
"m4a" | "mp4" | "alac" => "audio/mp4",
|
||||
"aac" => "audio/aac",
|
||||
"aiff" | "aif" => "audio/aiff",
|
||||
_ => "application/octet-stream",
|
||||
}
|
||||
}
|
||||
|
||||
/// Extension for a downloaded file, from the mime type the peer reported.
|
||||
fn extension_for_mime(mime: &str) -> &'static str {
|
||||
match mime {
|
||||
"audio/mpeg" => "mp3",
|
||||
"audio/flac" | "audio/x-flac" => "flac",
|
||||
"audio/ogg" => "ogg",
|
||||
"audio/opus" => "opus",
|
||||
"audio/wav" | "audio/x-wav" => "wav",
|
||||
"audio/mp4" | "audio/x-m4a" => "m4a",
|
||||
"audio/aac" => "aac",
|
||||
"audio/aiff" => "aiff",
|
||||
_ => "bin",
|
||||
}
|
||||
}
|
||||
|
||||
/// Reads one `\n`-terminated line, bounded by [`MAX_PROTOCOL_LINE`].
|
||||
async fn read_line<R: AsyncRead + Unpin>(reader: &mut R) -> Result<Vec<u8>> {
|
||||
let mut line = Vec::new();
|
||||
let mut byte = [0u8; 1];
|
||||
loop {
|
||||
let n = reader.read(&mut byte).await?;
|
||||
if n == 0 {
|
||||
anyhow::bail!("stream ended before the protocol line was complete");
|
||||
}
|
||||
if byte[0] == b'\n' {
|
||||
return Ok(line);
|
||||
}
|
||||
line.push(byte[0]);
|
||||
if line.len() > MAX_PROTOCOL_LINE {
|
||||
anyhow::bail!("protocol line exceeds {MAX_PROTOCOL_LINE} bytes");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
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?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Requesting side: download a track from its owner
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
/// Downloads a whole track from `owner` into `dir/<stem>.<ext>`; returns
|
||||
/// the file path and the mime type the peer reported. An already complete
|
||||
/// cached file is reused.
|
||||
pub async fn download_track(
|
||||
service: &MusicDhtService,
|
||||
owner: EndpointId,
|
||||
item_id_hex: &str,
|
||||
dir: &Path,
|
||||
stem: &str,
|
||||
) -> Result<(PathBuf, String)> {
|
||||
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,
|
||||
},
|
||||
)
|
||||
.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 stream: {}",
|
||||
header.error.unwrap_or_else(|| "unknown error".to_string())
|
||||
);
|
||||
}
|
||||
|
||||
let extension = extension_for_mime(&header.mime_type);
|
||||
let path = dir.join(format!("{stem}.{extension}"));
|
||||
if let Ok(metadata) = tokio::fs::metadata(&path).await
|
||||
&& metadata.len() == header.total_size
|
||||
&& header.total_size > 0
|
||||
{
|
||||
// Already fully downloaded earlier; no need to fetch again.
|
||||
return Ok((path, header.mime_type));
|
||||
}
|
||||
|
||||
let temp_path = dir.join(format!(".{stem}.{extension}.part"));
|
||||
let mut file = tokio::fs::File::create(&temp_path).await?;
|
||||
let mut received: u64 = 0;
|
||||
let mut chunk = vec![0u8; 64 * 1024];
|
||||
// quinn's inherent read returns None when the peer finished the stream.
|
||||
while let Some(n) = stream.recv.read(&mut chunk).await? {
|
||||
file.write_all(&chunk[..n]).await?;
|
||||
received += n as u64;
|
||||
}
|
||||
file.flush().await?;
|
||||
drop(file);
|
||||
if header.total_size > 0 && received != header.total_size {
|
||||
let _ = tokio::fs::remove_file(&temp_path).await;
|
||||
anyhow::bail!(
|
||||
"download incomplete: got {received} of {} bytes",
|
||||
header.total_size
|
||||
);
|
||||
}
|
||||
tokio::fs::rename(&temp_path, &path).await?;
|
||||
Ok((path, header.mime_type))
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Serving side: answer audio requests from other peers
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
/// Finds the local track whose derived DHT item id matches `item_id`.
|
||||
pub fn resolve_local_track_id(
|
||||
library: &Library,
|
||||
own: EndpointId,
|
||||
item_id: ItemId,
|
||||
) -> Result<Option<i64>> {
|
||||
let export = library.federation_export()?;
|
||||
for track in export.tracks {
|
||||
let derived = ItemId::derive(&own, ItemKind::Track, &format!("track:{}", track.id));
|
||||
if derived == item_id {
|
||||
return Ok(Some(track.id));
|
||||
}
|
||||
}
|
||||
Ok(None)
|
||||
}
|
||||
|
||||
fn resolve_local_file(
|
||||
library: &Library,
|
||||
own: EndpointId,
|
||||
item_id: ItemId,
|
||||
) -> Result<Option<String>> {
|
||||
let export = library.federation_export()?;
|
||||
for track in export.tracks {
|
||||
let derived = ItemId::derive(&own, ItemKind::Track, &format!("track:{}", track.id));
|
||||
if derived == item_id {
|
||||
return Ok(Some(track.file_path));
|
||||
}
|
||||
}
|
||||
Ok(None)
|
||||
}
|
||||
|
||||
/// Runs the accept loop of the audio protocol until the acceptor closes.
|
||||
/// Every track of the local library is downloadable by every peer of the
|
||||
/// network — the libraries of all participants are equal.
|
||||
pub async fn serve_peers(mut acceptor: StreamAcceptor, library: Arc<Library>, own: EndpointId) {
|
||||
while let Some(stream) = acceptor.accept().await {
|
||||
let library = Arc::clone(&library);
|
||||
tokio::spawn(async move {
|
||||
let peer = stream.peer_id;
|
||||
if let Err(err) = serve_one(stream, library, own).await {
|
||||
tracing::warn!(peer = %peer, "audio stream failed: {err:#}");
|
||||
}
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
async fn serve_one(mut stream: ByteStream, library: Arc<Library>, own: EndpointId) -> Result<()> {
|
||||
let request: AudioRequest = serde_json::from_slice(&read_line(&mut stream.recv).await?)?;
|
||||
tracing::info!(
|
||||
peer = %stream.peer_id,
|
||||
item = %request.item_id,
|
||||
offset = request.offset,
|
||||
"peer requested audio"
|
||||
);
|
||||
|
||||
let resolved = match hex_decode_item_id(&request.item_id) {
|
||||
Some(item_id) => {
|
||||
let library = Arc::clone(&library);
|
||||
match tokio::task::spawn_blocking(move || resolve_local_file(&library, own, item_id))
|
||||
.await
|
||||
{
|
||||
Ok(Ok(Some(path))) => Ok(path),
|
||||
Ok(Ok(None)) => Err("track not found in the library".to_string()),
|
||||
Ok(Err(err)) => Err(format!("library lookup failed: {err:#}")),
|
||||
Err(err) => Err(format!("lookup task failed: {err}")),
|
||||
}
|
||||
}
|
||||
None => Err("malformed item_id".to_string()),
|
||||
};
|
||||
let file_path = match resolved {
|
||||
Ok(path) => path,
|
||||
Err(message) => return refuse(stream, message).await,
|
||||
};
|
||||
|
||||
let path = PathBuf::from(&file_path);
|
||||
let mut file = match tokio::fs::File::open(&path).await {
|
||||
Ok(file) => file,
|
||||
Err(err) => return refuse(stream, format!("audio file is not readable: {err}")).await,
|
||||
};
|
||||
let total_size = file.metadata().await?.len();
|
||||
let offset = request.offset.min(total_size);
|
||||
if offset > 0 {
|
||||
file.seek(std::io::SeekFrom::Start(offset)).await?;
|
||||
}
|
||||
write_line(
|
||||
&mut stream.send,
|
||||
&AudioResponseHeader {
|
||||
ok: true,
|
||||
error: None,
|
||||
mime_type: guess_mime(&path).to_string(),
|
||||
total_size,
|
||||
offset,
|
||||
},
|
||||
)
|
||||
.await?;
|
||||
tokio::io::copy(&mut file, &mut stream.send).await?;
|
||||
stream.send.finish()?;
|
||||
// Wait until the peer read everything (or gave up) before dropping the
|
||||
// stream, otherwise the tail of the file is lost.
|
||||
let _ = stream.send.stopped().await;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
/// Sends a refusal header and waits until the peer read it.
|
||||
async fn refuse(mut stream: ByteStream, message: String) -> Result<()> {
|
||||
write_line(
|
||||
&mut stream.send,
|
||||
&AudioResponseHeader {
|
||||
ok: false,
|
||||
error: Some(message.clone()),
|
||||
mime_type: String::new(),
|
||||
total_size: 0,
|
||||
offset: 0,
|
||||
},
|
||||
)
|
||||
.await?;
|
||||
stream.send.finish()?;
|
||||
let _ = stream.send.stopped().await;
|
||||
anyhow::bail!("refused audio request: {message}");
|
||||
}
|
||||
@@ -0,0 +1,630 @@
|
||||
//! P2P federation for the TUI player.
|
||||
//!
|
||||
//! The player stays fully local, but can join a federated network of
|
||||
//! furumi instances (TUI or furumi-fd): it publishes its library index
|
||||
//! (names and small metadata, never files) into a shared DHT, searches the
|
||||
//! other participants' libraries and streams their audio over the same
|
||||
//! `furumi-fd/audio/1` protocol furumi-fd speaks — the two are wire
|
||||
//! compatible.
|
||||
//!
|
||||
//! Federated tracks are downloaded before playback: into a cache file, or —
|
||||
//! with "save on listen" enabled — straight into the local library (the
|
||||
//! file is imported like any local file, so this peer then serves it to
|
||||
//! the network too).
|
||||
|
||||
mod audio;
|
||||
|
||||
use std::path::PathBuf;
|
||||
use std::str::FromStr;
|
||||
use std::sync::Arc;
|
||||
use std::sync::atomic::{AtomicI64, Ordering};
|
||||
use std::time::Duration;
|
||||
|
||||
use anyhow::{Context, Result};
|
||||
use music_dht::{
|
||||
EndpointId, ItemKind, ItemSpec, MusicDhtConfig, MusicDhtService, NetworkId, PeerTicket,
|
||||
RendezvousConfig,
|
||||
};
|
||||
use serde::{Deserialize, Serialize};
|
||||
|
||||
use crate::library::Library;
|
||||
use crate::library::models::{ArtistRef, TrackItem};
|
||||
|
||||
pub use audio::AUDIO_ALPN;
|
||||
|
||||
/// How often the published library is re-synchronized with the local index.
|
||||
const SYNC_INTERVAL: Duration = Duration::from_secs(60);
|
||||
|
||||
/// Ephemeral (not-in-library) tracks get negative ids so the rest of the
|
||||
/// app can tell them apart from library rows (history, likes and release
|
||||
/// navigation skip them).
|
||||
static NEXT_EPHEMERAL_ID: AtomicI64 = AtomicI64::new(-1);
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Settings (persisted in <config dir>/federation.toml)
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
#[derive(Debug, Clone, PartialEq, Eq, Default, Serialize, Deserialize)]
|
||||
pub struct FedSettings {
|
||||
pub enabled: bool,
|
||||
#[serde(default)]
|
||||
pub network_id: String,
|
||||
/// Downloaded-for-playback federated tracks are imported into the local
|
||||
/// library (replication) instead of a throwaway cache.
|
||||
#[serde(default)]
|
||||
pub save_on_listen: bool,
|
||||
}
|
||||
|
||||
fn settings_path() -> Option<PathBuf> {
|
||||
crate::config::project_dirs().map(|dirs| dirs.config_dir().join("federation.toml"))
|
||||
}
|
||||
|
||||
pub fn load_settings() -> FedSettings {
|
||||
let Some(path) = settings_path() else {
|
||||
return FedSettings::default();
|
||||
};
|
||||
match std::fs::read_to_string(&path) {
|
||||
Ok(text) => toml::from_str(&text).unwrap_or_else(|err| {
|
||||
tracing::warn!(%err, "federation.toml is malformed; using defaults");
|
||||
FedSettings::default()
|
||||
}),
|
||||
Err(_) => FedSettings::default(),
|
||||
}
|
||||
}
|
||||
|
||||
fn save_settings(settings: &FedSettings) -> Result<()> {
|
||||
let path = settings_path().context("cannot determine the config directory")?;
|
||||
if let Some(parent) = path.parent() {
|
||||
std::fs::create_dir_all(parent)?;
|
||||
}
|
||||
std::fs::write(&path, toml::to_string_pretty(settings)?)?;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// Data shapes for the UI
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
/// A track found through federated search.
|
||||
#[derive(Debug, Clone, PartialEq, Eq)]
|
||||
pub struct FedTrack {
|
||||
/// Hex item id in the DHT (the key audio is requested by).
|
||||
pub item_id: String,
|
||||
/// Hex endpoint id of the owning peer.
|
||||
pub owner: String,
|
||||
/// The item is published by this very instance.
|
||||
pub own: bool,
|
||||
pub title: String,
|
||||
pub artist_names: Vec<String>,
|
||||
pub year: Option<i32>,
|
||||
pub duration_seconds: Option<i64>,
|
||||
}
|
||||
|
||||
impl FedTrack {
|
||||
pub fn artist_line(&self) -> String {
|
||||
self.artist_names.join(", ")
|
||||
}
|
||||
|
||||
pub fn owner_short(&self) -> String {
|
||||
self.owner.chars().take(10).collect()
|
||||
}
|
||||
|
||||
pub fn duration_label(&self) -> String {
|
||||
match self.duration_seconds {
|
||||
Some(total) => format!("{}:{:02}", total / 60, total % 60),
|
||||
None => String::new(),
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Live status snapshot rendered on the Federation tab.
|
||||
#[derive(Debug, Clone, Default)]
|
||||
pub struct FedStatus {
|
||||
pub running: bool,
|
||||
pub network: String,
|
||||
pub endpoint_id: String,
|
||||
pub connected_peers: Vec<String>,
|
||||
pub known_contacts: usize,
|
||||
pub published_items: usize,
|
||||
pub last_sync: Option<String>,
|
||||
pub last_error: Option<String>,
|
||||
}
|
||||
|
||||
/// Outcome of preparing a federated track for playback.
|
||||
#[derive(Debug)]
|
||||
pub struct FedPlayable {
|
||||
pub track: TrackItem,
|
||||
/// The file was imported into the local library (save-on-listen).
|
||||
pub imported: bool,
|
||||
}
|
||||
|
||||
// ---------------------------------------------------------------------------
|
||||
// The federation manager (lives in Runtime, not AppState)
|
||||
// ---------------------------------------------------------------------------
|
||||
|
||||
struct Running {
|
||||
service: Arc<MusicDhtService>,
|
||||
network_name: String,
|
||||
tasks: Vec<tokio::task::JoinHandle<()>>,
|
||||
}
|
||||
|
||||
pub struct Federation {
|
||||
library: Arc<Library>,
|
||||
data_dir: PathBuf,
|
||||
cache_dir: PathBuf,
|
||||
media_dir: PathBuf,
|
||||
settings: std::sync::Mutex<FedSettings>,
|
||||
running: tokio::sync::Mutex<Option<Running>>,
|
||||
last_sync: std::sync::Mutex<Option<String>>,
|
||||
last_error: std::sync::Mutex<Option<String>>,
|
||||
}
|
||||
|
||||
fn lock<T>(mutex: &std::sync::Mutex<T>) -> std::sync::MutexGuard<'_, T> {
|
||||
mutex.lock().unwrap_or_else(std::sync::PoisonError::into_inner)
|
||||
}
|
||||
|
||||
fn now_label() -> String {
|
||||
// Seconds since start of the day are enough for a status line without
|
||||
// pulling in a date-time crate.
|
||||
let secs = std::time::SystemTime::now()
|
||||
.duration_since(std::time::UNIX_EPOCH)
|
||||
.map(|d| d.as_secs())
|
||||
.unwrap_or(0);
|
||||
format!(
|
||||
"{:02}:{:02}:{:02} UTC",
|
||||
secs / 3600 % 24,
|
||||
secs / 60 % 60,
|
||||
secs % 60
|
||||
)
|
||||
}
|
||||
|
||||
impl Federation {
|
||||
pub fn new(library: Arc<Library>) -> Arc<Self> {
|
||||
let dirs = crate::config::project_dirs();
|
||||
let data_dir = dirs
|
||||
.as_ref()
|
||||
.map(|d| d.data_dir().join("federation"))
|
||||
.unwrap_or_else(|| PathBuf::from("federation"));
|
||||
let cache_dir = dirs
|
||||
.as_ref()
|
||||
.map(|d| d.cache_dir().join("fedcache"))
|
||||
.unwrap_or_else(|| PathBuf::from("fedcache"));
|
||||
let media_dir = dirs
|
||||
.as_ref()
|
||||
.map(|d| d.data_dir().join("federation-media"))
|
||||
.unwrap_or_else(|| PathBuf::from("federation-media"));
|
||||
Arc::new(Self {
|
||||
library,
|
||||
data_dir,
|
||||
cache_dir,
|
||||
media_dir,
|
||||
settings: std::sync::Mutex::new(load_settings()),
|
||||
running: tokio::sync::Mutex::new(None),
|
||||
last_sync: std::sync::Mutex::new(None),
|
||||
last_error: std::sync::Mutex::new(None),
|
||||
})
|
||||
}
|
||||
|
||||
pub fn settings(&self) -> FedSettings {
|
||||
lock(&self.settings).clone()
|
||||
}
|
||||
|
||||
fn set_error(&self, message: Option<String>) {
|
||||
*lock(&self.last_error) = message;
|
||||
}
|
||||
|
||||
/// Persists new settings and starts/stops/restarts the node to match.
|
||||
pub async fn apply_settings(self: &Arc<Self>, settings: FedSettings) -> Result<()> {
|
||||
anyhow::ensure!(
|
||||
!settings.enabled || !settings.network_id.trim().is_empty(),
|
||||
"set a network id before enabling federation"
|
||||
);
|
||||
if let Err(err) = save_settings(&settings) {
|
||||
tracing::warn!(%err, "saving federation settings failed");
|
||||
}
|
||||
*lock(&self.settings) = settings.clone();
|
||||
if settings.enabled {
|
||||
self.start(settings.network_id.trim().to_string()).await?;
|
||||
self.spawn_sync_soon().await;
|
||||
} else {
|
||||
self.stop().await;
|
||||
}
|
||||
Ok(())
|
||||
}
|
||||
|
||||
pub async fn start_if_enabled(self: &Arc<Self>) {
|
||||
let settings = self.settings();
|
||||
if settings.enabled && !settings.network_id.trim().is_empty() {
|
||||
if let Err(err) = self.start(settings.network_id.trim().to_string()).await {
|
||||
tracing::error!("federation autostart failed: {err:#}");
|
||||
self.set_error(Some(format!("autostart failed: {err}")));
|
||||
} else {
|
||||
self.spawn_sync_soon().await;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Starts the DHT node. Idempotent per network name.
|
||||
async fn start(self: &Arc<Self>, network_name: String) -> Result<()> {
|
||||
let mut guard = self.running.lock().await;
|
||||
if let Some(running) = guard.as_ref() {
|
||||
if running.network_name == network_name {
|
||||
return Ok(());
|
||||
}
|
||||
stop_running(guard.take()).await;
|
||||
}
|
||||
|
||||
let config = MusicDhtConfig::builder()
|
||||
.data_dir(&self.data_dir)
|
||||
.network_id(NetworkId::from_name(&network_name))
|
||||
// Peers of the network find each other knowing only its name.
|
||||
.rendezvous(RendezvousConfig::default())
|
||||
// Peers stream each other's audio over this protocol.
|
||||
.stream_protocol(AUDIO_ALPN)
|
||||
.build()
|
||||
.map_err(|err| anyhow::anyhow!("invalid federation config: {err}"))?;
|
||||
let (service, mut events) = MusicDhtService::start(config)
|
||||
.await
|
||||
.map_err(|err| anyhow::anyhow!("failed to start the DHT node: {err}"))?;
|
||||
let service = Arc::new(service);
|
||||
tracing::info!(
|
||||
endpoint_id = %service.endpoint_id(),
|
||||
network = %network_name,
|
||||
"federation started"
|
||||
);
|
||||
|
||||
// Drain DHT events into the log; the channel is bounded.
|
||||
let event_task = tokio::spawn(async move {
|
||||
while let Some(event) = events.recv().await {
|
||||
tracing::debug!("federation event: {event:?}");
|
||||
}
|
||||
});
|
||||
// Keep the published library in sync with the local index.
|
||||
let sync_self = Arc::clone(self);
|
||||
let sync_service = Arc::clone(&service);
|
||||
let sync_task = tokio::spawn(async move {
|
||||
let mut interval = tokio::time::interval(SYNC_INTERVAL);
|
||||
loop {
|
||||
interval.tick().await;
|
||||
sync_self.sync_once(&sync_service).await;
|
||||
}
|
||||
});
|
||||
// Serve audio requests from other peers of the network.
|
||||
let audio_acceptor = service
|
||||
.stream_acceptor(AUDIO_ALPN)
|
||||
.map_err(|err| anyhow::anyhow!("failed to take the audio acceptor: {err}"))?;
|
||||
let audio_task = tokio::spawn(audio::serve_peers(
|
||||
audio_acceptor,
|
||||
Arc::clone(&self.library),
|
||||
service.endpoint_id(),
|
||||
));
|
||||
|
||||
*guard = Some(Running {
|
||||
service,
|
||||
network_name,
|
||||
tasks: vec![event_task, sync_task, audio_task],
|
||||
});
|
||||
self.set_error(None);
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn stop(&self) {
|
||||
let mut guard = self.running.lock().await;
|
||||
stop_running(guard.take()).await;
|
||||
}
|
||||
|
||||
pub async fn shutdown(&self) {
|
||||
self.stop().await;
|
||||
}
|
||||
|
||||
async fn service(&self) -> Result<Arc<MusicDhtService>> {
|
||||
self.running
|
||||
.lock()
|
||||
.await
|
||||
.as_ref()
|
||||
.map(|running| Arc::clone(&running.service))
|
||||
.context("federation is not running")
|
||||
}
|
||||
|
||||
/// Publishes the library immediately (used right after start/settings).
|
||||
async fn spawn_sync_soon(self: &Arc<Self>) {
|
||||
if let Ok(service) = self.service().await {
|
||||
let fed = Arc::clone(self);
|
||||
tokio::spawn(async move { fed.sync_once(&service).await });
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn sync_now(self: &Arc<Self>) -> Result<()> {
|
||||
let service = self.service().await?;
|
||||
self.sync_once(&service).await;
|
||||
Ok(())
|
||||
}
|
||||
|
||||
async fn sync_once(&self, service: &MusicDhtService) {
|
||||
let library = Arc::clone(&self.library);
|
||||
let specs = tokio::task::spawn_blocking(move || collect_specs(&library)).await;
|
||||
let specs = match specs {
|
||||
Ok(Ok(specs)) => specs,
|
||||
Ok(Err(err)) => {
|
||||
tracing::warn!("federation sync: library read failed: {err:#}");
|
||||
self.set_error(Some(format!("library read failed: {err}")));
|
||||
return;
|
||||
}
|
||||
Err(err) => {
|
||||
tracing::warn!("federation sync task failed: {err}");
|
||||
return;
|
||||
}
|
||||
};
|
||||
match service.sync_library(specs).await {
|
||||
Ok(stats) => {
|
||||
*lock(&self.last_sync) = Some(format!(
|
||||
"{} (+{} ~{} −{}, unchanged {})",
|
||||
now_label(),
|
||||
stats.added,
|
||||
stats.updated,
|
||||
stats.removed,
|
||||
stats.unchanged
|
||||
));
|
||||
self.set_error(None);
|
||||
}
|
||||
Err(err) => {
|
||||
tracing::warn!("federation sync failed: {err}");
|
||||
self.set_error(Some(format!("sync failed: {err}")));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub async fn status(&self) -> FedStatus {
|
||||
let settings = self.settings();
|
||||
let guard = self.running.lock().await;
|
||||
let mut status = FedStatus {
|
||||
network: settings.network_id,
|
||||
last_sync: lock(&self.last_sync).clone(),
|
||||
last_error: lock(&self.last_error).clone(),
|
||||
..FedStatus::default()
|
||||
};
|
||||
if let Some(running) = guard.as_ref() {
|
||||
let service = &running.service;
|
||||
status.running = true;
|
||||
status.network = running.network_name.clone();
|
||||
status.endpoint_id = service.endpoint_id().to_string();
|
||||
status.connected_peers = service
|
||||
.connected_peers()
|
||||
.iter()
|
||||
.map(|p| p.to_string())
|
||||
.collect();
|
||||
status.known_contacts = service.known_peers().len();
|
||||
status.published_items = service
|
||||
.list_local_items()
|
||||
.await
|
||||
.map(|items| items.len())
|
||||
.unwrap_or(0);
|
||||
}
|
||||
status
|
||||
}
|
||||
|
||||
/// Searches the federated network for tracks matching `query`.
|
||||
pub async fn search(&self, query: &str) -> Result<Vec<FedTrack>> {
|
||||
let service = self.service().await?;
|
||||
let outcome = service
|
||||
.search_network(query)
|
||||
.await
|
||||
.map_err(|err| anyhow::anyhow!("federated search failed: {err}"))?;
|
||||
let own = service.endpoint_id();
|
||||
Ok(outcome
|
||||
.network_results
|
||||
.iter()
|
||||
.filter(|item| item.kind == ItemKind::Track)
|
||||
.map(|item| FedTrack {
|
||||
item_id: audio::hex_encode(item.id.as_bytes()),
|
||||
owner: item.owner.to_string(),
|
||||
own: item.owner == own,
|
||||
title: item.name.clone(),
|
||||
artist_names: item.artist_names.clone(),
|
||||
year: item.year,
|
||||
duration_seconds: item.duration_seconds.map(|d| d.round() as i64),
|
||||
})
|
||||
.collect())
|
||||
}
|
||||
|
||||
pub async fn ticket(&self) -> Result<String> {
|
||||
let service = self.service().await?;
|
||||
let ticket = service
|
||||
.ticket()
|
||||
.await
|
||||
.map_err(|err| anyhow::anyhow!("cannot create a ticket: {err}"))?;
|
||||
Ok(ticket.to_string())
|
||||
}
|
||||
|
||||
pub async fn connect(&self, ticket: &str) -> Result<String> {
|
||||
let service = self.service().await?;
|
||||
let ticket: PeerTicket = ticket
|
||||
.trim()
|
||||
.parse()
|
||||
.map_err(|err| anyhow::anyhow!("malformed ticket: {err}"))?;
|
||||
let peer = service
|
||||
.connect(ticket)
|
||||
.await
|
||||
.map_err(|err| anyhow::anyhow!("connect failed: {err}"))?;
|
||||
Ok(peer.to_string())
|
||||
}
|
||||
|
||||
/// Prepares a federated track for playback: local tracks resolve
|
||||
/// straight to the library; remote tracks are downloaded — into the
|
||||
/// library when save-on-listen is enabled, into the cache otherwise.
|
||||
pub async fn prepare_playback(self: &Arc<Self>, fed: &FedTrack) -> Result<FedPlayable> {
|
||||
let service = self.service().await?;
|
||||
let item_id =
|
||||
audio::hex_decode_item_id(&fed.item_id).context("malformed item id in the result")?;
|
||||
|
||||
if fed.own {
|
||||
let library = Arc::clone(&self.library);
|
||||
let own_id = service.endpoint_id();
|
||||
let track = tokio::task::spawn_blocking(move || -> Result<Option<TrackItem>> {
|
||||
let Some(track_id) = audio::resolve_local_track_id(&library, own_id, item_id)?
|
||||
else {
|
||||
return Ok(None);
|
||||
};
|
||||
Ok(library.tracks_by_ids(&[track_id])?.into_iter().next())
|
||||
})
|
||||
.await??
|
||||
.context("this track is no longer in the local library")?;
|
||||
return Ok(FedPlayable {
|
||||
track,
|
||||
imported: false,
|
||||
});
|
||||
}
|
||||
|
||||
let owner = EndpointId::from_str(&fed.owner)
|
||||
.map_err(|_| anyhow::anyhow!("malformed owner id '{}'", fed.owner))?;
|
||||
let save = self.settings().save_on_listen;
|
||||
let dir = if save { &self.media_dir } else { &self.cache_dir };
|
||||
tokio::fs::create_dir_all(dir).await?;
|
||||
|
||||
let (path, mime) =
|
||||
audio::download_track(&service, owner, &fed.item_id, dir, &download_stem(fed)).await?;
|
||||
tracing::info!(path = %path.display(), %mime, "federated track downloaded");
|
||||
|
||||
if save {
|
||||
let library = Arc::clone(&self.library);
|
||||
let import_path = path.clone();
|
||||
let imported = tokio::task::spawn_blocking(move || -> Result<Option<TrackItem>> {
|
||||
let import = crate::library::import::read_file(&import_path)?;
|
||||
let (track_id, _) = crate::library::import::upsert_track(&library, &import)?;
|
||||
Ok(library.tracks_by_ids(&[track_id])?.into_iter().next())
|
||||
})
|
||||
.await?;
|
||||
match imported {
|
||||
Ok(Some(track)) => {
|
||||
return Ok(FedPlayable {
|
||||
track,
|
||||
imported: true,
|
||||
});
|
||||
}
|
||||
Ok(None) => {}
|
||||
Err(err) => {
|
||||
tracing::warn!("importing the downloaded track failed: {err:#}; playing from the file");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
Ok(FedPlayable {
|
||||
track: ephemeral_track(fed, &path),
|
||||
imported: false,
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
async fn stop_running(running: Option<Running>) {
|
||||
let Some(running) = running else { return };
|
||||
for task in &running.tasks {
|
||||
task.abort();
|
||||
}
|
||||
if let Err(err) = running.service.shutdown().await {
|
||||
tracing::warn!("federation node shutdown reported an error: {err}");
|
||||
}
|
||||
tracing::info!("federation stopped");
|
||||
}
|
||||
|
||||
/// Reads the local library and converts it into DHT item specs. Only names
|
||||
/// and small metadata are shared — never file paths or the files themselves.
|
||||
fn collect_specs(library: &Library) -> Result<Vec<ItemSpec>> {
|
||||
let export = library.federation_export()?;
|
||||
let mut specs = Vec::new();
|
||||
for (id, name) in export.artists {
|
||||
specs.push(ItemSpec {
|
||||
local_key: format!("artist:{id}"),
|
||||
kind: ItemKind::Artist,
|
||||
name,
|
||||
artist_names: Vec::new(),
|
||||
year: None,
|
||||
release_type: None,
|
||||
duration_seconds: None,
|
||||
});
|
||||
}
|
||||
for release in export.releases {
|
||||
specs.push(ItemSpec {
|
||||
local_key: format!("release:{}", release.id),
|
||||
kind: ItemKind::Release,
|
||||
name: release.title,
|
||||
artist_names: release.artist_names,
|
||||
year: release.year,
|
||||
release_type: Some(release.release_type),
|
||||
duration_seconds: None,
|
||||
});
|
||||
}
|
||||
for track in export.tracks {
|
||||
specs.push(ItemSpec {
|
||||
local_key: format!("track:{}", track.id),
|
||||
kind: ItemKind::Track,
|
||||
name: track.title,
|
||||
artist_names: track.artist_names,
|
||||
year: track.year,
|
||||
release_type: None,
|
||||
duration_seconds: (track.duration_seconds > 0.0).then_some(track.duration_seconds),
|
||||
});
|
||||
}
|
||||
Ok(specs)
|
||||
}
|
||||
|
||||
fn sanitize_file_stem(value: &str) -> String {
|
||||
let cleaned: String = value
|
||||
.chars()
|
||||
.map(|c| match c {
|
||||
'/' | '\\' | ':' | '*' | '?' | '"' | '<' | '>' | '|' => '_',
|
||||
c if c.is_control() => '_',
|
||||
c => c,
|
||||
})
|
||||
.collect();
|
||||
let trimmed = cleaned.trim().trim_matches('.');
|
||||
let mut stem: String = trimmed.chars().take(120).collect();
|
||||
if stem.is_empty() {
|
||||
stem.push_str("track");
|
||||
}
|
||||
stem
|
||||
}
|
||||
|
||||
fn download_stem(fed: &FedTrack) -> String {
|
||||
let artists = fed.artist_line();
|
||||
if artists.is_empty() {
|
||||
sanitize_file_stem(&fed.title)
|
||||
} else {
|
||||
sanitize_file_stem(&format!("{artists} - {}", fed.title))
|
||||
}
|
||||
}
|
||||
|
||||
/// A playable TrackItem for a downloaded-but-not-imported federated track.
|
||||
fn ephemeral_track(fed: &FedTrack, path: &std::path::Path) -> TrackItem {
|
||||
let id = NEXT_EPHEMERAL_ID.fetch_sub(1, Ordering::Relaxed);
|
||||
let file_size = std::fs::metadata(path).map(|m| m.len() as i64).ok();
|
||||
TrackItem {
|
||||
id,
|
||||
title: fed.title.clone(),
|
||||
track_number: None,
|
||||
disc_number: None,
|
||||
duration_seconds: fed.duration_seconds.unwrap_or(0) as f64,
|
||||
artists: fed
|
||||
.artist_names
|
||||
.iter()
|
||||
.map(|name| ArtistRef {
|
||||
id: -1,
|
||||
name: name.clone(),
|
||||
})
|
||||
.collect(),
|
||||
featured_artists: Vec::new(),
|
||||
release_id: -1,
|
||||
release_title: format!("federation · {}", fed.owner_short()),
|
||||
release_year: fed.year,
|
||||
file_path: path.to_string_lossy().into_owned(),
|
||||
cover_path: None,
|
||||
audio_format: path
|
||||
.extension()
|
||||
.and_then(|e| e.to_str())
|
||||
.map(|e| e.to_string()),
|
||||
audio_bitrate: None,
|
||||
audio_sample_rate: None,
|
||||
audio_bit_depth: None,
|
||||
file_size_bytes: file_size,
|
||||
play_count: 0,
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user