First attempt at episode sync refactor

src/Ombi.Api.Emby/Models/Media/Tv/EmbyEpisodes.cs

@@ -1,9 +1,10 @@
 using Ombi.Api.Emby.Models.Movie;
+using Ombi.Api.MediaServer.Models.Media.Tv;
 using System;
 
 namespace Ombi.Api.Emby.Models.Media.Tv
 {
-    public class EmbyEpisodes
+    public class EmbyEpisodes : IMediaServerEpisodes
     {
         public string Name { get; set; }
         public string ServerId { get; set; }
@@ -41,5 +42,6 @@ namespace Ombi.Api.Emby.Models.Media.Tv
         public string MediaType { get; set; }
         public bool HasSubtitles { get; set; }
         public EmbyProviderids ProviderIds { get; set; }
+        int IMediaServerEpisodes.EpisodeNumber => IndexNumber;
     }
 }

src/Ombi.Api.Jellyfin/Models/Media/Tv/JellyfinEpisodes.cs

@@ -1,9 +1,10 @@
 using Ombi.Api.Jellyfin.Models.Movie;
+using Ombi.Api.MediaServer.Models.Media.Tv;
 using System;
 
 namespace Ombi.Api.Jellyfin.Models.Media.Tv
 {
-    public class JellyfinEpisodes
+    public class JellyfinEpisodes : IMediaServerEpisodes
     {
         public string Name { get; set; }
         public string ServerId { get; set; }
@@ -41,5 +42,6 @@ namespace Ombi.Api.Jellyfin.Models.Media.Tv
         public string MediaType { get; set; }
         public bool HasSubtitles { get; set; }
         public JellyfinProviderids ProviderIds { get; set; }
+        int IMediaServerEpisodes.EpisodeNumber => IndexNumber;
     }
 }

src/Ombi.Api.MediaServer/Models/Tv/IMediaServerEpisode.cs

@@ -0,0 +1,8 @@
+namespace Ombi.Api.MediaServer.Models.Media.Tv
+{
+    public interface IMediaServerEpisodes
+    {
+        public int EpisodeNumber { get; }
+        public string Name { get; }
+    }
+} 

src/Ombi.Api.MediaServer/Ombi.Api.MediaServer.csproj

@@ -13,6 +13,7 @@
   <ItemGroup>
     <ProjectReference Include="..\Ombi.Api\Ombi.Api.csproj" />
     <ProjectReference Include="..\Ombi.Helpers\Ombi.Helpers.csproj" />
+    <PackageReference Include="System.Linq.Async" Version="4.0.0" />
   </ItemGroup>
 
 </Project>

src/Ombi.Api.Plex/Models/Metadata.cs

@@ -1,8 +1,9 @@
 using System.Collections.Generic;
