add new scanner process (#1080)

* start moving local scans to separate process

* send progress updates to main process

* move scanners and tests

* simplify dependencies; sync search index

* commit search index more often when scanning

* support forced scan and cancellation

* use scanner process for plex libraries

* update changelog

* update dockerfiles

* fix search index for local folder scanning

* rework plex scanners

* rework scanner handlers

* emby works again

* sync jellyfin

* cleanup

* update build

* update changelog

* remove scanner dependency in pr and artifacts workflows

* fix mac sed syntax

* fix pr build

ErsatzTV/Services/EmbyService.cs

@@ -4,6 +4,7 @@ using ErsatzTV.Application;
 using ErsatzTV.Application.Emby;
 using ErsatzTV.Core;
 using ErsatzTV.Core.Domain;
+using ErsatzTV.Core.Interfaces.Locking;
 using MediatR;
 
 namespace ErsatzTV.Services;
@@ -58,9 +59,6 @@ public class EmbyService : BackgroundService
                             break;
                         default:
                             throw new NotSupportedException($"Unsupported request type: {request.GetType().Name}");
-                        case SynchronizeEmbyCollections synchronizeEmbyCollections:
-                            requestTask = SynchronizeEmbyCollections(synchronizeEmbyCollections, cancellationToken);
-                            break;
                     }
 
                     await requestTask;
@@ -132,6 +130,7 @@ public class EmbyService : BackgroundService
     {
         using IServiceScope scope = _serviceScopeFactory.CreateScope();
         IMediator mediator = scope.ServiceProvider.GetRequiredService<IMediator>();
+        IEntityLocker entityLocker = scope.ServiceProvider.GetRequiredService<IEntityLocker>();
 
         Either<BaseError, string> result = await mediator.Send(request, cancellationToken);
         result.BiIter(
@@ -140,21 +139,10 @@ public class EmbyService : BackgroundService
                 "Unable to synchronize emby library {LibraryId}: {Error}",
                 request.EmbyLibraryId,
                 error.Value));
-    }
-
-    private async Task SynchronizeEmbyCollections(
-        SynchronizeEmbyCollections request,
-        CancellationToken cancellationToken)
-    {
-        using IServiceScope scope = _serviceScopeFactory.CreateScope();
-        IMediator mediator = scope.ServiceProvider.GetRequiredService<IMediator>();
-
-        Either<BaseError, Unit> result = await mediator.Send(request, cancellationToken);
-        result.BiIter(
-            _ => _logger.LogDebug("Done synchronizing emby collections"),
-            error => _logger.LogWarning(
-                "Unable to synchronize emby collections for source {MediaSourceId}: {Error}",
-                request.EmbyMediaSourceId,
-                error.Value));
+        
+        if (entityLocker.IsLibraryLocked(request.EmbyLibraryId))
+        {
+            entityLocker.UnlockLibrary(request.EmbyLibraryId);
+        }
     }
 }

ErsatzTV/Services/JellyfinService.cs

@@ -4,6 +4,7 @@ using ErsatzTV.Application;
 using ErsatzTV.Application.Jellyfin;
 using ErsatzTV.Core;
 using ErsatzTV.Core.Domain;
+using ErsatzTV.Core.Interfaces.Locking;
 using MediatR;
 
 namespace ErsatzTV.Services;
@@ -59,11 +60,6 @@ public class JellyfinService : BackgroundService
                         case ISynchronizeJellyfinLibraryById synchronizeJellyfinLibraryById:
                             requestTask = SynchronizeJellyfinLibrary(synchronizeJellyfinLibraryById, cancellationToken);
                             break;
-                        case SynchronizeJellyfinCollections synchronizeJellyfinCollections:
-                            requestTask = SynchronizeJellyfinCollections(
-                                synchronizeJellyfinCollections,
-                                cancellationToken);
-                            break;
                         default:
                             throw new NotSupportedException($"Unsupported request type: {request.GetType().Name}");
                     }
@@ -159,6 +155,7 @@ public class JellyfinService : BackgroundService
     {
         using IServiceScope scope = _serviceScopeFactory.CreateScope();
         IMediator mediator = scope.ServiceProvider.GetRequiredService<IMediator>();
+        IEntityLocker entityLocker = scope.ServiceProvider.GetRequiredService<IEntityLocker>();
 
         Either<BaseError, string> result = await mediator.Send(request, cancellationToken);
         result.BiIter(
@@ -167,21 +164,10 @@ public class JellyfinService : BackgroundService
                 "Unable to synchronize jellyfin library {LibraryId}: {Error}",
                 request.JellyfinLibraryId,
                 error.Value));
-    }
-
-    private async Task SynchronizeJellyfinCollections(
-        SynchronizeJellyfinCollections request,
-        CancellationToken cancellationToken)
-    {
-        using IServiceScope scope = _serviceScopeFactory.CreateScope();
-        IMediator mediator = scope.ServiceProvider.GetRequiredService<IMediator>();
-
-        Either<BaseError, Unit> result = await mediator.Send(request, cancellationToken);
-        result.BiIter(
-            _ => _logger.LogDebug("Done synchronizing jellyfin collections"),
-            error => _logger.LogWarning(
-                "Unable to synchronize jellyfin collections for source {MediaSourceId}: {Error}",
-                request.JellyfinMediaSourceId,
-                error.Value));
+        
+        if (entityLocker.IsLibraryLocked(request.JellyfinLibraryId))
+        {
+            entityLocker.UnlockLibrary(request.JellyfinLibraryId);
+        }
     }
 }

