fix: enhance connection management with improved error handling and cleanup logic

Signed-off-by: Deluan <deluan@navidrome.org>

plugins/examples/discord-rich-presence-rs/src/lib.rs

@@ -208,14 +208,23 @@ impl CallbackProvider for DiscordPlugin {
         match req.payload.as_str() {
             PAYLOAD_HEARTBEAT => {
                 // Heartbeat callback - schedule_id is the username
-                rpc::handle_heartbeat_callback(&req.schedule_id)
-                    .map_err(|e| SchedulerError::new(e.to_string()))?;
+                if let Err(e) = rpc::handle_heartbeat_callback(&req.schedule_id) {
+                    // On heartbeat failure, clean up the connection (like the original Go plugin)
+                    // The next NowPlaying call will reconnect if needed
+                    warn!("Heartbeat failed for user {}, cleaning up connection: {:?}", req.schedule_id, e);
+                    rpc::cleanup_connection(&req.schedule_id);
+                    return Err(SchedulerError::new(format!("heartbeat failed, connection cleaned up: {}", e)));
+                }
             }
             PAYLOAD_CLEAR_ACTIVITY => {
                 // Clear activity callback - schedule_id is "username-clear"
                 let username = req.schedule_id.trim_end_matches("-clear");
+                info!("Removing presence for user {}", username);
                 rpc::handle_clear_activity_callback(username)
                     .map_err(|e| SchedulerError::new(e.to_string()))?;
+                info!("Disconnecting user {}", username);
+                rpc::disconnect(username)
+                    .map_err(|e| SchedulerError::new(e.to_string()))?;
             }
             _ => {
                 warn!("Unknown scheduler callback payload: {}", req.payload);
@@ -251,6 +260,8 @@ impl ErrorProvider for DiscordPlugin {
             "WebSocket error for connection '{}': {}",
             req.connection_id, req.error
         );
+        // Clean up all state associated with this connection since it's likely broken
+        rpc::handle_connection_close(&req.connection_id);
         Ok(())
     }
 }
@@ -261,6 +272,8 @@ impl CloseProvider for DiscordPlugin {
             "WebSocket connection '{}' closed with code {}: {}",
             req.connection_id, req.code, req.reason
         );
+        // Clean up all state associated with this connection
+        rpc::handle_connection_close(&req.connection_id);
         Ok(())
     }
 }

plugins/examples/discord-rich-presence-rs/src/rpc.rs

@@ -76,12 +76,6 @@ struct GatewayMessage<T> {
     d: T,
 }
 
-#[derive(Deserialize)]
-struct HelloMessage {
-    #[allow(dead_code)]
-    heartbeat_interval: i64,
-}
-
 #[derive(Deserialize)]
 struct GatewayResponse {
     op: i32,
@@ -123,8 +117,9 @@ fn is_connected(username: &str) -> bool {
     }
 }
 