+using Ombi.Api.MediaServer.Models.Media.Tv;
 
 namespace Ombi.Api.Plex.Models
 {
-    public class Metadata
+    public class Metadata : IMediaServerEpisodes
     {
         public int ratingKey { get; set; }
         public string key { get; set; }
@@ -39,6 +40,8 @@ namespace Ombi.Api.Plex.Models
         public string chapterSource { get; set; }
         public Medium[] Media { get; set; }
         public List<PlexGuids> Guid { get; set; } = new List<PlexGuids>();
+        int IMediaServerEpisodes.EpisodeNumber => index;
+        public string Name => title;
     }
 
     public class PlexGuids

src/Ombi.Api.Plex/Ombi.Api.Plex.csproj

@@ -12,6 +12,7 @@
 
   <ItemGroup>
     <ProjectReference Include="..\Ombi.Api\Ombi.Api.csproj" />
+    <ProjectReference Include="..\Ombi.Api.MediaServer\Ombi.Api.MediaServer.csproj" />
   </ItemGroup>
 
 </Project>

src/Ombi.Schedule/Jobs/Emby/EmbyEpisodeSync.cs

@@ -42,46 +42,51 @@ using Quartz;
 using Ombi.Schedule.Jobs.Ombi;
 using Ombi.Api.Emby.Models;
 using Ombi.Api.Emby.Models.Media.Tv;
+using Ombi.Schedule.Jobs.MediaServer;
 
 namespace Ombi.Schedule.Jobs.Emby
 {
-    public class EmbyEpisodeSync : IEmbyEpisodeSync
+    public class EmbyEpisodeSync : MediaServerEpisodeSync<EmbyEpisodes, EmbyEpisode, IEmbyContentRepository, EmbyContent>, IEmbyEpisodeSync
     {
         public EmbyEpisodeSync(ISettingsService<EmbySettings> s, IEmbyApiFactory api, ILogger<EmbyEpisodeSync> l, IEmbyContentRepository repo
-            , IHubContext<NotificationHub> notification)
+            , IHubContext<NotificationHub> notification) : base(l, repo, notification)
         {
             _apiFactory = api;
-            _logger = l;
             _settings = s;
-            _repo = repo;
-            _notification = notification;
         }
 
         private readonly ISettingsService<EmbySettings> _settings;
         private readonly IEmbyApiFactory _apiFactory;
-        private readonly ILogger<EmbyEpisodeSync> _logger;
+        private bool _recentlyAddedSearch = false;
-        private readonly IEmbyContentRepository _repo;
-        private readonly IHubContext<NotificationHub> _notification;
 
         private const int AmountToTake = 100;
 
         private IEmbyApi Api { get; set; }
 
 
-        public async Task Execute(IJobExecutionContext context)
+        public override async Task Execute(IJobExecutionContext context)
         {
             JobDataMap dataMap = context.MergedJobDataMap;
-            var recentlyAddedSearch = false;
             if (dataMap.TryGetValue(JobDataKeys.EmbyRecentlyAddedSearch, out var recentlyAddedObj))
             {
-                recentlyAddedSearch = Convert.ToBoolean(recentlyAddedObj);
+                _recentlyAddedSearch = Convert.ToBoolean(recentlyAddedObj);
+            }
+            await _notification.Clients.Clients(NotificationHub.AdminConnectionIds)
+                .SendAsync(NotificationHub.NotificationEvent, "Emby Episode Sync Started");
+
+            await CacheEpisodes();
+
+            await _notification.Clients.Clients(NotificationHub.AdminConnectionIds)
+                .SendAsync(NotificationHub.NotificationEvent, "Emby Episode Sync Finished");
+            _logger.LogInformation("Emby Episode Sync Finished - Triggering Metadata refresh");
+            await OmbiQuartz.TriggerJob(nameof(IRefreshMetadata), "System");
         }
 
+        protected async override IAsyncEnumerable<EmbyEpisodes> GetMediaServerEpisodes()
+        {
             var settings = await _settings.GetSettingsAsync();
 
             Api = _apiFactory.CreateClient(settings);
-            await _notification.Clients.Clients(NotificationHub.AdminConnectionIds)
-                .SendAsync(NotificationHub.NotificationEvent, "Emby Episode Sync Started");
             foreach (var server in settings.Servers)
             {
                 if (server.EmbySelectedLibraries.Any() && server.EmbySelectedLibraries.Any(x => x.Enabled))
@@ -90,25 +95,26 @@ namespace Ombi.Schedule.Jobs.Emby
                     foreach (var tvParentIdFilter in tvLibsToFilter)
                     {
                         _logger.LogInformation($"Scanning Lib for episodes '{tvParentIdFilter.Title}'");
-                        await CacheEpisodes(server, recentlyAddedSearch, tvParentIdFilter.Key);
+                        await foreach (var ep in GetEpisodesFromLibrary(server, tvParentIdFilter.Key))
+                        {
+                            yield return ep;
+                        }
                     }
                 }
                 else
                 {
-                    await CacheEpisodes(server, recentlyAddedSearch, string.Empty);
+                    await foreach (var ep in GetEpisodesFromLibrary(server, string.Empty))
+                    {
+                        yield return ep;
+                    }
                 }
             }
 
-            await _notification.Clients.Clients(NotificationHub.AdminConnectionIds)
-                .SendAsync(NotificationHub.NotificationEvent, "Emby Episode Sync Finished");
-            _logger.LogInformation("Emby Episode Sync Finished - Triggering Metadata refresh");
-            await OmbiQuartz.TriggerJob(nameof(IRefreshMetadata), "System");
         }
-
+        private async IAsyncEnumerable<EmbyEpisodes> GetEpisodesFromLibrary(EmbyServers server, string parentIdFilter)
-        private async Task CacheEpisodes(EmbyServers server, bool recentlyAdded, string parentIdFilter)
         {
             EmbyItemContainer<EmbyEpisodes> allEpisodes;
-            if (recentlyAdded)
+            if (_recentlyAddedSearch)
             {
                 var recentlyAddedAmountToTake = AmountToTake;
                 allEpisodes = await Api.RecentlyAddedEpisodes(server.ApiKey, parentIdFilter, 0, recentlyAddedAmountToTake, server.AdministratorId, server.FullUri);
@@ -123,7 +129,6 @@ namespace Ombi.Schedule.Jobs.Emby
             }
             var total = allEpisodes.TotalRecordCount;
             var processed = 0;
-            var epToAdd = new HashSet<EmbyEpisode>();
             while (processed < total)
             {
                 foreach (var ep in allEpisodes.Items)
@@ -145,22 +150,27 @@ namespace Ombi.Schedule.Jobs.Emby
                             ep.Name);
                         continue;
                     }
-
+                    yield return ep;
-                    var existingEpisode = await _repo.GetEpisodeByEmbyId(ep.Id);
+                }
-                    // Make sure it's not in the hashset too
+                if (!_recentlyAddedSearch)
-                    var existingInList = epToAdd.Any(x => x.EmbyId == ep.Id);
-
-                    if (existingEpisode == null && !existingInList)
                 {
-                        // Sanity checks
+                    allEpisodes = await Api.GetAllEpisodes(server.ApiKey, parentIdFilter, processed, AmountToTake, server.AdministratorId, server.FullUri);
-                        if (ep.IndexNumber == 0)
+                }
-                        {
+            }
-                            _logger.LogWarning($"Episode {ep.Name} has no episode number. Skipping.");
-                            continue;
         }
 
