improvements to plex connection management (#2020)

CHANGELOG.md

@@ -25,6 +25,11 @@ The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/).
 
 ### Changed
 - Start to make UI minimally responsive (functional on smaller screens)
+- Change how ETV determines which address to use for Plex connections
+  - The active Plex connection (address) will only be cached for 30 seconds
+  - When the connection is no longer cached, a ping will be sent to the last used address for Plex (the last address that had a successful ping)
+  - If the ping is successful, the address will be cached for another 30 seconds
+  - If the ping is not successful, all addresses will be checked again, and the first address to return a successful ping will be cached for 30 seconds
 
 ### Fixed
 - Fix error message about synchronizing Plex collections from a Plex server that has zero collections

ErsatzTV.Application/Plex/Commands/SynchronizePlexMediaSourcesHandler.cs

@@ -10,7 +10,7 @@ using Microsoft.Extensions.Logging;
 
 namespace ErsatzTV.Application.Plex;
 
-public class SynchronizePlexMediaSourcesHandler : IRequestHandler<SynchronizePlexMediaSources,
+public class SynchronizePlexMediaSourcesHandler : PlexBaseConnectionHandler, IRequestHandler<SynchronizePlexMediaSources,
     Either<BaseError, List<PlexMediaSource>>>
 {
     private const string LocalhostUri = "http://localhost:32400";
@@ -31,6 +31,7 @@ public class SynchronizePlexMediaSourcesHandler : IRequestHandler<SynchronizePle
         ChannelWriter<IScannerBackgroundServiceRequest> channel,
         IEntityLocker entityLocker,
         ILogger<SynchronizePlexMediaSourcesHandler> logger)
+        : base(plexServerApiClient, mediaSourceRepository, logger)
     {
         _mediaSourceRepository = mediaSourceRepository;
         _plexTvApiClient = plexTvApiClient;
@@ -102,63 +103,35 @@ public class SynchronizePlexMediaSourcesHandler : IRequestHandler<SynchronizePle
             var toRemove = existing.Connections
                 .Filter(connection => server.Connections.All(c => c.Uri != connection.Uri)).ToList();
             await _mediaSourceRepository.Update(existing, toAdd, toRemove);
-            await FindConnectionToActivate(existing);
+            Option<PlexServerAuthToken> maybeToken = await _plexSecretStore.GetServerAuthToken(server.ClientIdentifier);
+            if (maybeToken.IsNone)
+            {
+                _logger.LogError(
+                    "Unable to activate Plex connection for server {Server} without auth token",
+                    server.ServerName);
+            }
+
+            foreach (PlexServerAuthToken token in maybeToken)
+            {
+                await FindConnectionToActivate(existing, token);
+            }
         }
 
         if (maybeExisting.IsNone)
         {
             await _mediaSourceRepository.Add(server);
-            await FindConnectionToActivate(server);
+            Option<PlexServerAuthToken> maybeToken = await _plexSecretStore.GetServerAuthToken(server.ClientIdentifier);
-        }
+            if (maybeToken.IsNone)
-    }
-
-    private async Task FindConnectionToActivate(PlexMediaSource server)
-    {
-        var prioritized = server.Connections
-            .OrderByDescending(pc => pc.Uri == LocalhostUri)
-            .ThenByDescending(pc => pc.IsActive)
-            .ToList();
-
-        foreach (PlexConnection connection in server.Connections)
-        {
-            connection.IsActive = false;
-        }
-
-        Option<PlexServerAuthToken> maybeToken = await _plexSecretStore.GetServerAuthToken(server.ClientIdentifier);
-        foreach (PlexServerAuthToken token in maybeToken)
-        {
-            foreach (PlexConnection connection in prioritized)
             {
-                try
+                _logger.LogError(
-                {
+                    "Unable to activate Plex connection for server {Server} without auth token",
-                    _logger.LogDebug("Attempting to locate to Plex at {Uri}", connection.Uri);
+                    server.ServerName);
-                    if (await _plexServerApiClient.Ping(connection, token))
-                    {
-                        _logger.LogInformation("Located Plex at {Uri}", connection.Uri);
-                        connection.IsActive = true;
-                        break;
-                    }
-                }
-                catch
-                {
-                    // do nothing
-                }
             }
-        }
 
-        if (maybeToken.IsNone)
+            foreach (PlexServerAuthToken token in maybeToken)
-        {
+            {
-            _logger.LogError(
+                await FindConnectionToActivate(server, token);
-                "Unable to activate Plex connection for server {Server} without auth token",
+            }
-                server.ServerName);
-        }
-
-        if (server.Connections.All(c => !c.IsActive))
-        {
-            _logger.LogError("Unable to locate Plex");
-            server.Connections.Head().IsActive = true;
         }
-
-        await _mediaSourceRepository.Update(server, new List<PlexConnection>(), new List<PlexConnection>());
     }
 }

