From 3945aa7b58006a1f519522f71094d3f81daeb8d4 Mon Sep 17 00:00:00 2001 From: edde746 <86283021+edde746@users.noreply.github.com> Date: Sat, 22 Nov 2025 06:52:21 +0100 Subject: [PATCH] fix: multi-server connectivity optimization --- lib/services/multi_server_manager.dart | 190 +++++++++++++++++++- lib/services/server_connection_service.dart | 14 +- lib/services/storage_service.dart | 21 +++ 3 files changed, 219 insertions(+), 6 deletions(-) diff --git a/lib/services/multi_server_manager.dart b/lib/services/multi_server_manager.dart index 324ccacf..0799a6df 100644 --- a/lib/services/multi_server_manager.dart +++ b/lib/services/multi_server_manager.dart @@ -1,9 +1,12 @@ import 'dart:async'; +import 'package:connectivity_plus/connectivity_plus.dart'; + import '../client/plex_client.dart'; import '../config/plex_config.dart'; import '../utils/app_logger.dart'; import 'plex_auth_service.dart'; +import 'storage_service.dart'; /// Manages multiple Plex server connections simultaneously class MultiServerManager { @@ -22,6 +25,12 @@ class MultiServerManager { /// Stream of server status changes Stream> get statusStream => _statusController.stream; + /// Connectivity subscription for network monitoring + StreamSubscription>? _connectivitySubscription; + + /// Map of serverId to active optimization futures + final Map> _activeOptimizations = {}; + /// Get all registered server IDs List get serverIds => _servers.keys.toList(); @@ -100,14 +109,33 @@ class MultiServerManager { final baseUrl = workingConnection.uri; appLogger.d('Connected to ${server.name} at $baseUrl'); - // Create PlexClient with the working connection + // Get storage and load cached endpoint for this server + final storage = await StorageService.getInstance(); + final cachedEndpoint = storage.getServerEndpoint(serverId); + + // Create PlexClient with the working connection and failover support + final prioritizedEndpoints = server.prioritizedEndpointUrls( + preferredFirst: cachedEndpoint ?? baseUrl, + ); final config = await PlexConfig.create( baseUrl: baseUrl, token: server.accessToken, clientIdentifier: effectiveClientId, ); - final client = PlexClient(config); + final client = PlexClient( + config, + prioritizedEndpoints: prioritizedEndpoints, + onEndpointChanged: (newUrl) async { + await storage.saveServerEndpoint(serverId, newUrl); + appLogger.i( + 'Updated endpoint for ${server.name} after failover: $newUrl', + ); + }, + ); + + // Save the initial endpoint + await storage.saveServerEndpoint(serverId, baseUrl); // Store the client and server info _clients[serverId] = client; @@ -157,6 +185,11 @@ class MultiServerManager { 'Connected to $successCount/${servers.length} servers successfully', ); + // Start network monitoring if we have any connected servers + if (successCount > 0) { + startNetworkMonitoring(); + } + return successCount; } @@ -187,14 +220,33 @@ class MultiServerManager { final baseUrl = workingConnection.uri; - // Create PlexClient + // Get storage and load cached endpoint for this server + final storage = await StorageService.getInstance(); + final cachedEndpoint = storage.getServerEndpoint(serverId); + + // Create PlexClient with failover support + final prioritizedEndpoints = server.prioritizedEndpointUrls( + preferredFirst: cachedEndpoint ?? baseUrl, + ); final config = await PlexConfig.create( baseUrl: baseUrl, token: server.accessToken, clientIdentifier: effectiveClientId, ); - final client = PlexClient(config); + final client = PlexClient( + config, + prioritizedEndpoints: prioritizedEndpoints, + onEndpointChanged: (newUrl) async { + await storage.saveServerEndpoint(serverId, newUrl); + appLogger.i( + 'Updated endpoint for ${server.name} after failover: $newUrl', + ); + }, + ); + + // Save the initial endpoint + await storage.saveServerEndpoint(serverId, baseUrl); // Store _clients[serverId] = client; @@ -260,12 +312,142 @@ class MultiServerManager { await Future.wait(healthChecks); } + /// Start monitoring network connectivity for all servers + void startNetworkMonitoring() { + if (_connectivitySubscription != null) { + appLogger.d('Network monitoring already active'); + return; + } + + appLogger.i('Starting network monitoring for all servers'); + final connectivity = Connectivity(); + _connectivitySubscription = connectivity.onConnectivityChanged.listen( + (results) { + final status = results.isNotEmpty + ? results.first + : ConnectivityResult.none; + + if (status == ConnectivityResult.none) { + appLogger.w( + 'Connectivity lost, pausing optimization until network returns', + ); + return; + } + + appLogger.d( + 'Connectivity change detected, re-optimizing all servers', + error: { + 'status': status.name, + 'interfaces': results.map((r) => r.name).toList(), + 'serverCount': _servers.length, + }, + ); + + // Re-optimize all servers + _reoptimizeAllServers(reason: 'connectivity:${status.name}'); + }, + onError: (error, stackTrace) { + appLogger.w( + 'Connectivity listener error', + error: error, + stackTrace: stackTrace, + ); + }, + ); + } + + /// Stop monitoring network connectivity + void stopNetworkMonitoring() { + _connectivitySubscription?.cancel(); + _connectivitySubscription = null; + appLogger.i('Stopped network monitoring'); + } + + /// Re-optimize all connected servers + void _reoptimizeAllServers({required String reason}) { + for (final entry in _servers.entries) { + final serverId = entry.key; + final server = entry.value; + + // Skip if server is offline + if (!isServerOnline(serverId)) { + continue; + } + + // Skip if optimization already running for this server + if (_activeOptimizations.containsKey(serverId)) { + appLogger.d( + 'Optimization already running for ${server.name}, skipping', + error: {'reason': reason}, + ); + continue; + } + + // Run optimization + _activeOptimizations[serverId] = + _reoptimizeServer( + serverId: serverId, + server: server, + reason: reason, + ).whenComplete(() { + _activeOptimizations.remove(serverId); + }); + } + } + + /// Re-optimize connection for a specific server + Future _reoptimizeServer({ + required String serverId, + required PlexServer server, + required String reason, + }) async { + final storage = await StorageService.getInstance(); + final client = _clients[serverId]; + + try { + appLogger.d( + 'Starting connection optimization for ${server.name}', + error: {'reason': reason}, + ); + + await for (final connection in server.findBestWorkingConnection()) { + final newUrl = connection.uri; + + // Check if this is actually a better connection than current + if (client != null && client.config.baseUrl == newUrl) { + appLogger.d( + 'Already using optimal endpoint for ${server.name}: $newUrl', + ); + continue; + } + + // Save the new endpoint + await storage.saveServerEndpoint(serverId, newUrl); + + // If client has endpoint failover, it will automatically switch + // Otherwise, we might need to recreate the client (but failover should handle it) + appLogger.i( + 'Updated optimal endpoint for ${server.name}: $newUrl', + error: {'type': connection.displayType}, + ); + } + } catch (e, stackTrace) { + appLogger.w( + 'Connection optimization failed for ${server.name}', + error: e, + stackTrace: stackTrace, + ); + } + } + /// Disconnect all servers void disconnectAll() { appLogger.i('Disconnecting all servers'); + stopNetworkMonitoring(); _clients.clear(); _servers.clear(); _serverStatus.clear(); + _activeOptimizations.clear(); _statusController.add({}); } diff --git a/lib/services/server_connection_service.dart b/lib/services/server_connection_service.dart index 5f7d23f1..e226d06b 100644 --- a/lib/services/server_connection_service.dart +++ b/lib/services/server_connection_service.dart @@ -55,6 +55,7 @@ class ServerConnectionService { .findBestWorkingConnection() .asBroadcastStream(); PlexClient? client; + final serverId = server.clientIdentifier; final optimizationSubscription = connectionStream .skip(1) @@ -64,6 +65,7 @@ class ServerConnectionService { connection: connection, storage: storage, server: server, + serverId: serverId, client: client, reason: 'initial_latency_sweep', ); @@ -87,6 +89,7 @@ class ServerConnectionService { // Save server information to storage await storage.saveServerData(server.toJson()); await storage.saveServerUrl(connection.uri); + await storage.saveServerEndpoint(serverId, connection.uri); await storage.saveServerAccessToken(server.accessToken); // Save plex token if provided @@ -95,8 +98,9 @@ class ServerConnectionService { } // Create client with working connection + final cachedEndpoint = storage.getServerEndpoint(serverId); final prioritizedEndpoints = server.prioritizedEndpointUrls( - preferredFirst: connection.uri, + preferredFirst: cachedEndpoint ?? connection.uri, ); final config = await PlexConfig.create( baseUrl: connection.uri, @@ -108,6 +112,7 @@ class ServerConnectionService { prioritizedEndpoints: prioritizedEndpoints, onEndpointChanged: (newUrl) async { await storage.saveServerUrl(newUrl); + await storage.saveServerEndpoint(serverId, newUrl); appLogger.i( 'Updated stored server URL after failover', error: newUrl, @@ -265,6 +270,7 @@ class ServerConnectionService { required String reason, }) async { final storage = await StorageService.getInstance(); + final serverId = server.clientIdentifier; try { appLogger.d( 'Starting background connection optimization run', @@ -275,6 +281,7 @@ class ServerConnectionService { connection: connection, storage: storage, server: server, + serverId: serverId, client: client, reason: reason, ); @@ -292,13 +299,15 @@ class ServerConnectionService { required PlexConnection connection, required StorageService storage, required PlexServer server, + required String serverId, required PlexClient? client, required String reason, }) async { - final previousUrl = storage.getServerUrl(); + final previousUrl = storage.getServerEndpoint(serverId); final isNewEndpoint = previousUrl != connection.uri; await storage.saveServerUrl(connection.uri); + await storage.saveServerEndpoint(serverId, connection.uri); appLogger.d( 'Evaluated optimized endpoint candidate', error: { @@ -337,6 +346,7 @@ class ServerConnectionService { connection: upgraded, storage: storage, server: server, + serverId: serverId, client: client, reason: '$reason:https-upgrade', ); diff --git a/lib/services/storage_service.dart b/lib/services/storage_service.dart index 16aeeebd..ca257a49 100644 --- a/lib/services/storage_service.dart +++ b/lib/services/storage_service.dart @@ -54,6 +54,20 @@ class StorageService { return _prefs.getString(_keyServerUrl); } + // Per-Server Endpoint URL (for multi-server connection caching) + Future saveServerEndpoint(String serverId, String url) async { + await _prefs.setString('server_endpoint_$serverId', url); + LogRedactionManager.registerServerUrl(url); + } + + String? getServerEndpoint(String serverId) { + return _prefs.getString('server_endpoint_$serverId'); + } + + Future clearServerEndpoint(String serverId) async { + await _prefs.remove('server_endpoint_$serverId'); + } + // Server Access Token Future saveToken(String token) async { await _prefs.setString(_keyToken, token); @@ -377,10 +391,17 @@ class StorageService { /// Clear all multi-server data Future clearMultiServerData() async { + // Clear all server endpoint caches + final keys = _prefs.getKeys(); + final endpointKeys = keys.where( + (key) => key.startsWith('server_endpoint_'), + ); + await Future.wait([ clearServersList(), clearEnabledServers(), clearServerOrder(), + ...endpointKeys.map((key) => _prefs.remove(key)), ]); }