-                        _logger.LogDebug("Adding new episode {0} to parent {1}", ep.Name, ep.SeriesName);
+        protected override async Task<EmbyEpisode> GetExistingEpisode(EmbyEpisodes ep)
-                        // add it
+        {
+            return await _repo.GetEpisodeByEmbyId(ep.Id);
+        }
+
+        protected override bool IsIn(EmbyEpisodes ep, ICollection<EmbyEpisode> list)
+        {
+            return list.Any(x => x.EmbyId == ep.Id);
+        }
+
+        protected override void addEpisode(EmbyEpisodes ep, ICollection<EmbyEpisode> epToAdd)
+        {
             epToAdd.Add(new EmbyEpisode
             {
                 EmbyId = ep.Id,
@@ -191,38 +201,4 @@ namespace Ombi.Schedule.Jobs.Emby
             }
         }
     }
-
-                await _repo.AddRange(epToAdd);
-                epToAdd.Clear();
-                if (!recentlyAdded)
-                {
-                    allEpisodes = await Api.GetAllEpisodes(server.ApiKey, parentIdFilter, processed, AmountToTake, server.AdministratorId, server.FullUri);
-                }
-            }
-
-            if (epToAdd.Any())
-            {
-                await _repo.AddRange(epToAdd);
-            }
-        }
-
-        private bool _disposed;
-        protected virtual void Dispose(bool disposing)
-        {
-            if (_disposed)
-                return;
-
-            if (disposing)
-            {
-                //_settings?.Dispose();
-            }
-            _disposed = true;
-        }
-
-        public void Dispose()
-        {
-            Dispose(true);
-            GC.SuppressFinalize(this);
-        }
-    }
 }

src/Ombi.Schedule/Jobs/Jellyfin/JellyfinEpisodeSync.cs

@@ -40,36 +40,51 @@ using Ombi.Store.Entities;
 using Ombi.Store.Repository;
 using Quartz;
 using Ombi.Schedule.Jobs.Ombi;