ErsatzTV.Application/Plex/Commands/TryCompletePlexPinFlowHandler.cs

@@ -20,14 +20,15 @@ public class TryCompletePlexPinFlowHandler : IRequestHandler<TryCompletePlexPinF
     public async Task<Either<BaseError, bool>>
         Handle(TryCompletePlexPinFlow request, CancellationToken cancellationToken)
     {
-        var cts = new CancellationTokenSource(TimeSpan.FromMinutes(2));
+        using var cts = new CancellationTokenSource(TimeSpan.FromMinutes(2));
-        CancellationToken token = cts.Token;
+        using var linkedTokenSource = CancellationTokenSource.CreateLinkedTokenSource(cts.Token, cancellationToken);
+        CancellationToken token = linkedTokenSource.Token;
         while (!token.IsCancellationRequested)
         {
             bool result = await _plexTvApiClient.TryCompletePinFlow(request.AuthPin);
             if (result)
             {
-                await _channel.WriteAsync(new SynchronizePlexMediaSources(), cancellationToken);
+                await _channel.WriteAsync(new SynchronizePlexMediaSources(), token);
                 return true;
             }
 

ErsatzTV.Application/Plex/PlexBaseConnectionHandler.cs

@@ -0,0 +1,93 @@
+using System.Collections.Concurrent;
+using ErsatzTV.Core.Domain;
+using ErsatzTV.Core.Interfaces.Plex;
+using ErsatzTV.Core.Interfaces.Repositories;
+using ErsatzTV.Core.Plex;
+using Microsoft.Extensions.Logging;
+
+namespace ErsatzTV.Application.Plex;
+
+public abstract class PlexBaseConnectionHandler(
+    IPlexServerApiClient plexServerApiClient,
+    IMediaSourceRepository mediaSourceRepository,
+    ILogger logger)
+{
+    protected async Task<Option<PlexConnection>> FindConnectionToActivate(PlexMediaSource server, PlexServerAuthToken token)
+    {
+        Option<PlexConnection> result = Option<PlexConnection>.None;
+
+        foreach (PlexConnection connection in server.Connections)
+        {
+            connection.IsActive = false;
+        }
+
+        using var cts = new CancellationTokenSource();
+
+        ConcurrentDictionary<PlexConnection, TimeSpan> successfulTimes = new();
+        var tasks = server.Connections
+            .Map(connection => PingPlexConnection(connection, token, successfulTimes, cts.Token))
+            .ToList();
+
+        while (tasks.Count > 0)
+        {
+            Task completed = await Task.WhenAny(tasks);
+            if (completed.IsCompletedSuccessfully)
+            {
+                if (!successfulTimes.IsEmpty)
+                {
+                    await cts.CancelAsync();
+                    break;
+                }
+            }
+
+            tasks.Remove(completed);
+        }
+
+        Option<PlexConnection> maybeBest = successfulTimes.OrderByDescending(kv => kv.Value).Select(kvp => kvp.Key).HeadOrNone();
+        foreach (PlexConnection connection in maybeBest)
+        {
+            connection.IsActive = true;
+        }
+
+        if (server.Connections.All(c => !c.IsActive))
+        {
+            logger.LogError("Unable to locate Plex");
+            server.Connections.Head().IsActive = true;
+        }
+
+        await mediaSourceRepository.Update(server, [], []);
+
+        return result;
+    }
+
+    private async Task PingPlexConnection(PlexConnection connection, PlexServerAuthToken token, ConcurrentDictionary<PlexConnection, TimeSpan> successfulTimes, CancellationToken cancellationToken)
+    {
+        try
+        {
+            logger.LogDebug("Attempting to locate to Plex at {Uri}", connection.Uri);
+            var sw = new System.Diagnostics.Stopwatch();
+            sw.Start();
+            bool pingResult = await plexServerApiClient.Ping(connection, token, cancellationToken);
+            sw.Stop();
+            if (pingResult)
+            {
+                logger.LogInformation(
+                    "Located Plex at {Uri} in {Milliseconds} ms",
+                    connection.Uri,
+                    sw.ElapsedMilliseconds);
+                successfulTimes.TryAdd(connection, sw.Elapsed);
+            }
+            else
+            {
+                logger.LogDebug(
+                    "Unable to locate Plex at {Uri} after {Milliseconds} ms",
+                    connection.Uri,
+                    sw.ElapsedMilliseconds);
+            }
+        }
+        catch
+        {
+            // do nothing
+        }
+    }
+}

ErsatzTV.Application/Plex/Queries/GetPlexConnectionParametersHandler.cs

@@ -4,24 +4,32 @@ using ErsatzTV.Core.Interfaces.Plex;
 using ErsatzTV.Core.Interfaces.Repositories;
 using ErsatzTV.Core.Plex;
 using Microsoft.Extensions.Caching.Memory;
+using Microsoft.Extensions.Logging;
 
 namespace ErsatzTV.Application.Plex;
 
-public class GetPlexConnectionParametersHandler : IRequestHandler<GetPlexConnectionParameters,
+public class GetPlexConnectionParametersHandler : PlexBaseConnectionHandler, IRequestHandler<GetPlexConnectionParameters,
     Either<BaseError, PlexConnectionParametersViewModel>>
 {
     private readonly IMediaSourceRepository _mediaSourceRepository;
     private readonly IMemoryCache _memoryCache;
+    private readonly IPlexServerApiClient _plexServerApiClient;
     private readonly IPlexSecretStore _plexSecretStore;
+    private readonly ILogger<GetPlexConnectionParametersHandler> _logger;
 
     public GetPlexConnectionParametersHandler(
         IMemoryCache memoryCache,
+        IPlexServerApiClient plexServerApiClient,
         IMediaSourceRepository mediaSourceRepository,
-        IPlexSecretStore plexSecretStore)
+        IPlexSecretStore plexSecretStore,
+        ILogger<GetPlexConnectionParametersHandler> logger)
+        : base(plexServerApiClient, mediaSourceRepository, logger)
     {
         _memoryCache = memoryCache;
+        _plexServerApiClient = plexServerApiClient;
         _mediaSourceRepository = mediaSourceRepository;
         _plexSecretStore = plexSecretStore;
+        _logger = logger;
     }
 
     public async Task<Either<BaseError, PlexConnectionParametersViewModel>> Handle(
@@ -33,55 +41,43 @@ public class GetPlexConnectionParametersHandler : IRequestHandler<GetPlexConnect
             return parameters;
         }
 
-        Either<BaseError, PlexConnectionParametersViewModel> maybeParameters =
+        Option<PlexMediaSource> maybeMediaSource = await _mediaSourceRepository.GetPlex(request.PlexMediaSourceId);
-            await Validate(request)
+        foreach (PlexMediaSource mediaSource in maybeMediaSource)
-                .MapT(
+        {
-                    cp => new PlexConnectionParametersViewModel(
+            Option<PlexServerAuthToken> maybeToken =
-                        new Uri(cp.ActiveConnection.Uri),
+                await _plexSecretStore.GetServerAuthToken(mediaSource.ClientIdentifier);
-                        cp.PlexServerAuthToken.AuthToken))
+            foreach (PlexServerAuthToken token in maybeToken)
-                .Map(v => v.ToEither<PlexConnectionParametersViewModel>());
-
-        return maybeParameters.Match(
-            p =>
             {
-                _memoryCache.Set(request, p, TimeSpan.FromHours(1));
+                // try to keep the same connection
-                return maybeParameters;
+                Option<PlexConnection> maybeActiveConnection = mediaSource.Connections.Filter(c => c.IsActive).HeadOrNone();
-            },
+                foreach (PlexConnection activeConnection in maybeActiveConnection)
-            error => error);
+                {
-    }
+                    if (await _plexServerApiClient.Ping(activeConnection, token, cancellationToken))
+                    {
+                        _logger.LogDebug("Plex connection is still active at {Uri}", activeConnection.Uri);
+                        var p = new PlexConnectionParametersViewModel(new Uri(activeConnection.Uri), token.AuthToken);
+                        _memoryCache.Set(request, p, TimeSpan.FromSeconds(30));
+                        return p;
+                    }
+                }
 
-    private Task<Validation<BaseError, ConnectionParameters>> Validate(GetPlexConnectionParameters request) =>
+                _logger.LogInformation("Plex connection is no longer active, searching for a new connection");
-        PlexMediaSourceMustExist(request)
-            .BindT(MediaSourceMustHaveActiveConnection)
-            .BindT(MediaSourceMustHaveToken);
 
-    private Task<Validation<BaseError, PlexMediaSource>> PlexMediaSourceMustExist(
+                // check all connections for a working one
-        GetPlexConnectionParameters request) =>
+                Option<PlexConnection> maybeConnection = await FindConnectionToActivate(mediaSource, token);
-        _mediaSourceRepository.GetPlex(request.PlexMediaSourceId)
+                foreach (PlexConnection connection in maybeConnection)
-            .Map(
+                {
-                v => v.ToValidation<BaseError>(
+                    var p = new PlexConnectionParametersViewModel(new Uri(connection.Uri), token.AuthToken);
-                    $"Plex media source {request.PlexMediaSourceId} does not exist."));
+                    _memoryCache.Set(request, p, TimeSpan.FromMinutes(30));
+                    return p;
+                }
 
-    private Validation<BaseError, ConnectionParameters> MediaSourceMustHaveActiveConnection(
+                return BaseError.New($"Plex media source {request.PlexMediaSourceId} requires an active connection");
-        PlexMediaSource plexMediaSource)
+            }
-    {
-        Option<PlexConnection> maybeConnection =
-            plexMediaSource.Connections.SingleOrDefault(c => c.IsActive);
-        return maybeConnection.Map(connection => new ConnectionParameters(plexMediaSource, connection))
-            .ToValidation<BaseError>("Plex media source requires an active connection");
-    }
 
-    private async Task<Validation<BaseError, ConnectionParameters>> MediaSourceMustHaveToken(
+            return BaseError.New($"Plex media source {request.PlexMediaSourceId} requires a token");
-        ConnectionParameters connectionParameters)
+        }
-    {
-        Option<PlexServerAuthToken> maybeToken = await
-            _plexSecretStore.GetServerAuthToken(connectionParameters.PlexMediaSource.ClientIdentifier);
-        return maybeToken.Map(token => connectionParameters with { PlexServerAuthToken = token })
-            .ToValidation<BaseError>("Plex media source requires a token");
-    }
 
