using ErsatzTV.Core; using ErsatzTV.Core.Domain; using ErsatzTV.Core.Interfaces.Plex; using ErsatzTV.Core.Interfaces.Repositories; using ErsatzTV.Core.MediaSources; using ErsatzTV.Core.Plex; using Microsoft.Extensions.Logging; namespace ErsatzTV.Scanner.Core.Plex; public class PlexCollectionScanner : IPlexCollectionScanner { private readonly ILogger _logger; private readonly IMediator _mediator; private readonly IPlexCollectionRepository _plexCollectionRepository; private readonly IPlexServerApiClient _plexServerApiClient; public PlexCollectionScanner( IMediator mediator, IPlexCollectionRepository plexCollectionRepository, IPlexServerApiClient plexServerApiClient, ILogger logger) { _mediator = mediator; _plexCollectionRepository = plexCollectionRepository; _plexServerApiClient = plexServerApiClient; _logger = logger; } public async Task> ScanCollections( PlexConnection connection, PlexServerAuthToken token, CancellationToken cancellationToken) { try { var incomingKeys = new List(); // get all collections from db (key, etag) List existingCollections = await _plexCollectionRepository.GetCollections(); await foreach ((PlexCollection collection, int _) in _plexServerApiClient.GetAllCollections( connection, token, cancellationToken)) { incomingKeys.Add(collection.Key); Option maybeExisting = existingCollections.Find(c => c.Key == collection.Key); // skip if unchanged (etag) if (await maybeExisting.Map(e => e.Etag ?? string.Empty).IfNoneAsync(string.Empty) == collection.Etag) { _logger.LogDebug("Plex collection {Name} is unchanged", collection.Name); continue; } // add if new if (maybeExisting.IsNone) { _logger.LogDebug("Plex collection {Name} is new", collection.Name); await _plexCollectionRepository.AddCollection(collection); } await SyncCollectionItems(connection, token, collection, cancellationToken); // save collection etag await _plexCollectionRepository.SetEtag(collection); } // remove missing collections (and remove any lingering tags from those collections) foreach (PlexCollection collection in existingCollections.Filter(e => !incomingKeys.Contains(e.Key))) { await _plexCollectionRepository.RemoveCollection(collection); } } catch (Exception ex) { _logger.LogWarning(ex, "Failed to get collections from Plex"); return BaseError.New(ex.Message); } return Unit.Default; } private async Task SyncCollectionItems( PlexConnection connection, PlexServerAuthToken token, PlexCollection collection, CancellationToken cancellationToken) { try { // get collection items from Plex IAsyncEnumerable> items = _plexServerApiClient.GetCollectionItems( connection, token, collection.Key, cancellationToken); List removedIds = await _plexCollectionRepository.RemoveAllTags(collection); // sync tags on items var addedIds = new List(); await foreach ((MediaItem item, int _) in items) { addedIds.Add(await _plexCollectionRepository.AddTag(item, collection)); cancellationToken.ThrowIfCancellationRequested(); } _logger.LogDebug("Plex collection {Name} contains {Count} items", collection.Name, addedIds.Count); int[] changedIds = removedIds.Concat(addedIds).Distinct().ToArray(); await _mediator.Publish( new ScannerProgressUpdate(0, null, null, changedIds.ToArray(), Array.Empty()), CancellationToken.None); } catch (Exception ex) { _logger.LogWarning(ex, "Failed to synchronize Plex collection {Name}", collection.Name); } } }