+using Ombi.Schedule.Jobs.MediaServer;
+using Ombi.Api.Jellyfin.Models.Media.Tv;
 
 namespace Ombi.Schedule.Jobs.Jellyfin
 {
-    public class JellyfinEpisodeSync : IJellyfinEpisodeSync
+    public class JellyfinEpisodeSync : MediaServerEpisodeSync<JellyfinEpisodes, JellyfinEpisode, IJellyfinContentRepository, JellyfinContent>, IJellyfinEpisodeSync
     {
-        public JellyfinEpisodeSync(ISettingsService<JellyfinSettings> s, IJellyfinApiFactory api, ILogger<JellyfinEpisodeSync> l, IJellyfinContentRepository repo
+
-            , IHubContext<NotificationHub> notification)
+
+        public JellyfinEpisodeSync(
+            ISettingsService<JellyfinSettings> s,
+            IJellyfinApiFactory api,
+            ILogger<MediaServerEpisodeSync<JellyfinEpisodes, JellyfinEpisode, IJellyfinContentRepository, JellyfinContent>> l,
+            IJellyfinContentRepository repo,
+            IHubContext<NotificationHub> notification) : base(l, repo, notification)
         {
             _apiFactory = api;
-            _logger = l;
             _settings = s;
-            _repo = repo;
-            _notification = notification;
         }
-
         private readonly ISettingsService<JellyfinSettings> _settings;
         private readonly IJellyfinApiFactory _apiFactory;
-        private readonly ILogger<JellyfinEpisodeSync> _logger;
-        private readonly IJellyfinContentRepository _repo;
-        private readonly IHubContext<NotificationHub> _notification;
         private IJellyfinApi Api { get; set; }
 
-
+        public override async Task Execute(IJobExecutionContext job)
-        public async Task Execute(IJobExecutionContext job)
         {
             var settings = await _settings.GetSettingsAsync();
 
             Api = _apiFactory.CreateClient(settings);
             await _notification.Clients.Clients(NotificationHub.AdminConnectionIds)
                 .SendAsync(NotificationHub.NotificationEvent, "Jellyfin Episode Sync Started");
+
+            await CacheEpisodes();
+
+            await _notification.Clients.Clients(NotificationHub.AdminConnectionIds)
+                .SendAsync(NotificationHub.NotificationEvent, "Jellyfin Episode Sync Finished");
+            _logger.LogInformation("Jellyfin Episode Sync Finished - Triggering Metadata refresh");
+            await OmbiQuartz.TriggerJob(nameof(IRefreshMetadata), "System");
+        }
+
+        private int packageSize = 200;
+        protected override async IAsyncEnumerable<JellyfinEpisodes> GetMediaServerEpisodes()
+        {
+            var settings = await _settings.GetSettingsAsync();
+            Api = _apiFactory.CreateClient(settings);
+
             foreach (var server in settings.Servers)
             {
 
@@ -79,27 +94,27 @@ namespace Ombi.Schedule.Jobs.Jellyfin
                     foreach (var tvParentIdFilter in tvLibsToFilter)
                     {
                         _logger.LogInformation($"Scanning Lib for episodes '{tvParentIdFilter.Title}'");
-                        await CacheEpisodes(server, tvParentIdFilter.Key);
+                        await foreach (var ep in GetEpisodesFromLibrary(server, tvParentIdFilter.Key))
+                        {
+                            yield return ep;
+                        }
                     }
                 }
                 else
                 {
-                    await CacheEpisodes(server, string.Empty);
+                    await foreach (var ep in GetEpisodesFromLibrary(server, string.Empty))
-                }
+                    {
-            }
+                        yield return ep;
-
+                    }
-            await _notification.Clients.Clients(NotificationHub.AdminConnectionIds)
+                }
-                .SendAsync(NotificationHub.NotificationEvent, "Jellyfin Episode Sync Finished");
+            }
-            _logger.LogInformation("Jellyfin Episode Sync Finished - Triggering Metadata refresh");
+        }
-            await OmbiQuartz.TriggerJob(nameof(IRefreshMetadata), "System");
+        private async IAsyncEnumerable<JellyfinEpisodes> GetEpisodesFromLibrary(JellyfinServers server, string parentIdFilter)
-        }
-
-        private async Task CacheEpisodes(JellyfinServers server, string parentIdFilter)
         {
-            var allEpisodes = await Api.GetAllEpisodes(server.ApiKey, parentIdFilter, 0, 200, server.AdministratorId, server.FullUri);
-            var total = allEpisodes.TotalRecordCount;
             var processed = 0;
-            var epToAdd = new HashSet<JellyfinEpisode>();
+            var allEpisodes = await Api.GetAllEpisodes(server.ApiKey, parentIdFilter, processed, packageSize, server.AdministratorId, server.FullUri);
+
+            var total = allEpisodes.TotalRecordCount;
             while (processed < total)
             {
                 foreach (var ep in allEpisodes.Items)
@@ -121,19 +136,14 @@ namespace Ombi.Schedule.Jobs.Jellyfin
                             ep.Name);
                         continue;
                     }
-
+                    yield return ep;
-                    var existingEpisode = await _repo.GetEpisodeByJellyfinId(ep.Id);
-                    // Make sure it's not in the hashset too
-                    var existingInList = epToAdd.Any(x => x.JellyfinId == ep.Id);
-
-                    if (existingEpisode == null && !existingInList)
-                    {
-                        // Sanity checks
-                        if (ep.IndexNumber == 0) // no check on season number, Season 0 can be Specials
-                        {
-                            _logger.LogWarning($"Episode {ep.Name} has no episode number. Skipping.");
-                            continue;
                 }
+                allEpisodes = await Api.GetAllEpisodes(server.ApiKey, parentIdFilter, processed, packageSize, server.AdministratorId, server.FullUri);
+            }
+        }
+
+        protected override void addEpisode(JellyfinEpisodes ep, ICollection<JellyfinEpisode> epToAdd)
+        {
 
             _logger.LogDebug("Adding new episode {0} to parent {1}", ep.Name, ep.SeriesName);
             // add it
@@ -166,36 +176,15 @@ namespace Ombi.Schedule.Jobs.Jellyfin
                 });
             }
         }
-                }
 
-                await _repo.AddRange(epToAdd);
+        protected override async Task<JellyfinEpisode> GetExistingEpisode(JellyfinEpisodes ep)
-                epToAdd.Clear();
-                allEpisodes = await Api.GetAllEpisodes(server.ApiKey, parentIdFilter, processed, 200, server.AdministratorId, server.FullUri);
-            }
-
-            if (epToAdd.Any())
         {
-                await _repo.AddRange(epToAdd);
+            return await _repo.GetEpisodeByJellyfinId(ep.Id);
-            }
         }
 