-    private sealed record ConnectionParameters(PlexMediaSource PlexMediaSource, PlexConnection ActiveConnection)
+        return BaseError.New($"Plex media source {request.PlexMediaSourceId} does not exist.");
-    {
-        public PlexServerAuthToken PlexServerAuthToken { get; set; }
     }
 }

ErsatzTV.Core/Interfaces/Plex/IPlexServerApiClient.cs

@@ -7,7 +7,8 @@ public interface IPlexServerApiClient
 {
     Task<bool> Ping(
         PlexConnection connection,
-        PlexServerAuthToken token);
+        PlexServerAuthToken token,
+        CancellationToken cancellationToken);
 
     Task<Either<BaseError, List<PlexLibrary>>> GetLibraries(
         PlexConnection connection,

ErsatzTV.Infrastructure/Plex/IPlexServerApi.cs

@@ -9,7 +9,8 @@ public interface IPlexServerApi
     [Headers("Accept: text/xml")]
     public Task<PlexXmlMediaContainerPingResponse> Ping(
         [Query] [AliasAs("X-Plex-Token")]
-        string token);
+        string token,
+        CancellationToken cancellationToken);
 
     [Get("/library/sections")]
     [Headers("Accept: application/json")]

ErsatzTV.Infrastructure/Plex/PlexServerApiClient.cs