-/// Cleans up a failed connection for a user.
-fn cleanup_connection(username: &str) {
+/// Cleans up a connection for a user.
+/// Called when heartbeat fails or connection is lost.
+pub fn cleanup_connection(username: &str) {
     info!("Cleaning up failed connection for user {}", username);
     
     // Cancel the heartbeat schedule
@@ -139,6 +134,9 @@ fn cleanup_connection(username: &str) {
             if let Err(e) = websocket::close_connection(&conn_id, 1000, "Reconnecting") {
                 trace!("Failed to close WebSocket for user {}: {:?}", username, e);
             }
+            // Clean up reverse mapping
+            let reverse_key = format!("discord.reverse.{}", conn_id);
+            let _ = cache::remove(&reverse_key);
         }
     }
     
@@ -149,6 +147,39 @@ fn cleanup_connection(username: &str) {
     info!("Cleaned up connection for user {}", username);
 }
 
+/// Handles connection close by connection ID (called from WebSocket close callback).
+/// This cleans up all state associated with the connection.
+pub fn handle_connection_close(connection_id: &str) {
+    // Find the username for this connection using the reverse mapping
+    if let Ok(Some(username)) = find_username_for_connection(connection_id) {
+        info!("Connection closed for user {}, cleaning up", username);
+        
+        // Cancel the heartbeat schedule
+        if let Err(e) = scheduler::cancel_schedule(&username) {
+            // Not an error if schedule doesn't exist
+            trace!("Failed to cancel heartbeat schedule for user {}: {:?}", username, e);
+        }
+        
+        // Cancel any pending clear-activity schedule
+        let _ = scheduler::cancel_schedule(&format!("{}-clear", username));
+        
+        // Clean up cache entries
+        let conn_key = connection_key(&username);
+        let _ = cache::remove(&conn_key);
+        let _ = cache::remove(&sequence_key(&username));
+        
+        // Clean up reverse mapping
+        let reverse_key = format!("discord.reverse.{}", connection_id);
+        let _ = cache::remove(&reverse_key);
+        
+        info!("Cleaned up connection state for user {}", username);
+    } else {
+        // Just clean up the reverse mapping if we can't find the username
+        let reverse_key = format!("discord.reverse.{}", connection_id);
+        let _ = cache::remove(&reverse_key);
+    }
+}
+
 /// Connects to the Discord gateway for a user.
 pub fn connect(username: &str, token: &str) -> Result<(), Error> {
     // Check if already connected and connection is valid
@@ -165,10 +196,14 @@ pub fn connect(username: &str, token: &str) -> Result<(), Error> {
     // Store token for later use
     cache::set_string(&token_key(username), token, 86400)?;
 
+    // Get Discord Gateway URL
+    let gateway = get_discord_gateway()?;
+    info!("Using gateway: {}", gateway);
+
     // Connect to Discord gateway
     let headers = std::collections::HashMap::new();
     let conn_id = websocket::connect(
-        "wss://gateway.discord.gg/?v=10&encoding=json",
+        &gateway,
         headers,
         username, // Use username as connection ID for easy lookup
     )?;
@@ -237,7 +272,7 @@ pub fn handle_clear_activity_callback(username: &str) -> Result<(), Error> {
                 d: PresencePayload {
                     activities: vec![],
                     since: 0,
-                    status: "online".to_string(),
+                    status: "dnd".to_string(),
                     afk: false,
                 },
             };
@@ -252,6 +287,35 @@ pub fn handle_clear_activity_callback(username: &str) -> Result<(), Error> {
     Ok(())
 }
 
+/// Disconnects from Discord for a user.
+pub fn disconnect(username: &str) -> Result<(), Error> {
+    info!("Disconnecting from Discord for user {}", username);
+    
+    // Cancel the heartbeat schedule
+    if let Err(e) = scheduler::cancel_schedule(username) {
+        warn!("Failed to cancel heartbeat schedule: {:?}", e);
+    }
+    
+    // Close the WebSocket connection
+    let conn_key = connection_key(username);
+    if let Ok((conn_id, exists)) = cache::get_string(&conn_key) {
+        if exists && !conn_id.is_empty() {
+            if let Err(e) = websocket::close_connection(&conn_id, 1000, "Navidrome disconnect") {
+                warn!("Failed to close WebSocket connection: {:?}", e);
+            }
+            // Clean up reverse mapping
+            let reverse_key = format!("discord.reverse.{}", conn_id);
+            let _ = cache::remove(&reverse_key);
+        }
+    }
+    
+    // Clean up cache entries
+    let _ = cache::remove(&conn_key);
+    let _ = cache::remove(&sequence_key(username));
+    
+    Ok(())
+}
+
 /// Sends an activity update to Discord.
 pub fn send_activity(
     client_id: &str,
@@ -274,7 +338,7 @@ pub fn send_activity(
         d: PresencePayload {
             activities: vec![activity],
             since: 0,
-            status: "online".to_string(),
+            status: "dnd".to_string(),
             afk: false,
         },
     };
@@ -305,6 +369,27 @@ fn find_username_for_connection(connection_id: &str) -> Result<Option<String>, E
     Ok(None)
 }
 
+fn get_discord_gateway() -> Result<String, Error> {
+    let req = HttpRequest::new("https://discord.com/api/gateway")
+        .with_method("GET");
+
+    let resp = http::request::<String>(&req, None::<String>)?;
+    if resp.status_code() >= 400 {
+        return Err(Error::msg(format!(
+            "Failed to get Discord gateway: HTTP {}",
+            resp.status_code()
+        )));
+    }
+
+    let body = resp.body();
+    let data: std::collections::HashMap<String, String> = serde_json::from_slice(&body)
+        .map_err(|e| Error::msg(format!("Failed to parse gateway response: {}", e)))?;
+
+    data.get("url")
+        .map(|url| url.to_string())
+        .ok_or_else(|| Error::msg("No URL in gateway response"))
+}
+
 fn identify(username: &str) -> Result<(), Error> {
     info!("Identifying with Discord for user {}", username);
 
@@ -331,9 +416,9 @@ fn identify(username: &str) -> Result<(), Error> {
             token,
             intents: 0,
             properties: IdentifyProperties {
-                os: "navidrome".to_string(),
-                browser: "navidrome".to_string(),
-                device: "navidrome".to_string(),
+                os: "Windows 10".to_string(),
+                browser: "Discord Client".to_string(),
+                device: "Discord Client".to_string(),
             },
         },
     };

plugins/manager_watcher.go

@@ -91,7 +91,7 @@ func (m *Manager) handleWatcherEvent(event notify.EventInfo) {
 
 	pluginName := strings.TrimSuffix(filepath.Base(path), PackageExtension)
 
-	log.Debug(m.ctx, "Plugin file event", "plugin", pluginName, "event", event.Event(), "path", path)
+	log.Trace(m.ctx, "Plugin file event", "plugin", pluginName, "event", event.Event(), "path", path)
 
 	// Debounce: cancel any pending timer for this plugin and start a new one
 	m.debounceMu.Lock()