-        private bool _disposed;
+        protected override bool IsIn(JellyfinEpisodes ep, ICollection<JellyfinEpisode> list)
-        protected virtual void Dispose(bool disposing)
         {
-            if (_disposed)
+            return list.Any(x => x.JellyfinId == ep.Id);
-                return;
-
-            if (disposing)
-            {
-                //_settings?.Dispose();
-            }
-            _disposed = true;
-        }
-
-        public void Dispose()
-        {
-            Dispose(true);
-            GC.SuppressFinalize(this);
         }
     }
 }

src/Ombi.Schedule/Jobs/MediaServer/IMediaServerEpisodeSync.cs

@@ -0,0 +1,14 @@
+using System.Collections.Generic;
+using System.Threading.Tasks;
+using Ombi.Api.MediaServer.Models.Media.Tv;
+using Ombi.Store.Entities;
+
+namespace Ombi.Schedule.Jobs.MediaServer
+{
+    public interface IMediaServerEpisodeSync<T, U> : IBaseJob
+    where T:  IMediaServerEpisode
+    where U: IMediaServerEpisodes
+    {
+        Task<HashSet<T>> ProcessEpisodes(IAsyncEnumerable<U> serverEpisodes, ICollection<T> currentEpisodes);
+    }
+}

src/Ombi.Schedule/Jobs/MediaServer/MediaServerEpisodeSync.cs

@@ -0,0 +1,105 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Threading.Tasks;
+using Microsoft.AspNetCore.SignalR;
+using Microsoft.Extensions.Logging;
+using Ombi.Hubs;
+using Ombi.Store.Entities;
+using Ombi.Store.Repository;
+using Quartz;
+using Ombi.Api.MediaServer.Models.Media.Tv;
+
+namespace Ombi.Schedule.Jobs.MediaServer
+{
+    public abstract class MediaServerEpisodeSync<T, U, V, W> : IMediaServerEpisodeSync<U, T>
+    where T : IMediaServerEpisodes
+    where U : IMediaServerEpisode
+    where V : IMediaServerContentRepository<W>
+    where W : IMediaServerContent
+    {
+        public MediaServerEpisodeSync(
+            ILogger l,
+            V repo,
+            IHubContext<NotificationHub> notification)
+        {
+            _logger = l;
+            _repo = repo;
+            _notification = notification;
+        }
+
+        protected readonly IHubContext<NotificationHub> _notification;
+        protected readonly ILogger _logger;
+        protected readonly V _repo;
+        protected abstract IAsyncEnumerable<T> GetMediaServerEpisodes();
+        protected abstract Task<U> GetExistingEpisode(T ep);
+        protected abstract bool IsIn(T ep, ICollection<U> list);
+
+        protected async Task CacheEpisodes()
+        {
+            var epToAdd = new HashSet<U>();
+            await ProcessEpisodes(GetMediaServerEpisodes(), epToAdd);
+        }
+
+        protected abstract void addEpisode(T ep, ICollection<U> epToAdd);
+
+        private bool _disposed;
+        protected virtual void Dispose(bool disposing)
+        {
+            if (_disposed)
+                return;
+
+            if (disposing)
+            {
+                //_settings?.Dispose();
+            }
+            _disposed = true;
+        }
+
+        public void Dispose()
+        {
+            Dispose(true);
+            GC.SuppressFinalize(this);
+        }
+
+        public abstract Task Execute(IJobExecutionContext context);
+
+        public async Task<HashSet<U>> ProcessEpisodes(IAsyncEnumerable<T> serverEpisodes, ICollection<U> episodesAlreadyAdded)
+        {
+            var episodesBeingAdded = new HashSet<U>();
+
+            await foreach (var ep in GetMediaServerEpisodes())
+            {
+                if (IsIn(ep, episodesAlreadyAdded) || IsIn(ep, episodesBeingAdded))
+                {
+                    _logger.LogWarning($"Episode {ep.Name} already processed in this update.");
+                    continue;
+                }
+                // Sanity checks
+                if (ep.EpisodeNumber == 0) // no check on season number, Season 0 can be Specials
+                {
+                    _logger.LogWarning($"Episode {ep.Name} has no episode number. Skipping.");
+                    continue;
+                }
+
+                var existingEpisode = await GetExistingEpisode(ep);
+                if (existingEpisode == null)
+                {
+                    addEpisode(ep, episodesBeingAdded);
+                }
+                else
+                {
+                    _logger.LogWarning($"Episode {ep.Name} already exists in database.");
+                    // TODO: update if something changed on the media server
+                }
+            }
+
+
+            if (episodesBeingAdded.Any())
+            {
+                await _repo.AddRange((IEnumerable<IMediaServerEpisode>)episodesBeingAdded);
+            }
+            return episodesBeingAdded;
+        }
+    }
+}