@@ -23,12 +23,12 @@ public class PlexServerApiClient : IPlexServerApiClient
         _logger = logger;
     }
 
-    public async Task<bool> Ping(PlexConnection connection, PlexServerAuthToken token)
+    public async Task<bool> Ping(PlexConnection connection, PlexServerAuthToken token, CancellationToken cancellationToken)
     {
         try
         {
             IPlexServerApi service = XmlServiceFor(connection.Uri, TimeSpan.FromSeconds(5));
-            PlexXmlMediaContainerPingResponse pingResult = await service.Ping(token.AuthToken);
+            PlexXmlMediaContainerPingResponse pingResult = await service.Ping(token.AuthToken, cancellationToken);
             return token.ClientIdentifier == pingResult.MachineIdentifier;
         }
         catch (Exception)

ErsatzTV/Startup.cs

@@ -559,7 +559,8 @@ public class Startup
         app.Use(
             async (context, next) =>
             {
-                if (!context.Request.Path.StartsWithSegments("/iptv") &&
+                if (!context.Request.Host.Value.StartsWith("localhost", StringComparison.OrdinalIgnoreCase) &&
+                    !context.Request.Path.StartsWithSegments("/iptv") &&
                     context.Connection.LocalPort != Settings.UiPort)
                 {
                     context.Response.StatusCode = 404;