Bump protocols. Added proto status

Cargo.lock

@@ -1707,8 +1707,9 @@ dependencies = [
 
 [[package]]
 name = "federation-net"
-version = "0.1.0"
-source = "git+https://gt.hexor.cy/ab/frid.git#8de7d1292708fa0b225e5a4a9d5ab4f0676202d3"
+version = "0.2.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "15a8707baeccb46b5935138f9cb3df3c988c0730b807a2634d901f26b39250d6"
 dependencies = [
  "blake3",
  "data-encoding",
@@ -1835,7 +1836,7 @@ checksum = "e6d5a32815ae3f33302d95fdcb2ce17862f8c65363dcfd29360480ba1001fc9c"
 
 [[package]]
 name = "furumusic"
-version = "0.9.4"
+version = "0.9.6"
 dependencies = [
  "anyhow",
  "async-stream",
@@ -3613,8 +3614,9 @@ dependencies = [
 
 [[package]]
 name = "music-dht"
-version = "0.2.0"
-source = "git+https://gt.hexor.cy/ab/frid.git#8de7d1292708fa0b225e5a4a9d5ab4f0676202d3"
+version = "0.3.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "5f32daa9edf769fb5686e92ae6884e9fda6ea082452e4208309fc14301e26aef"
 dependencies = [
  "async-trait",
  "blake3",

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "furumusic"
-version = "0.9.5"
+version = "0.9.7"
 edition = "2024"
 description = "Reusable web-app boilerplate: auth, OIDC/SSO, admin panel, user management, i18n, PostgreSQL"
 
@@ -39,4 +39,4 @@ uuid = "1"
 librqbit = { version = "8.1.1", features = ["disable-upload"] }
 # P2P federation: publishes the library into a shared DHT and serves audio /
 # catalogs to furumi peers (TUI clients) over the frid stack.
-music-dht = { git = "https://gt.hexor.cy/ab/frid.git" }
+music-dht = "0.3"

src/federation/capabilities.rs

@@ -0,0 +1,66 @@
+//! Informational publication of the protocol versions exposed by this peer.
+
+use std::time::Duration;
+
+use anyhow::Result;
+use music_dht::StreamAcceptor;
+use music_dht::capabilities::{
+    CAPABILITIES_PROTOCOL_VERSION, CapabilityManifest, CapabilityMessage, JAM_ID, read_message,
+    write_message,
+};
+
+use super::serve::AUDIO_PROTOCOL_VERSION;
+
+fn local_manifest() -> CapabilityManifest {
+    CapabilityManifest::frid("furumusic", env!("CARGO_PKG_VERSION"))
+        // The web server does not expose federation Jam yet.
+        .without_protocol(JAM_ID)
+        .with_protocol("audio", AUDIO_PROTOCOL_VERSION)
+}
+
+pub async fn serve(mut acceptor: StreamAcceptor) {
+    while let Some(stream) = acceptor.accept().await {
+        tokio::spawn(async move {
+            if let Err(error) = serve_one(stream).await {
+                tracing::debug!("capability stream failed: {error:#}");
+            }
+        });
+    }
+}
+
+async fn serve_one(mut stream: music_dht::ByteStream) -> Result<()> {
+    let response = match read_message(&mut stream).await? {
+        CapabilityMessage::Get {
+            version: CAPABILITIES_PROTOCOL_VERSION,
+        } => CapabilityMessage::Manifest {
+            manifest: local_manifest(),
+        },
+        CapabilityMessage::Get { version } => CapabilityMessage::Error {
+            message: format!("unsupported capability protocol {version}"),
+        },
+        _ => CapabilityMessage::Error {
+            message: "expected capability request".to_string(),
+        },
+    };
+    write_message(&mut stream, &response).await?;
+    stream.send.finish()?;
+    let _ = tokio::time::timeout(Duration::from_secs(2), stream.send.stopped()).await;
+    Ok(())
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn manifest_describes_only_supported_player_protocols() {
+        let manifest = local_manifest();
+        assert_eq!(manifest.application, "furumusic");
+        assert_eq!(
+            manifest.protocols.get("audio"),
+            Some(&AUDIO_PROTOCOL_VERSION)
+        );
+        assert!(!manifest.protocols.contains_key(JAM_ID));
+        manifest.validate().unwrap();
+    }
+}

src/federation/mod.rs

@@ -12,6 +12,7 @@
 //! `federation_network_id`, `federation_save_on_listen`) and apply on the fly — saving the settings
 //! starts, stops or re-joins the node without a server restart.
 
+mod capabilities;
 pub mod client;
 pub mod devices;
 mod receive;
@@ -24,6 +25,7 @@ use std::sync::{Arc, OnceLock};
 use std::time::Duration;
 
 use anyhow::{Context, Result};
+use music_dht::capabilities::CAPABILITIES_ALPN;
 use music_dht::{
     ByteStream, ByteStreamConnectionStats, ItemKind, ItemSpec, MusicDhtConfig, MusicDhtService,
     NetworkId, PeerTicket, PublishStats, RendezvousConfig, SyncStats,
@@ -380,6 +382,7 @@ impl Federation {
             .stream_protocol(AUDIO_ALPN)
             .stream_protocol(CATALOG_ALPN)
             .stream_protocol(devices::SYNC_ALPN)
+            .schema_independent_stream_protocol(CAPABILITIES_ALPN)
             .build()
             .map_err(|err| anyhow::anyhow!("invalid federation config: {err}"))?;
         let (service, mut events) =
@@ -447,6 +450,10 @@ impl Federation {
             device_hub,
             Arc::clone(&self.transport_stats),
         ));
+        let capabilities_acceptor = service
+            .stream_acceptor(CAPABILITIES_ALPN)
+            .map_err(|err| anyhow::anyhow!("failed to take the capabilities acceptor: {err}"))?;
+        let capabilities_task = tokio::spawn(capabilities::serve(capabilities_acceptor));
 
         *guard = Some(Running {
             service,
@@ -458,6 +465,7 @@ impl Federation {
                 catalog_task,
                 device_task,
                 device_sync_task,
+                capabilities_task,
             ],
         });
         self.set_error(None);

src/federation/serve.rs

@@ -21,6 +21,8 @@ use super::{TransportStats, record_stream_transport};
 
 /// ALPN of the peer-to-peer audio streaming protocol.
 pub const AUDIO_ALPN: &[u8] = b"furumi-fd/audio/1";
+/// Version of the peer-to-peer audio streaming protocol.
+pub const AUDIO_PROTOCOL_VERSION: u16 = 1;
 
 /// Maximum size of a JSON protocol line (request or response header).
 const MAX_PROTOCOL_LINE: usize = 4096;

src/player/mod.rs

@@ -5602,19 +5602,62 @@ async fn history_list_handler(
     .await
     .map_err(|e| cot::Error::internal(e.to_string()))?;
 
+    // Listen metadata is an immutable snapshot. Once federated content has
+    // been materialized, resolve it through the content cache and use the
+    // normal local TrackItem as the presentation/action authority.
+    let content_ids = rows
+        .iter()
+        .map(|row| row.get::<String, _>("content_id"))
+        .collect::<Vec<_>>();
+    let local_rows = if content_ids.is_empty() {
+        Vec::new()
+    } else {
+        sqlx::query(
+            r#"SELECT DISTINCT ON (c.content_id) c.content_id, t.id AS track_id
+                 FROM furumusic__federation_content_id_cache c
+                 JOIN furumusic__media_file m
+                   ON m.id = c.media_file_id AND m.sha256_hash = c.sha256_hash
+                 JOIN furumusic__track t ON t.audio_file_id = m.id
+                 JOIN furumusic__release r ON r.id = t.release_id
+                WHERE c.content_id = ANY($1)
+                  AND t.is_hidden = false
+                  AND r.is_hidden = false
+                ORDER BY c.content_id, t.id"#,
+        )
+        .bind(&content_ids)
+        .fetch_all(pool)
+        .await
+        .map_err(|e| cot::Error::internal(e.to_string()))?
+    };
+    let local_by_content = local_rows
+        .into_iter()
+        .map(|row| {
+            (
+                row.get::<String, _>("content_id"),
+                row.get::<i64, _>("track_id"),
+            )
+        })
+        .collect::<HashMap<_, _>>();
+    let local_ids = local_by_content.values().copied().collect::<Vec<_>>();
+    let local_tracks = load_track_items_by_ids(pool, &local_ids)
+        .await?
+        .into_iter()
+        .map(|track| (track.id, track))
+        .collect::<HashMap<_, _>>();
+
     let items = rows
         .into_iter()
         .map(|row| {
             let listen_id: String = row.get("listen_id");
             let content_id: String = row.get("content_id");
-            let local_track_id: Option<i64> = row.get("local_track_id");
+            let local_track_id = local_by_content.get(&content_id).copied();
             let metadata: serde_json::Value = row.get("metadata_json");
-            let title = metadata
+            let snapshot_title = metadata
                 .get("title")
                 .and_then(serde_json::Value::as_str)
                 .unwrap_or("Unknown track")
                 .to_string();
-            let release_title = metadata
+            let snapshot_release_title = metadata
                 .get("release_title")
                 .and_then(serde_json::Value::as_str)
                 .map(ToOwned::to_owned);
@@ -5635,20 +5678,20 @@ async fn history_list_handler(
                 row.get("release_cover_file_id"),
                 "medium",
             );
-            let track = serde_json::json!({
+            let fallback_track = serde_json::json!({
                 "id": local_track_id.map_or_else(
                     || format!("history:{content_id}"),
                     |id| id.to_string(),
                 ),
                 "content_id": content_id,
-                "title": title,
+                "title": snapshot_title,
                 "track_number": null,
                 "disc_number": null,
                 "duration_seconds": duration_ms.unwrap_or_default() as f64 / 1000.0,
                 "artists": artist_values("artist_names"),
                 "featured_artists": artist_values("featured_artist_names"),
                 "release_id": release_id,
-                "release_title": release_title.clone().unwrap_or_default(),
+                "release_title": snapshot_release_title.clone().unwrap_or_default(),
                 "release_year": row.get::<Option<i32>, _>("release_year"),
                 "cover_url": cover_url,
                 "stream_url": local_track_id
@@ -5670,6 +5713,16 @@ async fn history_list_handler(
                     None
                 },
             });
+            let local_track = local_track_id.and_then(|id| local_tracks.get(&id));
+            let track = local_track
+                .and_then(|track| serde_json::to_value(track).ok())
+                .unwrap_or(fallback_track);
+            let title = local_track
+                .map(|track| track.title.clone())
+                .unwrap_or(snapshot_title);
+            let release_title = local_track
+                .map(|track| track.release_title.clone())
+                .or(snapshot_release_title);
             let started_at_ms: i64 = row.get("started_at_ms");
             let played_at = chrono::DateTime::from_timestamp_millis(started_at_ms)
                 .unwrap_or_else(chrono::Utc::now)

templates/base.html

@@ -3,6 +3,7 @@
 <head>
     <meta charset="UTF-8">
     <meta name="viewport" content="width=device-width, initial-scale=1.0">
+    <link rel="icon" type="image/svg+xml" href="data:image/svg+xml,%3Csvg xmlns='http://www.w3.org/2000/svg' viewBox='0 0 64 64'%3E%3Crect width='64' height='64' rx='14' fill='%23111827'/%3E%3Cpath d='M27 15v31.5a9 9 0 1 1-5-8.1V22l27-6v24.5a9 9 0 1 1-5-8.1V15.9L27 20.5' fill='%2367e8f9'/%3E%3C/svg%3E">
     <title>{% block title %}{{ t.site_name }}{% endblock title %}</title>
     {% block head_extra %}{% endblock head_extra %}
 </head>

templates/player/scripts.html

@@ -735,6 +735,7 @@ document.addEventListener('alpine:init', () => {
         _remoteStateReceivedAt: 0,
         _remoteStateTimer: null,
         _prefetchedQueueKey: null,
+        _baseDocumentTitle: '',
 
         _hasInitialShareLink() {
             const params = new URLSearchParams(window.location.search);
@@ -744,6 +745,10 @@ document.addEventListener('alpine:init', () => {
 
         init() {
             audio.volume = this.volume;
+            this._baseDocumentTitle = document.title.replace(/^▶\s*/, '');
+            Alpine.effect(() => {
+                document.title = `${this.isPlaying ? '▶ ' : ''}${this._baseDocumentTitle}`;
+            });
 
             audio.addEventListener('timeupdate', () => {
                 this.currentTime = audio.currentTime;
@@ -4019,6 +4024,9 @@ document.addEventListener('alpine:init', () => {
                         playable,
                     });
                     this.refreshMaterializedArtwork(track);
+                    Alpine.store('likes')?.reload();
+                    const history = Alpine.store('history');
+                    if (history?.modal) history.load(history.page);
                 }
                 const finalProgress = this.federationPreparing[contentId] || {};
                 const queueIndex = Number.isInteger(finalProgress.queueIndex)
@@ -4129,10 +4137,16 @@ document.addEventListener('alpine:init', () => {
         _set: new Set(),
 
         init() {
-            fetch('/api/player/likes')
-                .then(r => r.json())
-                .then(d => { this._set = new Set(d.track_ids || []); })
-                .catch(() => {});
+            this.reload();
+        },
+
+        async reload() {
+            try {
+                const response = await fetch('/api/player/likes');
+                if (!response.ok) return;
+                const data = await response.json();
+                this._set = new Set(data.track_ids || []);
+            } catch {}
         },
 
         has(trackId) {