src/Ombi.Schedule/Jobs/Plex/Interfaces/IPlexEpisodeSync.cs

@@ -1,14 +1,13 @@
-using System;
+using System.Collections.Generic;
-using System.Collections.Generic;
 using System.Linq;
 using System.Threading.Tasks;
 using Ombi.Api.Plex.Models;
+using Ombi.Schedule.Jobs.MediaServer;
 using Ombi.Store.Entities;
 
 namespace Ombi.Schedule.Jobs.Plex.Interfaces
 {
-    public interface IPlexEpisodeSync : IBaseJob
+    public interface IPlexEpisodeSync : IMediaServerEpisodeSync<PlexEpisode, Metadata>, IBaseJob
     {
-        Task<HashSet<PlexEpisode>> ProcessEpsiodes(Metadata[] episodes, IQueryable<PlexEpisode> currentEpisodes);
     }
 }

src/Ombi.Schedule/Jobs/Plex/PlexContentSync.cs

@@ -226,7 +226,7 @@ namespace Ombi.Schedule.Jobs.Plex
                     await Repo.SaveChangesAsync();
                     if (content.Metadata != null)
                     {
-                        var episodesAdded = await EpisodeSync.ProcessEpsiodes(content.Metadata, (IQueryable<PlexEpisode>)allEps);
+                        var episodesAdded = await EpisodeSync.ProcessEpisodes(content.Metadata.ToAsyncEnumerable(), (ICollection<PlexEpisode>)allEps.ToHashSet());
                         episodesProcessed.AddRange(episodesAdded.Select(x => x.Id));
                     }
                 }

src/Ombi.Schedule/Jobs/Plex/PlexEpisodeSync.cs

@@ -11,6 +11,7 @@ using Ombi.Core.Settings;
 using Ombi.Core.Settings.Models.External;
 using Ombi.Helpers;
 using Ombi.Hubs;
+using Ombi.Schedule.Jobs.MediaServer;
 using Ombi.Schedule.Jobs.Ombi;
 using Ombi.Schedule.Jobs.Plex.Interfaces;
 using Ombi.Store.Entities;
