furumi_tui/src/federation/mod.rs

//! 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;
pub mod catalog;

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, TrackMetadata};
pub use catalog::{CATALOG_ALPN, FedArtistCard, FedCardTrack, FedRelease};

/// 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
// ---------------------------------------------------------------------------

/// An artist surfaced by federated search — either an artist record, or
/// derived from the artist names of matching tracks/releases (so searching
/// a track title still leads to the artist's card).
#[derive(Debug, Clone, PartialEq, Eq)]
pub struct FedArtistHit {
    pub name: String,
    /// Distinct peers (other than this instance) holding the artist.
    pub peers: usize,
}

/// Federated search results for the UI.
#[derive(Debug, Clone, Default, PartialEq, Eq)]
pub struct FedSearchResults {
    pub artists: Vec<FedArtistHit>,
    pub tracks: Vec<FedTrack>,
}

/// 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>,
    /// Release context, known when the track came from an artist card.
    pub release_title: Option<String>,
    pub track_number: Option<i32>,
    pub disc_number: Option<i32>,
}

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)
            // ...and browse each other's per-artist catalogs over this one.
            .stream_protocol(CATALOG_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(),
        ));
        // Serve per-artist catalog requests (the federated artist card).
        let catalog_acceptor = service
            .stream_acceptor(CATALOG_ALPN)
            .map_err(|err| anyhow::anyhow!("failed to take the catalog acceptor: {err}"))?;
        let catalog_task = tokio::spawn(catalog::serve_peers(
            catalog_acceptor,
            Arc::clone(&self.library),
            service.endpoint_id(),
        ));

        *guard = Some(Running {
            service,
            network_name,
            tasks: vec![event_task, sync_task, audio_task, catalog_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 {}, failed {})",
                    now_label(),
                    stats.added,
                    stats.updated,
                    stats.removed,
                    stats.unchanged,
                    stats.failed
                ));
                if stats.failed > 0 {
                    self.set_error(Some(format!(
                        "{} item(s) failed to publish in the last sync",
                        stats.failed
                    )));
                } else {
                    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: matching tracks plus the artists a
    /// card can be assembled for (from artist records and from the artist
    /// names of matching tracks/releases).
    pub async fn search(&self, query: &str) -> Result<FedSearchResults> {
        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();
        let tracks: Vec<FedTrack> = 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),
                release_title: None,
                track_number: None,
                disc_number: None,
            })
            .collect();

        // Artists: normalized name -> (display name, distinct non-own peers).
        let mut artists: std::collections::HashMap<
            String,
            (String, std::collections::HashSet<EndpointId>),
        > = Default::default();
        for item in &outcome.network_results {
            if item.owner == own {
                continue;
            }
            let mut note = |name: &str| {
                let key = music_dht::normalize_name(name);
                if key.is_empty() {
                    return;
                }
                let entry = artists
                    .entry(key)
                    .or_insert_with(|| (name.to_string(), Default::default()));
                entry.1.insert(item.owner);
            };
            if item.kind == ItemKind::Artist {
                note(&item.name);
            }
            for artist in &item.artist_names {
                note(artist);
            }
        }
        let mut artists: Vec<FedArtistHit> = artists
            .into_values()
            .map(|(name, owners)| FedArtistHit {
                name,
                peers: owners.len(),
            })
            .collect();
        artists.sort_by(|a, b| b.peers.cmp(&a.peers).then_with(|| a.name.cmp(&b.name)));

        Ok(FedSearchResults { artists, tracks })
    }

    /// Assembles the federated artist card: finds the peers holding the
    /// artist through the DHT, asks each for its catalog slice directly and
    /// merges the answers (missing/slow peers are skipped).
    pub async fn artist_card(&self, name: &str) -> Result<FedArtistCard> {
        let service = self.service().await?;
        let own = service.endpoint_id();
        let normalized = music_dht::normalize_name(name);
        let outcome = service
            .search_network(name)
            .await
            .map_err(|err| anyhow::anyhow!("federated search failed: {err}"))?;
        let owners: std::collections::HashSet<EndpointId> = outcome
            .network_results
            .iter()
            .filter(|item| {
                (item.kind == ItemKind::Artist && item.normalized_name == normalized)
                    || item
                        .artist_names
                        .iter()
                        .any(|artist| music_dht::normalize_name(artist) == normalized)
            })
            .map(|item| item.owner)
            .filter(|owner| *owner != own)
            .collect();
        anyhow::ensure!(!owners.is_empty(), "no peers hold artist \"{name}\"");

        let mut requests = Vec::new();
        for owner in owners {
            let service = Arc::clone(&service);
            let name = name.to_string();
            requests.push(tokio::spawn(async move {
                let result = tokio::time::timeout(
                    Duration::from_secs(5),
                    catalog::fetch_catalog(&service, owner, &name),
                )
                .await;
                match result {
                    Ok(Ok(catalog)) => Some((owner.to_string(), catalog)),
                    Ok(Err(err)) => {
                        tracing::warn!(peer = %owner, "catalog fetch failed: {err:#}");
                        None
                    }
                    Err(_) => {
                        tracing::warn!(peer = %owner, "catalog fetch timed out");
                        None
                    }
                }
            }));
        }
        let mut catalogs = Vec::new();
        for request in requests {
            if let Ok(Some(catalog)) = request.await {
                catalogs.push(catalog);
            }
        }
        anyhow::ensure!(
            !catalogs.is_empty(),
            "none of the peers answered the catalog request"
        );
        Ok(catalog::merge_catalogs(name, catalogs))
    }

    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())
    }

    /// Directory for streamed (never library-imported) card artwork.
    fn art_cache_dir(&self) -> PathBuf {
        self.cache_dir.join("art")
    }

    /// Returns a cached-or-streamed image for the card: the artist image
    /// (`release: None`) or a release cover. Peers are tried in order until
    /// one answers with an image; the result lands in the art cache and its
    /// local path is returned.
    pub async fn card_image(
        &self,
        owners: &[String],
        artist: &str,
        release: Option<&str>,
    ) -> Option<String> {
        let dir = self.art_cache_dir();
        let stem = match release {
            Some(release) => format!(
                "cover-{}-{}",
                sanitize_file_stem(artist),
                sanitize_file_stem(release)
            ),
            None => format!("artist-{}", sanitize_file_stem(artist)),
        };
        // Reuse a previously streamed copy of any known image type.
        for extension in ["jpg", "png", "webp", "gif", "bmp"] {
            let path = dir.join(format!("{stem}.{extension}"));
            if path.is_file() {
                return Some(path.to_string_lossy().into_owned());
            }
        }
        let service = self.service().await.ok()?;
        tokio::fs::create_dir_all(&dir).await.ok()?;
        for owner in owners {
            let Ok(owner) = EndpointId::from_str(owner) else {
                continue;
            };
            let fetched = tokio::time::timeout(
                Duration::from_secs(5),
                catalog::fetch_image(&service, owner, artist, release),
            )
            .await;
            match fetched {
                Ok(Ok(Some((bytes, extension)))) => {
                    let path = dir.join(format!("{stem}.{extension}"));
                    if tokio::fs::write(&path, &bytes).await.is_ok() {
                        return Some(path.to_string_lossy().into_owned());
                    }
                }
                Ok(Ok(None)) => continue,
                Ok(Err(err)) => tracing::debug!(peer = %owner, "image fetch failed: {err:#}"),
                Err(_) => tracing::debug!(peer = %owner, "image fetch timed out"),
            }
        }
        None
    }

    /// Downloads a federated track straight into the local library
    /// (regardless of the save-on-listen setting) and returns the imported
    /// track. Own/already-local tracks resolve without downloading.
    pub async fn download_to_library(self: &Arc<Self>, fed: &FedTrack) -> Result<TrackItem> {
        let playable = self.fetch_playable(fed, true).await?;
        Ok(playable.track)
    }

    /// 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 save = self.settings().save_on_listen;
        self.fetch_playable(fed, save).await
    }

    async fn fetch_playable(self: &Arc<Self>, fed: &FedTrack, save: bool) -> 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 dir = if save { &self.media_dir } else { &self.cache_dir };
        tokio::fs::create_dir_all(dir).await?;

        let downloaded =
            audio::download_track(&service, owner, &fed.item_id, dir, &download_stem(fed)).await?;
        tracing::info!(
            path = %downloaded.path.display(),
            mime = %downloaded.mime_type,
            cover = downloaded.cover.is_some(),
            "federated track downloaded"
        );

        if save {
            let library = Arc::clone(&self.library);
            let import_path = downloaded.path.clone();
            let import_metadata = downloaded.metadata.clone();
            let import_cover = downloaded.cover.clone();
            let artist_image = downloaded.artist_image.clone();
            let fed_item_id = fed.item_id.clone();
            let imported = tokio::task::spawn_blocking(move || -> Result<Option<TrackItem>> {
                let mut import = crate::library::import::read_file(&import_path)?;
                // The owner's database is more authoritative than whatever
                // tags the file happens to carry (often none at all).
                if let Some(meta) = &import_metadata {
                    apply_remote_metadata(&mut import, meta);
                }
                // Same for the cover: the peer's library cover wins over an
                // embedded picture; embedded art stays as the fallback.
                if import_cover.is_some() {
                    import.cover = import_cover;
                }
                let (track_id, _) = crate::library::import::upsert_track(&library, &import)?;
                // A like that referenced the federated track moves onto the
                // freshly imported local row.
                if let Err(err) = library.transfer_fed_like(&fed_item_id, track_id) {
                    tracing::warn!(%err, "federated like transfer failed");
                }
                // The owner's artist image fills the gap for a freshly
                // created (or still image-less) main artist.
                if let (Some((bytes, extension)), Some(artist_name)) =
                    (&artist_image, import.artists.first())
                    && let Err(err) = save_artist_image(&library, artist_name, bytes, extension)
                {
                    tracing::warn!(%err, "saving the artist image failed");
                }
                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");
                }
            }
        }

        // Ephemeral playback: put the cover next to the cached audio so the
        // views can show it.
        let cover_path = match &downloaded.cover {
            Some((bytes, extension)) => {
                let path = downloaded.path.with_extension(format!("cover.{extension}"));
                match tokio::fs::write(&path, bytes).await {
                    Ok(()) => Some(path.to_string_lossy().into_owned()),
                    Err(err) => {
                        tracing::warn!(%err, "saving the cover failed");
                        None
                    }
                }
            }
            None => None,
        };
        let mut track =
            ephemeral_track(fed, downloaded.metadata.as_ref(), &downloaded.path);
        track.cover_path = cover_path;
        Ok(FedPlayable {
            track,
            imported: false,
        })
    }
}

