adding db watcher service, loading watchers from db on startup

Selector.CLI/DbWatcherService.cs

@@ -0,0 +1,170 @@
+using System;
+using System.Collections.Generic;
+using System.Linq;
+using System.Text;
+using System.Threading;
+using System.Threading.Tasks;
+using Microsoft.EntityFrameworkCore;
+using Microsoft.Extensions.Configuration;
+using Microsoft.Extensions.DependencyInjection;
+using Microsoft.Extensions.Hosting;
+using Microsoft.Extensions.Logging;
+using Microsoft.Extensions.Options;
+
+using Selector.Cache;
+using Selector.Model;
+using IF.Lastfm.Core.Api;
+using StackExchange.Redis;
+
+namespace Selector.CLI
+{
+    class DbWatcherService : IHostedService
+    {
+        private const int PollPeriod = 1000;
+
+        private readonly ILogger<LocalWatcherService> Logger;
+        private readonly ILoggerFactory LoggerFactory;
+        private readonly IServiceProvider ServiceProvider;
+        private readonly RootOptions Config;
+        private readonly IWatcherFactory WatcherFactory;
+        private readonly IWatcherCollectionFactory WatcherCollectionFactory;
+        private readonly IRefreshTokenFactoryProvider SpotifyFactory;
+        private readonly LastAuth LastAuth;
+        
+        private readonly IDatabaseAsync Cache;
+        private readonly ISubscriber Subscriber; 
+
+        private Dictionary<string, IWatcherCollection> Watchers { get; set; } = new();
+
+        public DbWatcherService(
+            IWatcherFactory watcherFactory,
+            IWatcherCollectionFactory watcherCollectionFactory,
+            IRefreshTokenFactoryProvider spotifyFactory,
+            ILoggerFactory loggerFactory,
+            IServiceProvider serviceProvider,
+            IOptions<RootOptions> config,
+            LastAuth lastAuth = null,
+            IDatabaseAsync cache = null,
+            ISubscriber subscriber = null
+        ) {
+            Logger = loggerFactory.CreateLogger<LocalWatcherService>();
+            LoggerFactory = loggerFactory;
+            Config = config.Value;
+            WatcherFactory = watcherFactory;
+            WatcherCollectionFactory = watcherCollectionFactory;
+            SpotifyFactory = spotifyFactory;
+            LastAuth = lastAuth;
+            ServiceProvider = serviceProvider;
+            Cache = cache;
+            Subscriber = subscriber;
+        }
+
+        public async Task StartAsync(CancellationToken cancellationToken)
+        {
+            Logger.LogInformation("Starting database watcher service...");
+
+            var watcherIndices = await InitInstances();
+
+            Logger.LogInformation($"Starting {watcherIndices.Count()} affected watcher collection(s)...");
+            StartWatcherCollections(watcherIndices);
+        }
+
+        private async Task<IEnumerable<string>> InitInstances()
+        {
+            using var scope = ServiceProvider.CreateScope();
+            var db = scope.ServiceProvider.GetService<ApplicationDbContext>();
+
+            var indices = new HashSet<string>();
+
+            foreach (var dbWatcher in db.Watcher.Include(w => w.User))
+            {
+                Logger.LogInformation($"Creating new [{dbWatcher.Type}] watcher");
+
+                var watcherCollectionIdx = dbWatcher.UserId;
+                indices.Add(watcherCollectionIdx);
+
+                if (!Watchers.ContainsKey(watcherCollectionIdx))
+                    Watchers[watcherCollectionIdx] = WatcherCollectionFactory.Get();
+
+                var watcherCollection = Watchers[watcherCollectionIdx];
+
+                Logger.LogDebug("Getting Spotify factory");
+                var spotifyFactory = await SpotifyFactory.GetFactory(dbWatcher.User.SpotifyRefreshToken);
+
+                IWatcher watcher = null;
+                List<IConsumer> consumers = new();
+
+                switch (dbWatcher.Type)
+                {
+                    case WatcherType.Player:
+                        watcher = await WatcherFactory.Get<PlayerWatcher>(spotifyFactory, id: dbWatcher.UserId, pollPeriod: PollPeriod);
+
+                        var featureInjector = new AudioFeatureInjectorFactory(LoggerFactory);
+                        consumers.Add(await featureInjector.Get(spotifyFactory));
+
+                        var featureInjectorCache = new CachingAudioFeatureInjectorFactory(LoggerFactory, Cache);
+                        consumers.Add(await featureInjectorCache.Get(spotifyFactory));
+
+                        var cacheWriter = new CacheWriterFactory(Cache, LoggerFactory);
+                        consumers.Add(await cacheWriter.Get());
+
+                        var pub = new PublisherFactory(Subscriber, LoggerFactory);
+                        consumers.Add(await pub.Get());
+
+                        if (!string.IsNullOrWhiteSpace(dbWatcher.User.LastFmUsername))
+                        {
+                            if (LastAuth is null) throw new ArgumentNullException("No Last Auth Injected");
+
+                            var client = new LastfmClient(LastAuth);
+
+                            var playCount = new PlayCounterFactory(LoggerFactory, client: client, creds: new() { Username = dbWatcher.User.LastFmUsername });
+                            consumers.Add(await playCount.Get());
+                        }
+                        else
+                        {
+                            Logger.LogDebug($"[{dbWatcher.User.UserName}] No Last.fm username, skipping play counter");
+                        }
+
+                        break;
+
+                    case WatcherType.Playlist:
+                        throw new NotImplementedException("Playlist watchers not implemented");
+                        // break;
+                }
+
+                watcherCollection.Add(watcher, consumers);
+            }
+
+            return indices;
+        }
+
+        private void StartWatcherCollections(IEnumerable<string> indices)
+        {
+            foreach (var index in indices)
+            {
+                try
+                {
+                    Logger.LogInformation($"Starting watcher collection [{index}]");
+                    Watchers[index].Start();
+                }
+                catch (KeyNotFoundException)
+                {
+                    Logger.LogError($"Unable to retrieve watcher collection [{index}] when starting");
+                }
+            }
+        }
+
+        public Task StopAsync(CancellationToken cancellationToken)
+        {
+            Logger.LogInformation("Shutting down");
+
+            foreach((var key, var watcher) in Watchers)
+            {
+                Logger.LogInformation($"Stopping watcher collection [{key}]");
+                watcher.Stop();
+            }
+
+            return Task.CompletedTask;
+        }
+    }
+}