@@ -19,111 +20,55 @@ using Quartz;
 
 namespace Ombi.Schedule.Jobs.Plex
 {
-    public class PlexEpisodeSync : IPlexEpisodeSync
+    public class PlexEpisodeSync : MediaServerEpisodeSync<Metadata, PlexEpisode, IPlexContentRepository, PlexServerContent>, IPlexEpisodeSync
     {
         public PlexEpisodeSync(ISettingsService<PlexSettings> s, ILogger<PlexEpisodeSync> log, IPlexApi plexApi,
-            IPlexContentRepository repo, IHubContext<NotificationHub> hub)
+            IPlexContentRepository repo, IHubContext<NotificationHub> hub) : base(log, repo, hub)
         {
             _settings = s;
-            _log = log;
             _api = plexApi;
-            _repo = repo;
-            _notification = hub;
             _settings.ClearCache();
         }
 
         private readonly ISettingsService<PlexSettings> _settings;
-        private readonly ILogger<PlexEpisodeSync> _log;
         private readonly IPlexApi _api;
-        private readonly IPlexContentRepository _repo;
-        private readonly IHubContext<NotificationHub> _notification;
 
-        public async Task Execute(IJobExecutionContext job)
+        public override async Task Execute(IJobExecutionContext job)
         {
             try
             {
-                var s = await _settings.GetSettingsAsync();
-                if (!s.Enable)
-                {
-                    return;
-                }
                 await _notification.Clients.Clients(NotificationHub.AdminConnectionIds)
                     .SendAsync(NotificationHub.NotificationEvent, "Plex Episode Sync Started");
 
-                foreach (var server in s.Servers)
+                await CacheEpisodes();
-                {
+                
-                    await Cache(server);
+                _logger.LogInformation(LoggingEvents.PlexEpisodeCacher, "We have finished caching the episodes.");
-                }
+                await _repo.SaveChangesAsync();
 
             }
             catch (Exception e)
             {
                 await _notification.Clients.Clients(NotificationHub.AdminConnectionIds)
                     .SendAsync(NotificationHub.NotificationEvent, "Plex Episode Sync Failed");
-                _log.LogError(LoggingEvents.Cacher, e, "Caching Episodes Failed");
+                _logger.LogError(LoggingEvents.Cacher, e, "Caching Episodes Failed");
             }
 
 
-            _log.LogInformation("Plex Episode Sync Finished - Triggering Metadata refresh");
+            _logger.LogInformation("Plex Episode Sync Finished - Triggering Metadata refresh");
             await OmbiQuartz.TriggerJob(nameof(IRefreshMetadata), "System");
 
             await _notification.Clients.Clients(NotificationHub.AdminConnectionIds)
                 .SendAsync(NotificationHub.NotificationEvent, "Plex Episode Sync Finished");
         }
 
-        private async Task Cache(PlexServers settings)
+        private async IAsyncEnumerable<Metadata> GetEpisodes(PlexServers settings, Directory section)
-        {
-            if (!Validate(settings))
-            {
-                _log.LogWarning("Validation failed");
-                return;
-            }
-
-            // Get the librarys and then get the tv section
-            var sections = await _api.GetLibrarySections(settings.PlexAuthToken, settings.FullUri);
-
-            // Filter the libSections
-            var tvSections = sections.MediaContainer.Directory.Where(x => x.type.Equals(PlexMediaType.Show.ToString(), StringComparison.CurrentCultureIgnoreCase));
-
-            foreach (var section in tvSections)
-            {
-                if (settings.PlexSelectedLibraries.Any())
-                {
-                    // Are any enabled?
-                    if (settings.PlexSelectedLibraries.Any(x => x.Enabled))
-                    {
-                        // Make sure we have enabled this 
-                        var keys = settings.PlexSelectedLibraries.Where(x => x.Enabled).Select(x => x.Key.ToString())
-                            .ToList();
-                        if (!keys.Contains(section.key))
-                        {
-                            // We are not monitoring this lib
-                            continue;
-                        }
-                    }
-                }
-
-                // Get the episodes
-                await GetEpisodes(settings, section);
-            }
-
-        }
-
-        private async Task GetEpisodes(PlexServers settings, Directory section)
         {
             var currentPosition = 0;
             var resultCount = settings.EpisodeBatchSize == 0 ? 150 : settings.EpisodeBatchSize;
             var currentEpisodes = _repo.GetAllEpisodes().Cast<PlexEpisode>();
             var episodes = await _api.GetAllEpisodes(settings.PlexAuthToken, settings.FullUri, section.key, currentPosition, resultCount);
-            _log.LogInformation(LoggingEvents.PlexEpisodeCacher, $"Total Epsiodes found for {episodes.MediaContainer.librarySectionTitle} = {episodes.MediaContainer.totalSize}");
+            _logger.LogInformation(LoggingEvents.PlexEpisodeCacher, $"Total Epsiodes found for {episodes.MediaContainer.librarySectionTitle} = {episodes.MediaContainer.totalSize}");
 
-            // Delete all the episodes because we cannot uniquly match an episode to series every time, 
-            // see comment below.
-
-            // 12.03.2017 - I think we should be able to match them now
-            //await _repo.ExecuteSql("DELETE FROM PlexEpisode");
-
-            await ProcessEpsiodes(episodes?.MediaContainer?.Metadata ?? new Metadata[] { }, currentEpisodes);
             currentPosition += resultCount;
 
             while (currentPosition < episodes.MediaContainer.totalSize)
@@ -131,35 +76,8 @@ namespace Ombi.Schedule.Jobs.Plex
                 var ep = await _api.GetAllEpisodes(settings.PlexAuthToken, settings.FullUri, section.key, currentPosition,
                     resultCount);
 
-                await ProcessEpsiodes(ep?.MediaContainer?.Metadata ?? new Metadata[] { }, currentEpisodes);
+                foreach (var episode in ep.MediaContainer.Metadata)
-                _log.LogInformation(LoggingEvents.PlexEpisodeCacher, $"Processed {resultCount} more episodes. Total Remaining {episodes.MediaContainer.totalSize - currentPosition}");
-                currentPosition += resultCount;
-            }
-
-            // we have now finished.
-            _log.LogInformation(LoggingEvents.PlexEpisodeCacher, "We have finished caching the episodes.");
-            await _repo.SaveChangesAsync();
-        }
-
-        public async Task<HashSet<PlexEpisode>> ProcessEpsiodes(Metadata[] episodes, IQueryable<PlexEpisode> currentEpisodes)
                 {
-            var ep = new HashSet<PlexEpisode>();
-            try
-            {
-                foreach (var episode in episodes)
-                {
-                    // I don't think we need to get the metadata, we only need to get the metadata if we need the provider id (TheTvDbid). Why do we need it for episodes?
-                    // We have the parent and grandparent rating keys to link up to the season and series
-                    //var metadata = _api.GetEpisodeMetaData(server.PlexAuthToken, server.FullUri, episode.ratingKey);
-
-                    // This does seem to work, it looks like we can somehow get different rating, grandparent and parent keys with episodes. Not sure how.
-                    var epExists = currentEpisodes.Any(x => episode.ratingKey == x.Key &&
-                                                              episode.grandparentRatingKey == x.GrandparentKey);
-                    if (epExists)
-                    {
-                        continue;
-                    }
-
                     // Let's check if we have the parent
                     var seriesExists = await _repo.GetByKey(episode.grandparentRatingKey);
                     if (seriesExists == null)
@@ -169,7 +87,7 @@ namespace Ombi.Schedule.Jobs.Plex
                             x.Title == episode.grandparentTitle);
                         if (seriesExists == null)
                         {
-                            _log.LogWarning(
+                            _logger.LogWarning(
                                 "The episode title {0} we cannot find the parent series. The episode grandparentKey = {1}, grandparentTitle = {2}",
                                 episode.title, episode.grandparentRatingKey, episode.grandparentTitle);
                             continue;
@@ -178,34 +96,15 @@ namespace Ombi.Schedule.Jobs.Plex
                         // Set the rating key to the correct one
                         episode.grandparentRatingKey = seriesExists.Key;
                     }
+                    yield return episode;
 
-                    // Sanity checks
+                }
-                    if (episode.index == 0)
+                _logger.LogInformation(LoggingEvents.PlexEpisodeCacher, $"Processed {resultCount} more episodes. Total Remaining {episodes.MediaContainer.totalSize - currentPosition}");
-                    {
+                currentPosition += resultCount;
-                        _log.LogWarning($"Episode {episode.title} has no episode number. Skipping.");
-                        continue;
             }
 
-                    ep.Add(new PlexEpisode
-                    {
-                        EpisodeNumber = episode.index,
-                        SeasonNumber = episode.parentIndex,
-                        GrandparentKey = episode.grandparentRatingKey,
-                        ParentKey = episode.parentRatingKey,
-                        Key = episode.ratingKey,
-                        Title = episode.title
-                    });
         }
 
-                await _repo.AddRange(ep);
-                return ep;
-            }
-            catch (Exception e)
-            {
-                Console.WriteLine(e);
-                throw;
-            }
-        }
 
         private bool Validate(PlexServers settings)
         {
@@ -217,23 +116,76 @@ namespace Ombi.Schedule.Jobs.Plex
             return true;
         }
 