/// Writes a received artist image into the covers directory and attaches it
/// to the artist unless one is already set.
fn save_artist_image(
    library: &Library,
    artist_name: &str,
    bytes: &[u8],
    extension: &str,
) -> Result<()> {
    let covers_dir = library.covers_dir();
    std::fs::create_dir_all(covers_dir)?;
    let path = covers_dir.join(format!(
        "artist-{}.{extension}",
        sanitize_file_stem(artist_name)
    ));
    // Write only if the artist actually lacks an image, to avoid litter.
    if library.artist_image_missing(artist_name)? {
        std::fs::write(&path, bytes)?;
        library.set_artist_image_if_missing(artist_name, &path.to_string_lossy())?;
    }
    Ok(())
}

/// Overlays the peer-supplied metadata onto tag-derived import data. Every
/// non-empty peer field wins; file tags only fill the gaps.
fn apply_remote_metadata(import: &mut crate::library::import::TrackImport, meta: &TrackMetadata) {
    let title = meta.title.trim();
    if !title.is_empty() {
        import.title = title.to_string();
    }
    if !meta.artists.is_empty() {
        import.artists = meta.artists.clone();
    }
    if !meta.featured_artists.is_empty() {
        import.featured_artists = meta.featured_artists.clone();
    }
    if !meta.album_artists.is_empty() {
        import.album_artists = meta.album_artists.clone();
    } else if !meta.artists.is_empty() {
        import.album_artists = meta.artists.clone();
    }
    let release_title = meta.release_title.trim();
    if !release_title.is_empty() {
        import.release_title = release_title.to_string();
    }
    if meta.release_type.is_some() {
        import.release_type = meta.release_type.clone();
    }
    if meta.year.is_some() {
        import.year = meta.year;
    }
    if meta.track_number.is_some() {
        import.track_number = meta.track_number;
    }
    if meta.disc_number.is_some() {
        import.disc_number = meta.disc_number;
    }
}

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 queueable placeholder for a federated track that has not been
/// downloaded yet: it behaves like a regular track everywhere (queue, info,
/// selection) and resolves to a local file when playback reaches it.
pub fn pending_track(fed: &FedTrack) -> TrackItem {
    let id = NEXT_EPHEMERAL_ID.fetch_sub(1, Ordering::Relaxed);
    let refs = |names: &[String]| -> Vec<ArtistRef> {
        names
            .iter()
            .map(|name| ArtistRef {
                id: -1,
                name: name.clone(),
            })
            .collect()
    };
    TrackItem {
        id,
        title: fed.title.clone(),
        track_number: fed.track_number,
        disc_number: fed.disc_number,
        duration_seconds: fed.duration_seconds.unwrap_or(0) as f64,
        artists: refs(&fed.artist_names),
        featured_artists: Vec::new(),
        release_id: -1,
        release_title: fed
            .release_title
            .clone()
            .unwrap_or_else(|| format!("federation · {}", fed.owner_short())),
        release_year: fed.year,
        file_path: String::new(),
        cover_path: None,
        audio_format: None,
        audio_bitrate: None,
        audio_sample_rate: None,
        audio_bit_depth: None,
        file_size_bytes: None,
        play_count: 0,
        fed: Some(fed.clone()),
    }
}

