using System.Threading.Channels; using ErsatzTV.Application.Emby; using ErsatzTV.Application.Jellyfin; using ErsatzTV.Application.MediaSources; using ErsatzTV.Application.Plex; using ErsatzTV.Core.Domain; using ErsatzTV.Core.Interfaces.Locking; using ErsatzTV.Infrastructure.Data; using ErsatzTV.Infrastructure.Extensions; using Microsoft.EntityFrameworkCore; using Microsoft.Extensions.Logging; namespace ErsatzTV.Application.Libraries; public class QueueLibraryScanByLibraryIdHandler( IDbContextFactory dbContextFactory, IEntityLocker locker, ChannelWriter scannerWorker, ILogger logger) : IRequestHandler { public async Task Handle( QueueLibraryScanByLibraryId request, CancellationToken cancellationToken) { await using TvContext dbContext = await dbContextFactory.CreateDbContextAsync(cancellationToken); Option maybeLibrary = await dbContext.Libraries .AsNoTracking() .SelectOneAsync(l => l.Id, l => l.Id == request.LibraryId, cancellationToken); foreach (Library library in maybeLibrary) { bool shouldSyncItems = library switch { PlexLibrary plexLibrary => plexLibrary.ShouldSyncItems, JellyfinLibrary jellyfinLibrary => jellyfinLibrary.ShouldSyncItems, EmbyLibrary embyLibrary => embyLibrary.ShouldSyncItems, _ => true }; if (!shouldSyncItems) { logger.LogWarning("Library sync is disabled for library id {Id}", library.Id); return QueueLibraryScanResult.SyncDisabled; } // A true from LockLibrary confers ownership of exactly one release; a false means a scan // is already in progress and we own no release. if (!locker.LockLibrary(library.Id)) { return QueueLibraryScanResult.AlreadyScanning; } try { logger.LogDebug("Queued library scan for library id {Id}", library.Id); switch (library) { case LocalLibrary: await scannerWorker.WriteAsync(new ForceScanLocalLibrary(library.Id), cancellationToken); break; case PlexLibrary: await scannerWorker.WriteAsync( new SynchronizePlexLibraries(library.MediaSourceId), cancellationToken); await scannerWorker.WriteAsync( new ForceSynchronizePlexLibraryById(library.Id, request.DeepScan), cancellationToken); break; case JellyfinLibrary: await scannerWorker.WriteAsync( new SynchronizeJellyfinLibraries(library.MediaSourceId), cancellationToken); await scannerWorker.WriteAsync( new ForceSynchronizeJellyfinLibraryById(library.Id, request.DeepScan), cancellationToken); break; case EmbyLibrary: await scannerWorker.WriteAsync( new SynchronizeEmbyLibraries(library.MediaSourceId), cancellationToken); await scannerWorker.WriteAsync( new ForceSynchronizeEmbyLibraryById(library.Id, request.DeepScan), cancellationToken); break; } } catch { // the scanner only unlocks when it receives the message; if enqueueing fails // (e.g. request aborted / channel completed) after we acquired the lock, release // it here or it is held forever (EnqueueWithTraktLock pattern). locker.UnlockLibrary(library.Id); throw; } return QueueLibraryScanResult.Queued; } return QueueLibraryScanResult.NotFound; } }