-        private bool _disposed;
+        protected override async IAsyncEnumerable<Metadata> GetMediaServerEpisodes()
-        protected virtual void Dispose(bool disposing)
         {
-            if (_disposed)
+            var s = await _settings.GetSettingsAsync();
-                return;
+            if (!s.Enable)
-
-            if (disposing)
             {
-                //_settings?.Dispose();
+                yield break;
             }
-            _disposed = true;
+            foreach (var server in s.Servers)
+            {
+                if (!Validate(server))
+                {
+                    _logger.LogWarning("Validation failed");
+                    continue;
                 }
 
-        public void Dispose()
+                // Get the librarys and then get the tv section
+                var sections = await _api.GetLibrarySections(server.PlexAuthToken, server.FullUri);
+
+                // Filter the libSections
+                var tvSections = sections.MediaContainer.Directory.Where(x => x.type.Equals(PlexMediaType.Show.ToString(), StringComparison.CurrentCultureIgnoreCase));
+
+                foreach (var section in tvSections)
                 {
-            Dispose(true);
+                    if (server.PlexSelectedLibraries.Any())
-            GC.SuppressFinalize(this);
+                    {
+                        // Are any enabled?
+                        if (server.PlexSelectedLibraries.Any(x => x.Enabled))
+                        {
+                            // Make sure we have enabled this 
+                            var keys = server.PlexSelectedLibraries.Where(x => x.Enabled).Select(x => x.Key.ToString())
+                                .ToList();
+                            if (!keys.Contains(section.key))
+                            {
+                                // We are not monitoring this lib
+                                continue;
                             }
                         }
+                    }
+
+                    // Get the episodes
+                    await foreach (var ep in GetEpisodes(server, section))
+                    {
+                        yield return ep;
+                    }
+                }
+            }
+        }
+
+        protected override Task<PlexEpisode> GetExistingEpisode(Metadata ep)
+        { 
+            return _repo.GetEpisodeByKey(ep.ratingKey);
+        }
+
+        protected override bool IsIn(Metadata ep, ICollection<PlexEpisode> list)
+        {
+            return false; // That check was never needed in Plex before refactoring
+        }
+
+        protected override void addEpisode(Metadata ep, ICollection<PlexEpisode> epToAdd)
+        {
+            epToAdd.Add(new PlexEpisode
+            {
+                EpisodeNumber = ep.index,
+                SeasonNumber = ep.parentIndex,
+                GrandparentKey = ep.grandparentRatingKey,
+                ParentKey = ep.parentRatingKey,
+                Key = ep.ratingKey,
+                Title = ep.title
+            });
+        }
+
+    }
 }