ErsatzTV/Services/PlexService.cs

@@ -4,6 +4,7 @@ using ErsatzTV.Application;
 using ErsatzTV.Application.Plex;
 using ErsatzTV.Core;
 using ErsatzTV.Core.Domain;
+using ErsatzTV.Core.Interfaces.Locking;
 using MediatR;
 
 namespace ErsatzTV.Services;
@@ -153,6 +154,7 @@ public class PlexService : BackgroundService
     {
         using IServiceScope scope = _serviceScopeFactory.CreateScope();
         IMediator mediator = scope.ServiceProvider.GetRequiredService<IMediator>();
+        IEntityLocker entityLocker = scope.ServiceProvider.GetRequiredService<IEntityLocker>();
 
         Either<BaseError, string> result = await mediator.Send(request, cancellationToken);
         result.BiIter(
@@ -161,5 +163,10 @@ public class PlexService : BackgroundService
                 "Unable to synchronize plex library {LibraryId}: {Error}",
                 request.PlexLibraryId,
                 error.Value));
+        
+        if (entityLocker.IsLibraryLocked(request.PlexLibraryId))
+        {
+            entityLocker.UnlockLibrary(request.PlexLibraryId);
+        }
     }
 }

ErsatzTV/Services/SearchIndexService.cs

@@ -0,0 +1,69 @@
+using System.Threading.Channels;
+using Bugsnag;
+using ErsatzTV.Application;
+using ErsatzTV.Application.Search;
+using MediatR;
+
+namespace ErsatzTV.Services;
+
+public class SearchIndexService : BackgroundService
+{
+    private readonly ChannelReader<ISearchIndexBackgroundServiceRequest> _channel;
+    private readonly ILogger<WorkerService> _logger;
+    private readonly IServiceScopeFactory _serviceScopeFactory;
+
+    public SearchIndexService(
+        ChannelReader<ISearchIndexBackgroundServiceRequest> channel,
+        IServiceScopeFactory serviceScopeFactory,
+        ILogger<WorkerService> logger)
+    {
+        _channel = channel;
+        _serviceScopeFactory = serviceScopeFactory;
+        _logger = logger;
+    }
+
+    protected override async Task ExecuteAsync(CancellationToken cancellationToken)
+    {
+        try
+        {
+            _logger.LogInformation("Search index worker service started");
+
+            await foreach (ISearchIndexBackgroundServiceRequest request in _channel.ReadAllAsync(cancellationToken))
+            {
+                using IServiceScope scope = _serviceScopeFactory.CreateScope();
+                IMediator mediator = scope.ServiceProvider.GetRequiredService<IMediator>();
+
+                try
+                {
+                    switch (request)
+                    {
+                        case ReindexMediaItems reindexMediaItems:
+                            await mediator.Send(reindexMediaItems, cancellationToken);
+                            break;
+                        case RemoveMediaItems removeMediaItems:
+                            await mediator.Send(removeMediaItems, cancellationToken);
+                            break;
+                    }
+                }
+                catch (Exception ex)
+                {
+                    _logger.LogWarning(ex, "Failed to handle search index worker request");
+
+                    try
+                    {
+                        IClient client = scope.ServiceProvider.GetRequiredService<IClient>();
+                        client.Notify(ex);
+                    }
+                    catch (Exception)
+                    {
+                        // do nothing
+                    }
+                }
+            }
+        }
+        catch (Exception ex) when (ex is TaskCanceledException or OperationCanceledException)
+        {
+            _logger.LogInformation("Search index worker service shutting down");
+        }
+    }
+}

ErsatzTV/Services/WorkerService.cs

@@ -7,6 +7,7 @@ using ErsatzTV.Application.MediaSources;
 using ErsatzTV.Application.Playouts;
 using ErsatzTV.Application.Search;
 using ErsatzTV.Core;
+using ErsatzTV.Core.Interfaces.Locking;
 using MediatR;
 
 namespace ErsatzTV.Services;
@@ -45,6 +46,7 @@ public class WorkerService : BackgroundService
                 try
                 {
                     IMediator mediator = scope.ServiceProvider.GetRequiredService<IMediator>();
+                    IEntityLocker entityLocker = scope.ServiceProvider.GetRequiredService<IEntityLocker>();
 
                     switch (request)
                     {
@@ -60,10 +62,10 @@ public class WorkerService : BackgroundService
                                     error.Value));
                             break;
                         case IScanLocalLibrary scanLocalLibrary:
-#if !DEBUG_NO_SYNC
                             Either<BaseError, string> scanResult = await mediator.Send(
                                 scanLocalLibrary,
                                 cancellationToken);
+
                             scanResult.BiIter(
                                 name => _logger.LogDebug(
                                     "Done scanning local library {Library}",
@@ -72,7 +74,11 @@ public class WorkerService : BackgroundService
                                     "Unable to scan local library {LibraryId}: {Error}",
                                     scanLocalLibrary.LibraryId,
                                     error.Value));
-#endif
+
+                            if (entityLocker.IsLibraryLocked(scanLocalLibrary.LibraryId))
+                            {
+                                entityLocker.UnlockLibrary(scanLocalLibrary.LibraryId);
+                            }
                             break;
                         case RebuildSearchIndex rebuildSearchIndex:
                             await mediator.Send(rebuildSearchIndex, cancellationToken);