Selector.CLI/LocalWatcherService.cs

@@ -16,11 +16,11 @@ using StackExchange.Redis;
 
 namespace Selector.CLI
 {
-    class WatcherService : IHostedService
+    class LocalWatcherService : IHostedService
     {
         private const string ConfigInstanceKey = "localconfig";
 
-        private readonly ILogger<WatcherService> Logger;
+        private readonly ILogger<LocalWatcherService> Logger;
         private readonly ILoggerFactory LoggerFactory;
         private readonly RootOptions Config;
         private readonly IWatcherFactory WatcherFactory;
@@ -33,7 +33,7 @@ namespace Selector.CLI
 
         private Dictionary<string, IWatcherCollection> Watchers { get; set; } = new();
 
-        public WatcherService(
+        public LocalWatcherService(
             IWatcherFactory watcherFactory,
             IWatcherCollectionFactory watcherCollectionFactory,
             IRefreshTokenFactoryProvider spotifyFactory,
@@ -43,7 +43,7 @@ namespace Selector.CLI
             IDatabaseAsync cache = null,
             ISubscriber subscriber = null
         ) {
-            Logger = loggerFactory.CreateLogger<WatcherService>();
+            Logger = loggerFactory.CreateLogger<LocalWatcherService>();
             LoggerFactory = loggerFactory;
             Config = config.Value;
             WatcherFactory = watcherFactory;
@@ -56,16 +56,15 @@ namespace Selector.CLI
 
         public async Task StartAsync(CancellationToken cancellationToken)
         {
-            Logger.LogInformation("Starting watcher service...");
+            Logger.LogInformation("Starting local watcher service...");
 
-            Logger.LogInformation("Loading config instances...");
+            var watcherIndices = await InitInstances();
-            var watcherIndices = await InitialiseConfigInstances();
 
             Logger.LogInformation($"Starting {watcherIndices.Count()} affected watcher collection(s)...");
             StartWatcherCollections(watcherIndices);
         }
 
-        private async Task<IEnumerable<string>> InitialiseConfigInstances()
+        private async Task<IEnumerable<string>> InitInstances()
         {
             var indices = new HashSet<string>();
 
@@ -143,7 +142,7 @@ namespace Selector.CLI
                             }
                             else 
                             {
-                                Logger.LogError("No Last.fm usernmae provided, skipping play counter");
+                                Logger.LogError("No Last.fm username provided, skipping play counter");
                             }
                             break;
                     }

Selector.CLI/Options.cs

@@ -52,6 +52,7 @@ namespace Selector.CLI
         public const string Key = "Watcher";
 
         public bool Enabled { get; set; } = true;
+        public bool LocalEnabled { get; set; } = true;
         public List<WatcherInstanceOptions> Instances { get; set; } = new();
     }
 

Selector.CLI/Program.cs

@@ -134,8 +134,17 @@ namespace Selector.CLI
             // HOSTED SERVICES
             if (config.WatcherOptions.Enabled)
             {
-                Console.WriteLine("> Adding Watcher Service");
+                if(config.WatcherOptions.LocalEnabled)
-                services.AddHostedService<WatcherService>();
+                {
+                    Console.WriteLine("> Adding Local Watcher Service");
+                    services.AddHostedService<LocalWatcherService>();
+                }
+
+                if(config.DatabaseOptions.Enabled)
+                {
+                    Console.WriteLine("> Adding Db Watcher Service");
+                    services.AddHostedService<DbWatcherService>();
+                }
             }
         }
 

Selector.CLI/appsettings.json

@@ -4,6 +4,7 @@
         "ClientSecret": "",
         "Equality": "uri",
         "Watcher": {
+            "localenabled":  false,
             "Instances": [
                 {
                     "type": "player",
@@ -14,7 +15,7 @@
             ]
         },
         "Database": {
-            "enabled": false
+            "enabled": true
         },
         "Redis": {
             "enabled": true

Selector.Cache/Selector.Cache.csproj

@@ -7,7 +7,7 @@
   </PropertyGroup>
 
   <ItemGroup>
-    <PackageReference Include="StackExchange.Redis" Version="2.2.79" />
+    <PackageReference Include="StackExchange.Redis" Version="2.2.88" />
     <PackageReference Include="Microsoft.Extensions.Logging" Version="6.0.0" />
     <PackageReference Include="SpotifyAPI.Web" Version="6.2.2" />
   </ItemGroup>

Selector.Tests/Selector.Tests.csproj

@@ -9,14 +9,14 @@
 
   <ItemGroup>
     <PackageReference Include="FluentAssertions" Version="6.2.0" />
-    <PackageReference Include="Microsoft.NET.Test.Sdk" Version="16.7.1" />
+    <PackageReference Include="Microsoft.NET.Test.Sdk" Version="17.0.0" />
     <PackageReference Include="Moq" Version="4.16.1" />
     <PackageReference Include="xunit" Version="2.4.1" />
     <PackageReference Include="xunit.runner.visualstudio" Version="2.4.3">
       <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
       <PrivateAssets>all</PrivateAssets>
     </PackageReference>
-    <PackageReference Include="coverlet.collector" Version="1.3.0">
+    <PackageReference Include="coverlet.collector" Version="3.1.0">
       <IncludeAssets>runtime; build; native; contentfiles; analyzers; buildtransitive</IncludeAssets>
       <PrivateAssets>all</PrivateAssets>
     </PackageReference>

Selector.Web/Selector.Web.csproj

@@ -3,6 +3,7 @@
   <PropertyGroup>
     <TargetFramework>net6.0</TargetFramework>
     <LangVersion>latest</LangVersion>
+    <TypeScriptCompileBlocked>true</TypeScriptCompileBlocked>
   </PropertyGroup>
 
   <ItemGroup>

Selector/Selector.csproj

@@ -7,7 +7,7 @@
   </PropertyGroup>
 
   <ItemGroup>
-    <PackageReference Include="Microsoft.Extensions.Logging" Version="5.0.0" />
+    <PackageReference Include="Microsoft.Extensions.Logging" Version="6.0.0" />
     <PackageReference Include="SpotifyAPI.Web" Version="6.2.2" />
     <PackageReference Include="Inflatable.Lastfm" Version="1.2.0" />
   </ItemGroup>

Selector/Watcher/PlayerWatcher.cs

@@ -152,7 +152,7 @@ namespace Selector
                             OnVolumeChange(ListeningChangeEventArgs.From(previous, Live, Past, id: Id, username: SpotifyUsername));
                         }
                     }
-                }                
+                }
             }
             catch(APIUnauthorizedException e)
             {
@@ -162,12 +162,13 @@ namespace Selector
             catch(APITooManyRequestsException e)
             {
                 Logger.LogDebug($"Too many requests error: [{e.Message}]");
-                throw e;
+                await Task.Delay(e.RetryAfter);
+                // throw e;
             }
             catch(APIException e)
             {
                 Logger.LogDebug($"API error: [{e.Message}]");
-                throw e;
+                // throw e;
             }
         }