/// A playable TrackItem for a downloaded-but-not-imported federated track.
fn ephemeral_track(
    fed: &FedTrack,
    metadata: Option<&TrackMetadata>,
    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();
    let refs = |names: &[String]| -> Vec<ArtistRef> {
        names
            .iter()
            .map(|name| ArtistRef {
                id: -1,
                name: name.clone(),
            })
            .collect()
    };
    let title = metadata
        .map(|m| m.title.trim())
        .filter(|t| !t.is_empty())
        .unwrap_or(&fed.title)
        .to_string();
    let artists = match metadata {
        Some(meta) if !meta.artists.is_empty() => refs(&meta.artists),
        _ => refs(&fed.artist_names),
    };
    let release_title = metadata
        .map(|m| m.release_title.trim())
        .filter(|t| !t.is_empty())
        .map(|t| t.to_string())
        .unwrap_or_else(|| format!("federation · {}", fed.owner_short()));
    TrackItem {
        id,
        title,
        track_number: metadata.and_then(|m| m.track_number),
        disc_number: metadata.and_then(|m| m.disc_number),
        duration_seconds: fed.duration_seconds.unwrap_or(0) as f64,
        artists,
        featured_artists: metadata.map(|m| refs(&m.featured_artists)).unwrap_or_default(),
        release_id: -1,
        release_title,
        release_year: metadata.and_then(|m| m.year).or(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,
        // Keep the federation reference: the cached track can still be
        // liked, re-downloaded and marked as federated in the lists.
        fed: Some(fed.clone()),
    }
}